saphid--frame-control/ui/frame_touch.py

"""Touch and direct input for the Steam Frame's panels. Frame Control's server runs this ON the Frame.

gamescope, the Frame's compositor, serves Valve's own input injection: an EIS
socket (libei's server side), which Steam uses to feed it Remote Play input.
This connects to it with libei, which is on the SteamOS image, and points,
clicks, scrolls and types into the panel that has focus in the headset: the
one the wearer last used. No install, and it reaches every panel, on either
of gamescope's X displays (see docs/streaming.md).

  python3 frame_touch.py focus   print the focused panel as JSON
  python3 frame_touch.py panels  print every app panel as JSON, and which has focus
  python3 frame_touch.py         read events on stdin, one JSON object (or list) per line:
    {"fx": 0.5, "fy": 0.2, "window": 123}  pointer to that fraction of the focused panel
                                           (ignored if another panel has focus since)
    {"dx": 4, "dy": -2}                    pointer by that much
    {"button": "left", "down": true}       left, right or middle; "down" false releases
    {"scroll": [0, 120]}                   by pixels; positive y scrolls down
    {"key": 30, "down": true}              a Linux (evdev) key code, as the page maps KeyboardEvent.code
    {"text": "hello"}                      printable ASCII, typed on a US layout

Status goes to stdout, one JSON object per line: {"state": "ready" | "error", ...}.
Standard library only (ctypes for libei), like the rest of what runs on the Frame.
"""
import ctypes
import json
import os
import select
import subprocess
import sys
import time

SOCKET = f"/run/user/{os.getuid()}/gamescope-0-ei"
BUTTONS = {"left": 0x110, "right": 0x111, "middle": 0x112}  # BTN_LEFT, BTN_RIGHT, BTN_MIDDLE
SHIFT = 42  # KEY_LEFTSHIFT
# Printable ASCII on a US layout: character -> (evdev key code, shifted).
ROWS = [("1234567890-=", "!@#$%^&*()_+", 2), ("qwertyuiop[]", "QWERTYUIOP{}", 16),
        ("asdfghjkl;'`", 'ASDFGHJKL:"~', 30), ("\\zxcvbnm,./", "|ZXCVBNM<>?", 43)]
ASCII = {" ": (57, False), "\n": (28, False), "\t": (15, False)}
for plain, shifted, first in ROWS:
    for i, (a, b) in enumerate(zip(plain, shifted)):
        ASCII[a], ASCII[b] = (first + i, False), (first + i, True)

# libei's event types and device capabilities (libei.h, libei 1.4).
EV_CONNECT, EV_DISCONNECT, EV_SEAT_ADDED, EV_DEVICE_ADDED, EV_DEVICE_REMOVED = 1, 2, 3, 5, 6
EV_DEVICE_PAUSED, EV_DEVICE_RESUMED = 7, 8
CAP_POINTER, CAP_ABSOLUTE, CAP_KEYBOARD, CAP_SCROLL, CAP_BUTTON = 1, 2, 4, 16, 32


def say(state, **more):
    print(json.dumps({"state": state, **more}), flush=True)


# ---- which panel has focus -----------------------------------------------------

def xprop_root(display, name):
    try:
        out = subprocess.run(["xprop", "-root", name], env=dict(os.environ, DISPLAY=display),
                             capture_output=True, text=True, timeout=5).stdout
    except (OSError, subprocess.SubprocessError):
        return []
    values = out.split("=", 1)[1] if "=" in out else ""
    return [int(v) for v in values.replace(",", " ").split() if v.isdigit()]


def window_info(display, window):
    """Name and geometry of a window on one X display, or None if it isn't there."""
    try:
        which = ["-root"] if window == "root" else ["-id", str(window)]
        out = subprocess.run(["xwininfo", *which], env=dict(os.environ, DISPLAY=display),
                             capture_output=True, text=True, timeout=5).stdout
    except (OSError, subprocess.SubprocessError):
        return None
    if "IsViewable" not in out:
        return None
    info = {}
    for line in out.splitlines():
        line = line.strip()
        if line.startswith("xwininfo: Window id:"):
            info["name"] = line.split('"', 1)[1].rsplit('"', 1)[0] if '"' in line else ""
        for key, field in (("Absolute upper-left X:", "x"), ("Absolute upper-left Y:", "y"),
                           ("Width:", "width"), ("Height:", "height")):
            if line.startswith(key):
                info[field] = int(line.split(":", 1)[1])
    return info if "width" in info else None


def displays():
    return sorted(f":{n[1:]}" for n in os.listdir("/tmp/.X11-unix") if n[1:].isdigit())


