ft-steam: open Steam's menu through Steam's own UI

steam/ft-steam menu opens the SteamVR dashboard on Steam's menu, or closes
the dashboard if it's up, without pointer mode: it asks Steam's UI over its
debugging port to show its dashboard overlay (ShowVROverlay, what Steam
calls itself) and focus the Steam frame's left menu (MenuStore.OpenMainMenu).
ft-steam check says whether those calls still exist, and update-check.py
runs it, since a Steam client update can rename them.

The CDP client moves from display-settings/steam_settings.py to
steam/steamui.py, so both use it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

display-settings/steam_settings.py

@@ -4,117 +4,26 @@ Steam, not systemd, puts the Frame to sleep: after "When Plugged In and Idle ->
 (an hour by default) without input, even while it charges. That's a Steam client setting,
 `system_idle_suspend_ac_sec` (0 = never), with no file or command line to change it. Steam
 on the Frame runs with -cef-enable-debugging, so its UI's JavaScript context
-(SharedJSContext) is reachable over the Chrome DevTools Protocol on 127.0.0.1:8080. There
+(SharedJSContext) is reachable over the Chrome DevTools Protocol on 127.0.0.1:8080
-`settingsStore.clientSettings` has the current values, and `SteamClient.Settings.SetSetting`
+(steam/steamui.py). There `settingsStore.clientSettings` has the current values, and
-takes a change as a serialized CMsgClientSettings protobuf, which is what Steam's own
+`SteamClient.Settings.SetSetting` takes a change as a serialized CMsgClientSettings protobuf,
-Settings -> Power page sends. Standard library only (a minimal WebSocket client).
+which is what Steam's own Settings -> Power page sends.
 """
-import base64
-import json
 import os
-import socket
+import sys
-import struct
+
-import urllib.request
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "steam"))
+from steamui import SteamUnreachable, evaluate  # noqa: E402,F401  (callers catch SteamUnreachable here)
 
-CDP_PORT = 8080
 # CMsgClientSettings field numbers (Steam's UI bundle maps the names to these).
 FIELDS = {"system_idle_suspend_ac_sec": 24004, "system_idle_suspend_battery_sec": 24003}
 
 
-class SteamUnreachable(Exception):
-    pass
-
-
-class _WebSocket:
-    def __init__(self, url, timeout=5):
-        host_port, path = url[len("ws://"):].split("/", 1)
-        host, port = host_port.rsplit(":", 1)
-        self.sock = socket.create_connection((host, int(port)), timeout=timeout)
-        key = base64.b64encode(os.urandom(16)).decode()
-        self.sock.sendall((f"GET /{path} HTTP/1.1\r\nHost: {host_port}\r\nUpgrade: websocket\r\n"
-                           f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())
-        head = b""
-        while b"\r\n\r\n" not in head:
-            chunk = self.sock.recv(4096)
-            if not chunk:
-                raise SteamUnreachable("Steam closed the connection")
-            head += chunk
-        if b" 101 " not in head.split(b"\r\n", 1)[0]:
-            raise SteamUnreachable(head.split(b"\r\n", 1)[0].decode(errors="replace"))
-        self.buf = head.split(b"\r\n\r\n", 1)[1]
-
-    def send(self, text):
-        data, mask = text.encode(), os.urandom(4)
-        n = len(data)
-        if n < 126:
-            head = struct.pack(">BB", 0x81, 0x80 | n)
-        elif n < 65536:
-            head = struct.pack(">BBH", 0x81, 0x80 | 126, n)
-        else:
-            head = struct.pack(">BBQ", 0x81, 0x80 | 127, n)
-        self.sock.sendall(head + mask + bytes(b ^ mask[i % 4] for i, b in enumerate(data)))
-
-    def _take(self, n):
-        while len(self.buf) < n:
-            chunk = self.sock.recv(65536)
-            if not chunk:
-                raise SteamUnreachable("Steam closed the connection")
-            self.buf += chunk
-        out, self.buf = self.buf[:n], self.buf[n:]
-        return out
-
-    def recv(self):
-        message = b""
-        while True:
-            b0, b1 = self._take(2)
-            n = b1 & 0x7F
-            if n == 126:
-                n = struct.unpack(">H", self._take(2))[0]
-            elif n == 127:
-                n = struct.unpack(">Q", self._take(8))[0]
-            message += self._take(n)
-            if b0 & 0x80:
-                return message.decode(errors="replace")
-
-    def close(self):
-        self.sock.close()
-
-
-def _evaluate(expression):
-    """Runs JavaScript in Steam's SharedJSContext and returns its (awaited) value."""
-    try:
-        with urllib.request.urlopen(f"http://127.0.0.1:{CDP_PORT}/json", timeout=3) as r:
-            targets = json.load(r)
-    except OSError as e:
-        raise SteamUnreachable(f"Steam isn't reachable on port {CDP_PORT} ({e})") from e
-    url = next((t["webSocketDebuggerUrl"] for t in targets if t.get("title") == "SharedJSContext"), None)
-    if not url:
-        raise SteamUnreachable("Steam's UI isn't running")
-    try:
-        ws = _WebSocket(url)
-        try:
-            ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate",
-                                "params": {"expression": expression, "awaitPromise": True, "returnByValue": True}}))
-            while True:
-                reply = json.loads(ws.recv())
-                if reply.get("id") == 1:
-                    break
-        finally:
-            ws.close()
-    except OSError as e:
-        raise SteamUnreachable(str(e)) from e
-    result = reply.get("result", {})
-    if "exceptionDetails" in result:
-        details = result["exceptionDetails"]
-        raise SteamUnreachable(details.get("exception", {}).get("description") or details.get("text", "error"))
-    return result.get("result", {}).get("value")
-
-
 def sleep_settings():
     """{"ac": seconds, "battery": seconds}: when Steam puts the Frame to sleep without input,
     plugged in and on battery (0 = never)."""
