feat: Add playtime tracking backend (#4108)

* Add support for pre_script_args and post_script_args in Manager and WineExecutor

* Add pre_script_args and post_script_args support in WineCommand and WineProgram

Enhance the WineCommand and WineProgram classes to accept pre-run and post-run script arguments. Update the UI to include fields for these arguments in the launch options dialog.

* Implement placeholder handling in WineExecutor and add unit tests

* Enhance BottleConfig with YAML registration and improve placeholder handling in WineExecutor

* Refactor BottleConfig to simplify YAML registration and enhance dataclass handling in yaml utility

* Update tooltip texts and add translations for pre-run and post-run script arguments in launch options dialog

* implement core infra for playtime tracking

* add tests

* implement signals

* Refactor playtime signal tests to use ProcessStartedPayload and ProcessFinishedPayload for improved clarity and structure.

* Use Result where needed

* add debug logs

* Add debug logging for playtime signal events and ensure proper database shutdown on exit

* Refactor tests to use mocker instead of monkeypatch for WineExecutor and WinePath, improving test clarity and maintainability.

* Add playtime tracking configuration options to gschema.xml, including enabling tracking and setting heartbeat interval.

* move integration tests

* Update development dependencies: add pytest-mock and freezegun to requirements.dev.txt

* reset config.py

* chore: change base to main

* refactor: remove unused imports in executor.py

* refactor: remove logging of executed command in WineCommand

---------

Co-authored-by: Mirko Brombin <[email protected]>

Author: evertonstz

bottles/backend/globals.py

@@ -48,6 +48,7 @@ class Paths:
     latencyflex = f"{base}/latencyflex"
     templates = f"{base}/templates"
     library = f"{base}/library.yml"
+    process_metrics = f"{base}/process_metrics.sqlite"
 
     @staticmethod
     def is_vkbasalt_available():

bottles/backend/managers/manager.py

@@ -44,6 +44,7 @@
 from bottles.backend.managers.importer import ImportManager
 from bottles.backend.managers.installer import InstallerManager
 from bottles.backend.managers.library import LibraryManager
+from bottles.backend.managers.playtime import ProcessSessionTracker
 from bottles.backend.managers.repository import RepositoryManager
 from bottles.backend.managers.steam import SteamManager
 from bottles.backend.managers.template import TemplateManager
@@ -52,6 +53,7 @@
 from bottles.backend.models.config import BottleConfig
 from bottles.backend.models.result import Result
 from bottles.backend.models.samples import Samples
+from bottles.backend.models.result import Result
 from bottles.backend.state import SignalManager, Signals, Events, EventManager
 from bottles.backend.utils import yaml
 from bottles.backend.utils.connection import ConnectionUtils
@@ -71,6 +73,11 @@
 from bottles.backend.wine.wineboot import WineBoot
 from bottles.backend.wine.winepath import WinePath
 from bottles.backend.wine.wineserver import WineServer
+from bottles.backend.state import SignalManager, Signals
+from bottles.backend.models.process import (
+    ProcessStartedPayload,
+    ProcessFinishedPayload,
+)
 
 logging = Logger()
 
@@ -103,6 +110,7 @@ class Manager(metaclass=Singleton):
     supported_latencyflex = {}
     supported_dependencies = {}
     supported_installers = {}
+    _playtime_signals_connected: bool = False
 
     def __init__(
         self,
@@ -154,6 +162,22 @@ def __init__(
         self.steam_manager = SteamManager()
         times["SteamManager"] = time.time()
 
+        # Initialize playtime tracker
+        playtime_enabled = self.settings.get_boolean("playtime-enabled")
+        playtime_interval = self.settings.get_int("playtime-heartbeat-interval")
+        self.playtime_tracker = ProcessSessionTracker(
+            enabled=playtime_enabled,
+            heartbeat_interval=playtime_interval if playtime_interval > 0 else 60,
+        )
+        self.playtime_tracker.recover_open_sessions()
+        times["PlaytimeTracker"] = time.time()
+
+        # Subscribe to playtime signals (connect once per process)
+        if not Manager._playtime_signals_connected:
+            SignalManager.connect(Signals.ProgramStarted, self._on_program_started)
+            SignalManager.connect(Signals.ProgramFinished, self._on_program_finished)
+            Manager._playtime_signals_connected = True
+
         if not self.is_cli:
             times.update(self.checks(install_latest=False, first_run=True).data)
         else:
@@ -288,6 +312,95 @@ def checks(
 
         return rv
 
+    def __del__(self):
+        # best-effort shutdown of playtime tracker
+        try:
+            if hasattr(self, "playtime_tracker") and self.playtime_tracker:
+                self.playtime_tracker.shutdown()
+        except Exception:
+            pass
+
+    # Playtime signal handlers
+    _launch_to_session: Dict[str, int] = {}
+
+    # Public Playtime API (wrap tracker with Result)
+    def playtime_start(
+        self,
+        *,
+        bottle_id: str,
+        bottle_name: str,
+        bottle_path: str,
+        program_name: str,
+        program_path: str,
+    ) -> Result[int]:
+        try:
+            sid = self.playtime_tracker.start_session(
+                bottle_id=bottle_id,
+                bottle_name=bottle_name,
+                bottle_path=bottle_path,
+                program_name=program_name,
+                program_path=program_path,
+            )
+            return Result(True, data=sid)
+        except Exception as e:
+            logging.exception(e)
+            return Result(False, message=str(e))
+
+    def playtime_finish(
+        self,
+        session_id: int,
+        *,
+        status: str = "success",
+        ended_at: Optional[int] = None,
+    ) -> Result[None]:
+        try:
+            if status == "success":
+                self.playtime_tracker.mark_exit(session_id, status="success", ended_at=ended_at)
+            else:
+                self.playtime_tracker.mark_failure(session_id, status=status)
+            return Result(True)
+        except Exception as e:
+            logging.exception(e)
+            return Result(False, message=str(e))
+
+    def _on_program_started(self, data: Optional[Result] = None) -> None:
+        try:
+            if not data or not data.data:
+                return
+            payload: ProcessStartedPayload = data.data  # type: ignore
+            logging.debug(
+                f"Playtime signal: started launch_id={payload.launch_id} bottle={payload.bottle_name} program={payload.program_name}"
+            )
+            res = self.playtime_start(
+                bottle_id=payload.bottle_id,
+                bottle_name=payload.bottle_name,
+                bottle_path=payload.bottle_path,
+                program_name=payload.program_name,
+                program_path=payload.program_path,
+            )
+            if not res.ok:
+                return
+            sid = int(res.data or -1)
+            self._launch_to_session[payload.launch_id] = sid
+        except Exception:
+            pass
+
+    def _on_program_finished(self, data: Optional[Result] = None) -> None:
+        try:
+            if not data or not data.data:
+                return
+            payload: ProcessFinishedPayload = data.data  # type: ignore
+            sid = self._launch_to_session.pop(payload.launch_id, -1)
+            if sid and sid > 0:
+                status = payload.status
+                ended_at = int(payload.ended_at or time.time())
+                logging.debug(
+                    f"Playtime signal: finished launch_id={payload.launch_id} status={status} sid={sid}"
+                )
+                self.playtime_finish(sid, status=status, ended_at=ended_at)
+        except Exception:
+            pass
+
     def __clear_temp(self, force: bool = False):
         """Clears the temp directory if user setting allows it. Use the force
         parameter to force clearing the directory.

bottles/backend/managers/meson.build

@@ -24,7 +24,8 @@ bottles_sources = [
   'origin.py',
   'queue.py',
   'steamgriddb.py',
-  'thumbnail.py'
+  'thumbnail.py',
+  'playtime.py'
 ]
 
 install_data(bottles_sources, install_dir: managersdir)