dotfiles/scripts/gamemode-watch.py

#!/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 fcntl
import json
import os
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


def acquire_single_instance():
    # Hold an exclusive lock so a second watcher (e.g. spawn-at-startup plus a
    # manual launch) exits instead of racing the same winwatch source. The open
    # file must stay referenced for the process lifetime to keep the lock.
    rt = os.environ.get("XDG_RUNTIME_DIR", "/tmp")
    f = open(os.path.join(rt, "gamemode-watch.lock"), "w")
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        return None
    return f


if __name__ == "__main__":
    _lock = acquire_single_instance()
    if _lock is None:
        sys.exit(0)  # another instance already running
    signal.signal(signal.SIGTERM, _on_sigterm)
    try:
        main()
    except KeyboardInterrupt:
        pass
    sys.exit(0)