-    value = _evaluate("(() => { const c = settingsStore.clientSettings; "
+    value = evaluate("(() => { const c = settingsStore.clientSettings; "
-                      "return {ac: c.system_idle_suspend_ac_sec, battery: c.system_idle_suspend_battery_sec}; })()")
+                     "return {ac: c.system_idle_suspend_ac_sec, battery: c.system_idle_suspend_battery_sec}; })()")
     if not isinstance(value, dict) or not all(isinstance(value.get(k), int) for k in ("ac", "battery")):
         raise SteamUnreachable("Steam's settings don't have the sleep timeouts")
     return value
@@ -125,7 +34,7 @@ def set_sleep_setting(name, seconds):
     field, seconds = FIELDS[name], int(seconds)
     if seconds < 0:
         raise ValueError("seconds must be 0 (never) or more")
-    ok = _evaluate(f"""(async () => {{
+    ok = evaluate(f"""(async () => {{
         const bytes = [];
         const varint = n => {{ while (n > 127) {{ bytes.push((n & 127) | 128); n = Math.floor(n / 128); }} bytes.push(n); }};
         varint({field} * 8); varint({seconds});

scripts/update-check.py

@@ -321,6 +321,30 @@ def check_services(sockets, desktop_up):
         report("FAIL", "XRService", "not running, so hand tracking has no cameras")
 
 
+def check_steam_ui():
+    """Steam's UI calls the Steam menu shortcut uses (steam/ft-steam). They're Steam client
+    internals, so a Steam update can change them too, not just a SteamOS one."""
+    # Run from stdin (over SSH) there's no __file__: try the usual checkouts.
+    repos = [os.path.dirname(os.path.dirname(os.path.abspath(__file__)))] if "__file__" in globals() else []
+    repos += [os.path.join(HOME, "frametop"), os.path.join(HOME, "dev/frametop")]
+    tool = next((t for t in (os.path.join(r, "steam/ft-steam") for r in repos) if os.path.exists(t)), None)
+    if not tool:
+        report("skip", "Steam menu shortcut", "steam/ft-steam isn't in a checkout here")
+        return
+    try:
+        p = subprocess.run([tool, "check"], capture_output=True, text=True, timeout=15)
+    except (OSError, subprocess.TimeoutExpired) as e:
+        report("warn", "Steam menu shortcut", f"ft-steam check didn't run ({e})")
+        return
+    detail = (p.stdout + p.stderr).strip()
+    if p.returncode == 0:
+        report("ok", "Steam menu shortcut", "Steam's UI still has the calls steam/ft-steam makes")
+    elif detail.startswith("ft-steam:"):
+        report("skip", "Steam menu shortcut", detail[len("ft-steam:"):].strip())
+    else:
+        report("warn", "Steam menu shortcut", f"{detail}; a Meta tap won't open the Steam menu (steam/ft-steam)")
+
+
 def check_eye_tracker():
     gaze = systemctl("is-enabled", "frametop-gaze") == "enabled"
     try:
@@ -369,6 +393,7 @@ def main():
         check_openvr()
         check_overlays(desktop_up)
         check_services(sockets, desktop_up)
+        check_steam_ui()
         check_eye_tracker()
 
     if args == ["--mark-good"]:

steam/ft-steam

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""ft-steam: Steam's menu and checks, through Steam's own UI (steamui.py).
+
+  ft-steam menu    open the SteamVR dashboard on Steam's menu, or close the dashboard if it's
+                   open (the Steam button's job, as a shortcut: the input relay's steam_menu)
+  ft-steam check   do Steam's UI calls that the above needs still exist (update-check.py)
+
+No pointer mode needed: it asks Steam to show its dashboard overlay (window.ShowVROverlay, what
+Steam itself calls) and to focus the Steam frame's left menu (MenuStore.OpenMainMenu, which in
+VR invokes the frame's focus_left_frame_menu_action). SteamClient.OpenVR.VROverlay says
+whether the dashboard is up and hides it. These are Steam's UI internals, not an API, so a
+Steam client update can rename them: `ft-steam check` says so.
+"""
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)))
+from steamui import SteamUnreachable, evaluate  # noqa: E402
+
+WINDOW = "SteamUIStore.WindowStore.VRGamepadUIMainWindowInstance"
+
+MENU = f"""(async () => {{
+    const vr = SteamClient.OpenVR.VROverlay;
+    if (await vr.IsDashboardVisible()) {{ vr.HideDashboard(); return "closed"; }}
+    const w = {WINDOW};
+    if (!w) throw new Error("Steam's VR dashboard window isn't open");
+    ShowVROverlay(w.GetVROverlayKey());
+    // The frame's menu takes focus only once the dashboard is up.
+    for (let i = 0; i < 40 && !(await vr.IsDashboardVisible()); i++) await new Promise(r => setTimeout(r, 25));
+    w.MenuStore.OpenMainMenu();
+    return "opened";
+}})()"""
+
+CHECK = f"""(() => {{
+    const missing = [];
+    const vr = SteamClient.OpenVR && SteamClient.OpenVR.VROverlay;
+    for (const f of ["IsDashboardVisible", "HideDashboard"]) if (!vr || typeof vr[f] !== "function") missing.push("SteamClient.OpenVR.VROverlay." + f);
+    if (typeof window.ShowVROverlay !== "function") missing.push("ShowVROverlay");
+    const w = {WINDOW};
+    if (!w) missing.push("{WINDOW}");
+    else {{
+        if (typeof w.GetVROverlayKey !== "function") missing.push("GetVROverlayKey");
+        if (!w.MenuStore || typeof w.MenuStore.OpenMainMenu !== "function") missing.push("MenuStore.OpenMainMenu");
+    }}
+    return missing;
+}})()"""
+
+
+def main():
+    cmd = sys.argv[1] if len(sys.argv) == 2 else ""
+    if cmd not in ("menu", "check"):
+        sys.exit(__doc__)
+    try:
+        result = evaluate(MENU if cmd == "menu" else CHECK)
+    except SteamUnreachable as e:
+        sys.exit(f"ft-steam: {e}")
+    if cmd == "check":
+        if result:
+            sys.exit("Steam's UI lacks " + ", ".join(result))
+        print("ok")
+    else:
+        print(result)
+
+
+if __name__ == "__main__":
+    main()

steam/steamui.py

@@ -0,0 +1,108 @@
+"""Run JavaScript in Steam's UI, standard library only.
+
+Steam on the Frame runs with -cef-enable-debugging, so its UI's JavaScript context
+(SharedJSContext) is reachable over the Chrome DevTools Protocol on 127.0.0.1:8080. Steam's
+own calls live there: SteamClient.* (settings, SteamVR's dashboard) and the UI's stores
+(SteamUIStore). Used by Frametop Display Settings (display-settings/steam_settings.py) and
+ft-steam (the Steam menu shortcut).
+
+  evaluate(expression)   the expression's value (awaited, JSON-able); SteamUnreachable when
+                         Steam or its UI isn't there, or the script threw
+"""
+import base64
+import json
+import os
+import socket
+import struct
+import urllib.request
+
+CDP_PORT = 8080
+
+
+class SteamUnreachable(Exception):
+    pass
+
+
+class _WebSocket:
+    def __init__(self, url, timeout=5):
+        host_port, path = url[len("ws://"):].split("/", 1)
+        host, port = host_port.rsplit(":", 1)
+        self.sock = socket.create_connection((host, int(port)), timeout=timeout)
+        key = base64.b64encode(os.urandom(16)).decode()
+        self.sock.sendall((f"GET /{path} HTTP/1.1\r\nHost: {host_port}\r\nUpgrade: websocket\r\n"
+                           f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())
+        head = b""
+        while b"\r\n\r\n" not in head:
+            chunk = self.sock.recv(4096)
+            if not chunk:
+                raise SteamUnreachable("Steam closed the connection")
+            head += chunk
+        if b" 101 " not in head.split(b"\r\n", 1)[0]:
+            raise SteamUnreachable(head.split(b"\r\n", 1)[0].decode(errors="replace"))
+        self.buf = head.split(b"\r\n\r\n", 1)[1]
+
+    def send(self, text):
+        data, mask = text.encode(), os.urandom(4)
+        n = len(data)
+        if n < 126:
+            head = struct.pack(">BB", 0x81, 0x80 | n)
+        elif n < 65536:
+            head = struct.pack(">BBH", 0x81, 0x80 | 126, n)
+        else:
+            head = struct.pack(">BBQ", 0x81, 0x80 | 127, n)
+        self.sock.sendall(head + mask + bytes(b ^ mask[i % 4] for i, b in enumerate(data)))
+
+    def _take(self, n):
+        while len(self.buf) < n:
+            chunk = self.sock.recv(65536)
+            if not chunk:
+                raise SteamUnreachable("Steam closed the connection")
+            self.buf += chunk
+        out, self.buf = self.buf[:n], self.buf[n:]
+        return out
+
+    def recv(self):
+        message = b""
+        while True:
+            b0, b1 = self._take(2)
+            n = b1 & 0x7F
+            if n == 126:
+                n = struct.unpack(">H", self._take(2))[0]
+            elif n == 127:
+                n = struct.unpack(">Q", self._take(8))[0]
+            message += self._take(n)
+            if b0 & 0x80:
+                return message.decode(errors="replace")
+
+    def close(self):
+        self.sock.close()
+
+
+def evaluate(expression):
+    """Runs JavaScript in Steam's SharedJSContext and returns its (awaited) value."""
+    try:
+        with urllib.request.urlopen(f"http://127.0.0.1:{CDP_PORT}/json", timeout=3) as r:
+            targets = json.load(r)
+    except OSError as e:
+        raise SteamUnreachable(f"Steam isn't reachable on port {CDP_PORT} ({e})") from e
+    url = next((t["webSocketDebuggerUrl"] for t in targets if t.get("title") == "SharedJSContext"), None)
+    if not url:
+        raise SteamUnreachable("Steam's UI isn't running")
+    try:
+        ws = _WebSocket(url)
+        try:
+            ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate",
+                                "params": {"expression": expression, "awaitPromise": True, "returnByValue": True}}))
+            while True:
+                reply = json.loads(ws.recv())
+                if reply.get("id") == 1:
+                    break
+        finally:
+            ws.close()
+    except OSError as e:
+        raise SteamUnreachable(str(e)) from e
+    result = reply.get("result", {})
+    if "exceptionDetails" in result:
+        details = result["exceptionDetails"]
+        raise SteamUnreachable(details.get("exception", {}).get("description") or details.get("text", "error"))
+    return result.get("result", {}).get("value")