SmartestHome/hosts/steam-tv-box/agent/steamtv_agent/session_mode.py

"""What is this box doing right now — gaming, media, or idle — and switching between.

WHY THIS EXISTS
---------------
Home Assistant needs to be able to answer "is somebody playing on the TV box?" without
guessing from power draw. That one sensor is worth a lot in automations: don't dim the
living room during a game, don't announce the doorbell over a raid, count the room as
occupied even though nobody has moved for forty minutes.

WHAT IT READS
-------------
Two cheap facts, both local:
  - is the Steam client running (pgrep);
  - which sway workspace is focused (swaymsg -t get_workspaces).

Deliberately NOT the window title. Steam's window titles and the existence of a
separate Big Picture window have both changed across client rewrites, and a sensor that
silently goes wrong after a Steam update is worse than one that is slightly coarse. The
workspace name is a contract this repo owns — see the header in
configs/session/session-watcher, which makes the same call for the same reason.

MODES
-----
  gaming  — Steam is running and 1:steam is focused. The screen is showing a game or
            Big Picture.
  steam   — Steam is running but the user has moved elsewhere (this is "exited Big
            Picture but left Steam up", the case session-watcher reacts to).
  media   — Steam is not running; the media session is what is on screen.
  idle    — nothing is up. Only really seen in the seconds after boot, or after
            somebody quit Steam and the media session has not been started yet.

SECURITY POSTURE, UNCHANGED
---------------------------
The two commands below take no arguments derived from any MQTT payload. The mode
*switch* is an enumerated action — see mqtt_discovery.py's module docstring.
"""

from __future__ import annotations

import json
import logging

from .sway_control import WS_STEAM

log = logging.getLogger(__name__)

MODE_GAMING = "gaming"
MODE_STEAM = "steam"
MODE_MEDIA = "media"
MODE_IDLE = "idle"

MODES = (MODE_GAMING, MODE_MEDIA)

STEAM_PROCESS = "steam"
MEDIA_SESSION = "/usr/local/bin/media-session"
STEAM_BIG_PICTURE = "/usr/local/bin/steam-big-picture"

# What media-session itself guards on. Kept in sync with that script by hand; there is
# no shared file to read, and inventing one to hold three strings would be worse.
MEDIA_PATTERNS = (
    "com.spotify.Client",
    "media-player-idle",
    "firefox.*--profile.*/firefox/web",
)


class SessionMode:
    def __init__(self, sway):
        self.sway = sway
        # What current() last computed, versus what was last put on the broker. Two
        # fields because current() updates the first every time it is called, so it
        # cannot also serve as "have we told HA about this yet".
        self.last_mode = MODE_IDLE
        self.last_published = ""

    # --- reading ------------------------------------------------------------
    def focused_workspace(self) -> str:
        output = self.sway.swaymsg("-t", "get_workspaces")
        if not output:
            return ""
        try:
            workspaces = json.loads(output)
        except ValueError:
            log.warning("could not parse get_workspaces output")
            return ""
        for workspace in workspaces:
            if isinstance(workspace, dict) and workspace.get("focused"):
                return str(workspace.get("name") or "")
        return ""

    def steam_running(self) -> bool:
        # is_process (pgrep -x), not is_running (pgrep -f) — see that method's comment
        # for why Steam specifically needs the exact-name match.
        return self.sway.is_process(STEAM_PROCESS)

    def media_running(self) -> bool:
        return any(self.sway.is_running(pattern) for pattern in MEDIA_PATTERNS)

    def current(self) -> str:
        steam = self.steam_running()
        if steam and self.focused_workspace() == WS_STEAM:
            mode = MODE_GAMING
        elif steam:
            mode = MODE_STEAM
        elif self.media_running():
            mode = MODE_MEDIA
        else:
            mode = MODE_IDLE
        self.last_mode = mode
        return mode

    def attributes(self) -> dict:
        return {
            "steam_running": self.steam_running(),
            "media_running": self.media_running(),
            "workspace": self.focused_workspace(),
        }

    # --- switching ----------------------------------------------------------
    def select(self, mode: str) -> str:
        """Handle the HA select. Enumerated: anything else is ignored, not guessed at.

        Note the asymmetry, which is intentional. Choosing "gaming" launches or focuses
        Steam and leaves the media apps alone — someone may well want music over a
        game, and killing a running Spotify because a game started would lose whatever
        was playing. Choosing "media" starts the media session and likewise does not
        quit Steam, because quitting Steam from a phone while somebody is mid-match is
        not a thing this should be able to do by accident. Both directions are additive;
        the only thing that shuts anything down is the explicit "Stop media apps"
        button.
        """
        mode = (mode or "").strip().lower()
        if mode == MODE_GAMING:
            log.info("session mode -> gaming")
            self.sway.launch_app([STEAM_BIG_PICTURE])
        elif mode == MODE_MEDIA:
            log.info("session mode -> media")
            self.sway.launch_app([MEDIA_SESSION, "start", "--focus"])
        else:
            log.warning("ignoring unknown session mode %r", mode)
        return self.current()

    def stop_media(self) -> str:
        log.info("stopping the media session")
        self.sway.launch_app([MEDIA_SESSION, "stop"])
        return self.current()