jessey
bf3ac4af1a
D6 trigger modes, no root: - core/service.py: write/enable `systemd --user` units; apply_mode(manual/ always-on/game-launch) reconciles the recorder + watcher services; status(). - core/watcher.py + `rigdoctor watch`: poll Steam RunningAppID, auto-bracket a focused capture (D12 zero-config fallback; wrapper stays primary). - CLI `rigdoctor service status|mode`; config `trigger_mode`. - GUI Settings: "Recording trigger" dropdown (Apply runs apply_mode off-thread). - Tests for unit generation, mode reconciliation, watcher transitions/parse. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
108 lines
3.1 KiB
Python
108 lines
3.1 KiB
Python
"""Zero-config game-launch watcher (D12 fallback): poll Steam's RunningAppID and
|
|
auto-bracket a focused capture around the running game.
|
|
|
|
For users who won't add the `rigdoctor wrap %command%` launch option. Less precise than the
|
|
wrapper (it depends on Steam writing RunningAppID to registry.vdf, and only covers Steam), so
|
|
the wrapper stays the primary mechanism. Stdlib only; safe to run as a `systemd --user` service
|
|
(the game-launch trigger mode).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import signal
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from . import reccontrol, steam
|
|
from .steam import _parse_vdf
|
|
|
|
_REGISTRY_CANDIDATES = ("~/.steam/registry.vdf", "~/.steam/steam/registry.vdf")
|
|
|
|
|
|
def _registry_path() -> Path | None:
|
|
for cand in _REGISTRY_CANDIDATES:
|
|
p = Path(os.path.expanduser(cand))
|
|
if p.exists():
|
|
return p
|
|
return None
|
|
|
|
|
|
def _find_key(data: dict, key: str):
|
|
"""Recursively find a (case-insensitive) scalar key in nested VDF dicts."""
|
|
target = key.lower()
|
|
for k, v in data.items():
|
|
if isinstance(v, dict):
|
|
found = _find_key(v, key)
|
|
if found is not None:
|
|
return found
|
|
elif k.lower() == target:
|
|
return v
|
|
return None
|
|
|
|
|
|
def running_appid() -> int:
|
|
"""The Steam appid currently running (0 if none / unknown)."""
|
|
path = _registry_path()
|
|
if path is None:
|
|
return 0
|
|
try:
|
|
data = _parse_vdf(path.read_text(encoding="utf-8", errors="replace"))
|
|
except OSError:
|
|
return 0
|
|
raw = _find_key(data, "RunningAppID")
|
|
try:
|
|
return int(raw)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def transition(prev: int, current: int) -> str | None:
|
|
"""'start' when a game begins, 'stop' when it ends, else None."""
|
|
if current and not prev:
|
|
return "start"
|
|
if prev and not current:
|
|
return "stop"
|
|
return None
|
|
|
|
|
|
def _name_for(appid: int) -> str:
|
|
target = str(appid)
|
|
for g in steam.cached_games() or steam.scan_games(steam.selected_library_paths()):
|
|
if g.appid == target:
|
|
return g.name
|
|
return f"Steam app {appid}"
|
|
|
|
|
|
def watch(interval: float = 5.0) -> int:
|
|
"""Poll for a running Steam game and bracket a capture around it. Blocks until signalled."""
|
|
from . import diagnostic
|
|
|
|
stop = {"flag": False}
|
|
|
|
def _on_signal(_sig, _frame):
|
|
stop["flag"] = True
|
|
|
|
signal.signal(signal.SIGTERM, _on_signal)
|
|
signal.signal(signal.SIGINT, _on_signal)
|
|
|
|
prev = 0
|
|
started = False
|
|
while not stop["flag"]:
|
|
current = running_appid()
|
|
action = transition(prev, current)
|
|
if action == "start" and not reccontrol.running_pid():
|
|
started = diagnostic.start(game=_name_for(current)) is not None
|
|
elif action == "stop" and started:
|
|
reccontrol.stop_background()
|
|
started = False
|
|
prev = current
|
|
# Sleep in small slices so a stop signal is handled promptly.
|
|
slept = 0.0
|
|
while slept < interval and not stop["flag"]:
|
|
time.sleep(min(0.25, interval - slept))
|
|
slept += 0.25
|
|
if started:
|
|
reccontrol.stop_background()
|
|
return 0
|