def focus():
    """The panel that has focus in the headset: its window, display, name and sizes.

    gamescope publishes it on :0's root (GAMESCOPE_FOCUSED_WINDOW). The window can be
    on any of its Xwayland displays; the pid in GAMESCOPE_FOCUSABLE_WINDOWS (window,
    app id, pid triples) tells which one when ids repeat across displays.
    """
    window = (xprop_root(":0", "GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
    if not window:
        return {"window": None}
    triples = xprop_root(":0", "GAMESCOPE_FOCUSABLE_WINDOWS")
    app = next((triples[i + 1] for i in range(0, len(triples) - 2, 3) if triples[i] == window), None)
    for display in displays():
        info = window_info(display, window)
        if info:
            root = window_info(display, "root") or {}
            return {"window": window, "display": display, "app": app, **info,
                    "root": [root.get("width", info["width"]), root.get("height", info["height"])]}
    return {"window": None}


def panels():
    """Every app panel (gamescope's focusable windows), for watching one that hasn't focus."""
    focused = (xprop_root(":0", "GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
    triples = xprop_root(":0", "GAMESCOPE_FOCUSABLE_WINDOWS")
    found, seen = [], set()
    for i in range(0, len(triples) - 2, 3):
        window, app = triples[i], triples[i + 1]
        if window in seen:
            continue
        seen.add(window)
        for display in displays():
            info = window_info(display, window)
            if info and info["width"] > 1 and info["height"] > 1:
                found.append({"window": window, "display": display, "app": app, **info,
                              "focused": window == focused})
                break
    return {"focus": focused or None, "panels": found}


def to_root(panel, fx, fy):
    """A point given as a fraction of the panel, in the root coordinates gamescope's pointer uses.

    gamescope fits each panel's window to its display, so a 1920x1080 window on a
    1280x720 display takes pointer positions at two thirds scale (verified 2026-09-29).
    """
    rw, rh = panel["root"]
    w, h = panel["width"], panel["height"]
    s = min(rw / w, rh / h)
    ox, oy = (rw - w * s) / 2, (rh - h * s) / 2
    fx, fy = min(max(fx, 0.0), 1.0), min(max(fy, 0.0), 1.0)
    return ox + fx * (w * s - 1), oy + fy * (h * s - 1)


# ---- gamescope's input socket ----------------------------------------------------

def libei():
    L = ctypes.CDLL("libei.so.1")
    vp, c = ctypes.c_void_p, ctypes
    sig = {
        "ei_new_sender": (vp, [vp]), "ei_configure_name": (None, [vp, c.c_char_p]),
        "ei_setup_backend_socket": (c.c_int, [vp, c.c_char_p]), "ei_get_fd": (c.c_int, [vp]),
        "ei_dispatch": (None, [vp]), "ei_get_event": (vp, [vp]), "ei_event_get_type": (c.c_int, [vp]),
        "ei_event_unref": (vp, [vp]), "ei_event_get_seat": (vp, [vp]), "ei_event_get_device": (vp, [vp]),
        "ei_device_has_capability": (c.c_bool, [vp, c.c_int]), "ei_now": (c.c_uint64, [vp]),
        "ei_device_start_emulating": (None, [vp, c.c_uint32]), "ei_device_stop_emulating": (None, [vp]),
        "ei_device_frame": (None, [vp, c.c_uint64]),
        "ei_device_pointer_motion": (None, [vp, c.c_double, c.c_double]),
        "ei_device_pointer_motion_absolute": (None, [vp, c.c_double, c.c_double]),
        "ei_device_button_button": (None, [vp, c.c_uint32, c.c_bool]),
        "ei_device_scroll_delta": (None, [vp, c.c_double, c.c_double]),
        "ei_device_keyboard_key": (None, [vp, c.c_uint32, c.c_bool]),
        "ei_unref": (vp, [vp]),
    }
    for name, (res, args) in sig.items():
        f = getattr(L, name)
        f.restype, f.argtypes = res, args
    return L


class Gamescope:
    """One connection to gamescope's EIS socket and its virtual input device."""

    def __init__(self):
        self.L = L = libei()
        self.ei = L.ei_new_sender(None)
        L.ei_configure_name(self.ei, b"Frame Control")
        if L.ei_setup_backend_socket(self.ei, SOCKET.encode()) != 0:
            raise RuntimeError("Couldn't reach gamescope's input socket. Is the headset on?")
        self.fd = L.ei_get_fd(self.ei)
        self.device, self.sequence, self.held = None, 0, set()

    def pump(self, wait=0.0):
        """Handle gamescope's events; False once it has disconnected."""
        select.select([self.fd], [], [], wait)
        self.L.ei_dispatch(self.ei)
        alive = True
        while True:
            ev = self.L.ei_get_event(self.ei)
            if not ev:
                return alive
            kind = self.L.ei_event_get_type(ev)
            if kind == EV_SEAT_ADDED:
                seat = self.L.ei_event_get_seat(ev)
                # Variadic, ending in 0 (NULL): ask for everything we send.
                self.L.ei_seat_bind_capabilities(ctypes.c_void_p(seat), *map(ctypes.c_int, (
                    CAP_POINTER, CAP_ABSOLUTE, CAP_BUTTON, CAP_SCROLL, CAP_KEYBOARD, 0)))
            elif kind == EV_DEVICE_RESUMED:
                device = self.L.ei_event_get_device(ev)
                if self.L.ei_device_has_capability(device, CAP_ABSOLUTE):
                    self.sequence += 1
                    self.L.ei_device_start_emulating(device, self.sequence)
                    self.device = device
            elif kind in (EV_DEVICE_PAUSED, EV_DEVICE_REMOVED):
                if self.L.ei_event_get_device(ev) == self.device:
                    self.device = None
            elif kind == EV_DISCONNECT:
                self.device, alive = None, False
            self.L.ei_event_unref(ev)

    def wait_ready(self, timeout=5):
        end = time.time() + timeout
        while self.device is None and time.time() < end:
            if not self.pump(0.1):
                break
        if self.device is None:
            raise RuntimeError("gamescope didn't offer an input device")

    def frame(self):
        self.L.ei_device_frame(self.device, self.L.ei_now(self.ei))
        self.L.ei_dispatch(self.ei)

    def move_to(self, x, y):
        self.L.ei_device_pointer_motion_absolute(self.device, x, y)
        self.frame()

    def move_by(self, dx, dy):
        self.L.ei_device_pointer_motion(self.device, dx, dy)
        self.frame()

    def button(self, name, down):
        code = BUTTONS[name]
        if down == (code in self.held):
            return  # already in that state
        self.L.ei_device_button_button(self.device, code, down)
        self.frame()
        (self.held.add if down else self.held.discard)(code)

    def scroll(self, dx, dy):
        self.L.ei_device_scroll_delta(self.device, dx, dy)
        self.frame()

    def key(self, code, down):
        self.L.ei_device_keyboard_key(self.device, code, down)
        self.frame()
        # Paced: a burst of keys can reach the app out of order (seen 2026-09-29).
        time.sleep(0.008)

    def text(self, text):
        for ch in text:
            if ch not in ASCII:
                continue
            code, shifted = ASCII[ch]
            if shifted:
                self.key(SHIFT, True)
            self.key(code, True)
            self.key(code, False)
            if shifted:
                self.key(SHIFT, False)

    def release_all(self):
        for code in list(self.held):
            name = next(n for n, c in BUTTONS.items() if c == code)
            self.button(name, False)


# ---- events from the server --------------------------------------------------------

def events(line):
    try:
        data = json.loads(line)
    except ValueError:
        return []
    return [e for e in (data if isinstance(data, list) else [data]) if isinstance(e, dict)]


def number(value, limit=100000.0):
    if isinstance(value, bool) or not isinstance(value, (int, float)) or value != value:
        raise ValueError("not a number")
    return max(-limit, min(limit, float(value)))


def apply(gs, event, panel):
    """Send one event; returns the panel it was aimed at (refreshed if focus moved)."""
    if "fx" in event:
        window = event.get("window")
        if not panel or panel.get("window") != window or time.time() - panel.get("_at", 0) > 1:
            panel = {**focus(), "_at": time.time()}
        if panel.get("window") is None or panel["window"] != window:
            say("ready", focus=panel.get("window"), stale=True)  # the page re-syncs its view
            return panel
        gs.move_to(*to_root(panel, number(event["fx"], 1), number(event["fy"], 1)))
    if "dx" in event or "dy" in event:
        gs.move_by(number(event.get("dx", 0), 2000), number(event.get("dy", 0), 2000))
    if event.get("button") in BUTTONS:
        gs.button(event["button"], event.get("down") is not False)
    if isinstance(event.get("scroll"), list) and len(event["scroll"]) == 2:
        gs.scroll(number(event["scroll"][0], 5000), number(event["scroll"][1], 5000))
    if isinstance(event.get("key"), int) and not isinstance(event["key"], bool) and 0 < event["key"] < 768:
        gs.key(event["key"], event.get("down") is not False)
    if isinstance(event.get("text"), str):
        gs.text(event["text"][:500])
    return panel


def main():
    if sys.argv[1:] == ["focus"]:
        print(json.dumps(focus()))
        return 0
    if sys.argv[1:] == ["panels"]:
        print(json.dumps(panels()))
        return 0
    try:
        gs = Gamescope()
        gs.wait_ready()
    except (OSError, RuntimeError) as e:
        say("error", message=str(e))
        return 1
    say("ready", focus=focus().get("window"))
    stdin, pending, panel = sys.stdin.fileno(), b"", None
    try:
        while True:
            ready, _, _ = select.select([stdin, gs.fd], [], [], 30)
            if gs.fd in ready and not gs.pump():
                say("error", message="gamescope closed its input socket")
                return 1
            if stdin not in ready:
                continue
            chunk = os.read(stdin, 65536)
            if not chunk:
                return 0  # the server went away
            *lines, pending = (pending + chunk).split(b"\n")
            for line in lines:
                for event in events(line):
                    if gs.device is None:
                        gs.wait_ready(2)
                    try:
                        panel = apply(gs, event, panel)
                    except (ValueError, KeyError, TypeError):
                        continue  # the server checks events; skip anything odd
    except RuntimeError as e:
        say("error", message=str(e))
        return 1
    finally:
        if gs.device is not None:
            gs.release_all()  # never leave a button held down in the headset


if __name__ == "__main__":
    sys.exit(main())