mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-18 11:01:18 +02:00
saphidandClaude Opus 5.5
a8d5c4b44d
Round-2 review of #83: the display search ran before the usual 5 s property and geometry reads, and its 30 s memory never helped because the server starts a fresh `python3 - panels` per request. With X tools answering in 4.6 s the run passed the server's 20 s. - One xprop call reads all of gamescope's root properties; :0 is read first and, when it has them, used with no other probe. Other displays are only asked (2 s each) if :0 has none. - focus() and focusable() reuse that reading instead of re-reading. - A `panels` run has one deadline (12 s); every xprop/xwininfo is cut to what's left, and past it the run returns what it found with a "why" saying the headset was too slow. - The display memory is documented as being for the long-lived input helper only. - Tests: `frame_touch.py panels` run as a fresh process with fake xprop and xwininfo on PATH (lists panels; 4.6 s tools return in about 12 s, where the previous head ran past 60 s), plus unit tests for the single read, probing, the deadline and partial results. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
584 lines
26 KiB
Python
584 lines
26 KiB
Python
"""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 re
|
|
import select
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
SOCKET = "/run/user/{uid}/gamescope-0-ei" # filled in on the Frame (Windows has no getuid; the tests import this)
|
|
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 -----------------------------------------------------
|
|
|
|
PROPS = ("GAMESCOPE_FOCUSABLE_WINDOWS", "GAMESCOPE_FOCUSED_WINDOW", "GAMESCOPE_FOCUS_DISPLAY")
|
|
TOOL_TIMEOUT = 5.0 # seconds for any one xprop or xwininfo
|
|
PANELS_DEADLINE = 12.0 # seconds for a whole `panels` run: well inside the server's 20 s for /api/panels
|
|
DEADLINE = [None] # when the current `panels` run must stop asking X tools (monotonic), if it has one
|
|
|
|
|
|
class OutOfTime(Exception):
|
|
"""The current run's deadline passed: stop and report what was found."""
|
|
|
|
|
|
def budget(limit=TOOL_TIMEOUT):
|
|
"""Seconds an X tool may take now: its own limit, cut to what's left of the run's deadline."""
|
|
if DEADLINE[0] is None:
|
|
return limit
|
|
left = DEADLINE[0] - time.monotonic()
|
|
if left <= 0:
|
|
raise OutOfTime()
|
|
return min(limit, left)
|
|
|
|
|
|
def read_props(display, limit=TOOL_TIMEOUT):
|
|
"""gamescope's root properties on one display, in one xprop call: {name: [values]} for
|
|
those the display has, {} if none (or xprop didn't answer)."""
|
|
try:
|
|
out = subprocess.run(["xprop", "-root", *PROPS], env=dict(os.environ, DISPLAY=display),
|
|
capture_output=True, text=True, timeout=budget(limit)).stdout
|
|
except (OSError, subprocess.SubprocessError):
|
|
return {}
|
|
props = {}
|
|
for line in out.splitlines():
|
|
name, sep, values = line.partition("=")
|
|
name = name.split("(", 1)[0].strip()
|
|
if sep and name in PROPS:
|
|
props[name] = [int(v) for v in values.replace(",", " ").split() if v.isdigit()]
|
|
return props
|
|
|
|
|
|
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=budget()).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():
|
|
"""Every X display on the Frame: the sockets in /tmp/.X11-unix, and the abstract ones
|
|
X servers on Linux usually listen on as well (in /proc/net/unix), in case a SteamOS
|
|
build stops sharing /tmp/.X11-unix with SSH sessions."""
|
|
found = set()
|
|
try:
|
|
found.update(int(n[1:]) for n in os.listdir("/tmp/.X11-unix") if n[:1] == "X" and n[1:].isdigit())
|
|
except OSError:
|
|
pass
|
|
try:
|
|
with open("/proc/net/unix") as f:
|
|
found.update(int(n) for n in re.findall(r"@/tmp/\.X11-unix/X(\d+)", f.read()))
|
|
except OSError:
|
|
pass
|
|
return [f":{n}" for n in sorted(found)]
|
|
|
|
|
|
# The input helper (`frame_touch.py` reading events) lives for a whole Control session and
|
|
# reads focus often, so it remembers which display has gamescope's properties: found, for
|
|
# good; not found, for ROOT_RETRY. A `panels` run is a fresh process each time, so this
|
|
# saves it nothing; it costs nothing either while :0 has the properties.
|
|
_ROOT = [] # [display, when it was given up on, or None if found]
|
|
ROOT_PROBE_TIMEOUT = 2.0 # seconds for each display other than :0, so a stalled one can't use the run
|
|
ROOT_RETRY = 30.0
|
|
|
|
|
|
def gamescope():
|
|
"""(display, properties): the X display whose root window carries gamescope's panel
|
|
properties (GAMESCOPE_FOCUSABLE_WINDOWS and friends), and what they say now.
|
|
|
|
That's :0 on every build seen so far (SteamOS 0.3.0 to 0.4.1), so :0 is read first and,
|
|
when it has them, used with no other probe. Only if it hasn't are the other displays
|
|
asked, briefly each; with none, :0 and no properties.
|
|
"""
|
|
if _ROOT and (_ROOT[1] is None or time.monotonic() - _ROOT[1] < ROOT_RETRY):
|
|
return _ROOT[0], read_props(_ROOT[0])
|
|
props, found = read_props(":0"), ":0"
|
|
if not props:
|
|
found = None
|
|
for display in displays():
|
|
if display != ":0":
|
|
props = read_props(display, ROOT_PROBE_TIMEOUT)
|
|
if props:
|
|
found = display
|
|
break
|
|
_ROOT[:] = [found, None] if found else [":0", time.monotonic()]
|
|
return _ROOT[0], props
|
|
|
|
|
|
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=budget()).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(props=None):
|
|
"""gamescope's focusable windows as (window, app id, pid), from props if already read."""
|
|
t = (gamescope()[1] if props is None else props).get("GAMESCOPE_FOCUSABLE_WINDOWS", [])
|
|
return [tuple(t[i:i + 3]) for i in range(0, len(t) - 2, 3)]
|
|
|
|
|
|
def focus_display(props=None):
|
|
"""The display of the focused window, from GAMESCOPE_FOCUS_DISPLAY on gamescope's root.
|
|
|
|
gamescope writes the name (":1") as 32-bit items, so its first four bytes land,
|
|
little-endian, in the first value: 12602 is 0x313A, ":1" (steamcompmgr.cpp;
|
|
seen 2026-09-29).
|
|
"""
|
|
values = (gamescope()[1] if props is None else props).get("GAMESCOPE_FOCUS_DISPLAY")
|
|
if not values:
|
|
return None
|
|
name = (values[0] & 0xFFFFFFFF).to_bytes(4, "little").split(b"\0", 1)[0].decode("ascii", "replace")
|
|
return name if name[:1] == ":" and name[1:].isdigit() else None
|
|
|
|
|
|
def focus_now():
|
|
"""Just which window and display have focus: one property read, for checking a press."""
|
|
props = gamescope()[1]
|
|
window = (props.get("GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
|
|
return (window or None, focus_display(props) if window else None)
|
|
|
|
|
|
def focus(props=None):
|
|
"""The panel that has focus in the headset: window, display, name and sizes (gamescope
|
|
publishes the window and its display on its root display, :0 so far). props: gamescope's
|
|
properties if already read."""
|
|
if props is None:
|
|
props = gamescope()[1]
|
|
window = (props.get("GAMESCOPE_FOCUSED_WINDOW") or [0])[0]
|
|
if not window:
|
|
return {"window": None}
|
|
app, pid = next(((a, p) for w, a, p in focusable(props) if w == window), (None, None))
|
|
display = focus_display(props)
|
|
info = window_info(display, window) if display else None
|
|
if info:
|
|
root = window_info(display, "root") or {}
|
|
panel = {"window": window, "display": display, **info,
|
|
"root": [root.get("width", info["width"]), root.get("height", info["height"])]}
|
|
else:
|
|
panel = locate(window, pid) # no display published: tell them apart by pid
|
|
return {**panel, "app": app} if panel else {"window": None}
|
|
|
|
|
|
def panels(limit=None):
|
|
"""Every app panel (gamescope's focusable windows), for watching one that hasn't focus.
|
|
|
|
One deadline (PANELS_DEADLINE) covers the whole run, inside the server's timeout: past
|
|
it, the panels found so far come back with a "why" saying the headset was too slow.
|
|
"""
|
|
DEADLINE[0] = time.monotonic() + (PANELS_DEADLINE if limit is None else limit)
|
|
now, found, unfound, late = {}, [], 0, False
|
|
try:
|
|
props = gamescope()[1]
|
|
now = focus(props)
|
|
for window, app, pid in focusable(props):
|
|
panel = locate(window, pid)
|
|
unfound += panel is None
|
|
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"))})
|
|
late = time.monotonic() >= DEADLINE[0] # the last read may have been cut short
|
|
except OutOfTime:
|
|
late = True
|
|
finally:
|
|
DEADLINE[0] = None
|
|
out = {"focus": now.get("window"), "focus_display": now.get("display"), "panels": found}
|
|
if late:
|
|
out["why"] = ("The headset took too long to list its panels (its X tools didn't answer in time). "
|
|
+ ("Some may be missing. " if found else "") + "Try again, or use the Headset view.")
|
|
elif not found:
|
|
out["why"] = why_none(unfound)
|
|
return out
|
|
|
|
|
|
# The headset's Linux desktop, by process name (docs/how-the-frame-works.md).
|
|
DESKTOP = ("plasmashell", "kwin_wayland", "kwin_wayland_wrapper", "steamos-nested-desktop", "startplasma-wayland")
|
|
|
|
|
|
def desktop_running():
|
|
"""Whether the headset's Linux desktop (the nested Plasma session) is running."""
|
|
try:
|
|
pids = [p for p in os.listdir("/proc") if p.isdigit()]
|
|
except OSError:
|
|
return False
|
|
for pid in pids:
|
|
try:
|
|
with open(f"/proc/{pid}/comm") as f:
|
|
if f.read().strip() in DESKTOP:
|
|
return True
|
|
except OSError:
|
|
continue
|
|
return False
|
|
|
|
|
|
def why_none(listed):
|
|
"""Why panels() found nothing, in words the person at the computer can act on.
|
|
listed: how many of the windows gamescope lists aren't X11 windows on any display."""
|
|
missing = [tool for tool in ("xprop", "xwininfo") if not shutil.which(tool)]
|
|
if missing:
|
|
return (f"This SteamOS build has no {' or '.join(missing)}, which Frame Control uses to find the "
|
|
"headset's panels. Use the Headset view instead, and please report your SteamOS version.")
|
|
if listed:
|
|
# gamescope lists windows, but none is an X11 window on any display: a Wayland
|
|
# window (a desktop drawn straight to gamescope, say) can't be found or captured this way.
|
|
return (f"The headset has {listed} panel{'s' if listed > 1 else ''} open, but Frame Control can't "
|
|
f"find {'its' if listed == 1 else 'their'} window. This SteamOS build may draw "
|
|
f"{'it' if listed == 1 else 'them'} in a way the Desktop view can't capture yet. "
|
|
"Use the Headset view instead, and please report your SteamOS version.")
|
|
if desktop_running():
|
|
return ("The headset's desktop is running, but SteamOS isn't listing it as a panel. Look at it in "
|
|
"the headset (or close and reopen it), then try again, or use the Headset view.")
|
|
return "No app panel is open in the headset. Open the desktop or an app in the headset, then try again."
|
|
|
|
|
|
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.format(uid=os.getuid()).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, self.keys, self.alive = None, 0, set(), set(), True
|
|
|
|
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
|
|
# Releases that arrived while it was paused were dropped: let go of
|
|
# everything now, so the headset and this agent agree nothing is held.
|
|
if self.held or self.keys:
|
|
self.release_all()
|
|
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, self.alive = None, False, 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 closed its input socket" if not self.alive
|
|
else "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()
|
|
(self.keys.add if down else self.keys.discard)(code)
|
|
# 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):
|
|
"""Let go of every button and key still down, so nothing stays held in the headset."""
|
|
for code in list(self.held):
|
|
name = next(n for n, c in BUTTONS.items() if c == code)
|
|
self.button(name, False)
|
|
for code in list(self.keys):
|
|
self.key(code, 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)))
|
|
|
|
|
|
STALE = [False] # whether the last status said a tap went nowhere
|
|
|
|
|
|
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.
|
|
"""
|
|
# Moves may use a focus reading up to a second old; anything that acts (a press, key,
|
|
# text or scroll) reads it afresh, so it can't land on a panel that took focus since.
|
|
acts = any(k in event for k in ("button", "key", "text", "scroll")) and event.get("down") is not False
|
|
if "window" in event:
|
|
if panel and acts and focus_now() == (panel.get("window"), panel.get("display")):
|
|
pass # still the same panel (its geometry is re-read on the usual one-second schedule)
|
|
elif acts or 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
|
|
STALE[0] = True
|
|
return panel
|
|
if STALE[0] and not stale:
|
|
# Anything that goes through (a trackpad move names no panel) means caught up: stop re-syncing.
|
|
STALE[0] = False
|
|
say("ready", focus=(panel or {}).get("window"), display=(panel or {}).get("display"))
|
|
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:
|
|
waited = False
|
|
for event in events(line):
|
|
if gs.device is None and waited:
|
|
continue # still paused: don't wait again for each event of this batch
|
|
if gs.device is None:
|
|
waited = True
|
|
# Paused (gamescope can pause the device): wait a moment; drop this
|
|
# event if it doesn't come back. Only a disconnect ends the session.
|
|
try:
|
|
gs.wait_ready(2)
|
|
except RuntimeError:
|
|
if not gs.alive:
|
|
raise
|
|
continue
|
|
try:
|
|
panel = apply(gs, event, panel)
|
|
except (ValueError, KeyError, TypeError, OSError):
|
|
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())
|