steam-frame/saphid--frame-control
1
0 Fork 0
mirror of https://github.com/saphid/frame-control.git synced 2026-10-06 08:00:32 +02:00
Code Issues Packages Projects Releases 6 Wiki Activity
Files
0f4f0b863f9109ebe7e453e54cc52c58368eb5dd
saphid--frame-control/ui/frame_touch.py
T
saphidandClaude Opus 5.5 0f4f0b863f Control: fixes from review 
- Keys held on the Frame are released with buttons when Control stops or
  the view loses focus; keys for the Frame no longer trigger Frame Control's
  own shortcuts.
- Taps only act when the picture on screen is the panel in use; positions,
  presses, keys, text and scrolls name their panel (display and window: ids
  repeat across :0 and :1, told apart by pid), and the Frame drops them if
  focus has moved on. Releases always go.
- While connecting, a tap keeps its position; on an error only releases wait
  and retries back off; trimming a long queue never drops a release.
- Lifting one of two scrolling fingers ends the scroll; a cancelled touch
  isn't a tap; clicks and holds on the bars around the picture do nothing.
- A capture loop from before a Live restart can't stop the new video.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-29 08:40:46 +10:00

386 lines
16 KiB
Python
Raw Blame History

"""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, "display": ":1"}
pointer to that fraction of that panel; any event can
name its panel, and goes nowhere if another has focus
{"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 window_pid(display, window):
try:
out = subprocess.run(["xprop", "-id", str(window), "_NET_WM_PID"], env=dict(os.environ, DISPLAY=display),
capture_output=True, text=True, timeout=5).stdout
except (OSError, subprocess.SubprocessError):
return None
value = out.rsplit("=", 1)[-1].strip() if "=" in out else ""
return int(value) if value.isdigit() else None
def locate(window, pid):
"""The display a focusable window is on, with its name and geometry.
Window ids are per X server, so :0 and :1 can both have one; the pid gamescope
lists with it (GAMESCOPE_FOCUSABLE_WINDOWS) tells them apart.
"""
found = []
for display in displays():
info = window_info(display, window)
if info:
found.append((display, info))
if len(found) > 1 and pid:
found = [f for f in found if window_pid(f[0], window) == pid] or found
if not found:
return None
display, info = found[0]
root = window_info(display, "root") or {}
return {"window": window, "display": display, **info,
"root": [root.get("width", info["width"]), root.get("height", info["height"])]}
def focusable():
"""gamescope's focusable windows as (window, app id, pid)."""
t = xprop_root(":0", "GAMESCOPE_FOCUSABLE_WINDOWS")
return [tuple(t[i:i + 3]) for i in range(0, len(t) - 2, 3)]
def focus():
"""The panel that has focus in the headset: window, display, name and sizes (gamescope
publishes the window on :0's root as GAMESCOPE_FOCUSED_WINDOW)."""
window = (xprop_root(":0", "GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
if not window:
return {"window": None}
app, pid = next(((a, p) for w, a, p in focusable() if w == window), (None, None))
panel = locate(window, pid)
return {**panel, "app": app} if panel else {"window": None}
def panels():
"""Every app panel (gamescope's focusable windows), for watching one that hasn't focus."""
now = focus()
found = []
for window, app, pid in focusable():
panel = locate(window, pid)
if panel and panel["width"] > 1 and panel["height"] > 1 and \
not any(f["window"] == window and f["display"] == panel["display"] for f in found):
panel.pop("root", None)
found.append({**panel, "app": app, "focused": (window, panel["display"]) ==
(now.get("window"), now.get("display"))})
return {"focus": now.get("window"), "focus_display": now.get("display"), "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 aimed_elsewhere(event, panel):
"""Whether an event names a panel that isn't the one with focus now."""
if "window" not in event:
return False
return (panel.get("window"), panel.get("display")) != (event.get("window"), event.get("display"))
def apply(gs, event, panel):
"""Send one event; returns the focused panel it checked against (looked up at most once a second).
Positions and presses name the panel they were meant for. If focus has moved to
another panel since, they go nowhere, so a tap can't land on the wrong one;
releases always go, so nothing stays held.
"""
if "window" in event and (not panel or time.time() - panel.get("_at", 0) > 1 or aimed_elsewhere(event, panel)):
panel = {**focus(), "_at": time.time()}
stale = aimed_elsewhere(event, panel) if "window" in event else False
if stale and not (event.get("down") is False and ("button" in event or "key" in event)):
say("ready", focus=panel.get("window"), display=panel.get("display"), stale=True) # the page re-syncs
return panel
if "fx" in event and panel and panel.get("window"):
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())
Reference in new issue View Git Blame Copy Permalink