clone

a desktop tv and program guide for over-the-air broadcasts: what's on air, what your tuner gets, live tv, recordings

main
airwaves / build / linux / steam-art / steamclient.py
steamclient.py 4.8K executable
1#!/usr/bin/env python32"""Small helpers run on the SteamOS box (python3 stdlib only).34 find print "<grid dir> <appid>" for the Airwaves non-Steam shortcut5 eval <expression> evaluate JS in Steam's SharedJSContext over CEF remote debugging6 (needs Developer > CEF remote debugging, which Decky Loader turns on)7 apply <grid> <appid>8 hand the art already copied into <grid> to the running Steam client so it9 shows without a restart, and point the shortcut's icon at <appid>_icon.png1011Going through Steam's own SteamClient API lets us set the shortcut icon while Steam runs,12without hand-editing shortcuts.vdf (which Steam would overwrite).13"""14import base64, glob, json, os, socket, struct, sys, urllib.request1516NAME = "Airwaves"171819def parse_vdf(d, i=0):20 out = {}21 while True:22 t = d[i]; i += 123 if t == 8:24 return out, i25 j = d.index(b"\0", i); key = d[i:j].decode("utf-8", "replace"); i = j + 126 if t == 0:27 out[key], i = parse_vdf(d, i)28 elif t == 1:29 j = d.index(b"\0", i); out[key] = d[i:j].decode("utf-8", "replace"); i = j + 130 elif t == 2:31 out[key] = struct.unpack("<I", d[i:i + 4])[0]; i += 432 else:33 raise ValueError(f"vdf type {t}")343536def find():37 for path in glob.glob(os.path.expanduser("~/.local/share/Steam/userdata/*/config/shortcuts.vdf")):38 shortcuts = parse_vdf(open(path, "rb").read())[0].get("shortcuts", {})39 for s in shortcuts.values():40 if s.get("appname") == NAME:41 print(os.path.join(os.path.dirname(path), "grid"), s["appid"])42 return43 sys.exit(f"no {NAME} shortcut found")444546def cef_eval(expr):47 targets = json.load(urllib.request.urlopen("http://127.0.0.1:8080/json", timeout=5))48 url = next(t["webSocketDebuggerUrl"] for t in targets if t["title"] == "SharedJSContext")49 hostport, path = url[len("ws://"):].split("/", 1)50 host, port = hostport.rsplit(":", 1)51 s = socket.create_connection((host, int(port)), timeout=20)52 key = base64.b64encode(os.urandom(16)).decode()53 s.sendall((f"GET /{path} HTTP/1.1\r\nHost: {hostport}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"54 f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n").encode())55 buf = b""56 while b"\r\n\r\n" not in buf:57 buf += s.recv(4096)58 head, buf = buf.split(b"\r\n\r\n", 1)59 if b" 101 " not in head.split(b"\r\n")[0]:60 sys.exit(head.decode())61 msg = json.dumps({"id": 1, "method": "Runtime.evaluate",62 "params": {"expression": expr, "awaitPromise": True, "returnByValue": True}}).encode()63 n, mask = len(msg), os.urandom(4)64 frame = bytearray([0x81])65 if n < 126:66 frame.append(0x80 | n)67 elif n < 65536:68 frame += bytes([0x80 | 126]) + struct.pack(">H", n)69 else:70 frame += bytes([0x80 | 127]) + struct.pack(">Q", n)71 s.sendall(bytes(frame) + mask + bytes(b ^ mask[i % 4] for i, b in enumerate(msg)))7273 def need(k):74 nonlocal buf75 while len(buf) < k:76 chunk = s.recv(65536)77 if not chunk:78 raise EOFError("CEF closed the socket")79 buf += chunk8081 while True:82 need(2)83 op, n, off = buf[0] & 0x0F, buf[1] & 0x7F, 284 if n == 126:85 need(4); n = struct.unpack(">H", buf[2:4])[0]; off = 486 elif n == 127:87 need(10); n = struct.unpack(">Q", buf[2:10])[0]; off = 1088 need(off + n)89 payload, buf = buf[off:off + n], buf[off + n:]90 if op == 1:91 reply = json.loads(payload)92 if reply.get("id") == 1:93 result = reply.get("result", {})94 if "exceptionDetails" in result or "error" in reply:95 sys.exit(json.dumps(reply))96 return result.get("result", {}).get("value")979899# SetCustomArtworkForApp asset types, by grid file suffix.100ASSETS = {"p": 0, "_hero": 1, "_logo": 2, "": 3}101102103def apply(grid, appid):104 for suffix, kind in ASSETS.items():105 path = os.path.join(grid, f"{appid}{suffix}.png")106 if os.path.exists(path):107 data = base64.b64encode(open(path, "rb").read()).decode()108 cef_eval(f'SteamClient.Apps.SetCustomArtworkForApp({int(appid)}, "{data}", "png", {kind})')109 print(f"set {os.path.basename(path)}")110 icon = os.path.join(grid, f"{appid}_icon.png")111 if os.path.exists(icon):112 cef_eval(f"SteamClient.Apps.SetShortcutIcon({int(appid)}, {json.dumps(icon)})")113 print(f"set icon {icon}")114115116if __name__ == "__main__":117 if sys.argv[1:2] == ["find"]:118 find()119 elif sys.argv[1:2] == ["eval"]:120 print(json.dumps(cef_eval(sys.argv[2])))121 elif sys.argv[1:2] == ["apply"]:122 apply(sys.argv[2], sys.argv[3])123 else:124 sys.exit(__doc__)