DawnsorrowandCursor
69ee41dc91
Record Jorg desk compatdata locations for satellite setup; devices list terra; tighten watchers so syncgames argv is not mistaken for Eden. Co-authored-by: Cursor <[email protected]>
70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from syncgames.config import GameConfig
|
|
|
|
try:
|
|
import psutil
|
|
except ImportError: # pragma: no cover
|
|
psutil = None # type: ignore
|
|
|
|
|
|
def _norm(name: str) -> str:
|
|
return name.lower().removesuffix(".exe")
|
|
|
|
|
|
def is_game_running(game: GameConfig) -> bool:
|
|
if psutil is None:
|
|
return False
|
|
names = {_norm(n) for n in game.process_names}
|
|
# Steam AppID heuristic: look for steam apps running with appid in cmdline
|
|
appid = game.steam_appid
|
|
for proc in psutil.process_iter(["name", "cmdline"]):
|
|
try:
|
|
pname = _norm(proc.info["name"] or "")
|
|
if names and pname in names:
|
|
return True
|
|
cmd = proc.info.get("cmdline") or []
|
|
# Match executable basenames only — never substring-search the full
|
|
# cmdline (e.g. `syncgames end eden-saves` must not look like Eden).
|
|
for part in cmd[:2]:
|
|
if not part:
|
|
continue
|
|
if _norm(Path(part).name) in names:
|
|
return True
|
|
if appid:
|
|
joined = " ".join(cmd).lower()
|
|
if (
|
|
f"appid={appid}" in joined
|
|
or f"steam_appid={appid}" in joined
|
|
or f"appid={appid}" in joined
|
|
):
|
|
return True
|
|
except (psutil.Error, TypeError):
|
|
continue
|
|
return False
|
|
|
|
|
|
def watch_loop(
|
|
game: GameConfig,
|
|
on_start: Callable[[], None],
|
|
on_end: Callable[[], None],
|
|
*,
|
|
poll_s: float = 3.0,
|
|
) -> None:
|
|
"""Block forever: fire on_start when process appears, on_end when it exits."""
|
|
was_running = False
|
|
while True:
|
|
running = is_game_running(game)
|
|
if running and not was_running:
|
|
on_start()
|
|
elif not running and was_running:
|
|
# brief settle for flush
|
|
time.sleep(2.0)
|
|
on_end()
|
|
was_running = running
|
|
time.sleep(poll_s)
|