From a8d5c4b44d30f18c8baad42fb3fef6674e2720ce Mon Sep 17 00:00:00 2001 From: saphid <4596216+saphid@users.noreply.github.com> Date: Mon, 12 Oct 2026 14:36:53 +1100 Subject: [PATCH] Panels: one deadline for the whole run; read gamescope's properties once 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 --- docs/streaming.md | 7 +- tests/test_touch.py | 175 +++++++++++++++++++++++++++++++------------ ui/frame_touch.py | 178 ++++++++++++++++++++++++++------------------ 3 files changed, 240 insertions(+), 120 deletions(-) diff --git a/docs/streaming.md b/docs/streaming.md index ad0073a..05a95c6 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -265,9 +265,12 @@ How (**verified 2026-09-29**, SteamOS 0.4.1, build 20260925.6191901): Desktop view shows that instead of a bare "no app panel": X tools missing, gamescope listing windows that aren't X11 windows on any display, the desktop running but not listed, or nothing open. It reads gamescope's properties from - `:0` first and probes the other displays if `:0` has none. Added for + `:0` first (one `xprop` call) and probes the other displays, 2 s each, only if `:0` + has none. A whole run stops after 12 s, inside the server's 20 s, and says the headset + was too slow. Added for [#73](https://github.com/saphid/frame-control/issues/73) (SteamOS 0.4.5, desktop - open but not found); the cause on that build isn't confirmed yet. + open but not found). The likely cause there was the page itself: a second + `loadPanels` (the panel switcher's) replaced the Desktop view's, fixed in #84. - Tested from the iPhone app (Simulator): a tap on the Desktop view focused a text box in the panel and the text field typed into it; a trackpad move went exactly (+40, +25). diff --git a/tests/test_touch.py b/tests/test_touch.py index 10aacb0..b9ff731 100644 --- a/tests/test_touch.py +++ b/tests/test_touch.py @@ -3,9 +3,13 @@ Run: python3 -m unittest discover -s tests """ import io +import json +import os import shutil import subprocess import sys +import tempfile +import time import unittest from pathlib import Path @@ -52,12 +56,9 @@ class Mapping(unittest.TestCase): self.assertEqual(t.locate(5, None)["display"], ":0") def test_focus_display_is_decoded(self): - t = self.t - saved = t.xprop_root - self.addCleanup(lambda: setattr(t, "xprop_root", saved)) for values, want in (([12602, 0, 58], ":1"), ([12346], ":0"), ([], None), ([0x41], None)): - t.xprop_root = lambda d, n, v=values: v - self.assertEqual(t.focus_display(), want) + self.assertEqual(self.t.focus_display({"GAMESCOPE_FOCUS_DISPLAY": values}), want) + self.assertIsNone(self.t.focus_display({})) def test_ascii_table_covers_printable_characters(self): for code in range(0x20, 0x7F): @@ -270,11 +271,11 @@ class FindingPanels(unittest.TestCase): def setUp(self): t = self.t - names = ("xprop_root", "window_info", "window_pid", "displays", "desktop_running", "has_gamescope") + names = ("read_props", "window_info", "window_pid", "displays", "desktop_running") saved = {n: getattr(t, n) for n in names} saved_which = t.shutil.which self.real_displays = saved["displays"] - self.real_has = saved["has_gamescope"] + self.real_read_props = saved["read_props"] def restore(): for n, v in saved.items(): @@ -285,9 +286,8 @@ class FindingPanels(unittest.TestCase): t._ROOT.clear() self.props = {} # (display, property) -> values self.windows = {} # (display, window) -> info - t.xprop_root = lambda d, n: list(self.props.get((d, n), [])) - t.has_gamescope = lambda d, timeout: any((d, n) in self.props for n in - ("GAMESCOPE_FOCUSABLE_WINDOWS", "GAMESCOPE_FOCUSED_WINDOW")) + self.reads = [] # displays whose gamescope properties were read + t.read_props = lambda d, limit=5.0: self.reads.append(d) or {n: list(v) for (dd, n), v in self.props.items() if dd == d} t.window_info = lambda d, w: self.windows.get((d, w)) t.window_pid = lambda d, w: None t.displays = lambda: [":0", ":1", ":2"] @@ -308,58 +308,79 @@ class FindingPanels(unittest.TestCase): self.assertEqual([(p["window"], p["display"], p["focused"]) for p in out["panels"]], [(4194307, ":0", True)]) self.assertNotIn("why", out) + def test_display_0_costs_one_read(self): + # The layout of every build seen so far: one read of :0's properties for the whole run. + self.plasma_on(":0") + self.t.panels() + self.assertEqual(self.reads, [":0"]) + def test_gamescope_properties_on_another_display_are_found(self): # A build that numbers its displays differently: :0 has no gamescope properties. self.plasma_on(":1") - self.assertEqual(self.t.gamescope_root(), ":1") + self.assertEqual(self.t.gamescope()[0], ":1") + self.reads.clear() + self.t._ROOT.clear() self.assertEqual(len(self.t.panels()["panels"]), 1) + self.assertEqual(self.reads, [":0", ":1"]) # read once, then reused by focus() and focusable() - def test_root_search_shares_one_deadline(self): - # Every display stalls: the search gives up within ROOT_DEADLINE, each probe within - # ROOT_PROBE_TIMEOUT, and the answer (:0) is remembered rather than searched for again. - t, now, probes = self.t, [0.0], [] - t.displays = lambda: [":0", ":1", ":2", ":3", ":4", ":5"] + def test_other_displays_are_probed_briefly(self): + limits = [] + self.t.read_props = lambda d, limit=5.0: limits.append((d, limit)) or {} + self.assertEqual(self.t.gamescope(), (":0", {})) + self.assertEqual(limits, [(":0", 5.0), (":1", self.t.ROOT_PROBE_TIMEOUT), (":2", self.t.ROOT_PROBE_TIMEOUT)]) - def stalled(display, timeout): - probes.append((display, timeout)) - now[0] += timeout - return False - t.has_gamescope = stalled - self.assertEqual(t.gamescope_root(clock=lambda: now[0]), ":0") - self.assertLessEqual(now[0], t.ROOT_DEADLINE) - self.assertTrue(all(0 < timeout <= t.ROOT_PROBE_TIMEOUT for _, timeout in probes)) - self.assertLess(len(probes), 6) - count = len(probes) - now[0] += 1 - self.assertEqual(t.gamescope_root(clock=lambda: now[0]), ":0") - self.assertEqual(len(probes), count) # remembered - now[0] += t.ROOT_RETRY - t.gamescope_root(clock=lambda: now[0]) - self.assertGreater(len(probes), count) # tried again later - - def test_root_found_is_remembered(self): + def test_input_helper_remembers_the_display(self): + # The input helper lives for a session: a found display is kept, a failed search retried later. + t = self.t self.plasma_on(":1") - calls = [] - has = self.t.has_gamescope - self.t.has_gamescope = lambda d, timeout: calls.append(d) or has(d, timeout) - self.assertEqual(self.t.gamescope_root(), ":1") - self.t.panels() - self.assertEqual(calls, [":0", ":1"]) # one search for the whole request + t.gamescope(); t.gamescope() + self.assertEqual(self.reads, [":0", ":1", ":1"]) + self.props.clear() + self.reads.clear() + t._ROOT.clear() + t.gamescope(); t.gamescope() + self.assertEqual(self.reads, [":0", ":1", ":2", ":0"]) # given up: :0 for ROOT_RETRY + t._ROOT[1] -= t.ROOT_RETRY + self.reads.clear() + t.gamescope() + self.assertEqual(self.reads, [":0", ":1", ":2"]) - def test_has_gamescope_reads_xprop_output(self): + def test_read_props_parses_xprop(self): t = self.t saved = t.subprocess.run self.addCleanup(setattr, t.subprocess, "run", saved) - for out, want in (("GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 1, 0, 2\nGAMESCOPE_FOCUSED_WINDOW(CARDINAL) = 1\n", True), - ("GAMESCOPE_FOCUSABLE_WINDOWS: not found.\nGAMESCOPE_FOCUSED_WINDOW(CARDINAL) = 0\n", True), - ("GAMESCOPE_FOCUSABLE_WINDOWS: not found.\nGAMESCOPE_FOCUSED_WINDOW: not found.\n", False)): - t.subprocess.run = lambda *a, o=out, **k: subprocess.CompletedProcess(a, 0, o, "") - self.assertEqual(self.real_has(":0", 1), want, out) + out = ("GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 1, 0, 2, 3, 769, 4\n" + "GAMESCOPE_FOCUSED_WINDOW(CARDINAL) = 0\nGAMESCOPE_FOCUS_DISPLAY: not found.\n") + t.subprocess.run = lambda *a, **k: subprocess.CompletedProcess(a, 0, out, "") + self.assertEqual(self.real_read_props(":0"), {"GAMESCOPE_FOCUSABLE_WINDOWS": [1, 0, 2, 3, 769, 4], + "GAMESCOPE_FOCUSED_WINDOW": [0]}) + t.subprocess.run = lambda *a, **k: subprocess.CompletedProcess(a, 0, "GAMESCOPE_FOCUSED_WINDOW: not found.\n", "") + self.assertEqual(self.real_read_props(":0"), {}) def hang(*a, **k): raise subprocess.TimeoutExpired(a[0], k.get("timeout")) t.subprocess.run = hang - self.assertFalse(self.real_has(":0", 1)) + self.assertEqual(self.real_read_props(":0"), {}) + + def test_deadline_returns_what_was_found(self): + t, clock = self.t, [0.0] + self.plasma_on(":0") + self.props[(":0", "GAMESCOPE_FOCUSABLE_WINDOWS")] = [4194307, 0, 2222, 5, 0, 1, 6, 0, 1] + saved = t.time.monotonic + self.addCleanup(setattr, t.time, "monotonic", saved) + t.time.monotonic = lambda: clock[0] + info = self.windows.get + + def slow(d, w): # every xwininfo takes 1 s: the run is out of time after a few + t.budget() + clock[0] += 1 + return info((d, w)) + t.window_info = slow + out = t.panels(limit=7) # focus: 2 reads; the first panel: 3 displays and its root + self.assertEqual([p["window"] for p in out["panels"]], [4194307]) + self.assertIn("took too long", out["why"]) + self.assertIn("Some may be missing", out["why"]) + self.assertIsNone(t.DEADLINE[0]) # the next run (or the input helper) has no deadline left over def test_listed_window_that_isnt_x11_explains_itself(self): self.plasma_on(":0") @@ -410,6 +431,66 @@ class FindingPanels(unittest.TestCase): builtins.open, t.os.listdir = real_open, real_listdir +FAKE_XPROP = """#!/bin/sh +sleep "${FAKE_X_DELAY:-0}" +if [ "$1" = "-root" ] && [ "$DISPLAY" = ":0" ]; then + echo "GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 11, 0, 101, 12, 769, 102" + echo "GAMESCOPE_FOCUSED_WINDOW(CARDINAL) = 11" + echo "GAMESCOPE_FOCUS_DISPLAY(CARDINAL) = 12346" +fi +""" +FAKE_XWININFO = """#!/bin/sh +sleep "${FAKE_X_DELAY:-0}" +[ "$DISPLAY" = ":0" ] || exit 1 +echo 'xwininfo: Window id: 0xb "Desktop"' +echo ' Absolute upper-left X: 0' +echo ' Absolute upper-left Y: 0' +echo ' Width: 1280' +echo ' Height: 800' +echo ' Map State: IsViewable' +""" + + +@unittest.skipIf(sys.platform == "win32", "fake X tools are shell scripts; frame_touch.py runs on the Frame") +class PanelsAsAFreshProcess(unittest.TestCase): + """`frame_touch.py panels` the way the server runs it: a new process each request, with + fake xprop and xwininfo first on PATH (two displays, :0 and :1).""" + + def run_panels(self, delay): + with tempfile.TemporaryDirectory() as bin_dir: + for name, body in (("xprop", FAKE_XPROP), ("xwininfo", FAKE_XWININFO)): + path = Path(bin_dir) / name + path.write_text(body) + path.chmod(0o755) + env = dict(os.environ, PATH=bin_dir + os.pathsep + os.environ.get("PATH", ""), FAKE_X_DELAY=str(delay), + PYTHONDONTWRITEBYTECODE="1") + code = ("import sys; sys.path.insert(0, sys.argv[1]); import frame_touch as t; " + "t.displays = lambda: [':0', ':1']; t.desktop_running = lambda: False; " + "sys.argv = ['frame_touch.py', 'panels']; sys.exit(t.main())") + start = time.monotonic() + r = subprocess.run([sys.executable, "-c", code, str(ROOT / "ui")], env=env, capture_output=True, + text=True, timeout=60) + took = time.monotonic() - start + self.assertEqual(r.returncode, 0, r.stderr) + return json.loads(r.stdout), took + + def test_lists_panels(self): + out, _ = self.run_panels(0) + self.assertEqual(out["focus"], 11) + self.assertEqual(out["focus_display"], ":0") + self.assertEqual([(p["window"], p["display"], p["focused"]) for p in out["panels"]], + [(11, ":0", True), (12, ":0", False)]) + self.assertNotIn("why", out) + + def test_slow_x_tools_stay_inside_the_servers_timeout(self): + # Review of #83: X tools answering in 4.6 s each would take well over the server's + # 20 s; one deadline for the run (PANELS_DEADLINE, 12 s) returns in time, saying why. + out, took = self.run_panels(4.6) + self.assertLess(took, 15, out) + self.assertEqual(out["panels"], []) + self.assertIn("took too long", out["why"]) + + @unittest.skipUnless(shutil.which("node"), "needs node") class PageGestures(unittest.TestCase): """Control's gestures and queue (tests/page/ctrl_gestures.mjs runs the real functions from index.html).""" diff --git a/ui/frame_touch.py b/ui/frame_touch.py index e5e21d9..8f1be7d 100644 --- a/ui/frame_touch.py +++ b/ui/frame_touch.py @@ -55,14 +55,41 @@ def say(state, **more): # ---- which panel has focus ----------------------------------------------------- -def xprop_root(display, name): +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", name], env=dict(os.environ, DISPLAY=display), - capture_output=True, text=True, timeout=5).stdout + 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 [] - values = out.split("=", 1)[1] if "=" in out else "" - return [int(v) for v in values.replace(",", " ").split() if v.isdigit()] + 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): @@ -70,7 +97,7 @@ def window_info(display, window): 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 + capture_output=True, text=True, timeout=budget()).stdout except (OSError, subprocess.SubprocessError): return None if "IsViewable" not in out: @@ -104,52 +131,42 @@ def displays(): return [f":{n}" for n in sorted(found)] -_ROOT = [] # [display, when found or None]: the display that carries gamescope's properties -ROOT_DEADLINE = 6.0 # seconds for the whole search: well inside the server's 20 s for /api/panels -ROOT_PROBE_TIMEOUT = 2.0 # seconds for any one display, so one stalled display can't use it all -ROOT_RETRY = 30.0 # seconds before a search that found nothing is tried again (the input helper lives on) +# 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 has_gamescope(display, timeout): - """Whether this display's root has gamescope's properties: one xprop call, bounded.""" - try: - out = subprocess.run(["xprop", "-root", "GAMESCOPE_FOCUSABLE_WINDOWS", "GAMESCOPE_FOCUSED_WINDOW"], - env=dict(os.environ, DISPLAY=display), capture_output=True, text=True, - timeout=timeout).stdout - except (OSError, subprocess.SubprocessError): - return False - return any(line.startswith("GAMESCOPE_") and "=" in line for line in out.splitlines()) +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. - -def gamescope_root(clock=time.monotonic): - """The X display whose root window carries gamescope's panel properties - (GAMESCOPE_FOCUSABLE_WINDOWS and friends). - - That's :0 on every build seen so far (SteamOS 0.3.0 to 0.4.1), so :0 is asked first - and is the answer when nothing publishes them. If a SteamOS update numbers its - displays differently, the others are probed rather than reporting no panels. The - search shares one deadline and is remembered: found for good, not found for ROOT_RETRY. + 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. """ - now = clock() - if _ROOT and (_ROOT[1] is None or now - _ROOT[1] < ROOT_RETRY): - return _ROOT[0] - deadline = now + ROOT_DEADLINE - found = None - for display in [":0"] + [d for d in displays() if d != ":0"]: - left = deadline - clock() - if left <= 0: - break - if has_gamescope(display, min(ROOT_PROBE_TIMEOUT, left)): - found = display - break - _ROOT[:] = [found, None] if found else [":0", clock()] - return _ROOT[0] + 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=5).stdout + capture_output=True, text=True, timeout=budget()).stdout except (OSError, subprocess.SubprocessError): return None value = out.rsplit("=", 1)[-1].strip() if "=" in out else "" @@ -177,20 +194,20 @@ def locate(window, pid): "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(gamescope_root(), "GAMESCOPE_FOCUSABLE_WINDOWS") +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(): - """The display of the focused window, from GAMESCOPE_FOCUS_DISPLAY on :0's root. +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 = xprop_root(gamescope_root(), "GAMESCOPE_FOCUS_DISPLAY") + 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") @@ -198,19 +215,23 @@ def focus_display(): def focus_now(): - """Just which window and display have focus: two property reads, for checking a press.""" - window = (xprop_root(gamescope_root(), "GAMESCOPE_FOCUSED_WINDOW") or [0])[0] - return (window or None, focus_display() if window else None) + """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(): +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).""" - window = (xprop_root(gamescope_root(), "GAMESCOPE_FOCUSED_WINDOW") or [0])[0] + 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() if w == window), (None, None)) - display = focus_display() + 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 {} @@ -221,20 +242,35 @@ def focus(): 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, unfound = [], 0 - for window, app, pid in focusable(): - 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"))}) +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 not 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