gamemode: global auto-trigger via niri window watch
Add scripts/gamemode-watch.py, launched at niri startup, which subscribes to niri's event stream and toggles gamemode whenever a game window (steam_app_*, RuneLite, gamescope) is open — so Steam games trigger gamemode with zero per-game setup instead of needing gamemoderun %command%. 'winwatch' is a third reference-counted source alongside feral + manual. Pure Watcher class is unit-tested; SIGTERM/finally guarantee cleanup so the watcher stopping never leaves gamemode stuck on. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
funman300
Executable
+142
@@ -0,0 +1,142 @@
|
||||
#!/usr/bin/env python3
|
||||
"""gamemode-watch — drive gamemode-session from niri's window event stream.
|
||||
|
||||
Subscribes to `niri msg --json event-stream`, tracks which open windows are
|
||||
games (matched by app_id), and calls `gamemode-session add/del winwatch` on the
|
||||
empty<->non-empty transition. 'winwatch' is just another reference-counted
|
||||
source, so it coexists with the Mod+G toggle and the Feral gamemode.ini hooks.
|
||||
|
||||
Launched at niri startup (spawn-at-startup "gamemode-watch"). Reconnects if the
|
||||
event stream drops while niri is still alive, and always releases the source on
|
||||
exit so gamemode can never get stuck on when the watcher stops.
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
# app_id patterns that count as "a game" — mirrors the niri window rules.
|
||||
GAME_PATTERNS = [
|
||||
re.compile(r"steam_app_"),
|
||||
re.compile(r"^(net-runelite-client-RuneLite|runelite|RuneLite)$"),
|
||||
re.compile(r"gamescope", re.IGNORECASE),
|
||||
]
|
||||
|
||||
|
||||
def is_game(app_id):
|
||||
if not app_id:
|
||||
return False
|
||||
return any(p.search(app_id) for p in GAME_PATTERNS)
|
||||
|
||||
|
||||
class Watcher:
|
||||
"""Pure event logic. `emit(action)` is called with "add"/"del" on the
|
||||
transition into/out of "at least one game window open"."""
|
||||
|
||||
def __init__(self, emit):
|
||||
self.games = set() # ids of currently-open game windows
|
||||
self.active = False
|
||||
self.emit = emit
|
||||
|
||||
def _reconcile(self):
|
||||
want = len(self.games) > 0
|
||||
if want and not self.active:
|
||||
self.active = True
|
||||
self.emit("add")
|
||||
elif not want and self.active:
|
||||
self.active = False
|
||||
self.emit("del")
|
||||
|
||||
def handle(self, ev):
|
||||
if "WindowsChanged" in ev: # full snapshot (sent on connect)
|
||||
self.games = {
|
||||
w["id"]
|
||||
for w in ev["WindowsChanged"]["windows"]
|
||||
if is_game(w.get("app_id"))
|
||||
}
|
||||
self._reconcile()
|
||||
elif "WindowOpenedOrChanged" in ev: # fired on open AND on any change
|
||||
w = ev["WindowOpenedOrChanged"]["window"]
|
||||
if is_game(w.get("app_id")):
|
||||
self.games.add(w["id"])
|
||||
else:
|
||||
self.games.discard(w["id"])
|
||||
self._reconcile()
|
||||
elif "WindowClosed" in ev: # only carries the id, no app_id
|
||||
self.games.discard(ev["WindowClosed"]["id"])
|
||||
self._reconcile()
|
||||
|
||||
|
||||
def call(action):
|
||||
subprocess.run(
|
||||
["gamemode-session", action, "winwatch"],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
|
||||
def niri_alive():
|
||||
try:
|
||||
return subprocess.run(
|
||||
["niri", "msg", "version"],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
).returncode == 0
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
|
||||
|
||||
def stream_lines():
|
||||
proc = subprocess.Popen(
|
||||
["niri", "msg", "--json", "event-stream"],
|
||||
stdout=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
try:
|
||||
for line in proc.stdout:
|
||||
yield line
|
||||
finally:
|
||||
proc.terminate()
|
||||
try:
|
||||
proc.wait(timeout=2)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
|
||||
|
||||
def main():
|
||||
watcher = Watcher(call)
|
||||
try:
|
||||
while True:
|
||||
for line in stream_lines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
ev = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
watcher.handle(ev)
|
||||
# stream ended; stop if niri is gone, else reconnect
|
||||
if not niri_alive():
|
||||
break
|
||||
time.sleep(2)
|
||||
finally:
|
||||
if watcher.active:
|
||||
call("del") # never leave gamemode stuck on
|
||||
|
||||
|
||||
def _on_sigterm(signum, frame):
|
||||
# SIGTERM (how niri/logout stops us) -> KeyboardInterrupt so main()'s
|
||||
# `finally` runs and releases the winwatch source — never leave gamemode on.
|
||||
raise KeyboardInterrupt
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
signal.signal(signal.SIGTERM, _on_sigterm)
|
||||
try:
|
||||
main()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
sys.exit(0)
|
||||
@@ -0,0 +1,96 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for gamemode-watch's pure event logic (no niri, no subprocess).
|
||||
Loads the hyphenated script by path and drives Watcher with synthetic events,
|
||||
asserting the exact sequence of add/del calls."""
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
MOD_PATH = os.path.join(HERE, "..", "gamemode-watch.py")
|
||||
spec = importlib.util.spec_from_file_location("gamemode_watch", MOD_PATH)
|
||||
gw = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(gw)
|
||||
|
||||
fails = 0
|
||||
|
||||
|
||||
def check(desc, got, want):
|
||||
global fails
|
||||
if got == want:
|
||||
print(f"ok - {desc}")
|
||||
else:
|
||||
print(f"FAIL - {desc} (got {got!r} want {want!r})")
|
||||
fails = 1
|
||||
|
||||
|
||||
# --- is_game ---------------------------------------------------------------
|
||||
check("steam_app_ matches", gw.is_game("steam_app_480"), True)
|
||||
check("bare steam_app_ matches", gw.is_game("steam_app_"), True)
|
||||
check("RuneLite matches", gw.is_game("RuneLite"), True)
|
||||
check("runelite matches", gw.is_game("runelite"), True)
|
||||
check("runelite full id matches", gw.is_game("net-runelite-client-RuneLite"), True)
|
||||
check("gamescope matches", gw.is_game("gamescope"), True)
|
||||
check("Gamescope (case) matches", gw.is_game("Gamescope"), True)
|
||||
check("Alacritty is not a game", gw.is_game("Alacritty"), False)
|
||||
check("bolt-launcher excluded", gw.is_game("bolt-launcher"), False)
|
||||
check("empty app_id is not a game", gw.is_game(""), False)
|
||||
check("None app_id is not a game", gw.is_game(None), False)
|
||||
|
||||
|
||||
# --- Watcher transitions ---------------------------------------------------
|
||||
def drive(events):
|
||||
calls = []
|
||||
w = gw.Watcher(lambda action: calls.append(action))
|
||||
for ev in events:
|
||||
w.handle(ev)
|
||||
return calls
|
||||
|
||||
|
||||
def win(wid, app_id):
|
||||
return {"WindowOpenedOrChanged": {"window": {"id": wid, "app_id": app_id}}}
|
||||
|
||||
|
||||
def closed(wid):
|
||||
return {"WindowClosed": {"id": wid}}
|
||||
|
||||
|
||||
def snapshot(pairs):
|
||||
return {"WindowsChanged": {"windows": [{"id": i, "app_id": a} for i, a in pairs]}}
|
||||
|
||||
|
||||
check("empty snapshot: no calls", drive([snapshot([])]), [])
|
||||
check("non-game window: no calls", drive([win(1, "Alacritty")]), [])
|
||||
check("one game opens: add", drive([win(1, "steam_app_10")]), ["add"])
|
||||
check(
|
||||
"second game keeps active (single add)",
|
||||
drive([win(1, "steam_app_10"), win(2, "RuneLite")]),
|
||||
["add"],
|
||||
)
|
||||
check(
|
||||
"close one of two: still active",
|
||||
drive([win(1, "steam_app_10"), win(2, "RuneLite"), closed(2)]),
|
||||
["add"],
|
||||
)
|
||||
check(
|
||||
"close last game: add then del",
|
||||
drive([win(1, "steam_app_10"), closed(1)]),
|
||||
["add", "del"],
|
||||
)
|
||||
check(
|
||||
"snapshot already containing a game: add (restart mid-game)",
|
||||
drive([snapshot([(1, "Alacritty"), (5, "steam_app_9")])]),
|
||||
["add"],
|
||||
)
|
||||
check(
|
||||
"game window changes to non-game app_id: del",
|
||||
drive([win(1, "steam_app_10"), win(1, "Alacritty")]),
|
||||
["add", "del"],
|
||||
)
|
||||
check(
|
||||
"repeated OpenedOrChanged for same game: single add",
|
||||
drive([win(1, "gamescope"), win(1, "gamescope"), win(1, "gamescope")]),
|
||||
["add"],
|
||||
)
|
||||
|
||||
sys.exit(fails)
|
||||
Reference in New Issue
Block a user