steam-frame/MichaelScottsman--LiquidAss-Frame
1
0 Fork 0
mirror of https://github.com/MichaelScottsman/LiquidAss-Frame.git synced 2026-10-19 16:01:10 +02:00
Code Issues Packages Projects Releases 7 Wiki Activity
Files
main
MichaelScottsman--LiquidAss…/device/lgs_shell.py
T
MichaelScottsman 8b422d1cd6 AssPod in + > Launch Program: the installer adds it once VLC is there, and it opens the assPod with the theme on or off 
lgs pod: with the theme on, the bar's asspod.open; with it off, only the assPod's runtime modules
in Steam (rt.podOnly, no liveness watch on the CSS core) and lgs-shell in pod mode (no theme, the
assPod's page script only), which exits once the assPod is away with nothing playing. lgs on
replaces a pod unit with a full one.
2026-10-10 14:37:53 -04:00

4551 lines
233 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""Glass Shell native layer daemon: the transient user unit "lgs-shell".
While the theme is on, this keeps SteamVR's own pages themed (what the old
"lgs-vr" watcher did). With the native layer opted in (lgs on --native), it
also joins three things (docs/NATIVE.md):
Steam (devtools 8080, SharedJSContext)
device/lgs_layers.js reports the elements that pop out, through the CDP
binding "lgsLayers"; html.lgs-native is set on a window, and popped
elements are acknowledged (data-lgs-pop), only once the compositor
shows their glass
glassd (native/glassd/glassd, a child process)
renders the room-refracting window glass and the Liquid Glass slabs
into dmabuf overlays "glassd.<surface>"; it reads /dev/shm/lgs/glassd.json
and answers in /dev/shm/lgs/glassd-out.json
SteamVR systemui (devtools 8090)
device/lgs_sg.js builds the scene-graph nodes that layer glassd's glass
and crops of Steam's own textures in stereo depth (window.__LGS_SG)
Native mode is decided once when the daemon starts (native = auto | on | off;
auto reads /tmp/lgs/flags.json, then device/defaults.json, and is CSS only
unless one of them says native: on): a glassd binary that appears later is
never picked up, and one that disappears ends native mode for the session.
Without it (the default, and what the "+ > Launch Program" toggle runs) the
daemon is CSS only: SteamVR page theming, no reporter in Steam, no glassd, no
camera, and in systemui only what a flagged transform override or a window
dim needs (lgs_sg.js, contracts/sg.md §4-5).
In both modes it also (docs/phase2/contracts/daemon.md):
- publishes geometry (S, r, Hm, the frame menu's panel) and the shown frame
page from systemui to Steam's runtime bridge (__LGS_RT.bridge);
- relays Steam's "lgsAction" binding calls to device/shell_ext plugins,
strictly validated and rate-limited;
- injects SteamVR page scripts device/vr/<page>.<name>.js and removes them
again.
Pod mode (--pod, what `lgs pod` starts while the theme is off; + > Launch
Program > AssPod): no theme anywhere. SteamVR's pages are not themed, only
the assPod's page script goes in, Steam runs only the assPod's runtime
modules (rt.podOnly, loaded by lgs pod), and a theme that is off is not a
reason to go dormant. It exits once the assPod has been away with nothing
playing for POD_IDLE_S, at once after a Steam restart, and on lgs on (which
starts a full unit in its place) or lgs off.
Nothing persists: the unit is transient (systemd-run --collect) and writes no
bytecode (python3 -B); lgs off stops it at once. A theme off without lgs off
(a lab --stock step) makes it dormant: native layer, glassd, reporter and scene
graph off, SteamVR page theming kept; it resumes when the theme is back and
exits after shellThemeGraceS (600 s), at once after a Steam restart, or after
120 s without Steam (contracts/daemon.md §1). On exit it removes every node,
class, binding global and file it made. If it dies without cleaning up,
lgs-native clears itself within 3 s and the scene-graph nodes within 12 s
(heartbeat watchdogs).
lgs_shell.py start [--native|--css|--auto|--pod] [--glassd PATH|none]
[--glassd-args "ARGS"] [--feed] [--stay]
[--test-report PATH] [--assume-caps CAPS]
start the unit; lgs on does this.
--native opts in to the native layer
(from this command glassd gets
--no-feed unless --feed or
--glassd-args is given); --css is CSS
only; neither (or --auto) keeps a
running unit and resolves "auto" for
a new one. --stay (lab tests and
native-session only): dormant up to
STAY_MAX_S while the theme is off,
and native only for STAY_MAX_S.
--test-report (lab) replaces the
reporter with a report JSON file.
lgs_shell.py stop stop it (clean teardown)
lgs_shell.py status unit + live daemon status (JSON)
lgs_shell.py selftest NAME lab checks: actions (offline), dm1,
dm2, dm3, dm4, dm5, dm6, grace,
plates (contract §11)
lgs_shell.py daemon [--native] [--glassd PATH|none] ...
the unit's process (foreground)
"""
import sys
sys.dont_write_bytecode = True # no __pycache__ next to the code (hard rule 7); the unit also runs -B
import asyncio # noqa: E402
import collections # noqa: E402
import hashlib # noqa: E402
import json # noqa: E402
import os # noqa: E402
import shutil # noqa: E402
import signal # noqa: E402
import subprocess # noqa: E402
import time # noqa: E402
import traceback # noqa: E402
import urllib.request # noqa: E402
import weakref # noqa: E402
HERE = os.path.dirname(os.path.realpath(__file__))
sys.path.insert(0, HERE)
import lgs # noqa: E402
import lgs_vr # noqa: E402
import shell_ext # noqa: E402
UNIT = "lgs-shell"
SHM = "/dev/shm/lgs"
GLASSD_JSON = os.path.join(SHM, "glassd.json")
GLASSD_OUT = os.path.join(SHM, "glassd-out.json")
STATUS = os.path.join(SHM, "shell.json")
SG_FLAGS = "/tmp/lgs/sg-flags.json" # optional lgs_sg.js panel flags (wearer tests; transient)
FLAGS_FILE = "/tmp/lgs/flags.json" # session flags (RAM; P10 writes it for a locked step)
DEFAULTS_FILE = os.path.join(HERE, "defaults.json") # V1's shipped defaults (PLAN §1.17)
GLASSD_BIN = os.path.join(lgs.ROOT, "native", "glassd", "glassd")
LAYERS_JS = os.path.join(HERE, "lgs_layers.js")
LAYERS_CFG = os.path.join(lgs.THEME_DIR, "layers.json") # Phase 1 configuration (retired by P6)
LAYERS_DIR = os.path.join(lgs.THEME_DIR, "layers") # Phase 2 fragments, name order
SG_JS = os.path.join(HERE, "lgs_sg.js")
SG_RULES_DIR = os.path.join(lgs.THEME_DIR, "sg") # transform overrides (contracts/sg.md §4)
MOTION_JS = os.path.join(HERE, "shared", "motion.js") # P5's springs, prepended to lgs_sg.js
VR_SCRIPT_DIRS = (os.path.join(HERE, "vr"), "/tmp/lgs/vr-scripts") # page scripts; the second for tests
VR_PAUSE = lgs_vr.PAUSE
VR_PAUSE_MAX_S = 300 # a lab --theme off pause on vr: pages older than this is stale (REQ C6b->P8)
BINDING = "lgsLayers"
ACTION_BINDING = "lgsAction" # Steam -> daemon actions (shell_ext)
REPORTER = "__LGS_LAYERS" # global installed by lgs_layers.js
STEAM_MAIN_KEY = "valve.steam.gamepadui.main"
DAEMON_BEAT_S = 2.0 # bridge 'daemon' heartbeat; its ttlMs is 3x this
GEOM_POLL_S = 1.0 # systemui geometry and page poll
GEOM_DUMP_S = 15.0 # DumpLaserOverlays (a ~100 ms compositor round trip) as a safety net this often;
GEOM_DUMP_AFTER = (1.0, 3.0, 6.0) # ... and these many s after the frame menu's node changed (the dump lags it)
ACTION_TICK_S = 1.0 # plugin scan and tick
RAMP_IN_MS = {"window": 735, "thick": 735} # GM §5.2: covers and thick ride sheet-in (settles ~0.73 s)
RAMP_OUT_MS = {"window": 514, "thick": 514} # sheet-out (~0.51 s)
RAMP_IN_DEFAULT_MS, RAMP_OUT_DEFAULT_MS, RAMP_REDUCED_MS = 250, 350, 180
PAGE_OUT_MS = 150 # --lgs-d-page-out: a leaving route's fade; plates removed with it fade on it
TUNE_SAVE_S = 0.4 # the wearer's tune is saved (and the dial file mirrored) this long after the last change
# glassd.json v3 fields copied from the report, and the glassd cap each needs
# (contracts/glassd.md; contracts/daemon.md §7)
SURF_V3 = {"plates": "plates", "coverDz": "coverDz", "masks": "masks", "scaleFrom": "scaleFrom"}
SURF_V2 = ("quad", "phase", "appear", "phaseMs")
SLAB_V3 = {"tint": "tint", "hole": "holes", "ox": "offset", "oy": "offset"}
SLAB_V2 = ("phase", "appear", "phaseMs")
PLATE_KEYS = ("id", "x", "y", "w", "h", "r", "material", "phase", "appear", "phaseMs", "appearAt", "tint", "fill",
"occluder", "shadow", "exitAt", "exitMs")
TOP_V3 = {"unitM": "unitM", "masks": "masks", "roomDim": "roomDim"}
COVER_DZ, BASE_DZ = 0.001, 0.002 # metres toward the viewer (NATIVE.md)
SG_MAX_PUSH_HZ = 15 # lgs_sg.js scheduler pushes per second, at most
SPEC_MIN_INTERVAL = 1 / SG_MAX_PUSH_HZ
GLASSD_JSON_MIN_INTERVAL = 0.25 # position-only changes to glassd.json
WATCHDOG_MS = 12000 # lgs_sg.js clears its nodes without a heartbeat this long
NATIVE_TTL_MS = 3000 # lgs-native clears itself without a heartbeat this long (< WATCHDOG_MS)
NATIVE_BEAT_S = 1.0 # lgs-native / ack heartbeat
SHOWN_MS = 350 # a pushed scene-graph change is on screen after this (fact 4: ~0.3 s)
SG_BEAT_FRESH_S = 2.5 # lgs-native needs a systemui heartbeat this recent
NATIVE_GRACE_S = 1.5 # a window keeps lgs-native this long after glassd last drew glass for it
MORPH_TOKENS = ("interactive", "hover-in", "fade", "snappy", "morph-open", "morph-close", "sheet-in", "sheet-out",
"page", "depth") # native/shared/motion_tokens.h: the springs a cover morph may ride
STILL_S = 0.15 # a layer pops once its rect has been still this long
STILL_PX = 2.0 # ... within this many texture px
OVERLAP_CLIP_PX = 2.5 # overlaps this thin are trimmed off, wider ones skip the layer
SLAB_SIZE_TOL_PX = 2.5 # glassd slab vs element x backdropScale
THEME_OFF_POLLS = 3 # Steam theme seen off this many polls (1 s) -> dormant (or exit)
THEME_OFF_GRACE_S = 600 # ... dormant this long before exiting (flag shellThemeGraceS; 0 = exit)
STAY_MAX_S = 2400 # --stay: dormant at most this long, and native at most this long after start
POD_SCRIPTS = ("asspod",) # --pod: the only SteamVR page scripts it installs
POD_IDLE_S = 30.0 # --pod: exit once the assPod is away and nothing plays this long
POD_FIRST_OPEN_S = 90.0 # --pod: ... or when it was never taken out this long after start
POD_RT_TEARDOWN_JS = ("(async () => { const r = window.__LGS_RT; if (!r || !r.podOnly || typeof r.teardown !== 'function')"
" return 'absent'; await r.teardown('pod daemon exit'); return 'removed'; })()")
# (glass.py's native-session client gives up after 2400 s; a unit older than
# that has lost its client, e.g. a usage-limit cut-off)
STEAM_GONE_S = 120 # Steam's devtools unreachable this long -> exit (also with --stay)
STEAM_SLOW_MAX = 3 # evaluations timing out in a row before the Steam socket is dropped
GLASSD_MAX_CRASHES = 5 # within GLASSD_CRASH_WINDOW s -> give up (CSS only)
GLASSD_CRASH_WINDOW = 120
GLASSD_TEMP_EXIT = 75 # EX_TEMPFAIL from glassd: SteamVR gone, lock held; not a crash
GLASSD_STALE_S = 6.0 # glassd rewrites glassd-out.json every 2 s ("updated")
GLASSD_FIRST_OUT_S = 20.0 # time for glassd's first glassd-out.json
STALL_S = 1.5 # the daemon's event loop did not run this long: it was stopped (SIGSTOP) or
# starved; its own silence is not glassd's or Steam's (REQ P7->P8, SG-6)
MAIN_QUAD_W = 0.98 # main window's nominal world width (m), NATIVE.md fact 4
MAIN_QUAD_TOL = 0.12 # glassd's quadW off by more than this: main stays CSS only
BUG_LIMIT, BUG_WINDOW = 5, 120 # unexpected exceptions in one loop before it is fatal
SPEC_LOG_MIN_S = 1.0 # "systemui: spec" log lines at most this often
START_LOCK = "/tmp/lgs/shell-start.lock" # serializes start() (two `lgs on` at once; RAM)
TEST_EXT_DIR = "/tmp/lgs/shell-ext-test" # lab: RAM plugins (kinds echo / read only), selftest dm4
def _net_errors():
errs = (ConnectionError, OSError, RuntimeError, asyncio.TimeoutError, TimeoutError, ValueError, EOFError)
try:
import aiohttp
errs += (aiohttp.ClientError,) # WSServerHandshakeError, ServerDisconnectedError, ...
except ImportError:
pass
return errs
NET_ERRORS = _net_errors() # a lost or refused devtools connection: retry
def log(msg):
print(time.strftime("%H:%M:%S ") + msg, flush=True)
lgs.log("shell: " + msg)
def write_json_atomic(path, obj):
os.makedirs(os.path.dirname(path), exist_ok=True)
tmp = f"{path}.{os.getpid()}.tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(obj, f, separators=(",", ":"))
os.replace(tmp, path)
def read_json(path):
try:
with open(path, encoding="utf-8") as f:
return json.load(f)
except (OSError, ValueError):
return None
def remove_glassd_files():
for p in (GLASSD_JSON, GLASSD_OUT):
try:
os.remove(p)
except OSError:
pass
def read_dial():
return lgs.read_dial()
def _mtime(path):
try:
return os.stat(path).st_mtime_ns
except OSError:
return None
def http_json(url, timeout=3):
with urllib.request.urlopen(url, timeout=timeout) as r:
return json.load(r)
# ------------------------------------------------------------------ flags and the native decision
def flat_flags(d):
"""A flags file is flat ({"native": "on"}) or nested ({"flags": {...}})."""
if not isinstance(d, dict):
return {}
if isinstance(d.get("flags"), dict):
out = {k: v for k, v in d.items() if k != "flags"}
out.update(d["flags"])
return out
return dict(d)
class FlagReader:
"""defaults.json < /tmp/lgs/flags.json, re-read when either file changes."""
def __init__(self, paths=(DEFAULTS_FILE, FLAGS_FILE)):
self.paths = paths
self.cache = {}
self.merged = {}
self.sig = None
def read(self):
sig = []
for p in self.paths:
try:
sig.append(os.stat(p).st_mtime_ns)
except OSError:
sig.append(None)
sig = tuple(sig)
if sig != self.sig:
self.sig = sig
m = {}
for p, s in zip(self.paths, sig):
if s is not None:
m.update(flat_flags(read_json(p)))
self.merged = m
return self.merged
def native_word(v):
if v is True or v in ("on", "true", "1", 1):
return "on"
if v is False or v in ("off", "false", "0", 0):
return "off"
return "auto"
def resolve_native(requested):
"""(native: bool, source, reason) for start(native=requested). 'auto' reads
/tmp/lgs/flags.json, then device/defaults.json; 'auto' in a file means
"not decided" and falls through; the built-in end of the chain is CSS only."""
req = native_word(requested) if requested is not None else "auto"
if req != "auto":
return req == "on", "cli", None if req == "on" else "CSS only requested (lgs on --css)"
for src, path in (("flags", FLAGS_FILE), ("defaults", DEFAULTS_FILE)):
v = native_word(flat_flags(read_json(path)).get("native"))
if v != "auto":
return v == "on", src, None if v == "on" else f"native=auto: native off in {path}"
return False, "builtin", "native=auto: the native gate has not passed (device/defaults.json has no native: on)"
def css_reason_for(req, src):
if req == "off":
return "CSS only requested (lgs on --css)"
if req == "auto":
if src == "builtin":
return "native=auto: the native gate has not passed (device/defaults.json has no native: on)"
return f"native=auto: native off in {FLAGS_FILE if src == 'flags' else DEFAULTS_FILE}"
return "native layer not requested; lgs on --native opts in"
def json_files(folder):
"""*.json in name order, skipping _wip/ and dot files."""
try:
names = sorted(n for n in os.listdir(folder) if n.endswith(".json") and not n.startswith((".", "_")))
except OSError:
return []
return [os.path.join(folder, n) for n in names]
def layers_config():
"""The reporter's configuration: theme/layers/*.json as [{file, ...}] in name
order (contracts/reporter.md §3.1), else Phase 1's theme/layers.json object.
Returns (config, errors, version)."""
errors, frags = [], []
for p in json_files(LAYERS_DIR):
d = read_json(p)
if isinstance(d, dict):
frags.append(dict(d, file=os.path.basename(p)))
else:
errors.append(f"{os.path.basename(p)}: not a JSON object")
if frags:
cfg = frags
else:
cfg = read_json(LAYERS_CFG) if os.path.exists(LAYERS_CFG) else None
if os.path.exists(LAYERS_CFG) and cfg is None:
errors.append("theme/layers.json: not valid JSON")
ver = hashlib.sha1(json.dumps(cfg, sort_keys=True).encode()).hexdigest()[:10] if cfg is not None else None
return cfg, errors, ver
def sg_rules(flags):
"""Active transform overrides from theme/sg/*.json (contracts/sg.md §4.1):
merged in name order, `supersedes` drops earlier ids, a rule with a `flag`
is kept only while that flag is true. Returns (rules, errors)."""
rules, errors, order = {}, [], []
for p in json_files(SG_RULES_DIR):
d = read_json(p)
if not isinstance(d, dict) or not isinstance(d.get("rules", []), list):
errors.append(f"{os.path.basename(p)}: not a fragment")
continue
for rid in d.get("supersedes") or []:
if rid in rules:
del rules[rid]
order.remove(rid)
for r in d.get("rules") or []:
if not isinstance(r, dict) or not isinstance(r.get("id"), str):
errors.append(f"{os.path.basename(p)}: rule without id")
continue
if r["id"] in rules:
errors.append(f"{os.path.basename(p)}: duplicate rule {r['id']}")
continue
rules[r["id"]] = dict(r, fragment=os.path.basename(p))
order.append(r["id"])
active = [rules[i] for i in order if not rules[i].get("flag") or flags.get(rules[i]["flag"]) is True]
return active, errors
def vr_scripts():
"""SteamVR page scripts: {name: {page, path, version, flag, src}} from
device/vr/<page>.<name>.js and the test folder (the test folder wins)."""
out = {}
for folder in VR_SCRIPT_DIRS:
try:
names = sorted(os.listdir(folder))
except OSError:
continue
for n in names:
if not n.endswith(".js") or n.startswith((".", "_")) or n.count(".") < 2:
continue
page, name = n[:-3].split(".", 1)
p = os.path.join(folder, n)
try:
with open(p, encoding="utf-8") as f:
src = f.read()
except OSError:
continue
flag = None
for line in src.splitlines()[:5]:
rest = line.split("@lgs-flag", 1)[1].split() if "@lgs-flag" in line else []
if rest:
flag = rest[0]
out[name] = {"page": page, "path": p, "flag": flag, "src": src.strip().rstrip(";"),
"version": hashlib.sha1(src.encode()).hexdigest()[:10]}
return out
def _pdeathsig():
"""preexec for glassd: SIGTERM it when the daemon dies (a hand-run daemon
killed outside systemd must not leave the camera-reading child behind)."""
try:
import ctypes
ctypes.CDLL("libc.so.6", use_errno=True).prctl(1, int(signal.SIGTERM)) # PR_SET_PDEATHSIG
except Exception: # noqa: BLE001 - best effort
pass
# ------------------------------------------------------------------ CDP
class CDP:
"""A persistent devtools websocket: concurrent calls plus event handlers."""
live = weakref.WeakSet() # open sockets, closed at the daemon's exit (no aiohttp warnings)
def __init__(self, name, ws_url):
self.name = name
self.ws_url = ws_url
self.n = 0
self.pending = {}
self.handlers = {}
self.closed = False
self.ws = None
self.http = None
self.reader = None
async def open(self):
import aiohttp
self.http = aiohttp.ClientSession()
CDP.live.add(self)
try:
self.ws = await self.http.ws_connect(self.ws_url, max_msg_size=0)
except BaseException:
await self.http.close()
raise
self.reader = asyncio.create_task(self._read())
return self
async def _read(self):
import aiohttp
try:
async for msg in self.ws:
if msg.type != aiohttp.WSMsgType.TEXT:
if msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
break
continue
d = json.loads(msg.data)
if "id" in d:
fut = self.pending.pop(d["id"], None)
if fut and not fut.done():
fut.set_result(d)
else:
cb = self.handlers.get(d.get("method"))
if cb:
try:
cb(d.get("params") or {})
except Exception as e: # noqa: BLE001 - a handler bug must not kill the socket
log(f"{self.name}: handler {d.get('method')}: {e!r}")
except Exception: # noqa: BLE001 - socket torn down
pass
finally:
self.closed = True
for fut in self.pending.values():
if not fut.done():
fut.set_exception(ConnectionError(f"{self.name}: devtools socket closed"))
self.pending.clear()
async def send(self, method, params=None, timeout=20):
if self.closed:
raise ConnectionError(f"{self.name}: devtools socket closed")
self.n += 1
i = self.n
fut = asyncio.get_running_loop().create_future()
self.pending[i] = fut
try:
await self.ws.send_json({"id": i, "method": method, "params": params or {}})
d = await asyncio.wait_for(fut, timeout)
finally:
self.pending.pop(i, None)
if "error" in d:
raise RuntimeError(f"{method}: {d['error']}")
return d["result"]
async def eval(self, expr, timeout=20, by_value=True):
r = await self.send("Runtime.evaluate", {
"expression": expr, "returnByValue": by_value, "awaitPromise": True}, timeout)
if "exceptionDetails" in r:
ex = r["exceptionDetails"]
raise RuntimeError("JS: " + str(ex.get("exception", {}).get("description") or ex.get("text")))
return r["result"].get("value") if by_value else r["result"]
async def close(self):
self.closed = True
CDP.live.discard(self)
try:
if self.ws is not None:
await self.ws.close()
except Exception: # noqa: BLE001
pass
try:
if self.http is not None:
await self.http.close()
except Exception: # noqa: BLE001
pass
if self.reader is not None:
self.reader.cancel()
async def eval_once(ws_url, expr, timeout=20, name="page"):
"""One evaluation on a fresh socket that is always closed again."""
c = CDP(name, ws_url)
try:
await c.open()
return await c.eval(expr, timeout)
finally:
await c.close()
# ------------------------------------------------------------------ Steam-side helpers
# Sets html.lgs-native on the Steam windows whose glassd cover the compositor
# shows, and acknowledges the popped layers it shows to the reporter
# (__LGS_LAYERS.ack: only those get data-lgs-pop). Only while the theme
# (__LGS) is on. Without a heartbeat (set()) for ttl ms it removes the class
# again, so a dead daemon never leaves windows unpainted.
# `all`: native mode is live (glassd and the scene graph healthy): every Steam
# window is native, pooled popups included, before they first show, so their
# glass is glassd's from the first frame (lgs_core.js also sets the class on a
# window it injects into while h.live()).
NATIVE_JS = r"""
(function (keys, ttl, acks, plates, covers, all) {
const W = window;
let h = W.__LGS_NATIVE;
if (!h) {
const ALIAS = { 'valve.steam.gamepadui.main': /^VR_uid/, 'valve.steam.gamepadui.keyboard': /^VRKeyboard_uid/,
'valve.steam.gamepadui.notifications': /^VRNotificationToasts_uid/ };
const matches = (name, key) => (ALIAS[key] ? ALIAS[key].test(name) : (name.startsWith(key + '.') || name.startsWith(key + '_')));
h = W.__LGS_NATIVE = { keys: [], acks: {}, plates: {}, covers: [], all: false, last: 0, ttl, timer: 0, applied: [], tagged: null };
h.live = () => (h.all || h.keys.length > 0) && Date.now() - h.last < h.ttl
&& !!(W.__LGS && W.__LGS.state && W.__LGS.state.enabled);
h.sweep = () => {
const live = h.live();
const applied = [];
let pops = [];
try { pops = [...g_PopupManager.m_mapPopups.values()]; } catch (_) { /* not ready */ }
for (const p of pops) {
let doc = null;
try { doc = p.window && p.window.document; } catch (_) { continue; }
if (!doc || !doc.documentElement) continue;
const on = live && (h.all || h.keys.some((k) => matches(p.m_strName, k)));
const cl = doc.documentElement.classList;
if (on !== cl.contains('lgs-native')) cl.toggle('lgs-native', on);
if (on) applied.push(p.m_strName.replace(/_uid\d+$/, ''));
}
h.applied = applied;
if (!live && h.timer) { clearInterval(h.timer); h.timer = 0; }
return applied;
};
// v3 reporters (status().v >= 3) take {key: {cover, pops, plates}}; older
// ones {key: [pop ids]} (contracts/reporter.md §7)
h.ack = () => {
try {
const L = W.__LGS_LAYERS;
if (!L || typeof L.ack !== 'function') return;
let v3 = false;
try { v3 = typeof L.status === 'function' && (L.status().v | 0) >= 3; } catch (_) { /* old */ }
const map = {};
if (h.keys.length) {
for (const k of h.keys) {
map[k] = v3 ? { cover: h.covers.includes(k), pops: h.acks[k] || [], plates: h.plates[k] || [] }
: (h.acks[k] || []);
}
}
h.tagged = L.ack(map);
} catch (_) { /* reporter restarting */ }
};
h.set = (ks, ac, pl, cv, al) => {
h.keys = ks || []; h.acks = ac || {}; h.plates = pl || {}; h.covers = cv || h.keys; h.all = !!al; h.last = Date.now();
if (!h.timer && (h.all || h.keys.length)) h.timer = setInterval(h.sweep, 1000);
const r = h.sweep(); h.ack(); return r;
};
h.clear = () => { h.keys = []; h.acks = {}; h.plates = {}; h.covers = []; h.all = false; const r = h.sweep(); h.ack(); if (h.timer) clearInterval(h.timer); delete W.__LGS_NATIVE; return r; };
}
h.ttl = ttl;
const applied = h.set(keys, acks, plates, covers, all);
return JSON.stringify({ applied, tagged: h.tagged });
})
"""
# __LGS_RT.bridge (P1's runtime): set() each [name, value]; 'none' without a runtime
BRIDGE_SET_JS = r"""
(function (pairs) {
const R = window.__LGS_RT, b = R && R.bridge;
if (!b || typeof b.set !== 'function') return 'none';
let n = 0;
for (const [k, v] of pairs) { try { b.set(k, v); n++; } catch (_) { /* a listener threw */ } }
return 'ok:' + n;
})
"""
# Names the runtime does not hold (it was reinstalled): null without a runtime
BRIDGE_MISSING_JS = r"""
(function (names) {
const R = window.__LGS_RT, b = R && R.bridge;
if (!b || typeof b.set !== 'function') return null;
if (typeof b.get !== 'function') return names;
return names.filter((n) => { try { return b.get(n) === undefined; } catch (_) { return true; } });
})
"""
# Our CDP binding globals out of SharedJSContext: Runtime.removeBinding does not
# delete the function a closed session installed (G-REMOVE, review R1 M1). Only
# native-code functions (bindings); a test's own JS stand-in is left alone.
# Returns the names deleted, joined ('' when none).
DELETE_BINDINGS_JS = r"""
(function (names) {
const out = [];
for (const n of names) {
const f = window[n];
if (typeof f === 'function' && /\[native code\]/.test(Function.prototype.toString.call(f))) {
try { delete window[n]; out.push(n); } catch (_) { /* non-configurable */ }
}
}
return out.join(',');
})
"""
# Steam tick: theme on, Reduce Motion
STEAM_TICK_JS = ("JSON.stringify([!!(window.__LGS && window.__LGS.state && window.__LGS.state.enabled), "
"!!(window.matchMedia && matchMedia('(prefers-reduced-motion: reduce)').matches)])")
# systemui, read-only: geometry (the SP §1.2 snippet), the Steam frame's shown
# page (CC15) and, with dump, the frame menu's laser panel (WN §3.3.2)
GEOM_JS = r"""
(async function (wantDump) {
const out = { geom: null, page: null, err: null };
try {
const FS = window.FrameStore, DS = window.DashboardStore;
if (!FS || !DS || !FS.frames || !FS.frames.length) { out.err = 'no FrameStore'; return JSON.stringify(out); }
const MAIN = 'valve.steam.gamepadui.main';
// Steam's page by its summon key (page ids are handed out at run time: page 3
// once, page 4 now), not /page:3/ or the frame's active page, which is SteamVR
// Settings or the binding UI while those show (REQ P7->P8, sg.md §4.2)
let f = null, sp = null;
for (const x of FS.frames) {
const ps = x.pages ? [...(typeof x.pages.values === 'function' ? x.pages.values() : x.pages)] : [];
const p = ps.find((q) => q && q.m_sSummonOverlayKey === MAIN);
if (p) { f = x; sp = p; break; }
}
if (!f) { f = FS.frames[0]; sp = f.activePage; }
const z = sp.size, S = DS.dashboardScale, H0 = f.size.mainPanelHeightOverride || 1.5;
const r = z.latestMeasuredPanelLocalHeight / H0, Hw = z.latestMeasuredPanelWorldHeight;
const g = { S, H0, r, Hm: Hw, unitM: S * r, mmPerCssPx: 1500 * Hw / 1080, dashDist: DS.dashboardDistance, frameMenu: null };
const sf = FS.frames.find((x) => x.m_mapPages && [...x.m_mapPages.values()].some((p) => p.m_sSummonOverlayKey === MAIN)) || f;
const pages = {};
if (sf.m_mapPages) for (const [id, p] of sf.m_mapPages) pages[id] = p.m_sSummonOverlayKey;
const ap = sf.activePage;
out.page = { frameID: sf.m_unFrameID, activePageID: sf.m_unActivePageID, summonKey: ap ? ap.m_sSummonOverlayKey : null,
steam: !!ap && ap.m_sSummonOverlayKey === MAIN, pages };
const el = document.querySelector('[id^="PooledPopup-valve.steam.gamepadui.frame.menu"]');
if (el && typeof el.buildNode === 'function') {
const p = el.buildNode({}, el)[1].properties;
g.frameMenu = { key: p.key, mpp: p['meters-per-pixel'], uv: [p.uv_min[0], p.uv_min[1], p.uv_max[0], p.uv_max[1]],
visible: p.visibility === 0 || p.visibility === undefined };
// what can change the laser panel's size: a change re-dumps (not published)
out.fmSig = JSON.stringify(['key', 'uv_min', 'uv_max', 'meters-per-pixel', 'visibility', 'scale-index',
'frame-resize-scale-factor', 'origin', 'curvature'].map((k) => p[k]));
if (wantDump && window.OverlayStore && OverlayStore.DumpLaserOverlays) {
const d = await OverlayStore.DumpLaserOverlays();
const ov = d && d.overlays || {};
const k = Object.keys(ov).find((n) => n.startsWith(p.key + ' ') || n.startsWith(p.key + '_'));
const sp = k && ov[k].scene_graph_panel;
if (sp) { g.frameMenu.fWidth = sp.fWidth; g.frameMenu.fHeight = sp.fHeight; g.frameMenu.dumped = true; }
}
}
out.geom = g;
} catch (e) { out.err = String(e && e.message || e).slice(0, 200); }
return JSON.stringify(out);
})
"""
# systemui: our scene graph out (overrides restored), and P5's springs that
# the daemon prepended (P5's REQ: no global left behind)
SG_DESTROY_JS = r"""
(() => {
let r = 'absent';
if (window.__LGS_SG) { try { window.__LGS_SG.destroy(); r = 'cleared'; } catch (e) { r = 'error: ' + e; } }
const M = globalThis.__LGS_MOTION;
if (M) { try { if (typeof M.remove === 'function') M.remove(); } catch (_) { /* best effort */ } delete globalThis.__LGS_MOTION; }
return r;
})()
"""
# SteamVR page scripts (contracts/daemon.md §6). VRX_INSTALL_JS(name, version,
# ctx) wraps the script's function expression.
VRX_PROBE_JS = r"""
(function (want) {
const R = window.__LGS_VRX || {};
const out = {};
for (const [n, v] of Object.entries(want)) out[n] = R[n] ? (R[n].version === v ? 'ok' : 'stale') : 'need';
for (const n of Object.keys(R)) if (!(n in want)) out[n] = 'extra';
return JSON.stringify(out);
})
"""
VRX_REMOVE_JS = r"""
(function (names) {
const R = window.__LGS_VRX;
if (!R) return 0;
let n = 0;
for (const k of Object.keys(R)) {
if (names && !names.includes(k)) continue;
try { if (R[k].api && typeof R[k].api.remove === 'function') R[k].api.remove(); } catch (_) { /* its bug */ }
delete R[k]; n++;
}
if (!Object.keys(R).length) delete window.__LGS_VRX;
return n;
})
"""
def vrx_install_js(name, version, ctx, src):
return (f"(() => {{ const R = window.__LGS_VRX || (window.__LGS_VRX = {{}}); const N = {json.dumps(name)};\n"
f"if (R[N] && R[N].version === {json.dumps(version)}) return 'ok';\n"
"if (R[N]) { try { R[N].api && R[N].api.remove && R[N].api.remove(); } catch (_) {} delete R[N]; }\n"
f"let api;\ntry {{ api = ({src}\n)({json.dumps(ctx)}); }}\n"
"catch (e) { if (!Object.keys(R).length) delete window.__LGS_VRX; return 'error: ' + String(e && e.message || e).slice(0, 160); }\n"
f"R[N] = {{ version: {json.dumps(version)}, api: api || {{}}, at: Date.now() }};\nreturn 'installed'; }})()")
MAIN_SIZE_JS = r"""
(() => { try {
const w = SteamUIStore.WindowStore.VRGamepadUIMainWindowInstance.BrowserWindow;
return [Math.round(w.innerWidth * w.devicePixelRatio), Math.round(w.innerHeight * w.devicePixelRatio)];
} catch (e) { return null; } })()
"""
REPORTER_STOP_JS = f"""
(() => {{ const r = window.{REPORTER}; if (!r) return 'absent';
for (const m of ['stop', 'disable', 'destroy']) if (typeof r[m] === 'function') {{ r[m](); return m; }}
return 'no stop method'; }})()
"""
# The reporter's heartbeat (it stops itself when pings stop), sent only to our
# own instance. 'ok' | 'foreign' (running, but another binding: someone's
# test) | 'waiting' (for the theme's token index) | 'stopped' | 'absent' | 'error'
REPORTER_PING_JS = f"""
(() => {{ const r = window.{REPORTER}; if (!r || typeof r.status !== 'function') return 'absent';
try {{ const s = r.status();
if (!s.running) return s.waiting ? 'waiting' : 'stopped';
if (s.binding !== {json.dumps(BINDING)}) return 'foreign';
if (typeof r.ping === 'function') r.ping();
return 'ok';
}} catch (e) {{ return 'error'; }} }})()
"""
REPORTER_RESEND_JS = f"""
(() => {{ const r = window.{REPORTER}; if (!r) return 'absent';
for (const m of ['resend', 'flush', 'report', 'refresh']) if (typeof r[m] === 'function') {{ try {{ r[m](true); }} catch (_) {{}} return m; }}
return 'none'; }})()
"""
def file_version(path):
try:
with open(path, "rb") as f:
return hashlib.sha1(f.read()).hexdigest()[:10]
except OSError:
return None
def material_for(surface):
if surface.get("material"):
return surface["material"]
return "window" if surface.get("name") == "main" else "panel"
# ------------------------------------------------------------------ report -> glassd.json v3 checks
def _num(v, lo=-1e7, hi=1e7):
"""A finite number within [lo, hi], else None (bools are not numbers)."""
if isinstance(v, bool) or not isinstance(v, (int, float, str)):
return None
try:
f = float(v)
except ValueError:
return None
return f if lo <= f <= hi and f == f else None
def _nums(v, n, lo=-1e7, hi=1e7):
"""A list of exactly n finite numbers (floats), else None."""
if not isinstance(v, (list, tuple)) or len(v) != n:
return None
out = [_num(x, lo, hi) for x in v]
return None if None in out else out
def _round(f):
return round(f, 4) if f != int(f) else int(f)
def clean_color(c):
"""A CSS colour string or [r, g, b(, a)] in 0..1 (contracts/glassd.md units)."""
if isinstance(c, str) and 0 < len(c) <= 64:
return c
if isinstance(c, list) and len(c) in (3, 4):
vals = [_num(v, 0, 1) for v in c]
if all(v is not None for v in vals):
return [_round(v) for v in vals]
return None
def clean_rect(d, keys=("x", "y", "w", "h", "r"), optional=("r",)):
out = {}
for k in keys:
if k not in d and k in optional:
continue
v = _num(d.get(k))
if v is None:
return None
out[k] = _round(round(v, 5 if k == "dz" else 2))
return out
def clean_phase_fields(src, dst):
if "phase" in src and _num(src["phase"], 0, 1) is not None:
dst["phase"] = _round(_num(src["phase"], 0, 1))
if src.get("appear") in ("materialize", "dematerialize"):
dst["appear"] = src["appear"]
if "phaseMs" in src and _num(src["phaseMs"], 0, 10000) is not None:
dst["phaseMs"] = int(_num(src["phaseMs"], 0, 10000))
# the epoch ms the materialize started at (the page's own open animation): glassd ramps from then
if "appearAt" in src and _num(src["appearAt"], 0, 1e15) is not None:
dst["appearAt"] = int(_num(src["appearAt"], 0, 1e15))
def clean_plate(p, i):
if not isinstance(p, dict):
return None
out = clean_rect(p)
if out is None or out["w"] < 2 or out["h"] < 2:
return None
pid = p.get("id")
out["id"] = str(pid)[:64] if pid is not None else f"p{i}"
if isinstance(p.get("material"), str) and len(p["material"]) <= 16:
out["material"] = p["material"]
clean_phase_fields(p, out)
for k in ("tint", "fill"):
if k in p and clean_color(p[k]) is not None:
out[k] = clean_color(p[k])
if isinstance(p.get("occluder"), bool):
out["occluder"] = p["occluder"]
if "shadow" in p and _num(p["shadow"], 0, 1) is not None:
out["shadow"] = _round(_num(p["shadow"], 0, 1))
# its content fading out (a route leaving): from that epoch ms over that long (contracts/glassd.md §1.3)
if _num(p.get("exitAt"), 1, 1e15) is not None:
ms = _num(p.get("exitMs"), 0, 2000)
out["exitAt"] = int(_num(p["exitAt"], 1, 1e15))
out["exitMs"] = int(ms) if ms is not None else PAGE_OUT_MS
return out
def clean_edges(e):
"""hole.edges (contracts/glassd.md §1.4, cap holeEdges): {top, right, bottom,
left}, each a colour or a list of 1-8 colours; unparsable colours and empty
edges are dropped. None when nothing is left."""
if not isinstance(e, dict):
return None
out = {}
for k in ("top", "right", "bottom", "left"):
v = e.get(k)
if isinstance(v, list) and v and not (len(v) in (3, 4) and all(_num(x) is not None for x in v)):
cs = [c for c in (clean_color(x) for x in v[:8]) if c is not None]
if cs:
out[k] = cs
elif v is not None and clean_color(v) is not None:
out[k] = clean_color(v)
return out or None
def clean_hole(h, clip=None, edges=False):
if h is True:
return {"clip": [_round(round(v, 2)) for v in clip]} if clip else True
if not isinstance(h, dict):
return None
out = {}
for k, lo, hi in (("shadow", 0, 1), ("y", -500, 500), ("blur", 0, 500)):
if k in h and _num(h[k], lo, hi) is not None:
out[k] = _round(_num(h[k], lo, hi))
if "fill" in h and clean_color(h["fill"]) is not None:
out["fill"] = clean_color(h["fill"])
if edges and clean_edges(h.get("edges")) is not None: # REQ P9->P8, only with glassd's holeEdges cap
out["edges"] = clean_edges(h["edges"])
c = h.get("clip")
if isinstance(c, list) and len(c) == 4 and all(_num(v) is not None for v in c):
out["clip"] = [_round(round(_num(v), 2)) for v in c]
elif clip:
out["clip"] = [_round(round(v, 2)) for v in clip]
return out or True
def clean_masks(ms, world=False):
"""World quads {O, U, V} (top level) or surface rects {x, y, w, h, dz?}, ≤ 8."""
out = []
for m in (ms if isinstance(ms, list) else [])[:8]:
if not isinstance(m, dict):
continue
if world:
q = {}
for k in ("O", "U", "V"):
v = m.get(k)
if isinstance(v, list) and len(v) == 3 and all(_num(x) is not None for x in v):
q[k] = [_round(round(_num(x), 5)) for x in v]
if len(q) == 3:
out.append(q)
else:
r = clean_rect(m, ("x", "y", "w", "h", "dz"), optional=("dz",))
if r is not None:
out.append(r)
return out
def clean_quad(q):
m = clean_masks([q], world=True)
return m[0] if m else None
# ------------------------------------------------------------------ the daemon
class Shell:
def __init__(self, glassd_path, stay=False, glassd_args=None, native=False, native_req=None, native_src=None,
test_report=None, assume_caps=None, pod=False):
self.pod = bool(pod) # pod mode: the assPod alone, no theme (lgs pod)
if self.pod:
native = False
self.glassd_path = glassd_path
self.glassd_args = list(glassd_args or [])
self.stay = stay # lab aid: go dormant instead of exiting on theme off
self.native_requested = bool(native)
self.native_req = native_req or ("on" if native else "off") # what start() was asked: auto | on | off
self.native_src = native_src or "cli" # where auto was decided
self.test_report = test_report # lab: a report file replaces the reporter
self.assume_caps = set(assume_caps or []) # lab: act as if glassd announced these caps
self.test_report_mtime = None
# Read once: a binary built mid-session is never picked up.
if not self.native_requested:
self.native_enabled, self.css_reason = False, css_reason_for(self.native_req, self.native_src)
elif not self.glassd_path:
self.native_enabled, self.css_reason = False, "glassd disabled"
elif not os.access(self.glassd_path, os.X_OK) and not ensure_glassd(self.glassd_path):
self.native_enabled, self.css_reason = False, "no glassd binary at start; python glass.py native-build"
else:
self.native_enabled, self.css_reason = True, None
self.stopping = False
self.stop_task = None
self.tasks = []
self.exit_code = 0
self.torn = False
self.since = time.time()
self.errors = collections.deque(maxlen=12)
self.bugs = {} # loop -> [times of unexpected exceptions]
self.changed = asyncio.Event() # spec inputs changed
self.wake_at = None # a layer becomes still enough to pop
# Steam side
self.steam = None
self.steam_ok = False
self.theme_on = None
self.theme_off_polls = 0
self.dormant_since = None # theme off, waiting up to the grace for it to come back
self.steam_target_id = None # SharedJSContext's target id: a new one = Steam restarted
self.steam_restarted = False # ... and the theme has not been seen on since
self.steam_lost_at = None # Steam's devtools unreachable since
self.slow_evals = 0 # Steam evaluations that timed out (connection kept)
self.loop_beat = time.monotonic() # stall_loop's last tick
self.stall_at = -1e9 # monotonic time the last stall of our own loop ended
self.stalls, self.last_stall_s = 0, None
self.report = None
self.report_at = 0.0
self.reports = 0
self.dash_hidden = False
self.layer_still = {} # (surface, id) -> (rect, still since)
self.reporter_state = "not injected"
self.reporter_injected_at = 0.0
self.reporter_version = None
self.fallback = None # synthesized report when no reporter
self.native_applied = []
self.native_tagged = None
self.acked = {}
self.native_force = False
# systemui side
self.sysui = None
self.sysui_ok = False
self.sg_version = file_version(SG_JS)
self.sg_status = None
self.sg_injected_at = 0.0
self.sg_nosched = 0
self.sg_reset = False
self.sysui_ctx = None
self.sg_beat = None
self.sg_beat_at = 0.0
self.spec_sent = None
self.spec_sent_obj = None
self.spec_sent_at = 0.0
self.spec_seq = 0
self.spec_summary = None
self.spec_logged_at = 0.0
self.spec_log_sig = None
self.spec_log_skipped = 0
self.flags_mtime = None
self.flags = None
self.resized_noted = None
# glassd
self.gproc = None
self.g_pid = None
self.g_started = 0.0
self.g_crashes = []
self.g_restarts = 0
self.g_temp_exits = 0
self.g_last_exit = None
self.g_given_up = False
self.g_kill_reason = None # 'dormant' while we stop glassd on purpose
self.g_frames = False
self.gout = None
self.gout_mtime = None
self.gout_seen = 0.0
self.g_stale = False
self.g_files_removed = False
self.gseq = 0
self.gjson_struct = None
self.gjson_full = None
self.gjson_written_at = 0.0
self.dial_mtime = None
# the wearer's glass tune (lgs.TUNE, kept across reboots): glassd.json's dial and tune, the bridge
# value 'tune'; saved TUNE_SAVE_S after the last change (a slider streams changes)
self.tune = lgs.read_tune()
self.tune_mtime = _mtime(lgs.TUNE)
self.dial_v = lgs.read_dial()
self.tune_save = None # the pending save (asyncio TimerHandle)
self.tune_mirrored = (self.dial_v, self.tune["hue"], self.tune["hueK"])
# SteamVR page theming and page scripts
self.vr_version = None
self.vr_pages = {}
self.vrx_status = {} # page -> {script name: state}
self.vrx_failed = {} # (page, name) -> version that threw
self.vrx_sig, self.vrx_scripts = None, {} # page script files: (folder, name, mtime) list, parsed
self.vr_pause_since = None # a lab --theme off pause seen since (vr_paused)
# flags (defaults.json < /tmp/lgs/flags.json)
self.flagreader = FlagReader()
self.rt_flags = {}
self.profile = "default"
self.reduce_motion = False
# bridge to Steam's runtime (__LGS_RT.bridge)
self.bridge_values = {"tune": dict(self.tune)} # name -> value
self.bridge_sent = {} # name -> (json without 'at', time)
self.bridge_runtime = None # True / False once checked
self.bridge_beat_at = 0.0
# actions
self.registry = shell_ext.Registry(self, test_dir=TEST_EXT_DIR) # + RAM test plugins (echo / read)
self.registry.add_builtin("sgwindow", {
"args": {"dim": {"type": "num", "min": 0, "max": 1, "optional": True},
"recede": {"type": "num", "min": 0, "max": 0.3, "optional": True},
"motion": {"type": "str", "max": 24, "re": r"^[a-z][a-z-]{0,23}$", "optional": True},
"ttlMs": {"type": "int", "min": 500, "max": 60000, "optional": True},
"reset": {"type": "bool", "optional": True}},
"sources": ["main", "bar", "barpopup", "ccpopup"], "rate": 0.25, "kind": "ui"},
self.action_sgwindow)
# the wearer's glass tune (the bar's paintbrush panel, rt 33-tune.js): every field optional, a partial
# update merged; a slider streams calls, so its own rate (25 a second) is its only limit
self.registry.add_builtin("glass.tune", {
"args": {"dial": {"type": "num", "min": 0, "max": 1, "optional": True},
"hueK": {"type": "num", "min": 0, "max": 1, "optional": True},
"refract": {"type": "num", "min": 0, "max": 2, "optional": True},
"frost": {"type": "num", "min": 0, "max": 2, "optional": True},
"light": {"type": "num", "min": 0, "max": 2, "optional": True},
"hue": {"type": "str", "max": 7, "re": r"^(#[0-9a-fA-F]{6})?$", "optional": True},
"reset": {"type": "bool", "optional": True}},
"sources": ["bar", "barpopup", "main"], "rate": 0.04, "kind": "ui", "global": False},
self.action_glass_tune)
# echo restricted to the main window: the wrong-source probe of DM-4
self.registry.add_builtin("echo.main", {"args": {"text": {"type": "str", "max": 200, "optional": True}},
"sources": ["main"], "rate": 0.2, "kind": "echo"},
shell_ext.BUILTIN["echo"][1])
self.trusted_ctx = None # SharedJSContext's default execution context id
self.ctx_probe = None # (nonce, future) while probing
self.ctx_probe_lock = asyncio.Lock()
self.replies = collections.deque(maxlen=10)
self.action_tasks = set()
# geometry and page (systemui, read-only)
self.geom = None
self.page = None
self.geom_err = None
self.geom_at = 0.0
self.geom_dump_at = 0.0 # last DumpLaserOverlays read (status geomDumpAt; DM-3 waits for one)
self.geom_ok = False
self.vr_conns = {} # SteamVR page target id -> CDP (kept open; m4)
self.vr_targets = {} # SteamVR page title -> devtools target (the page loop's last list)
self.vr_http = None # aiohttp session for 8090's /json/list (keep-alive)
# glassd.json v3 and the materialize policy
self.items_first = {} # (surface, kind, id) -> time first written
self.native_seen = {} # overlay key -> when glassd last drew glass for it (NATIVE_GRACE_S)
self.vr_panels = {} # SteamVR panels: name -> {w, h, material, glass: {x, y, w, h, r}} (lgs_sg.js beat)
self.vr_shown = set() # ... the ones on the page now
self.items_last = {} # (surface, kind, id) -> last item written (for its fade-out)
self.fading = {} # (surface, kind, id) -> (item with phase 0, until)
self.cover_last = {} # surface -> had a cover in the last glassd.json
self.sg_crops = {} # Steam overlay key -> [u0, u1] its scene-graph panel shows
self.plate_acks = {}
self.cover_keys = []
self.pop_clips = {} # (surface, id) -> clip rect trimmed by build_spec (hole.clip)
# transform overrides and the window state (lgs_sg.js v2)
self.sg_rules = []
self.sg_rule_errors = []
self.sg_rules_sig = None
self.sg_rules_sent = None
self.window_req = None # {dim, recede, motion, until} from the sgwindow action
self.layers_version = None
# ---------------------------------------------------------------- state
def mode(self):
if self.pod:
return "pod"
if self.dormant_since is not None: # theme off without lgs off: waiting for it (contract §1)
return "dormant"
if not self.native_enabled:
return f"css-only ({self.css_reason})"
if self.native_ok():
return "native"
return "starting"
def native_possible(self):
"""Native mode opted in, with a glassd binary at start, not given up and
not removed since. Otherwise the daemon only themes SteamVR pages (CSS
only) and injects neither the reporter nor lgs_sg.js."""
return self.native_enabled
def disable_native(self, reason):
if self.native_enabled:
self.native_enabled, self.css_reason = False, reason
self.note(f"native layer off for this session: {reason}")
self.changed.set()
def glassd_running(self):
return bool(self.gproc and self.gproc.returncode is None)
def glassd_ready(self):
"""glassd draws: frames out, not exiting, healthy, output not stale."""
g = self.gout
return bool(g and self.g_frames and not g.get("exiting") and g.get("healthy", True) is not False
and not self.g_stale)
def native_ok(self):
return bool(self.native_possible() and self.glassd_running() and self.glassd_ready()
and self.steam_ok and self.theme_on)
def native_why(self):
"""Why native_ok() is false (logs and status), or None."""
for ok, why in ((self.native_possible(), "native off"), (self.glassd_running(), "glassd not running"),
(self.glassd_ready(), "glassd not ready"), (self.steam_ok, "steam not connected"),
(self.theme_on, "theme off")):
if not ok:
return why
return None
def current_report(self):
if self.report is not None:
return self.report
return self.fallback
def page_away(self, overlay_key):
"""True for Steam's main window while the dashboard frame shows another
page (SteamVR Settings, the binding UI): Steam's window is not on screen
then, so its cover, plates, base and pops must not be either (REQ C6b->P8:
glassd's main cover stayed drawn over SteamVR's settings panel). Unknown
(systemui unreachable, steam None) is not away."""
p = self.page
return overlay_key == STEAM_MAIN_KEY and isinstance(p, dict) and p.get("steam") is False
def note(self, msg):
self.errors.append(time.strftime("%H:%M:%S ") + msg)
log(msg)
def bug(self, where, e):
"""An unexpected exception in a loop: log it, back off; fatal (re-raised)
after BUG_LIMIT within BUG_WINDOW s. Returns the delay before a retry."""
now = time.time()
times = [t for t in self.bugs.get(where, []) if now - t < BUG_WINDOW] + [now]
self.bugs[where] = times
tb = traceback.format_exc().strip().splitlines()
self.note(f"{where}: unexpected {e!r} ({len(times)}/{BUG_LIMIT} in {BUG_WINDOW} s) at {' | '.join(tb[-3:])[:300]}")
if len(times) >= BUG_LIMIT:
raise e
return min(30, 2 ** len(times))
def caps(self):
"""glassd's v3 features (glassd-out.json caps); empty for a v2 binary."""
c = (self.gout or {}).get("caps")
caps = {x for x in c if isinstance(x, str)} if isinstance(c, list) else set()
return caps | self.assume_caps # lab only (--assume-caps): test the writer against an older glassd
def refresh_flags(self):
"""Re-read the flags; returns True when the depth profile changed."""
self.rt_flags = self.flagreader.read()
prof = "wearer" if self.rt_flags.get("interactivePops") is True else "default"
changed = prof != self.profile
if changed:
log(f"flags: profile {self.profile} -> {prof}")
self.profile = prof
rules, errs = sg_rules(self.rt_flags)
sig = json.dumps(rules, sort_keys=True)
if sig != self.sg_rules_sig:
self.sg_rules_sig, self.sg_rules, self.sg_rule_errors = sig, rules, errs
log(f"systemui: transform overrides {[r.get('id') for r in rules] or 'none'}"
+ (f" (errors {errs})" if errs else ""))
self.changed.set()
if self.window_req and self.window_req.get("until", 0) < time.time():
log("systemui: window state request expired; restoring")
self.window_req = None
self.changed.set()
return changed
def window_state(self):
"""The window dim / recede to request (contracts/sg.md §5): the report's
`window` in native mode wins over the sgwindow action."""
rep = self.current_report() if self.native_possible() else None
w = (rep or {}).get("window") if isinstance(rep, dict) else None
if isinstance(w, dict):
return {k: w[k] for k in ("dim", "recede", "motion") if k in w}
if self.window_req:
return {k: self.window_req[k] for k in ("dim", "recede", "motion") if k in self.window_req}
return None
def theme_grace_s(self):
"""How long to stay dormant while the Steam theme is off (flag
shellThemeGraceS, 0..3600 s; 0 = exit after THEME_OFF_POLLS as in Phase 1).
A --stay unit (lab, native-session) waits STAY_MAX_S whatever the flag."""
if self.stay:
return STAY_MAX_S
v = _num(self.rt_flags.get("shellThemeGraceS"), 0, 3600)
return THEME_OFF_GRACE_S if v is None else v
def check_stay_limit(self):
"""A --stay unit is native for at most STAY_MAX_S: after that its
native-session client is gone for sure (it gives up after 2400 s), so
the native layer ends and the unit follows the normal rules (contract §1)."""
if self.stay and time.time() - self.since >= STAY_MAX_S:
self.stay = False
self.note(f"--stay limit: running {STAY_MAX_S} s; its client is gone. Native layer off, normal "
"theme-off rules from now on")
self.disable_native(f"--stay limit ({STAY_MAX_S} s) reached")
def sg_wanted(self):
"""lgs_sg.js belongs in systemui: native mode, or (CSS only) an active
transform override or a window state (P7's REQ, contracts/sg.md §1)."""
if self.dormant_since is not None: # theme off: Steam's own panels until it is back
return False
return self.native_possible() or bool(self.sg_rules) or self.window_state() is not None
def daemon_value(self):
return {"v": 1, "at": int(time.time() * 1000), "ttlMs": int(DAEMON_BEAT_S * 3000), "pid": os.getpid(),
"mode": self.mode(), "native": bool(self.native_ok()), "profile": self.profile,
"steamvr": bool(self.geom_ok), "actions": sorted(self.registry.types),
"dryRun": bool(self.rt_flags.get("actionsDryRun"))}
# ---------------------------------------------------------------- plugin host (shell_ext.Ctx)
async def ext_steam_eval(self, expr, timeout=10):
s = self.steam
if self.steam_ok and s is not None and not s.closed:
return await s.eval(expr, timeout)
t = await self.steam_target()
return await eval_once(t["webSocketDebuggerUrl"], expr, timeout, "steam-ext")
async def ext_vr_eval(self, page, expr, timeout=10):
"""Over the page's kept socket (the page loop's target list, else a fresh one): plugins that
poll a page several times a second (asspod) cost one message per call, not a new socket."""
t = self.vr_targets.get(page)
if t is None:
loop = asyncio.get_running_loop()
t = next((x for x in await loop.run_in_executor(None, lgs_vr.targets) if (x.get("title") or "") == page), None)
if t is None:
raise ConnectionError(f"no SteamVR page {page!r}")
try:
return await self.vr_page_eval(t, expr, timeout)
except RuntimeError as e:
if str(e).startswith("JS: "):
raise
self.vr_targets.pop(page, None)
raise
except Exception:
self.vr_targets.pop(page, None)
raise
def ext_vr_pages(self):
return sorted(self.vr_pages)
def ext_flags(self):
return self.rt_flags
def ext_geom(self):
return self.geom
def ext_page(self):
return self.page
def ext_log(self, msg):
log(msg)
async def ext_actions_logger(self, atype, args):
"""True when P1's action logger (rt.test.actions) is on: the call is
recorded there and not run (the never-list in lab steps)."""
js = ("(function (t, a) { const R = window.__LGS_RT, A = R && R.test && R.test.actions;"
" if (!A || typeof A.enabled !== 'function' || !A.enabled()) return false;"
" try { A.record({ fn: 'daemon.' + t, arg: a }); } catch (_) { /* old runtime */ } return true; })"
f"({json.dumps(atype)}, {json.dumps(args)})")
return bool(await self.ext_steam_eval(js, 5))
async def action_sgwindow(self, ctx, atype, args, call):
"""Built-in 'sgwindow': Steam's window dim / recede for a while (TTL),
also in CSS-only mode (contracts/sg.md §5; contract §5)."""
if args.get("reset"):
self.window_req = None
else:
w = {k: args[k] for k in ("dim", "recede", "motion") if k in args}
if not w:
raise ValueError("nothing to set")
self.window_req = dict(w, until=time.time() + args.get("ttlMs", 10000) / 1000, src=call.get("src"))
self.changed.set()
return {"window": self.window_state()}
async def action_glass_tune(self, ctx, atype, args, call):
"""Built-in 'glass.tune': the wearer's glass tune from the bar's paintbrush panel (dial, colour,
refraction, frost, highlights; reset). glassd.json and the bridge follow at once (glassd eases
to it); tune.json, kept across reboots, TUNE_SAVE_S after the last change."""
if args.get("reset"):
t = dict(lgs.TUNE_DEFAULT)
else:
t = dict(self.tune)
for k in ("dial", "hueK", "refract", "frost", "light"):
if k in args:
t[k] = args[k]
if "hue" in args:
t["hue"] = args["hue"] or None
self.set_tune(lgs.clean_tune(t), save=True)
return {"tune": self.tune}
def stereo_on(self):
"""Per-eye glass unless the flag glassStereo is false."""
return (self.rt_flags or {}).get("glassStereo") is not False
def set_tune(self, t, save):
"""A new tune (the panel's, or tune.json changed by `lgs dial`): glassd.json, the bridge, and
(save) tune.json plus the dial file a moment later."""
self.tune = t
self.dial_v = t["dial"] if save or os.path.exists(lgs.TUNE) else lgs.read_dial()
self.bridge_values["tune"] = dict(t)
self.write_glassd_json()
self.changed.set()
if save:
if self.tune_save is not None:
self.tune_save.cancel()
self.tune_save = asyncio.get_running_loop().call_later(TUNE_SAVE_S, self.save_tune)
def save_tune(self):
"""tune.json (removed at the defaults) and, when the dial or the colour changed, the dial file:
the theme bundle carries both, and lgs_vr re-themes the SteamVR pages on the dial file."""
self.tune_save = None
try:
lgs.write_tune(self.tune)
mirror = (self.tune["dial"], self.tune["hue"], self.tune["hueK"])
if mirror != self.tune_mirrored:
with open(lgs.DIAL, "w", encoding="utf-8") as f:
f.write(f"{self.tune['dial']:g}\n")
self.tune_mirrored = mirror
except OSError as e:
self.note(f"tune: not saved ({e})")
self.tune_mtime = _mtime(lgs.TUNE)
self.dial_mtime = _mtime(lgs.DIAL)
def status(self):
g = self.gout or {}
sg = self.sg_status or {}
b = self.sg_beat or {}
rep = self.current_report() or {}
return {
"pid": os.getpid(), "since": time.strftime("%H:%M:%S", time.localtime(self.since)),
"updated": time.time(), "mode": self.mode(), "pod": self.pod,
"native": {"requested": self.native_req, "resolved": "on" if self.native_requested else "off",
"source": self.native_src, "enabled": self.native_enabled, "reason": self.css_reason},
"flags": self.rt_flags, "profile": self.profile, "reduceMotion": self.reduce_motion,
"testReport": self.test_report,
"bridge": {"runtime": self.bridge_runtime,
"sent": {k: round(time.time() - t, 1) for k, (_, t) in self.bridge_sent.items()}},
"geom": self.geom, "page": self.page, "geomError": self.geom_err,
"mainAway": self.page_away(STEAM_MAIN_KEY),
"geomDumpAt": round(self.geom_dump_at, 2) if self.geom_dump_at else None,
"actions": dict(self.registry.status(), replies=list(self.replies), trustedCtx=self.trusted_ctx),
"vrScripts": self.vrx_status,
"sgRules": {"active": [r.get("id") for r in self.sg_rules], "errors": self.sg_rule_errors,
"window": self.window_req},
"glassd": {"path": self.glassd_path, "args": self.glassd_args, "pid": self.g_pid,
"running": self.glassd_running(), "ready": self.glassd_ready(),
"healthy": g.get("healthy"), "stale": self.g_stale, "feed": g.get("feed"),
"restarts": self.g_restarts, "tempExits": self.g_temp_exits, "lastExit": self.g_last_exit,
"givenUp": self.g_given_up, "frames": self.g_frames, "seq": g.get("seq"), "fps": g.get("fps"),
"gpu_ms": g.get("gpu_ms"), "surfaces": sorted((g.get("surfaces") or {}).keys()),
"configSeq": self.gseq, "version": g.get("version"), "caps": sorted(self.caps()),
"fading": len(self.fading)},
"dormant": ({"since": round(self.dormant_since, 1), "graceS": self.theme_grace_s(),
"stay": self.stay} if self.dormant_since is not None else None),
"steam": {"connected": self.steam_ok, "themeOn": self.theme_on, "reporter": self.reporter_state,
"restarted": self.steam_restarted, "slowEvals": self.slow_evals,
"reports": self.reports, "reportSeq": (self.report or {}).get("seq"),
"reportAgeS": round(time.time() - self.report_at, 1) if self.report_at else None,
"dashboardHidden": self.dash_hidden,
"source": "reporter" if self.report is not None else ("fallback" if self.fallback else None),
"surfaces": [s.get("name") for s in rep.get("surfaces", [])],
"lgsNative": self.native_applied, "acked": self.acked, "plateAcks": self.plate_acks,
"tagged": self.native_tagged, "layersVersion": self.layers_version},
"systemui": {"connected": self.sysui_ok, "sgVersion": self.sg_version, "spec": self.spec_summary,
"beatAgeS": round(time.time() - self.sg_beat_at, 1) if self.sg_beat_at else None,
"items": b.get("items", sg.get("items")), "panels": sg.get("panels"), "nodes": sg.get("nodes"),
"attached": b.get("attached", sg.get("attached")), "parents": sg.get("parents"),
"relayouts": sg.get("relayouts"), "pushes": sg.get("pushes"),
"pushesLastSec": sg.get("pushesLastSec"), "reattaches": sg.get("reattaches"),
"scheduler": sg.get("scheduler"), "expired": b.get("expired", sg.get("expired")),
"expiries": sg.get("expiries"), "rebuilds": sg.get("rebuilds"),
"flags": self.flags, "sgErrors": sg.get("errors")},
"steamvrPages": {"paused": os.path.exists(VR_PAUSE), "version": self.vr_version, "pages": self.vr_pages,
"pausedS": round(time.time() - self.vr_pause_since, 1) if self.vr_pause_since else None},
"stalls": {"count": self.stalls, "lastS": self.last_stall_s,
"agoS": round(time.monotonic() - self.stall_at, 1) if self.stalls else None},
"errors": list(self.errors),
}
# ---------------------------------------------------------------- glassd.json
def ramp_ms(self, material, out, kind="slab"):
if self.reduce_motion:
return RAMP_REDUCED_MS
if material == "dim" and kind == "slab": # a dim slab rides sheet-in / sheet-out (glassd §1.4)
material = "thick"
if out:
return RAMP_OUT_MS.get(material, RAMP_OUT_DEFAULT_MS)
return RAMP_IN_MS.get(material, RAMP_IN_DEFAULT_MS)
def glassd_config(self):
"""glassd.json from the report: Phase 1 fields, the v3 fields glassd's
caps announce, and the materialize policy (contracts/daemon.md §7)."""
now = time.time()
caps = self.caps()
rep = self.current_report() or {"surfaces": []}
surfaces, by_name, current, shown, gained = [], {}, {}, set(), set()
for s in rep.get("surfaces", []):
if not isinstance(s, dict) or not s.get("name") or not s.get("overlayKey"):
continue
name = str(s["name"])[:64]
# every number type-checked (contract §7): a bad field drops its item, never the daemon
texW, texH, radius = _num(s.get("texW") or 0, 0, 16384), _num(s.get("texH") or 0, 0, 16384), \
_num(s.get("radius") or 0, 0, 8192)
if texW is None or texH is None or radius is None or not isinstance(s["overlayKey"], str):
continue
# Steam's page away (SteamVR Settings or the binding UI in its frame): glassd keeps drawing
# main's glass, ready for its return, and lgs_sg.js takes main's nodes off the scene graph
# itself in the same update the page goes in, and back in the one it returns in (it reads
# SteamVR's active page every frame); this 1 s poll only follows it for status
slabs = []
for L in s.get("layers") if isinstance(s.get("layers"), list) else []:
if not isinstance(L, dict) or L.get("id") is None:
continue
w, h = _num(L.get("w"), 0, 16384), _num(L.get("h"), 0, 16384)
r, x, y = _num(L.get("r") or 0, 0, 8192), _num(L.get("x") or 0), _num(L.get("y") or 0)
dz = _num(L.get("dz") or 0, -1, 1)
if None in (w, h, r, x, y, dz) or w < 2 or h < 2:
continue
mat = L.get("material")[:16] if isinstance(L.get("material"), str) else "liquid"
if mat == "none" and "none" not in caps:
continue # an older glassd would draw glass over art: the element stays flat
if mat == "dim" and "dimSlab" not in caps:
continue # ... or glass where a dark room-dim cell was asked for
sl = {"id": str(L["id"])[:64], "w": round(w), "h": round(h), "r": round(r),
"material": mat or "liquid",
# for glassd's slab world point:
"x": round(x), "y": round(y), "dz": dz}
clean_phase_fields(L, sl)
if "tint" in caps and clean_color(L.get("tint")) is not None:
sl["tint"] = clean_color(L["tint"])
if mat == "dim" and clean_color(L.get("fill")) is not None:
sl["fill"] = clean_color(L["fill"])
if "holes" in caps and L.get("hole") not in (None, False):
hole = clean_hole(L["hole"], self.pop_clips.get((name, sl["id"])), edges="holeEdges" in caps)
if hole is not None:
sl["hole"] = hole
if "offset" in caps:
for k in ("ox", "oy"):
if _num(L.get(k)) is not None:
sl[k] = _round(round(_num(L[k]), 2))
slabs.append(sl)
mat_s = material_for(s)
surf = {"name": name, "overlayKey": s["overlayKey"][:128],
"texW": round(texW), "texH": round(texH), "radius": round(radius),
"material": mat_s[:16] if isinstance(mat_s, str) else "panel",
"visible": bool(s.get("visible", True)), "slabs": slabs}
# a hidden pooled popup whose content is ready: glassd draws its glass ahead (contracts/glassd.md)
if s.get("armed") is True and not surf["visible"]:
surf["armed"] = True
# the cover's rounded shapes (bar segments, a popup's card); glassd
# covers the whole texture with `radius` when absent
if isinstance(s.get("shapes"), list):
shapes = []
for sh in s["shapes"]:
if not isinstance(sh, dict):
continue
vals = {k: _num(sh.get(k) or 0) for k in ("x", "y", "w", "h", "r")}
if None not in vals.values():
shapes.append({k: round(v) for k, v in vals.items()})
surf["shapes"] = shapes
clean_phase_fields(s, surf) # the cover's own phase: passed through, never set by P8
# a morphing cover (the reporter's cover.morph): glassd moves the shapes it shows toward these
# on the named spring from `at` (epoch ms), with the page's own motion
m = s.get("morph")
if "morph" in caps and isinstance(m, dict) and m.get("token") in MORPH_TOKENS \
and _num(m.get("at"), 0, 1e15) is not None:
surf["morph"] = {"token": m["token"], "at": int(_num(m["at"], 0, 1e15))}
q = clean_quad(s.get("quad")) if s.get("quad") is not None else None
if q:
surf["quad"] = q
plates = [p for p in (clean_plate(p, i) for i, p in enumerate(s.get("plates") or [])
if isinstance(s.get("plates"), list)) if p]
if "dim" not in caps: # a dim plate is a flat dark tone: never glass instead
plates = [p for p in plates if p.get("material") != "dim"]
if "exitAt" not in caps:
for p in plates:
p.pop("exitAt", None)
p.pop("exitMs", None)
if plates and "plates" in caps:
surf["plates"] = plates[:32]
elif plates and isinstance(surf.get("shapes"), list):
# a v2 glassd: plates become cover shapes (≤ 8), as P6 suggests
surf["shapes"] = (surf["shapes"] + [{k: round(p[k]) for k in ("x", "y", "w", "h")}
| {"r": round(p.get("r", min(p["w"], p["h"]) / 2))}
for p in plates])[:8]
if "coverDz" in caps and _num(s.get("coverDz"), -1, 1) is not None:
surf["coverDz"] = _round(round(_num(s["coverDz"], -1, 1), 5))
if "masks" in caps and s.get("masks") is not None:
surf["masks"] = clean_masks(s.get("masks"))
if "scaleFrom" in caps and s.get("scaleFrom") in ("main", "overlay"):
surf["scaleFrom"] = s["scaleFrom"]
# per-eye glass where the scene graph shows the surface whole (Steam's window; the bar, the frame
# menu and the popups show crops of theirs, which a stereo panel cannot split per eye)
# stereo where the scene graph says which part of the texture its panel shows (glassd builds
# the eye pair around that crop; the panel spans the pair): Steam's window, the bar, the frame
# menu, the footer, popups once shown
# not the frame menu: it opens by animating its width and Steam widens its crop along with it, which
# reaches glassd a beat (~1 s) late, so a pair built around the old crop cut the opening glass off
crop = (self.sg_crops or {}).get(s["overlayKey"]) if "stereo" in caps and self.stereo_on() and not name.startswith("frame.menu") else None
if isinstance(crop, list) and len(crop) == 2 and None not in (_num(crop[0], 0, 1), _num(crop[1], 0, 1)) and crop[1] - crop[0] > 0.01:
surf["stereo"] = True
surf["crop"] = [round(float(crop[0]), 3), round(float(crop[1]), 3)]
# materialize: a new slab or plate id appears with materialize (GM §5.4)
for kind, items in (("slab", surf["slabs"]), ("plate", surf.get("plates") or [])):
for it in items:
key = (name, kind, it["id"])
if key not in self.items_first:
self.items_first[key] = now
if "appear" not in it and "phase" not in it:
it["appear"] = "materialize"
current[key] = it
if surf["visible"]:
shown.add(name)
# a cover is "shapes" non-empty, or (absent) the whole texture of a window-material surface
cover = bool(surf["shapes"]) if isinstance(surf.get("shapes"), list) else surf["material"] == "window"
if cover and self.cover_last.get(name) is False:
gained.add(name)
self.cover_last[name] = cover
surfaces.append(surf)
by_name[name] = surf
# ... and one that leaves dematerializes: phase 0 for its out-ramp, then gone. A plate whose content
# was fading out (exitAt) stays for exactly that fade instead, glassd fading it with the page; on a
# surface that just gained its cover (Home -> a window route) a plate removed before any fade was
# seen fades on the page's leave (PAGE_OUT_MS) from now
for key, it in self.items_last.items():
if key in current or key in self.fading:
continue
if key[0] not in shown: # its surface closed or hid: nothing to fade on
continue
if key[1] == "plate" and "exitAt" in caps:
if "exitAt" not in it and key[0] in gained:
it = it | {"exitAt": int(now * 1000), "exitMs": PAGE_OUT_MS}
if "exitAt" in it:
until = (it["exitAt"] + it["exitMs"]) / 1000
if until > now:
self.fading[key] = ({k: v for k, v in it.items() if k != "appear"}, min(until, now + 2.0))
continue
until = now + self.ramp_ms(it.get("material"), True, key[1]) / 1000
self.fading[key] = ({k: v for k, v in it.items() if k != "appear"} | {"phase": 0}, until)
for key, (it, until) in list(self.fading.items()):
surf = by_name.get(key[0])
if key in current or until <= now or surf is None or not surf["visible"]:
del self.fading[key]
continue
surf.setdefault("slabs" if key[1] == "slab" else "plates", []).append(dict(it))
self.items_last = current
for key in [k for k in self.items_first if k not in current and k not in self.fading]:
del self.items_first[key]
# SteamVR Settings' window glass (a SteamVR panel, not a Steam window): placed by main's overlay
# (the page takes main's place in the frame), drawn ahead ("armed") while the page is away, so the
# glass is there in the update the page mounts in (lgs_sg.js adds the cover node then)
for vn, vp in self.vr_panels.items() if "armed" in caps else []:
g = vp.get("glass") or {}
vals = [_num(vp.get("w"), 2, 8192), _num(vp.get("h"), 2, 8192)] + \
[_num(g.get(k), 0, 8192) for k in ("x", "y", "w", "h", "r")]
if None in vals:
continue
w, h, gx, gy, gw, gh, gr = (round(v) for v in vals)
on = vn in self.vr_shown
surfaces.append({"name": "vr." + vn, "overlayKey": STEAM_MAIN_KEY, "texW": w, "texH": h,
"radius": gr, "material": vp["material"], "visible": on, "armed": not on, "slabs": [],
**({"stereo": True} if "stereo" in caps and self.stereo_on() else {}),
"shapes": [{"x": gx, "y": gy, "w": gw, "h": gh, "r": gr}]})
cfg = {"dial": self.dial_v, "reduceMotion": bool(self.reduce_motion), "surfaces": surfaces}
# per-eye glass (glassd renders each eye's view side by side; the scene graph shows each eye its half):
# on unless the flag glassStereo is false (contracts/glassd.md §1.1)
if "stereo" in caps:
cfg["stereo"] = self.stereo_on()
if "tune" in caps: # the wearer's tune (glassd eases to it; contracts/glassd.md §1.1)
t = self.tune
cfg["tune"] = {"refract": t["refract"], "frost": t["frost"], "light": t["light"]}
if t["hue"]:
cfg["tune"].update(hue=t["hue"], hueK=t["hueK"])
if "unitM" in caps and self.geom and _num(self.geom.get("unitM"), 0.01, 10) is not None:
cfg["unitM"] = self.geom["unitM"]
if "masks" in caps and rep.get("masks") is not None:
cfg["masks"] = clean_masks(rep.get("masks"), world=True)
if "roomDim" in caps and _num(rep.get("roomDim"), 0, 0.9) is not None:
cfg["roomDim"] = _round(_num(rep["roomDim"], 0, 0.9))
return cfg
def write_glassd_json(self, force=False):
if not self.native_possible() or (self.dormant_since is not None and not self.glassd_running()):
return False # dormant: glassd is stopped and its files are gone until the theme is back
cfg = self.glassd_config()
full = json.dumps(cfg, sort_keys=True)
moving = ("x", "y", "dz", "ox", "oy")
struct = json.dumps(dict(cfg, surfaces=[
dict(s, slabs=[{k: v for k, v in sl.items() if k not in moving} for sl in s["slabs"]],
plates=[{k: v for k, v in p.items() if k not in moving} for p in s.get("plates") or []])
for s in cfg["surfaces"]]), sort_keys=True)
if not force and full == self.gjson_full:
return False
now = time.time()
# Positions alone change while scrolling: write those at most 4x a second.
if not force and struct == self.gjson_struct and now - self.gjson_written_at < GLASSD_JSON_MIN_INTERVAL:
return False
self.gseq += 1
write_json_atomic(GLASSD_JSON, dict(cfg, seq=self.gseq))
self.gjson_full, self.gjson_struct, self.gjson_written_at = full, struct, now
return True
# ---------------------------------------------------------------- spec
def track_layers(self, rep, now):
"""Since when each reported layer's rect has been still (within STILL_PX)."""
seen = {}
for s in rep.get("surfaces") or []:
if not isinstance(s, dict):
continue
for L in s.get("layers") if isinstance(s.get("layers"), list) else []:
try:
k = (str(s.get("name")), str(L["id"]))
r = (float(L["x"]), float(L["y"]), float(L["w"]), float(L["h"]))
except (KeyError, TypeError, ValueError):
continue
prev = self.layer_still.get(k)
if prev is not None and max(abs(a - b) for a, b in zip(prev[0], r)) <= STILL_PX:
seen[k] = prev
else:
seen[k] = (r, now)
self.layer_still = seen
def read_flags(self):
"""Optional panel flags for lgs_sg.js (spec.flags), from /tmp/lgs/sg-flags.json."""
try:
m = os.stat(SG_FLAGS).st_mtime_ns
except OSError:
m = None
if m != self.flags_mtime:
self.flags_mtime = m
f = read_json(SG_FLAGS) if m is not None else None
self.flags = f if isinstance(f, dict) and f else None
if m is not None:
log(f"systemui: panel flags {self.flags}")
return self.flags
def main_resized(self, g):
"""glassd's world width of the main quad, if it reports one (quadW), far
from nominal: the user resized the window. Until crops are verified to
follow a resize, main then stays CSS only."""
q = _num(g.get("quadW"), 0.01, 100)
if q is None:
return False
bad = abs(q / MAIN_QUAD_W - 1) > MAIN_QUAD_TOL
if bad != bool(self.resized_noted):
self.resized_noted = bad
if bad:
self.note(f"main window is {q:.2f} m wide (nominal {MAIN_QUAD_W}): resized; main stays CSS only")
else:
log(f"main window back to {q:.2f} m wide: native again")
return bad
def spec_top(self):
"""The spec's top-level v2 fields (contracts/sg.md §3.1)."""
top = {"profile": self.profile, "reduceMotion": bool(self.reduce_motion),
"depthMotion": "none" if self.rt_flags.get("sgDepthAnim") is False else "depth"}
if self.geom and _num(self.geom.get("unitM"), 0.01, 10) is not None:
top["unitM"] = self.geom["unitM"]
w = self.window_state()
if w is not None:
top["window"] = w
return top
def build_spec(self):
self.wake_at = None
if not self.native_ok():
return dict({"M": None, "surfaces": []}, **self.spec_top())
rep = self.current_report() or {"surfaces": []}
clips = {}
gsurf = (self.gout or {}).get("surfaces")
gsurf = gsurf if isinstance(gsurf, dict) else {}
now = time.time()
out = []
wake = None
for s in rep.get("surfaces") if isinstance(rep.get("surfaces"), list) else []:
if not isinstance(s, dict) or not isinstance(s.get("name"), str) or not isinstance(s.get("overlayKey"), str):
continue
g = gsurf.get(s["name"])
if not isinstance(g, dict):
continue
if not s.get("visible", True):
# a hidden pooled popup: its nodes wait in lgs_sg.js (standby) and are built the moment
# Steam puts its panel back on the page, in the same scene-graph update as Steam's own
if s.get("pooled") is True:
sb = self.standby_entry(s, g)
if sb:
out.append(sb)
continue
if s.get("name") == "main" and self.main_resized(g):
continue
texW, texH = _num(s.get("texW"), 1, 16384), _num(s.get("texH"), 1, 16384)
scale = _num(g.get("backdropScale") or 0.75, 0.05, 4)
gW, gH = _num(g.get("texW"), 1, 16384), _num(g.get("texH"), 1, 16384)
backdrop = _nums(g.get("backdrop"), 4)
if None in (texW, texH, scale, gW, gH) or backdrop is None:
continue
slabs = g.get("slabs") if isinstance(g.get("slabs"), dict) else {}
popped, taken, dims = [], [], []
for L in s.get("layers") if isinstance(s.get("layers"), list) else []:
if not isinstance(L, dict) or L.get("id") is None:
continue
lid = str(L["id"])[:64]
x, y, w, h = (_num(L.get(k)) for k in ("x", "y", "w", "h"))
if None in (x, y, w, h):
continue
uv = _nums(slabs.get(lid), 4)
if uv is None:
continue # no slab yet: the element stays flat in the base (with its CSS glass)
if L.get("material") == "dim":
# a room-dim cell (glassd §1.4): no element to crop; P7 places
# the cell (behind the window, scaled up). Older lgs_sg.js ignore it.
dz = _num(L.get("dz") or 0, -1, 1)
if dz is not None:
dims.append({"id": lid, "x": round(x, 2), "y": round(y, 2), "w": round(w, 2),
"h": round(h, 2), "dz": dz, "slab": uv})
continue
# glassd draws a slab at the element's size x backdropScale; if
# the element was resized since, wait for the new slab.
if abs((uv[2] - uv[0]) * gW - w * scale) > SLAB_SIZE_TOL_PX or \
abs((uv[3] - uv[1]) * gH - h * scale) > SLAB_SIZE_TOL_PX:
continue
# pop only once the rect has been still: a moving element
# (scrolling, a focus change) stays flat instead of being
# cropped at a stale rect for a scene-graph round trip
still = self.layer_still.get((str(s.get("name")), lid))
if still is None:
continue
if now - still[1] < STILL_S:
t = still[1] + STILL_S
wake = t if wake is None else min(wake, t)
continue
# an overlap of a few px with an already popped element (the
# reporter allows 2 px) is trimmed off; a wider one skips it
rect = [x, y, x + w, y + h]
clipped, conflict = False, False
for t in taken:
ix = min(rect[2], t[2]) - max(rect[0], t[0])
iy = min(rect[3], t[3]) - max(rect[1], t[1])
if ix <= 0 or iy <= 0:
continue
if min(ix, iy) > OVERLAP_CLIP_PX:
conflict = True
break
clipped = True
if ix <= iy:
if t[0] <= rect[0] < t[2]:
rect[0] = t[2]
else:
rect[2] = t[0]
else:
if t[1] <= rect[1] < t[3]:
rect[1] = t[3]
else:
rect[3] = t[1]
if conflict or rect[2] - rect[0] < 2 or rect[3] - rect[1] < 2:
continue
dz = _num(L.get("dz") or 0.012, -1, 1)
if dz is None:
continue
taken.append(tuple(rect))
p = {"id": lid, "x": round(x, 2), "y": round(y, 2), "w": round(w, 2), "h": round(h, 2),
"dz": dz, "slab": uv}
if clipped:
p["clip"] = [round(v, 2) for v in rect]
clips[(str(s.get("name")), lid)] = p["clip"]
# depth-channel fields (contracts/sg.md §3.3); interactive only in the wearer profile
if L.get("interactive") is True and self.profile == "wearer":
p["interactive"] = True
if L.get("from") == "cut" or _num(L.get("from"), -1, 1) is not None:
p["from"] = L["from"] if L.get("from") == "cut" else _num(L["from"], -1, 1)
if isinstance(L.get("motion"), str) and len(L["motion"]) <= 24:
p["motion"] = L["motion"]
if isinstance(L.get("sink"), bool):
p["sink"] = L["sink"]
popped.append(p)
sp = {"steamKey": s["overlayKey"], "texW": texW, "texH": texH, "visible": True,
"glassd": {"key": g.get("key") or "glassd." + s["name"], "backdrop": backdrop, "scale": scale,
"stereo": g.get("stereo") is True, "crop": g.get("crop")},
"coverDz": COVER_DZ, "baseDz": BASE_DZ, "popped": popped}
if _num(s.get("coverDz"), -1, 1) is not None:
sp["coverDz"] = _num(s["coverDz"], -1, 1)
if _num(s.get("dim"), 0, 1) is not None:
sp["dim"] = _num(s["dim"], 0, 1)
# mosaic bands only without a cover (contracts/reporter.md §2.3): with a cover Steam's real
# panel is hidden behind it, so a base limited to bands would hide everything outside them
# (REQ C2a->P8 #17; P6's reporter no longer sends both, this guards older copies)
shp = s.get("shapes")
has_cover = (isinstance(shp, list) and len(shp) > 0) or (shp is None and material_for(s) == "window")
if isinstance(s.get("mosaic"), list) and not has_cover:
sp["mosaic"] = [r for r in (clean_rect(m, ("x", "y", "w", "h"), ()) for m in s["mosaic"]
if isinstance(m, dict)) if r][:16]
# slabs glassd is fading out (contracts/daemon.md §7): no crop, their cell
outs = []
for (sname, kind, sid), (it, until) in self.fading.items():
if sname != s.get("name") or kind != "slab":
continue
uv = _nums(slabs.get(sid), 4)
if uv is not None:
outs.append({"id": sid, "x": it.get("x"), "y": it.get("y"), "w": it.get("w"), "h": it.get("h"),
"dz": it.get("dz"), "slab": uv, "until": int(until * 1000)})
if outs:
sp["slabsOut"] = outs
if dims:
sp["dimSlabs"] = dims
out.append(sp)
self.wake_at = wake
self.pop_clips = clips
spec = dict({"M": None, "surfaces": out}, **self.spec_top())
vr = []
for vn in self.vr_panels:
gv = gsurf.get("vr." + vn)
if isinstance(gv, dict) and _nums(gv.get("backdrop"), 4) is not None:
vr.append({"name": vn, "glassd": {"key": gv.get("key") or "glassd.vr." + vn,
"backdrop": _nums(gv.get("backdrop"), 4),
"scale": _num(gv.get("backdropScale") or 0.75, 0.05, 4),
"stereo": gv.get("stereo") is True}})
if vr:
spec["vr"] = vr
flags = self.read_flags()
if flags and out:
spec["flags"] = flags
return spec
def standby_entry(self, s, g):
texW, texH = _num(s.get("texW"), 1, 16384), _num(s.get("texH"), 1, 16384)
scale = _num(g.get("backdropScale") or 0.75, 0.05, 4)
backdrop = _nums(g.get("backdrop"), 4)
if None in (texW, texH, scale) or backdrop is None:
return None
sp = {"steamKey": s["overlayKey"], "texW": texW, "texH": texH, "visible": False, "standby": True,
"glassd": {"key": g.get("key") or "glassd." + s["name"], "backdrop": backdrop, "scale": scale,
"stereo": g.get("stereo") is True, "crop": g.get("crop")},
"coverDz": COVER_DZ, "baseDz": BASE_DZ, "popped": []}
if _num(s.get("coverDz"), -1, 1) is not None:
sp["coverDz"] = _num(s["coverDz"], -1, 1)
return sp
def native_live(self):
"""Native mode is live: glassd produces frames and the scene graph follows our spec (the same
preconditions native_state needs). Then every Steam window is lgs-native."""
if not (self.native_ok() and self.sysui_ok and self.sg_beat and self.spec_sent_obj):
return False
b = self.sg_beat
return not (time.time() - self.sg_beat_at > SG_BEAT_FRESH_S or b.get("expired") or not b.get("attached")
or not b.get("push"))
def native_state(self):
"""(overlay keys that get lgs-native, {key: popped ids shown}, {key: plate
ids shown}, keys whose cover glass is shown), from the latest systemui
heartbeat, the spec it reflects and glassd's output (contract §8)."""
if not (self.native_ok() and self.sysui_ok and self.sg_beat and self.spec_sent_obj):
return [], {}, {}, []
b = self.sg_beat
if time.time() - self.sg_beat_at > SG_BEAT_FRESH_S or b.get("expired") or not b.get("attached") \
or not b.get("push"):
return [], {}, {}, []
gsurf = (self.gout or {}).get("surfaces")
gsurf = gsurf if isinstance(gsurf, dict) else {}
names = {s["overlayKey"]: s["name"] for s in (self.current_report() or {}).get("surfaces", [])
if isinstance(s, dict) and isinstance(s.get("overlayKey"), str) and isinstance(s.get("name"), str)}
spec = {s["steamKey"]: {p["id"] for p in s["popped"]} for s in self.spec_sent_obj.get("surfaces", [])}
now = time.time()
now_ms = now * 1000
keys, acks, plates, covers = [], {}, {}, []
surf = b.get("surf") if isinstance(b.get("surf"), dict) else {}
for key, sm in surf.items():
if key not in spec or not isinstance(sm, dict):
continue
name = names.get(key)
g = gsurf.get(name) if isinstance(gsurf.get(name), dict) else {}
at = _num(sm.get("coverAt") or 0, 0, 1e15) or 0 # epoch ms
if not at or now_ms - at < SHOWN_MS:
continue
# plates: drawn by glassd (listed in its output), in the cover's
# region (pushed), and their materialize ramp has had time to end
pl = []
for pid in g.get("plates") if isinstance(g.get("plates"), list) else []:
first = self.items_first.get((name, "plate", str(pid)))
if first is None or (name, "plate", str(pid)) in self.fading:
continue
item = self.items_last.get((name, "plate", str(pid))) or {}
if now - max(first, at / 1000) >= (self.ramp_ms(item.get("material"), False, "plate") + SHOWN_MS) / 1000:
pl.append(str(pid))
has_cover = g.get("cover") != 0
if not has_cover and not pl:
# glassd draws no glass for it right now: between two glass modes (main: window ->
# windowless, its plates still materializing) the window stays native a moment, so its
# CSS glass never flashes back in between
if now - self.native_seen.get(key, 0) > NATIVE_GRACE_S:
continue
else:
self.native_seen[key] = now
keys.append(key)
if has_cover:
covers.append(key)
if pl:
plates[key] = sorted(pl)
pops = sm.get("pops") if isinstance(sm.get("pops"), dict) else {}
acks[key] = sorted(i for i, t in pops.items()
if _num(t, 0, 1e15) and now_ms - _num(t, 0, 1e15) >= SHOWN_MS and i in spec[key])
return sorted(keys), acks, plates, sorted(covers)
# ---------------------------------------------------------------- Steam
def on_binding(self, params):
if params.get("name") == ACTION_BINDING:
payload = params.get("payload")
probe = self.ctx_probe
if probe and isinstance(payload, str) and payload == json.dumps({"probe": probe[0]}):
if not probe[1].done():
probe[1].set_result(params.get("executionContextId"))
return
t = asyncio.get_running_loop().create_task(self.handle_action(payload, params.get("executionContextId")))
self.action_tasks.add(t)
t.add_done_callback(self.action_tasks.discard)
return
if params.get("name") != BINDING or self.test_report:
return
self.take_report(params.get("payload"))
def take_report(self, payload):
try:
rep = json.loads(payload or "null")
except ValueError as e:
self.note(f"bad layer report: {e}")
return
if not isinstance(rep, dict) or not isinstance(rep.get("surfaces"), list):
return
now = time.time()
self.report_at = now
self.reports += 1
# its last report (every surface invisible) when it stops: re-injected
# on the next steam tick
self.reporter_state = "stopped (reporter)" if rep.get("stopped") else "reporting"
hidden = rep.get("dash") is False
if hidden != self.dash_hidden:
log("steam: dashboard " + ("hidden: glass layout kept as is" if hidden else "shown"))
self.dash_hidden = hidden
# While the dashboard is hidden every surface reports invisible: keep
# the last layout (nodes, covers, glassd's atlas) so reopening shows the
# glass at once instead of rebuilding it.
if hidden and self.report is not None:
return
self.track_layers(rep, now)
self.report = rep
self.changed.set()
def reporter_key(self):
"""What the injected reporter depends on: its file, the layer fragments
and the depth profile. A change re-injects it."""
self.layers_version = layers_config()[2]
return f"{file_version(LAYERS_JS)}/{self.layers_version}/{self.profile}"
def read_test_report(self):
"""Lab: the report comes from a file (--test-report), re-read on change."""
try:
m = os.stat(self.test_report).st_mtime_ns
except OSError:
m = None
if m == self.test_report_mtime:
return
self.test_report_mtime = m
if m is None:
if self.report is not None:
self.report = None
self.changed.set()
self.reporter_state = "test report missing"
return
try:
with open(self.test_report, encoding="utf-8") as f:
payload = f.read()
except OSError:
return
self.reporter_state = "test report"
self.take_report(payload)
async def steam_target(self):
loop = asyncio.get_running_loop()
ts = await loop.run_in_executor(None, lambda: http_json(lgs.CDP + "/json/list", 5))
for t in ts:
if t.get("title") == "SharedJSContext":
return t
raise ConnectionError("no SharedJSContext target")
# ---------------------------------------------------------------- actions (contract §5)
async def probe_context(self):
"""Learn SharedJSContext's default execution context id: our own
evaluation (which runs there) calls the binding with a nonce."""
async with self.ctx_probe_lock:
s = self.steam
if not (self.steam_ok and s is not None and not s.closed):
return None
nonce = os.urandom(8).hex()
fut = asyncio.get_running_loop().create_future()
self.ctx_probe = (nonce, fut)
try:
await s.eval(f"(window.{ACTION_BINDING}({json.dumps(json.dumps({'probe': nonce}))}), 0)", 5)
self.trusted_ctx = await asyncio.wait_for(fut, 3)
except (asyncio.TimeoutError, *NET_ERRORS) as e:
self.note(f"actions: context probe failed: {e!r}")
finally:
self.ctx_probe = None
return self.trusted_ctx
async def handle_action(self, payload, ctx_id):
try:
if self.trusted_ctx is None or ctx_id != self.trusted_ctx:
await self.probe_context() # first call, or the page's context changed
reply = await self.registry.call(payload, from_default_context=ctx_id == self.trusted_ctx)
if reply is None:
return
self.replies.append({k: reply.get(k) for k in ("id", "ok", "error", "dry")} | {"at": time.strftime("%H:%M:%S")})
await self.bridge_send([["reply", reply]])
except asyncio.CancelledError:
raise
except Exception as e: # noqa: BLE001 - an action must never take the daemon down
self.note(f"actions: {e!r}")
async def actions_loop(self):
"""Plugins: load new or changed files, unload removed ones, tick them."""
while not self.stopping:
try:
ch = await self.registry.scan()
if ch:
log(f"actions: plugins {ch}; types {sorted(self.registry.types)}")
await self.registry.tick()
except asyncio.CancelledError:
raise
except Exception as e: # noqa: BLE001
await asyncio.sleep(self.bug("actions", e))
await asyncio.sleep(ACTION_TICK_S)
# ---------------------------------------------------------------- bridge (contract §3)
async def bridge_send(self, pairs):
s = self.steam
if not pairs or not (self.steam_ok and s is not None and not s.closed):
return False
r = await s.eval(f"({BRIDGE_SET_JS})({json.dumps(pairs)})", 5)
ok = isinstance(r, str) and r.startswith("ok")
if ok != self.bridge_runtime:
log(f"steam: runtime bridge {'present' if ok else 'absent'}")
self.bridge_runtime = ok
now = time.time()
if ok:
for name, value in pairs:
if name != "reply":
self.bridge_sent[name] = (json.dumps({k: v for k, v in value.items() if k != "at"}
if isinstance(value, dict) else value, sort_keys=True), now)
return ok
async def bridge_tick(self):
"""Send what changed, what the runtime lost (reinstalled), and the
daemon heartbeat every DAEMON_BEAT_S."""
s = self.steam
if not (self.steam_ok and s is not None and not s.closed):
return
now = time.time()
self.bridge_values["daemon"] = self.daemon_value()
if self.geom is not None:
self.bridge_values["geom"] = self.geom
if self.page is not None:
self.bridge_values["page"] = self.page
names = sorted(self.bridge_values)
missing = await s.eval(f"({BRIDGE_MISSING_JS})({json.dumps(names)})", 5)
if missing is None:
if self.bridge_runtime is not False:
log("steam: runtime bridge absent (no __LGS_RT.bridge); values kept for when it appears")
self.bridge_runtime = False
self.bridge_sent = {}
return
pairs = []
for name in names:
v = self.bridge_values[name]
js = json.dumps({k: x for k, x in v.items() if k != "at"} if isinstance(v, dict) else v, sort_keys=True)
prev = self.bridge_sent.get(name)
due = name in (missing or []) or prev is None or prev[0] != js or \
(name == "daemon" and now - prev[1] >= DAEMON_BEAT_S - 0.05)
if due:
pairs.append([name, v])
if pairs:
await self.bridge_send(pairs)
async def inject_reporter(self):
if not os.path.exists(LAYERS_JS):
if self.reporter_state != "missing":
log("lgs_layers.js not installed: main window glass only, nothing pops out")
self.reporter_state = "missing"
return False
with open(LAYERS_JS, encoding="utf-8") as f:
src = f.read()
cfg, errs, _ = layers_config()
for e in errs:
self.note(f"layers: {e}")
opts = {"binding": BINDING, "layers": cfg, "version": file_version(LAYERS_JS), "ackMode": True,
"profile": self.profile}
if self.geom and _num(self.geom.get("S"), 0.01, 10) is not None and _num(self.geom.get("r"), 0.05, 10):
opts["geom"] = {"S": self.geom["S"], "r": self.geom["r"]}
self.reporter_injected_at = time.time()
self.reporter_version = self.reporter_key()
try:
await self.steam.eval(f"(window.__LGS_LAYERS_OPTS = {json.dumps(opts)}, 0)")
r = await self.steam.eval(src, timeout=30, by_value=False)
if r.get("type") == "function" and r.get("objectId"): # a factory: call it with the options
await self.steam.send("Runtime.callFunctionOn", {
"objectId": r["objectId"], "functionDeclaration": "function (o) { return this(o); }",
"arguments": [{"value": opts}], "returnByValue": True, "awaitPromise": True})
if r.get("objectId"):
await self.steam.send("Runtime.releaseObject", {"objectId": r["objectId"]})
except RuntimeError as e: # their bug: report it, retry only when the file changes
self.reporter_state = f"error: {str(e)[:160]}"
self.note(f"reporter: {e}")
return False
present = await self.steam.eval(f"typeof window.{REPORTER}")
self.reporter_state = "injected" if present != "undefined" else "injected (no global)"
self.native_force = True # a fresh reporter: send the acknowledgements again
resend = await self.steam.eval(REPORTER_RESEND_JS)
log(f"reporter injected ({self.reporter_state}, resend: {resend})")
return True
async def steam_loop(self):
while not self.stopping:
delay = 2
try:
try:
t = await self.steam_target()
except NET_ERRORS:
# Steam gone (quit, restarting): nothing of ours is left in it;
# wait a while for it, then exit (never outlive a Steam restart)
self.steam_lost_at = self.steam_lost_at or time.time()
if time.time() - self.steam_lost_at >= STEAM_GONE_S: # --stay too (review R1 M3)
log(f"steam: unreachable for {STEAM_GONE_S} s; exiting")
self.request_stop("steam gone", strip_vr=True)
return
raise
self.steam_lost_at = None
if self.steam_target_id and t.get("id") != self.steam_target_id:
# a new SharedJSContext: Steam (or its UI) restarted. Until
# the theme is seen on again, theme off means exit at once.
self.steam_restarted = True
log("steam: SharedJSContext is a new target (Steam restarted or its UI reloaded)")
self.steam_target_id = t.get("id")
self.steam = await CDP("steam", t["webSocketDebuggerUrl"]).open()
self.steam.handlers["Runtime.bindingCalled"] = self.on_binding
# lgsLayers only when a reporter can be injected (native mode);
# a leftover one from an older unit is removed (G-REMOVE, review R1 M1)
if self.native_possible():
await self.steam.send("Runtime.addBinding", {"name": BINDING})
else:
r = await self.steam.eval(f"({DELETE_BINDINGS_JS})({json.dumps([BINDING])})", 10)
if r:
log(f"steam: removed a leftover binding global {r}")
await self.steam.send("Runtime.addBinding", {"name": ACTION_BINDING})
self.steam_ok = True
self.trusted_ctx, self.bridge_sent = None, {}
log("steam: connected")
await self.probe_context()
self.refresh_flags()
if self.native_possible() and not self.test_report:
await self.inject_reporter()
tick, slow = 0, 0
while not self.stopping and not self.steam.closed:
try:
r = await self.steam.eval(STEAM_TICK_JS, 10)
on, rm = json.loads(r) if isinstance(r, str) else (False, False)
if bool(rm) != self.reduce_motion:
self.reduce_motion = bool(rm)
log(f"steam: Reduce Motion {'on' if rm else 'off'}")
self.changed.set()
if self.refresh_flags() and self.native_possible() and self.reporter_version \
and not self.test_report:
await self.inject_reporter() # the depth profile changed
self.theme_on = bool(on)
self.theme_off_polls = 0 if on else self.theme_off_polls + 1
if on:
self.steam_restarted = False
if self.dormant_since is not None:
log(f"steam: theme is back after {time.time() - self.dormant_since:.0f} s; resuming")
self.dormant_since = None
self.native_force, self.bridge_sent = True, {}
self.changed.set()
self.check_stay_limit()
if self.theme_off_polls >= THEME_OFF_POLLS and not self.pod: # pod: no theme, by design
# A theme off without `lgs off` (which stops us first) is a
# test toggling it (lab --stock, P1's selftests) or a Steam
# restart: stay dormant for the grace, exit on a restart.
# --stay only lengthens the grace (STAY_MAX_S): a Steam
# restart ends a --stay unit too (review R1 M3).
grace = self.theme_grace_s()
if self.steam_restarted or grace <= 0 or (
self.dormant_since is not None and time.time() - self.dormant_since >= grace):
log("steam: theme is off" + (" after a Steam restart" if self.steam_restarted else
f" (dormant {grace:.0f} s)" if grace > 0 else "")
+ "; tearing down")
self.request_stop("steam theme off", strip_vr=True)
return
if self.dormant_since is None:
self.dormant_since = time.time()
log(f"steam: theme is off; dormant for up to {grace:.0f} s"
+ (" (--stay)" if self.stay else "")
+ " (native layer, glassd and reporter off; lgs off stops the daemon at once)")
self.changed.set()
# reporter installed later, changed on disk, or lost (page reload);
# stopped again when native mode ended (CSS only) or while dormant
if (tick % 3 == 0 or self.dormant_since is not None) and self.reporter_version and (
not self.native_possible() or self.dormant_since is not None):
r = await self.steam.eval(REPORTER_STOP_JS, 10)
log(f"steam: {'dormant' if self.native_possible() else 'native layer off'}; "
f"reporter stopped ({r})")
self.reporter_version = None
self.reporter_state = "stopped (dormant)" if self.native_possible() else "stopped (css only)"
self.report, self.fallback = None, None
elif self.dormant_since is not None:
pass
elif self.native_possible() and self.test_report:
self.read_test_report()
elif self.native_possible() and os.path.exists(LAYERS_JS):
if tick % 3 == 0 and self.reporter_key() != self.reporter_version:
await self.inject_reporter()
elif self.reporter_state == "stopped (reporter)":
log("steam: reporter stopped; re-injecting")
await self.inject_reporter()
elif self.reporter_state in ("injected", "reporting", "foreign"):
rs = await self.steam.eval(REPORTER_PING_JS, 10) # every tick: its heartbeat
if rs in ("absent", "stopped", "error") and tick % 3 == 0:
# page reloaded, it stopped itself (no pings or
# heartbeat while we stalled), or someone stopped it
log(f"steam: reporter {rs}; re-injecting")
await self.inject_reporter()
elif rs == "foreign" and self.reporter_state != "foreign":
self.note("steam: another lgs_layers instance (other binding) is running; "
"not replacing it; no layer reports meanwhile")
self.reporter_state = "foreign"
elif rs == "ok" and self.reporter_state == "foreign":
self.reporter_state = "injected"
# no reporter (or it never reported): glass for the main window only
if not self.native_possible() or self.test_report or self.dormant_since is not None:
pass
elif self.report is None and (self.reporter_state == "missing"
or time.time() - self.reporter_injected_at > 10):
size = await self.steam.eval(MAIN_SIZE_JS, 10)
fb = None
if size:
fb = {"seq": 0, "surfaces": [{"name": "main", "overlayKey": "valve.steam.gamepadui.main",
"visible": True, "texW": size[0], "texH": size[1],
"radius": 48, "layers": []}]}
if fb != self.fallback:
self.fallback = fb
self.changed.set()
elif self.report is not None:
self.fallback = None
self.write_glassd_json()
await self.bridge_tick()
slow = 0
except (asyncio.TimeoutError, TimeoutError):
# Steam's JS thread busy (another agent's audit, a runtime
# reload): one late answer is not a lost connection. Closing
# the socket here took the native layer down for 2-4 s.
slow += 1
if self.steam.closed or slow >= STEAM_SLOW_MAX:
raise
if slow == 1:
self.slow_evals += 1
log("steam: an evaluation timed out (Steam busy?); connection kept")
tick += 1
await asyncio.sleep(1.0)
except asyncio.CancelledError:
raise
except NET_ERRORS as e:
if not self.stopping:
self.note(f"steam: {e!r}")
if self.steam_ok:
delay = 0.5 # a connection that worked: try again at once
except Exception as e: # noqa: BLE001 - see bug()
if not self.stopping:
delay = self.bug("steam", e)
finally:
self.steam_ok = False
self.changed.set()
if self.steam:
await self.steam.close()
if not self.stopping:
await asyncio.sleep(delay)
async def native_loop(self):
"""html.lgs-native per window and the popped-layer acknowledgements:
sent when they change (within 0.1 s) and as a 1 s heartbeat while any
window is native. Cleared at once when systemui or glassd drops out."""
sent, sent_at = None, 0.0
while not self.stopping:
# 10 Hz only while there is a native layer to follow (CSS only: 1 Hz; review R1 m4)
await asyncio.sleep(0.1 if self.native_possible() or self.native_applied or self.native_force else 1.0)
s = self.steam
if not self.steam_ok or s is None or s.closed:
sent = None
continue
try:
keys, acks, plates, covers = self.native_state()
live = self.native_live()
payload = (keys, acks, plates, covers, live)
now = time.time()
if not self.native_force:
if payload == sent and (not (keys or live) or now - sent_at < NATIVE_BEAT_S):
continue
if sent is None and not keys and not live and not self.native_applied:
sent = payload # nothing native and nothing applied: no call needed
continue
self.native_force = False
r = await s.eval(f"({NATIVE_JS})({json.dumps(keys)}, {NATIVE_TTL_MS}, {json.dumps(acks)}, "
f"{json.dumps(plates)}, {json.dumps(covers)}, {json.dumps(live)})", 5)
res = json.loads(r) if isinstance(r, str) else {}
sent, sent_at = payload, now
applied = res.get("applied") or []
if applied != self.native_applied:
log(f"steam: lgs-native on {applied or 'no windows'}")
self.native_applied = applied
self.native_tagged = res.get("tagged")
if (acks, plates) != (self.acked, self.plate_acks):
self.bridge_values["acks"] = {"pops": acks, "plates": plates}
self.acked, self.plate_acks, self.cover_keys = acks, plates, covers
except asyncio.CancelledError:
raise
except NET_ERRORS as e:
if not self.stopping:
self.note(f"steam: lgs-native: {e!r}")
sent = None
# right after a stall of ours the late answer was ours: again at once (SG-6)
await asyncio.sleep(0 if self.stalled_recently() else 1)
except Exception as e: # noqa: BLE001 - see bug()
if not self.stopping:
sent = None
await asyncio.sleep(self.bug("lgs-native", e))
# ---------------------------------------------------------------- systemui
async def sysui_target(self):
loop = asyncio.get_running_loop()
ts = await loop.run_in_executor(None, lambda: http_json(lgs_vr.VR_CDP + "/json/list", 3))
for t in ts:
if t.get("type") == "page" and t.get("title") == "systemui":
return t
for t in ts:
if t.get("type") == "page" and "systemui" in (t.get("url") or ""):
return t
raise ConnectionError("no systemui page on 8090")
def on_sysui_context(self, params, cleared=False):
"""systemui's page context went away (reload) while the devtools socket
stays open: our nodes are gone with it. Drop lgs-native at once and
re-inject (sysui_loop)."""
ctx = (params or {}).get("executionContextId")
if cleared or (ctx is not None and ctx == self.sysui_ctx):
self.sysui_ctx = None
self.sg_beat, self.sg_beat_at = None, 0.0
self.sg_reset = True
self.changed.set()
def on_sysui_context_created(self, params):
c = (params or {}).get("context") or {}
if (c.get("auxData") or {}).get("isDefault"):
self.sysui_ctx = c.get("id")
@staticmethod
def sg_source_version():
"""lgs_sg.js and, when present, P5's motion.js (prepended): a change to
either re-injects."""
v = file_version(SG_JS)
m = file_version(MOTION_JS)
return f"{v}+{m}" if m else v
async def send_overrides(self, force=False):
"""theme/sg rules, filtered by flags (refresh_flags), to lgs_sg.js v2."""
if not force and self.sg_rules_sent == self.sg_rules_sig:
return
payload = json.dumps({"seq": self.spec_seq, "rules": self.sg_rules})
r = await self.sysui.eval(f"(window.__LGS_SG && typeof window.__LGS_SG.overrides === 'function') ? "
f"JSON.stringify(window.__LGS_SG.overrides({payload})) : 'unsupported'", 10)
self.sg_rules_sent = self.sg_rules_sig
if self.sg_rules or r != "unsupported":
log(f"systemui: overrides {[x.get('id') for x in self.sg_rules]} -> {str(r)[:200]}")
async def inject_sg(self, force=False):
with open(SG_JS, encoding="utf-8") as f:
src = f.read().strip().rstrip(";")
motion = None
if os.path.exists(MOTION_JS):
try:
with open(MOTION_JS, encoding="utf-8") as f:
motion = f.read()
except OSError:
motion = None
self.sg_version = self.sg_source_version()
opts = {"version": self.sg_version, "watchdogMs": WATCHDOG_MS, "maxPushHz": SG_MAX_PUSH_HZ}
if force:
opts["force"] = True
if motion:
# the depth channel: P5's springs in the same scope (contracts/sg.md §1)
opts["motion"] = True
expr = f"(() => {{\n{motion}\n;return ({src})({json.dumps(opts)});\n}})()"
else:
expr = f"({src})({json.dumps(opts)})"
r = await self.sysui.eval(expr, 20)
self.sg_rules_sent = None # a fresh instance: send the overrides again
st = json.loads(r) if isinstance(r, str) else r
self.sg_status = st
self.sg_injected_at = time.time()
self.sg_beat, self.sg_beat_at = None, 0.0
sch = (st or {}).get("scheduler") or {}
log(f"systemui: lgs_sg {self.sg_version} installed (scheduler module {sch.get('module')}, "
f"retire {sch.get('retire')}{', ' + sch['error'] if sch.get('error') else ''})")
self.spec_sent = None
def set_beat(self, b):
if isinstance(b, dict) and "surf" in b:
self.sg_beat, self.sg_beat_at = b, time.time()
# the part of each Steam texture its panel shows: kept (a surface hidden for a moment keeps its
# crop), a change rewrites glassd.json (the eye pair is built around it)
c = b.get("crops")
if isinstance(c, dict):
crops = dict(self.sg_crops or {})
for k, v in c.items():
if isinstance(k, str) and isinstance(v, list) and len(v) == 2:
crops[k] = [round(float(x), 3) for x in v if isinstance(x, (int, float))][:2]
if crops != self.sg_crops:
self.sg_crops = crops
self.changed.set()
# SteamVR Settings' panel (lgs_sg.js measures it while the page shows): its size and glass rect
# are kept, so glassd's "vrsettings" glass stays drawn ahead for the next time the page shows
v = b.get("vr")
shown = set()
for it in v if isinstance(v, list) else []:
if isinstance(it, dict) and isinstance(it.get("name"), str) and isinstance(it.get("glass"), dict) \
and it.get("material") in ("window", "panel", "thick", "liquid"):
name = it["name"][:24]
shown.add(name)
if self.vr_panels.get(name) != it:
self.vr_panels[name] = it
self.changed.set()
if shown != self.vr_shown:
self.vr_shown = shown
self.changed.set()
async def send_spec(self, spec, js):
self.spec_seq += 1
payload = json.dumps(dict(spec, seq=self.spec_seq), separators=(",", ":"))
r = await self.sysui.eval(f"window.__LGS_SG ? JSON.stringify(window.__LGS_SG.update({payload})) : null", 10)
if r is None:
await self.inject_sg()
r = await self.sysui.eval(f"JSON.stringify(window.__LGS_SG.update({payload}))", 10)
self.spec_sent, self.spec_sent_obj, self.spec_sent_at = js, spec, time.time()
res = json.loads(r) if r else {}
self.set_beat(res)
self.spec_summary = {"seq": self.spec_seq, "surfaces": len(spec["surfaces"]),
"popped": sum(len(s["popped"]) for s in spec["surfaces"]),
"items": res.get("items"), "counts": res.get("counts")}
# Log structural changes only, at most once a second (scrolling moves
# popped elements many times a second).
sig = (self.spec_summary["surfaces"], self.spec_summary["popped"], json.dumps(res.get("counts"), sort_keys=True))
if res.get("changed") and sig != self.spec_log_sig:
now = time.time()
if now - self.spec_logged_at >= SPEC_LOG_MIN_S:
more = f" (+{self.spec_log_skipped} more)" if self.spec_log_skipped else ""
why = "" if spec["surfaces"] else f" [{self.native_why() or 'no surface ready'}]"
log(f"systemui: spec -> {self.spec_summary}{more}{why}")
self.spec_logged_at, self.spec_log_sig, self.spec_log_skipped = now, sig, 0
else:
self.spec_log_skipped += 1
return res
async def sysui_loop(self):
while not self.stopping:
delay = 2
try:
if not self.sg_wanted(): # CSS only and no override or window state: leave systemui alone
await asyncio.sleep(1)
continue
ok = await asyncio.get_running_loop().run_in_executor(None, lgs_vr.available)
if not ok:
await asyncio.sleep(3)
continue
t = await self.sysui_target()
self.sysui = await CDP("systemui", t["webSocketDebuggerUrl"]).open()
self.sysui_ctx, self.sg_reset = None, False
self.sysui.handlers["Runtime.executionContextCreated"] = self.on_sysui_context_created
self.sysui.handlers["Runtime.executionContextDestroyed"] = self.on_sysui_context
self.sysui.handlers["Runtime.executionContextsCleared"] = lambda p: self.on_sysui_context(p, True)
await self.sysui.send("Runtime.enable")
await self.inject_sg()
self.sysui_ok = True
log("systemui: connected")
next_beat, next_status, it_at, slow = 0.0, 0.0, time.monotonic(), 0
while not self.stopping and not self.sysui.closed:
try:
now = time.time()
# wake for a layer that becomes still and for a fade-out that ends
# (its phase-0 item leaves glassd.json and slabsOut on time)
fade_end = min((u for _, u in self.fading.values()), default=now + 0.5) + 0.01
until = min(next_beat, now + 0.5, self.wake_at or now + 0.5, fade_end)
try:
await asyncio.wait_for(self.changed.wait(), max(0.02, until - now))
except asyncio.TimeoutError:
pass
self.changed.clear()
if self.stopping:
break
gap, it_at = time.monotonic() - it_at, time.monotonic()
if gap > STALL_S and self.sg_wanted():
# we were stopped (SIGSTOP / SIGCONT) or starved: the heartbeat
# first, so lgs_sg.js rebuilds its last spec at once (REQ P7->P8, SG-6)
r = await self.sysui.eval("window.__LGS_SG ? JSON.stringify(window.__LGS_SG.ping()) : null", 10)
if r:
self.set_beat(json.loads(r))
log(f"systemui: heartbeat sent first after a {gap:.1f} s stall of ours")
next_beat = 0.0
if not self.sg_wanted(): # glassd gave up, its binary was removed, overrides off
await self.sysui.eval(SG_DESTROY_JS, 10)
log("systemui: nothing left to show (native off, no overrides); scene graph removed")
break
if self.sg_reset: # page reloaded under us
self.sg_reset = False
log("systemui: page context reset (reload?); re-injecting")
await asyncio.sleep(0.5) # let the new page come up
await self.inject_sg()
next_beat = 0.0
if self.sg_source_version() != self.sg_version:
await self.inject_sg()
next_beat = 0.0
await self.send_overrides()
self.write_glassd_json()
spec = self.build_spec()
js = json.dumps(spec, separators=(",", ":"))
if js != self.spec_sent:
wait = self.spec_sent_at + SPEC_MIN_INTERVAL - time.time()
if wait > 0:
await asyncio.sleep(wait)
spec = self.build_spec()
js = json.dumps(spec, separators=(",", ":"))
res = await self.send_spec(spec, js)
# look again right after lgs_sg.js pushed, to learn when
pin = res.get("pushIn")
next_beat = min(next_beat, time.time() + (pin if isinstance(pin, (int, float)) else 50) / 1000 + 0.04)
now = time.time()
if now >= next_beat:
r = await self.sysui.eval("window.__LGS_SG ? JSON.stringify(window.__LGS_SG.ping()) : null", 10)
if r is None:
log("systemui: __LGS_SG gone (page reloaded?); re-injecting")
await self.inject_sg()
next_beat = 0.0
continue
b = json.loads(r)
self.set_beat(b)
if b.get("push"):
self.sg_nosched = 0
elif now - self.sg_injected_at > min(60, 3 * 2 ** self.sg_nosched):
# installed before the page's bundle was ready (or the
# module moved): no scheduler, so nothing reaches the
# compositor and nothing is native. Retry, backing off.
self.sg_nosched += 1
self.note(f"systemui: scene-graph scheduler not found; re-injecting (try {self.sg_nosched})")
await self.inject_sg(force=True)
next_beat = time.time() + 1
continue
if b.get("rebuilt") or b.get("expired") or (not b.get("items") and spec["surfaces"]):
if b.get("rebuilt") or b.get("expired"):
self.note(f"systemui: scene-graph watchdog had fired (rebuilt {b.get('rebuilt')}); "
"sending the spec again")
self.spec_sent = None
self.changed.set()
# poll fast only while a push is pending shortly after a change
fast = b.get("pending") and now - self.spec_sent_at < 3
next_beat = now + (0.1 if fast else NATIVE_BEAT_S)
if now >= next_status:
next_status = now + 2
r = await self.sysui.eval("window.__LGS_SG ? JSON.stringify(window.__LGS_SG.status()) : null", 10)
if r:
self.sg_status = json.loads(r)
slow = 0
except (asyncio.TimeoutError, TimeoutError):
# an answer that was due while we were stopped: not a lost page.
# Keep the socket and ping at once (REQ P7->P8, SG-6); twice in a row
# without a stall drops it as before
slow += 1
if self.sysui.closed or (slow >= 2 and not self.stalled_recently()):
raise
log("systemui: an evaluation timed out" + (" across a stall of ours" if self.stalled_recently()
else " (systemui busy?)") + "; connection kept")
next_beat, it_at = 0.0, time.monotonic()
except asyncio.CancelledError:
raise
except NET_ERRORS as e:
if not self.stopping:
self.note(f"systemui: {e!r}")
except Exception as e: # noqa: BLE001 - see bug()
if not self.stopping:
delay = self.bug("systemui", e)
finally:
self.sysui_ok = False
self.sg_beat, self.sg_beat_at = None, 0.0
if self.sysui:
await self.sysui.close()
if not self.stopping:
await asyncio.sleep(delay)
# ---------------------------------------------------------------- glassd
async def glassd_out_loop(self):
last_check = 0.0
while not self.stopping:
await asyncio.sleep(0.1 if self.native_possible() or self.glassd_running() else 1.0) # CSS only: 1 Hz
now = time.time()
if now - last_check >= 1.0:
last_check = now
try:
await self.supervise_glassd(now)
except asyncio.CancelledError:
raise
except Exception as e: # noqa: BLE001 - see bug()
await asyncio.sleep(self.bug("glassd supervisor", e))
try:
m = os.stat(GLASSD_OUT).st_mtime_ns
except OSError:
if self.gout is not None:
self.gout = None
self.changed.set()
continue
if m == self.gout_mtime:
continue
self.gout_mtime = m
try:
with open(GLASSD_OUT, encoding="utf-8") as f:
d = json.load(f)
except (OSError, ValueError):
continue # mid-write; next poll
if not isinstance(d, dict) or not self.glassd_running():
continue
self.gout_seen = now
if self.g_stale:
self.g_stale = False
log("glassd: output fresh again")
was_ready = self.glassd_ready()
self.gout = d
if not self.g_frames and ((_num(d.get("fps") or 0) or 0) > 0 or (_num(d.get("frames") or 0, 0, 1e15) or 0) > 0):
self.g_frames = True
log(f"glassd: producing frames (fps {d.get('fps')}, gpu {d.get('gpu_ms')} ms); native mode")
if was_ready != self.glassd_ready():
log(f"glassd: {'ready' if self.glassd_ready() else 'not ready'} (healthy {d.get('healthy')}, "
f"exiting {d.get('exiting', False)})")
self.changed.set()
def stalled_recently(self, within=3.0):
"""True while our own event loop is (or just was) stalled: SIGSTOP /
SIGCONT (NAT freeze), a starved CPU. Read before stall_loop has run
again, it still sees the gap (loop_beat is old)."""
now = time.monotonic()
return now - self.loop_beat > STALL_S or now - self.stall_at < within
async def stall_loop(self):
"""Notices that the daemon itself did not run (REQ P7->P8, SG-6): the
silence of glassd and the late Steam / systemui answers during that
time are ours, so glassd is not restarted as hung, and systemui gets
its heartbeat first (sysui_loop)."""
self.loop_beat = time.monotonic()
while not self.stopping:
await asyncio.sleep(0.25)
now = time.monotonic()
gap = now - self.loop_beat
self.loop_beat = now
if gap > STALL_S:
self.stall_at, self.stalls, self.last_stall_s = now, self.stalls + 1, round(gap, 1)
self.gout_seen = time.time() # glassd gets a fresh window to show it is alive
log(f"daemon: our event loop stalled {gap:.1f} s (stopped or starved); not held against "
"glassd or Steam")
self.changed.set()
async def supervise_glassd(self, now):
"""Once a second: binary removed -> CSS only; dormant -> glassd stopped
(glassd_loop starts it again when the theme is back); output stale ->
restart (not right after a stall of our own)."""
if not self.native_possible():
if self.glassd_running():
await self.kill_glassd()
if self.gseq and not self.g_files_removed: # native mode ended mid-session
self.g_files_removed = True
remove_glassd_files()
# only the binary was missing (an install swapping the folder under us, a toggle run before the
# install finished): native comes back as soon as a glassd is in place again
if self.native_requested and self.glassd_path and str(self.css_reason or "").startswith(
("glassd binary removed", "no glassd binary at start")) and ensure_glassd(self.glassd_path):
log("glassd: binary in place again; native layer back on")
self.native_enabled, self.css_reason = True, None
self.g_files_removed = False
self.g_given_up = False
self.tasks.append(asyncio.create_task(self.glassd_loop()))
self.changed.set()
return
if not os.access(self.glassd_path, os.X_OK) and not ensure_glassd(self.glassd_path):
self.disable_native("glassd binary removed")
await self.kill_glassd()
return
if self.dormant_since is not None:
# theme off without lgs off: nothing to draw, so no camera reader and
# no render loop while we wait (review R1 M2); not counted as a crash
if self.glassd_running():
await self.kill_glassd("dormant")
remove_glassd_files()
self.gjson_full = self.gjson_struct = None
return
if not self.glassd_running():
return
if self.stalled_recently():
# we were stopped (SIGSTOP) or starved, not glassd: its output time
# restarts now (it rewrites glassd-out.json within 2 s if alive)
self.gout_seen = max(self.gout_seen, time.time())
return
# The real glassd rewrites glassd-out.json every 2 s ("updated"); the
# stand-in (fakeglassd) only on changes, so staleness is not checked for it.
if self.gout and "updated" in self.gout and now - self.gout_seen > GLASSD_STALE_S:
if not self.g_stale:
self.g_stale = True
self.changed.set()
self.note(f"glassd: no output for {now - self.gout_seen:.0f} s (hung?); restarting it")
await self.kill_glassd()
elif not self.gout and now - self.g_started > GLASSD_FIRST_OUT_S:
self.note(f"glassd: no glassd-out.json after {GLASSD_FIRST_OUT_S:.0f} s; restarting it")
await self.kill_glassd()
async def pipe_glassd(self, proc):
while True:
try:
line = await proc.stdout.readline()
except ValueError: # a line over the stream limit (the reader dropped it)
print("glassd: [over-long output line dropped]", flush=True)
continue
if not line:
return
print("glassd: " + line.decode(errors="replace").rstrip(), flush=True)
async def glassd_loop(self):
if not self.native_possible():
log(f"CSS only + SteamVR page theming ({self.css_reason})")
return
temp_streak = 0
while not self.stopping and self.native_possible():
if not os.access(self.glassd_path, os.X_OK) and not ensure_glassd(self.glassd_path):
self.disable_native("glassd binary removed")
return
# glassd needs SteamVR and something to cover (and the theme on: not dormant)
if not (self.sysui_ok and self.steam_ok and self.theme_on and self.dormant_since is None
and self.current_report()):
await asyncio.sleep(1)
continue
for p in (GLASSD_OUT,):
try:
os.remove(p)
except OSError:
pass
self.gout, self.gout_mtime, self.g_frames, self.g_stale = None, None, False, False
self.write_glassd_json(force=True)
args = [self.glassd_path] + self.glassd_args + os.environ.get("LGS_GLASSD_ARGS", "").split()
try:
self.gproc = await asyncio.create_subprocess_exec(
*args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT,
stdin=asyncio.subprocess.DEVNULL, start_new_session=False, limit=1 << 20,
preexec_fn=_pdeathsig)
except OSError as e:
self.note(f"glassd: cannot start: {e}")
self.g_given_up = True
self.disable_native(f"glassd cannot start: {e}")
return
self.g_pid = self.gproc.pid
self.g_started = time.time()
self.gout_seen = self.g_started
log(f"glassd: started pid {self.g_pid} ({' '.join(args)})")
piper = asyncio.create_task(self.pipe_glassd(self.gproc))
rc = await self.gproc.wait()
await asyncio.gather(piper, return_exceptions=True)
self.g_last_exit = rc
self.g_frames = False
self.gout = None
self.changed.set()
if self.stopping or not self.native_possible():
return
if self.g_kill_reason == "dormant": # we stopped it: wait for the theme (top of the loop)
self.g_kill_reason = None
temp_streak = 0
continue
self.g_kill_reason = None
self.g_restarts += 1
if rc == GLASSD_TEMP_EXIT: # SteamVR went away, or another glassd holds the lock
self.g_temp_exits += 1
temp_streak += 1
if temp_streak == 1 or temp_streak % 10 == 0:
self.note(f"glassd: exited {rc} (temporary; SteamVR gone or lock held), "
f"retrying ({temp_streak} in a row)")
await asyncio.sleep(min(30, 3 * temp_streak))
continue
temp_streak = 0
now = time.time()
self.g_crashes = [t for t in self.g_crashes if now - t < GLASSD_CRASH_WINDOW] + [now]
self.note(f"glassd: exited with {rc} after {now - self.g_started:.1f} s")
if len(self.g_crashes) >= GLASSD_MAX_CRASHES:
self.g_given_up = True
self.disable_native(f"glassd gave up: {GLASSD_MAX_CRASHES} exits in {GLASSD_CRASH_WINDOW} s")
return
await asyncio.sleep(min(30, 2 ** len(self.g_crashes)))
async def kill_glassd(self, reason=None):
"""SIGTERM glassd (SIGKILL after 4 s). reason 'dormant' = on purpose,
not a crash (glassd_loop does not count it nor back off)."""
p = self.gproc
if not p or p.returncode is not None:
return
self.g_kill_reason = reason
if reason:
log(f"glassd: stopping it ({reason})")
try:
p.terminate()
await asyncio.wait_for(p.wait(), 4)
except asyncio.TimeoutError:
p.kill()
await p.wait()
except ProcessLookupError:
pass
log(f"glassd: stopped ({p.returncode})")
# ---------------------------------------------------------------- geometry and page (contract §4)
@staticmethod
def round_geom(g, prev):
def rn(v, nd):
f = _num(v)
return None if f is None else round(f, nd)
out = {"v": 1, "S": rn(g.get("S"), 4), "H0": rn(g.get("H0"), 4), "r": rn(g.get("r"), 4),
"Hm": rn(g.get("Hm"), 4), "unitM": rn(g.get("unitM"), 4), "mmPerCssPx": rn(g.get("mmPerCssPx"), 4),
"dashDist": rn(g.get("dashDist"), 3), "frameMenu": None}
fm = g.get("frameMenu")
if isinstance(fm, dict) and fm.get("key"):
m = {"key": fm["key"], "mpp": rn(fm.get("mpp"), 7), "visible": bool(fm.get("visible")),
"uv": [rn(v, 4) for v in fm.get("uv") or []]}
pf = (prev or {}).get("frameMenu") or {}
if fm.get("dumped"):
m["fWidth"], m["fHeight"] = rn(fm.get("fWidth"), 4), rn(fm.get("fHeight"), 4)
elif pf.get("key") == m["key"]:
m["fWidth"], m["fHeight"] = pf.get("fWidth"), pf.get("fHeight")
if m.get("fHeight") and m["mpp"]:
m["clipTexH"] = round(m["fHeight"] / m["mpp"], 1)
m["clipCssH"] = round(m["clipTexH"] / 1.5, 1)
if out["S"] and m["clipCssH"]:
m["mpMm"] = round(1000 * m["fHeight"] * out["S"] / m["clipCssH"], 3)
out["frameMenu"] = m
return out
async def geom_loop(self):
"""Read-only geometry and page from systemui, in both modes, published
on the bridge when they change (polled 1/s). DumpLaserOverlays (the
frame menu's panel size) runs 1, 3 and 6 s after the frame menu's node
or another value changed (the dump lags the node), and every
GEOM_DUMP_S as a safety net."""
loop = asyncio.get_running_loop()
conn = None
last_dump, dumps_at, fm_sig = 0.0, [0.0], None
while not self.stopping:
delay = 3
try:
if not await loop.run_in_executor(None, lgs_vr.available):
await asyncio.sleep(3)
continue
t = await self.sysui_target()
conn = await CDP("systemui-geom", t["webSocketDebuggerUrl"]).open()
self.geom_ok = True
dumps_at = [0.0] # a fresh connection: dump at once
while not self.stopping and not conn.closed:
now = time.time()
want = now - last_dump >= GEOM_DUMP_S or any(a <= now for a in dumps_at)
r = await conn.eval(f"({GEOM_JS})({'true' if want else 'false'})", 10)
d = json.loads(r) if isinstance(r, str) else {}
if want:
last_dump = now
dumps_at = [a for a in dumps_at if a > now]
if isinstance(d.get("geom"), dict) and ((d["geom"].get("frameMenu") or {}).get("dumped")
or not d["geom"].get("frameMenu")):
self.geom_dump_at = now
if d.get("fmSig") != fm_sig:
if fm_sig is not None:
dumps_at = [now + a for a in GEOM_DUMP_AFTER] # the panel may have changed size
fm_sig = d.get("fmSig")
self.geom_err = d.get("err")
if isinstance(d.get("geom"), dict):
g = self.round_geom(d["geom"], self.geom)
old = {k: v for k, v in (self.geom or {}).items() if k != "at"}
if g != old:
if self.geom is not None and not want and not dumps_at:
dumps_at = [now + a for a in GEOM_DUMP_AFTER] # something moved: the panel too
if (self.geom or {}).get("unitM") != g.get("unitM"):
self.changed.set() # glassd's unitM and the spec follow
self.geom = dict(g, at=int(now * 1000))
self.bridge_values["geom"] = self.geom
if isinstance(d.get("page"), dict):
p = dict(d["page"], v=1)
p["pages"] = {str(k): v for k, v in (p.get("pages") or {}).items()}
if p != {k: v for k, v in (self.page or {}).items() if k != "at"}:
if self.page is not None:
log(f"systemui: frame page {self.page.get('summonKey')} -> {p.get('summonKey')}")
if (self.page or {}).get("steam") is not p.get("steam"):
self.changed.set() # Steam's window shown / hidden: glassd.json and the spec follow
self.page = dict(p, at=int(now * 1000))
self.bridge_values["page"] = self.page
await asyncio.sleep(GEOM_POLL_S)
except asyncio.CancelledError:
raise
except NET_ERRORS as e:
if not self.stopping:
self.note(f"geometry: {e!r}")
except Exception as e: # noqa: BLE001 - see bug()
if not self.stopping:
delay = self.bug("geometry", e)
finally:
self.geom_ok = False
if self.page is not None:
if self.page.get("steam") is False:
self.changed.set() # unknown again: Steam's window is no longer held hidden
self.page = {"v": 1, "at": int(time.time() * 1000), "frameID": None, "activePageID": None,
"summonKey": None, "steam": None, "pages": {}}
self.bridge_values["page"] = self.page
if conn:
await conn.close()
conn = None
if not self.stopping:
await asyncio.sleep(delay)
# ---------------------------------------------------------------- SteamVR page scripts (contract §6)
def vr_paused(self):
"""A lab `--theme off` on a vr: page (lgs_vr.stop) pauses page theming and the page scripts. The
lab's `--theme on` (lgs_vr.apply_once) ends the pause; one older than VR_PAUSE_MAX_S is stale (a step
that turned the theme off and never on again, which left the pause for good): it is removed and
theming and the page scripts resume (REQ C6b->P8: a flagged page script was never installed)."""
try:
age = time.time() - os.stat(VR_PAUSE).st_mtime
except OSError:
if self.vr_pause_since is not None:
log(f"vr: page theming resumed (paused {time.time() - self.vr_pause_since:.0f} s)")
self.vr_pause_since = None
return False
if age > VR_PAUSE_MAX_S:
try:
os.remove(VR_PAUSE)
except OSError:
pass
log(f"vr: a page-theming pause {age:.0f} s old ended (a lab --theme off without its --theme on); "
"theming and page scripts resume")
self.vr_pause_since = None
return False
if self.vr_pause_since is None:
self.vr_pause_since = time.time() - age
log("vr: page theming paused (lab --theme off on a vr: page): page scripts removed with the theme")
return True
async def vr_scripts_for(self, t, scripts):
"""Install, refresh or remove the page scripts of one SteamVR page."""
name = t.get("title") or t.get("url")
# the flags files themselves (a stat, cached by mtime), not the Steam loop's copy: a lab step's
# --flags reaches the page within one poll even while Steam's loop is slow or reconnecting
flags = self.flagreader.read()
st = self.vrx_status.setdefault(name, {})
mine = {n: s for n, s in scripts.items() if s["page"] == name}
want = {n: s for n, s in mine.items() if not s["flag"] or flags.get(s["flag"]) is True}
if not want and not any(v in ("installed", "ok") for v in st.values()):
for n, s in mine.items():
st[n] = "flag-off"
for n in [n for n in st if n not in mine]:
del st[n]
return
r = await self.vr_page_eval(t, f"({VRX_PROBE_JS})({json.dumps({n: s['version'] for n, s in want.items()})})", 10)
probe = json.loads(r) if isinstance(r, str) else {}
extra = [n for n, v in probe.items() if v == "extra"]
if extra:
await self.vr_page_eval(t, f"({VRX_REMOVE_JS})({json.dumps(extra)})", 10)
log(f"vr: {name}: page scripts removed {extra}")
for n in extra:
st[n] = "flag-off" if n in mine else "removed"
for n in [n for n in st if n not in mine and n not in extra]:
del st[n]
for n, s in mine.items():
if n not in want:
st[n] = "flag-off"
for n, s in want.items():
if probe.get(n) == "ok":
st[n] = "ok"
continue
if self.vrx_failed.get((name, n)) == s["version"]:
continue
ctx = {"name": n, "page": name, "version": s["version"], "flags": dict(flags)}
res = await self.vr_page_eval(t, vrx_install_js(n, s["version"], ctx, s["src"]), 15)
st[n] = res if isinstance(res, str) else str(res)
if st[n].startswith("error"):
self.vrx_failed[(name, n)] = s["version"]
self.note(f"vr: {name}: page script {n} failed: {st[n][:160]}")
else:
self.vrx_failed.pop((name, n), None)
log(f"vr: {name}: page script {n} {st[n]} ({s['version']})")
async def vr_scripts_remove_all(self):
loop = asyncio.get_running_loop()
out = {}
try:
if not await loop.run_in_executor(None, lgs_vr.available):
return out
for t in await loop.run_in_executor(None, lgs_vr.targets):
try:
out[t.get("title") or t.get("url")] = await eval_once(
t["webSocketDebuggerUrl"], f"({VRX_REMOVE_JS})(null)", 10)
except Exception: # noqa: BLE001 - page closing
pass
except Exception as e: # noqa: BLE001
log(f"vr: removing page scripts: {e!r}")
self.vrx_status = {}
return out
# ---------------------------------------------------------------- SteamVR page theming
async def vr_list(self):
"""SteamVR's page targets (8090 /json/list) over one kept-alive HTTP
session (no new TCP connection per poll; review R1 m4), or None while
SteamVR's devtools are unreachable."""
import aiohttp
if self.vr_http is None or self.vr_http.closed:
self.vr_http = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=3))
try:
async with self.vr_http.get(lgs_vr.VR_CDP + "/json/list") as r:
ts = await r.json(content_type=None)
except NET_ERRORS:
return None
return [t for t in ts if isinstance(t, dict) and t.get("type") == "page"
and t.get("webSocketDebuggerUrl")] if isinstance(ts, list) else None
async def vr_page_eval(self, t, expr, timeout=10):
"""Evaluate in a SteamVR page over its kept socket (opened on first
use; dropped when it fails, times out or the page goes). A JS error
keeps the socket."""
tid = t.get("id") or t["webSocketDebuggerUrl"]
c = self.vr_conns.get(tid)
if c is None or c.closed:
c = await CDP("vr:" + str(t.get("title") or tid), t["webSocketDebuggerUrl"]).open()
self.vr_conns[tid] = c
try:
return await c.eval(expr, timeout)
except RuntimeError as e:
if c.closed or not str(e).startswith("JS: "):
self.vr_conns.pop(tid, None)
await c.close()
raise
except BaseException:
self.vr_conns.pop(tid, None)
await c.close()
raise
def vr_scripts_cached(self):
"""vr_scripts(), re-read only when a file in its folders changed."""
sig = []
for folder in VR_SCRIPT_DIRS:
try:
for n in sorted(os.listdir(folder)):
if n.endswith(".js"):
sig.append((folder, n, os.stat(os.path.join(folder, n)).st_mtime_ns))
except OSError:
continue
if sig != self.vrx_sig:
self.vrx_sig, self.vrx_scripts = sig, vr_scripts()
return self.vrx_scripts
async def vr_conns_close(self, keep=()):
for tid in [k for k in self.vr_conns if k not in keep]:
await self.vr_conns.pop(tid).close()
async def vr_theme_loop(self):
sig, full, probe = None, None, None
while not self.stopping:
await asyncio.sleep(lgs_vr.POLL)
if self.vr_paused(): # lab "--theme off" on a vr: page (page scripts stripped with it)
self.vrx_status, self.vr_pages, self.vr_targets = {}, {}, {}
await self.vr_conns_close()
continue
try:
ts = await self.vr_list()
if ts is None:
self.vr_targets = {}
await self.vr_conns_close()
continue
if self.pod: # no page theming: the assPod's page script only
self.vr_targets = {(t.get("title") or t.get("url")): t for t in ts}
scripts = {n: v for n, v in self.vr_scripts_cached().items() if n in POD_SCRIPTS}
await self.vr_conns_close(keep={t.get("id") or t["webSocketDebuggerUrl"] for t in ts})
pages = {}
for t in ts:
name = t.get("title") or t.get("url")
try:
await self.vr_scripts_for(t, scripts)
pages[name] = "pod"
except Exception: # noqa: BLE001 - page reloading or closing
pages[name] = "error"
self.vr_pages = pages
for gone in [n for n in self.vrx_status if n not in pages]:
del self.vrx_status[gone]
continue
s = lgs_vr.signature()
if s != sig:
sig = s
p = lgs_vr.payload()
full = lgs_vr.core_call(p)
probe = (f"(window.__LGS_VR && window.__LGS_VR.version === {json.dumps(p['version'])})"
" ? (window.__LGS_VR.apply() ? 'ok' : 'retry') : 'need'")
self.vr_version = p["version"]
log(f"vr: theme {p['version']} ({len(p['css'])} bytes)")
self.vr_targets = {(t.get("title") or t.get("url")): t for t in ts}
pages = {}
scripts = self.vr_scripts_cached()
await self.vr_conns_close(keep={t.get("id") or t["webSocketDebuggerUrl"] for t in ts})
for t in ts:
name = t.get("title") or t.get("url")
try:
r = await self.vr_page_eval(t, probe, 10)
if r == "need":
r = await self.vr_page_eval(t, full, 30)
log(f"vr: {name}: {r}")
pages[name] = r
if r in ("ok", "applied"): # page scripts go in after the page's theme
await self.vr_scripts_for(t, scripts)
except Exception: # noqa: BLE001 - page reloading or closing
pages[name] = "error"
self.vr_pages = pages
for gone in [n for n in self.vrx_status if n not in pages]:
del self.vrx_status[gone]
except asyncio.CancelledError:
raise
except Exception as e: # noqa: BLE001 - best effort, like lgs-vr
self.note(f"vr: {e!r}")
# ---------------------------------------------------------------- pod mode
async def pod_loop(self):
"""--pod: end once the assPod is away and nothing plays (POD_IDLE_S), when it was never taken out
(POD_FIRST_OPEN_S), or at once after a Steam restart (its pod runtime went with the old page)."""
opened, idle_since = False, None
while not self.stopping:
await asyncio.sleep(1.0)
if self.steam_restarted:
log("pod: Steam restarted; exiting")
self.request_stop("pod: steam restarted", strip_vr=True)
return
p = self.registry.plugins.get("asspod") or {}
st = p["ctx"].state if p.get("ctx") is not None else {}
busy = bool(st.get("open") or st.get("playing") or st.get("starting"))
opened = opened or bool(st.get("open"))
now = time.time()
if busy:
idle_since = None
continue
if not opened:
if now - self.since >= POD_FIRST_OPEN_S:
log(f"pod: the assPod was not taken out within {POD_FIRST_OPEN_S:.0f} s; exiting")
self.request_stop("pod: never opened", strip_vr=True)
return
continue
idle_since = idle_since or now
if now - idle_since >= POD_IDLE_S:
log(f"pod: the assPod has been away with nothing playing for {POD_IDLE_S:.0f} s; exiting")
self.request_stop("pod: idle", strip_vr=True)
return
# ---------------------------------------------------------------- status
async def status_loop(self):
while not self.stopping:
# tune.json written by someone else (`lgs dial`): taken over; our own saves set tune_mtime
tm = _mtime(lgs.TUNE)
if tm != self.tune_mtime and self.tune_save is None:
self.tune_mtime = tm
self.set_tune(lgs.read_tune(), save=False)
m = _mtime(lgs.DIAL)
if m != self.dial_mtime:
self.dial_mtime = m
self.dial_v = lgs.read_dial()
self.write_glassd_json()
try:
write_json_atomic(STATUS, self.status())
except OSError:
pass
await asyncio.sleep(2)
# ---------------------------------------------------------------- lifecycle
def request_stop(self, reason, strip_vr=False, code=0):
"""Start the shutdown once (signal handlers, theme off, fatal errors)."""
if self.stop_task is None:
self.stop_task = asyncio.get_running_loop().create_task(self.stop(reason, strip_vr, code))
return self.stop_task
async def stop(self, reason, strip_vr=False, code=0):
# Cancel the workers first, so a late spec push cannot re-create the
# nodes that teardown removes; teardown uses fresh devtools sessions.
self.stopping = True
self.exit_code = code
log(f"stopping: {reason}")
try: # tells "lgs on" (start()) to wait for us instead of calling us running
st = self.status()
st.update(mode="stopping", stopping=reason)
write_json_atomic(STATUS, st)
except OSError:
pass
self.changed.set()
for t in self.tasks:
t.cancel()
await asyncio.gather(*self.tasks, return_exceptions=True)
await self.teardown(strip_vr)
async def teardown(self, strip_vr=False):
if self.torn:
return
self.torn = True
loop = asyncio.get_running_loop()
try: # the kept SteamVR page sockets and 8090 HTTP session (teardown uses fresh ones)
await self.vr_conns_close()
if self.vr_http is not None:
await self.vr_http.close()
except Exception as e: # noqa: BLE001
log(f"teardown: closing page sockets: {e!r}")
# 0. plugins undo what they did (e.g. a deep link switched SteamVR's
# frame page), then the SteamVR page scripts come out
try:
await asyncio.wait_for(self.registry.stop(), 15)
log("teardown: action plugins stopped")
except Exception as e: # noqa: BLE001
log(f"teardown: plugins: {e!r}")
for t in list(self.action_tasks):
t.cancel()
r0 = await self.vr_scripts_remove_all()
if any(r0.values()):
log(f"teardown: page scripts removed {r0}")
# 1. scene graph: back to Steam's own panels (also in CSS-only mode:
# transform overrides restore with destroy())
try:
if await loop.run_in_executor(None, lgs_vr.available):
t = await self.sysui_target()
r = await eval_once(t["webSocketDebuggerUrl"],
SG_DESTROY_JS, 10)
log(f"teardown: scene graph {r}")
except Exception as e: # noqa: BLE001
log(f"teardown: scene graph: {e!r}")
# 2. Steam: lgs-native off, reporter off, binding removed
try:
t = await self.steam_target()
s = await CDP("steam-teardown", t["webSocketDebuggerUrl"]).open()
try:
r = await s.eval("window.__LGS_NATIVE ? JSON.stringify(window.__LGS_NATIVE.clear()) : 'absent'", 10)
rr = await s.eval(REPORTER_STOP_JS, 10)
for b in (BINDING, ACTION_BINDING):
try:
await s.send("Runtime.removeBinding", {"name": b})
except RuntimeError:
pass
# the bridge learns the daemon is gone at once (not after its TTL)
gone = dict(self.daemon_value(), mode="stopped", native=False, actions=[], steamvr=False, ttlMs=0)
await s.eval(f"({BRIDGE_SET_JS})({json.dumps([['daemon', gone]])})", 5)
# the inert binding functions (both: removeBinding leaves them):
# gone, so callers see no daemon and nothing of ours stays (G-REMOVE)
rb = await s.eval(f"({DELETE_BINDINGS_JS})({json.dumps([BINDING, ACTION_BINDING])})", 5)
if self.pod: # the assPod's own runtime (lgs pod) goes with us; a full one stays
rp = await s.eval(POD_RT_TEARDOWN_JS, 30)
log(f"teardown: pod runtime {rp}")
finally:
await s.close()
log(f"teardown: lgs-native {r}, reporter {rr}, binding globals deleted [{rb}]")
except Exception as e: # noqa: BLE001
log(f"teardown: steam: {e!r}")
# 3. glassd
await self.kill_glassd()
remove_glassd_files()
# 4. SteamVR pages (only when the Steam theme went away)
if strip_vr:
try:
await loop.run_in_executor(None, lgs_vr.strip)
log("teardown: SteamVR pages stripped")
except Exception as e: # noqa: BLE001
log(f"teardown: strip: {e!r}")
try:
st = self.status()
st.update(mode="stopped", stopped=time.time())
write_json_atomic(STATUS, st)
except OSError:
pass
async def run(self):
loop = asyncio.get_running_loop()
for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
loop.add_signal_handler(sig, lambda s=sig: self.request_stop(signal.Signals(s).name))
os.makedirs(SHM, exist_ok=True)
log(f"lgs-shell starting (pid {os.getpid()}, " + ("pod mode: the assPod alone, no theme" if self.pod else
(f"native, glassd {self.glassd_path} {' '.join(self.glassd_args)}".rstrip() if self.native_enabled
else f"css only: {self.css_reason}")) + ")")
self.refresh_flags()
self.tasks = [loop.create_task(c) for c in (
self.stall_loop(), self.steam_loop(), self.native_loop(), self.sysui_loop(), self.glassd_loop(),
self.glassd_out_loop(),
self.vr_theme_loop(), self.status_loop(), self.geom_loop(), self.actions_loop())]
if self.pod:
self.tasks.append(loop.create_task(self.pod_loop()))
pending = set(self.tasks)
while self.stop_task is None and pending:
done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
if self.stop_task is not None:
break
for t in done:
if not t.cancelled() and t.exception() is not None: # a bug, not a lost connection
log(f"fatal: {t.exception()!r}")
self.request_stop(f"fatal {t.exception()!r}", code=1)
break
# a loop that ends normally (CSS only, glassd given up) is fine
if self.stop_task is None:
self.request_stop("all loops ended")
await self.stop_task
for c in list(CDP.live): # sockets a cancelled task left open
await c.close()
await asyncio.sleep(0)
log("lgs-shell exited")
return self.exit_code
# ------------------------------------------------------------------ unit control
def user_env():
return lgs_vr.user_env()
def unit_info():
"""(ActiveState, MainPID) of the unit."""
r = subprocess.run(["systemctl", "--user", "show", "-p", "ActiveState", "-p", "MainPID", UNIT],
capture_output=True, text=True, env=user_env())
d = dict(line.split("=", 1) for line in r.stdout.splitlines() if "=" in line)
try:
pid = int(d.get("MainPID") or 0)
except ValueError:
pid = 0
return d.get("ActiveState") or "inactive", pid
def unit_active():
return unit_info()[0] in ("active", "activating", "reloading")
def daemon_argv(pid):
try:
with open(f"/proc/{pid}/cmdline", "rb") as f:
return [a.decode(errors="replace") for a in f.read().split(b"\0") if a]
except OSError:
return []
def daemon_stopping(pid):
st = read_json(STATUS) or {}
return bool(pid) and st.get("pid") == pid and st.get("mode") in ("stopping", "stopped")
def settle(timeout=20.0):
"""Wait while the unit is stopping (deactivating, or its daemon tearing
itself down after the theme went off). Returns (state, pid)."""
end = time.time() + timeout
while True:
state, pid = unit_info()
if not (state == "deactivating" or (state in ("active", "activating") and daemon_stopping(pid))):
return state, pid
if time.time() > end:
subprocess.run(["systemctl", "--user", "stop", UNIT], capture_output=True, env=user_env())
return unit_info()
time.sleep(0.25)
def stay_arg(argv):
return "--stay" in argv
def native_arg(argv):
"""--native: opt in to the native layer; --css: CSS only; neither: None."""
return True if "--native" in argv else (False if "--css" in argv else None)
def glassd_args_arg(argv):
"""--glassd-args "ARGS": extra glassd arguments (tests: --no-feed, --key-prefix)."""
if "--glassd-args" in argv:
return argv[argv.index("--glassd-args") + 1].split()
return None
def glassd_arg(argv):
if "--glassd" in argv:
v = argv[argv.index("--glassd") + 1]
return None if v == "none" else os.path.abspath(v)
return GLASSD_BIN
NATIVE_MODES = ("auto", "on", "off") # start(native=...) takes these (P1 checks for it)
def core_args(argv):
"""A daemon's arguments without the bookkeeping ones (--req, --src)."""
out, skip = [], False
for a in argv:
if skip:
skip = False
continue
if a in ("--req", "--src"):
skip = True
continue
out.append(a)
return out
GLASSD_PREBUILT = os.path.join(lgs.ROOT, "native", "glassd", "prebuilt", "glassd")
def ensure_glassd(path=GLASSD_BIN):
"""The shipped glassd (native/glassd/prebuilt/glassd) in place when no build is there yet, so native
glass works out of the box and a Liquid Glass toggle that runs before a build finishes is never stuck
CSS-only. True when an executable glassd is in place."""
if os.access(path, os.X_OK):
return True
if path != GLASSD_BIN or not os.access(GLASSD_PREBUILT, os.R_OK):
return False
try:
tmp = path + ".tmp"
shutil.copyfile(GLASSD_PREBUILT, tmp)
os.chmod(tmp, 0o755)
os.replace(tmp, path)
log("glassd: no build here; the shipped binary is in place")
return True
except OSError as e:
log(f"glassd: could not put the shipped binary in place: {e}")
return False
def start(glassd=GLASSD_BIN, stay=False, glassd_args=None, native=None, test_report=None, assume_caps=None, pod=False):
"""Start the unit. Also resumes SteamVR page theming that a lab "--theme
off" on a vr: page paused.
native: 'on' / True runs the native layer, 'off' / False CSS only; both
restart a unit that runs otherwise. None or 'auto' keeps a running unit as
it is, whatever its mode (a plain `lgs on` never ends another agent's
native session); a new unit resolves auto (resolve_native: flags.json,
then defaults.json, else CSS only). A unit that is stopping is waited for
and then started fresh. stay=True (lab testing only) keeps it running,
dormant, while the Steam theme is off. test_report (lab) replaces the
reporter with a report JSON file. pod=True (lgs pod, theme off) runs pod
mode: a running pod unit is kept, any other unit is replaced by one; a
running pod unit is replaced by a full one for any other request.
Serialized by START_LOCK: two `lgs on` at once (one lock-free) made the
loser's systemd-run fail with "already loaded" although a unit ran (review
R1 m5); if the lock cannot be had in 60 s, it goes on and treats that
error as running."""
if glassd and glassd != "none":
ensure_glassd(glassd)
try:
lk = FileLock(START_LOCK, 60).__enter__()
except (OSError, TimeoutError):
lk = None
try:
return _start(glassd, stay, glassd_args, native, test_report, assume_caps, pod)
finally:
if lk is not None:
lk.__exit__()
def _start(glassd, stay, glassd_args, native, test_report, assume_caps, pod=False):
if pod:
native = "off"
req = "auto" if native is None else native_word(native)
try:
os.remove(VR_PAUSE)
except OSError:
pass
# the pre-native watcher, if an older lgs started it
subprocess.run(["systemctl", "--user", "stop", lgs_vr.UNIT], capture_output=True, env=user_env())
def daemon_args(nat):
return ["--glassd", (glassd if nat and glassd else "none")] + (["--native"] if nat else []) + \
(["--stay"] if stay else []) + (["--glassd-args", " ".join(glassd_args)] if nat and glassd_args else []) + \
(["--test-report", os.path.abspath(test_report)] if test_report else []) + \
(["--assume-caps", ",".join(assume_caps)] if assume_caps else []) + \
(["--pod"] if pod else [])
state, pid = settle()
if state in ("active", "activating", "reloading"):
argv = daemon_argv(pid)
cur = "--native" in argv
if pod and "--pod" in argv:
return "running (pod)"
was_pod = "--pod" in argv
running = core_args(argv[argv.index("daemon") + 1:]) if "daemon" in argv else None
# an explicit request restarts a unit that runs differently
st = read_json(STATUS) or {}
stay_over = req == "on" and st.get("pid") == pid and \
str((st.get("native") or {}).get("reason") or "").startswith("--stay limit")
if (req == "auto" or running == daemon_args(req == "on")) and not stay_over and not was_pod and not pod:
if cur and st.get("pid") == pid and not (st.get("native") or {}).get("enabled", True):
return f"running (native requested; {st.get('mode')})"
return "running (native)" if cur else "running (css only)"
stop() # (a --stay unit past its limit is restarted for a new explicit native request)
settle()
nat, src, reason = resolve_native(req)
subprocess.run(["systemctl", "--user", "reset-failed", UNIT], capture_output=True, env=user_env())
cmd = ["systemd-run", "--user", "--unit", UNIT, "--collect", "--quiet",
"-p", "KillMode=mixed", "-p", "TimeoutStopSec=15", "-E", "PYTHONDONTWRITEBYTECODE=1",
"--description=Glass Shell: native glass layer + SteamVR page theming (transient)",
"/usr/bin/python3", "-B", "-u", os.path.realpath(__file__), "daemon"] + daemon_args(nat) + \
["--req", req, "--src", src]
r = subprocess.run(cmd, capture_output=True, text=True, env=user_env())
if r.returncode != 0:
if "already loaded" in r.stderr or "already exists" in r.stderr:
# another start won a race (one without the start lock): a unit runs
state, pid = settle()
if state in ("active", "activating", "reloading"):
cur = "--native" in daemon_argv(pid)
return "running (native, started concurrently)" if cur else "running (css only, started concurrently)"
return f"failed: {r.stderr.strip()}"
if pod:
return "started (pod)"
if nat and not (glassd and os.access(glassd, os.X_OK)):
return "started (css only: native requested but no glassd binary)"
return "started (native)" if nat else "started (css only)"
def stop():
"""Stop the unit; its SIGTERM handler removes every node, class and file."""
was = unit_active()
subprocess.run(["systemctl", "--user", "stop", UNIT], capture_output=True, env=user_env())
subprocess.run(["systemctl", "--user", "stop", lgs_vr.UNIT], capture_output=True, env=user_env())
return "stopped" if was else "not running"
def status():
state, pid = unit_info()
out = {"unit": "active" if state in ("active", "activating", "reloading") else state}
st = read_json(STATUS)
if isinstance(st, dict) and (out["unit"] == "active" or st.get("mode") == "stopped"):
age = time.time() - st.get("updated", 0)
out.update(st)
out["statusAgeS"] = round(age, 1)
return out
def brief(st):
"""One-level summary for lgs status."""
if st.get("unit") != "active":
return {"unit": st.get("unit")}
g, s, u = st.get("glassd", {}), st.get("steam", {}), st.get("systemui", {})
a, geo = st.get("actions") or {}, st.get("geom") or {}
return {"unit": "active", "mode": st.get("mode"), "native": st.get("native"), "dormant": st.get("dormant"),
"glassd": {k: g.get(k) for k in ("running", "ready", "pid", "restarts", "fps", "gpu_ms", "surfaces",
"caps")},
"layers": {"source": s.get("source"), "reporter": s.get("reporter"), "surfaces": s.get("surfaces")},
"lgsNative": s.get("lgsNative"),
"sceneGraph": {"items": u.get("items"), "panels": u.get("panels"), "attached": u.get("attached"),
"overrides": (st.get("sgRules") or {}).get("active")},
"steamvrPages": len((st.get("steamvrPages") or {}).get("pages") or {}),
"vrScripts": st.get("vrScripts"),
"bridge": (st.get("bridge") or {}).get("runtime"),
"geom": {k: geo.get(k) for k in ("S", "r", "Hm", "unitM")} if geo else None,
"page": (st.get("page") or {}).get("summonKey"),
"actions": {"types": a.get("types"), "calls": a.get("calls"), "failed": a.get("failed")},
"profile": st.get("profile"),
"errors": (st.get("errors") or [])[-3:]}
# ------------------------------------------------------------------ lab self-tests (contract §11)
TEST_SCRIPT_JS = r"""// lgs_shell.py selftest dm5: a test page script (RAM only; the test removes it)
(function (ctx) {
const el = document.documentElement;
el.dataset.lgsEcho = ctx.version;
return { remove() { delete el.dataset.lgsEcho; }, status() { return { echo: el.dataset.lgsEcho || null }; } };
})
"""
TEST_SCRIPT = "/tmp/lgs/vr-scripts/systemui.p8echo.js"
# the SP §1.2 snippet, with Steam's page found by its summon key (REQ P7->P8: page ids change at run time)
SP12_SNIPPET = ("(()=>{let f=null,p=null; for (const x of FrameStore.frames) { for (const q of x.pages.values()) "
"if (q.m_sSummonOverlayKey==='valve.steam.gamepadui.main') { f=x; p=q; } } "
"const z=p.size, S=DashboardStore.dashboardScale, "
"H0=f.size.mainPanelHeightOverride||1.5; const r=z.latestMeasuredPanelLocalHeight/H0, "
"Hw=z.latestMeasuredPanelWorldHeight; return JSON.stringify({S,H0,r,Hw,unitToM:S*r,"
"mmPerCssPx:1500*Hw/1080,dashDist:DashboardStore.dashboardDistance});})()")
FAKEGLASSD = os.path.join(lgs.ROOT, "native", "spike", "fakeglassd")
class FileLock:
"""flock with a timeout (the lab's lock files; order native -> lab -> lab-vr)."""
def __init__(self, path, timeout):
self.path, self.timeout, self.f = path, timeout, None
def __enter__(self):
import fcntl
os.makedirs(os.path.dirname(self.path), exist_ok=True)
self.f = open(self.path, "a")
end = time.time() + self.timeout
while True:
try:
fcntl.flock(self.f, fcntl.LOCK_EX | fcntl.LOCK_NB)
return self
except OSError:
if time.time() > end:
self.f.close()
raise TimeoutError(f"lock busy: {self.path}")
time.sleep(0.5)
def __exit__(self, *exc):
import fcntl
fcntl.flock(self.f, fcntl.LOCK_UN)
self.f.close()
def _wait(pred, timeout, step=0.25):
end = time.time() + timeout
while True:
v = pred()
if v or time.time() > end:
return v
time.sleep(step)
def _vr_js(page, expr, timeout=10):
for t in lgs_vr.targets():
if t.get("title") == page:
return asyncio.run(eval_once(t["webSocketDebuggerUrl"], expr, timeout))
raise RuntimeError(f"no SteamVR page {page!r}")
def _steam_js(expr, timeout=10):
return lgs.run_js("SharedJSContext", expr, timeout)
def _close(a, b, tol):
try:
return abs(float(a) - float(b)) <= tol
except (TypeError, ValueError):
return False
def _journal(since):
r = subprocess.run(["journalctl", "--user", "-u", UNIT, "--no-pager", "-o", "cat", "--since", since],
capture_output=True, text=True, env=user_env())
return r.stdout.splitlines()
TEST_REPORT = "/tmp/lgs/p8-test-report.json"
TEST_DUMP = "/tmp/lgs/p8-dump"
MAIN_NATIVE_JS = ("(() => { try { for (const p of g_PopupManager.m_mapPopups.values()) if (/^VR_uid/.test(p.m_strName)) "
"return p.window.document.documentElement.classList.contains('lgs-native'); } catch (e) {} return null; })()")
def _main_native():
try:
return _steam_js(MAIN_NATIVE_JS, 5)
except Exception: # noqa: BLE001
return None
def _bindings():
"""typeof the daemon's binding globals in SharedJSContext."""
try:
return json.loads(_steam_js(f"JSON.stringify({{{BINDING}: typeof window.{BINDING}, "
f"{ACTION_BINDING}: typeof window.{ACTION_BINDING}}})", 5))
except Exception as e: # noqa: BLE001
return {"error": repr(e)}
def _alive(pid):
try:
os.kill(int(pid), 0)
return True
except (OSError, TypeError, ValueError):
return False
def _sg_items():
try:
r = _vr_js("systemui", "window.__LGS_SG ? JSON.stringify((s => ({items: s.items, expired: s.expired}))"
"(__LGS_SG.status())) : null", 5)
return json.loads(r) if r else None
except Exception: # noqa: BLE001
return None
def _main_size():
try:
r = _steam_js(MAIN_SIZE_JS, 5)
if isinstance(r, list) and len(r) == 2:
return r
except Exception: # noqa: BLE001
pass
return [1920, 1080]
def _native_start(report, caps=None, gargs=None, glassd=None):
write_json_atomic(TEST_REPORT, report)
print("start native:", start(glassd=glassd or FAKEGLASSD, stay=True, glassd_args=gargs or [], native="on",
test_report=TEST_REPORT, assume_caps=caps), flush=True)
st = _wait(lambda: (lambda s: s if s.get("mode") == "native" else None)(status()), 45, 0.5)
return st or status()
def _report_window(W, H, layers=True):
return {"seq": 1, "dash": True, "surfaces": [
{"name": "main", "overlayKey": STEAM_MAIN_KEY, "visible": True, "texW": W, "texH": H, "radius": 81,
"material": "window", "shapes": [{"x": 0, "y": 0, "w": W, "h": round(H * 656 / 720), "r": 81}],
"layers": [{"id": "p8-cap", "x": round(W * 0.42), "y": round(H * 0.05), "w": 300, "h": 90, "r": 45,
"dz": 0.0271, "material": "liquid"}] if layers else []}]}
def _journal_times(lines, needle):
out = []
for x in lines:
if needle in x:
try:
h, m, s = x[:8].split(":")
out.append(int(h) * 3600 + int(m) * 60 + int(s))
except ValueError:
pass
return out
def _selftest_dm1(check, args):
"""NAT regressions with fakeglassd: freeze (SIGSTOP 15 s), teardown order,
restart backoff and give-up."""
import signal as sg
W, H = _main_size()
st = _native_start(_report_window(W, H))
check("native mode with fakeglassd", st.get("mode") == "native", st.get("mode"))
check("lgs-native on main", _wait(lambda: _main_native() is True, 20, 0.5))
s0 = _sg_items()
check("scene graph built", bool(s0 and s0.get("items")), s0)
# (a) freeze
pid = status().get("pid")
t_off = t_nodes = None
os.kill(pid, sg.SIGSTOP)
t0 = time.time()
try:
while time.time() - t0 < 15:
if t_off is None and _main_native() is False:
t_off = round(time.time() - t0, 1)
if t_nodes is None:
s = _sg_items()
if s is None or not s.get("items") or s.get("expired"):
t_nodes = round(time.time() - t0, 1)
time.sleep(0.25)
finally:
os.kill(pid, sg.SIGCONT)
t1 = time.time()
t_nb = t_sb = None
while time.time() - t1 < 15 and (t_nb is None or t_sb is None):
si = _sg_items() or {}
if t_sb is None and (si.get("items") or 0) > 0 and si.get("expired") is False: # rebuilt, not just kept
t_sb = round(time.time() - t1, 1)
if t_nb is None and _main_native() is True:
t_nb = round(time.time() - t1, 1)
time.sleep(0.1)
back = t_nb is not None and t_sb is not None
t_back = max(t_nb, t_sb) if back else None
print(f"NOTE after SIGCONT: nodes back {t_sb} s, lgs-native back {t_nb} s", flush=True)
check("freeze: lgs-native off within 4.5 s", t_off is not None and t_off <= 4.5, t_off)
check("freeze: nodes off between 11 and 14 s", t_nodes is not None and 11 <= t_nodes <= 14, t_nodes)
check("freeze: both back within 3 s of SIGCONT", bool(back) and t_back <= 3, t_back)
check("freeze: lgs-native back within 1.5 s of SIGCONT (NAT: about 1 s)", t_nb is not None and t_nb <= 1.5, t_nb)
# (b) teardown order
since = time.strftime("%Y-%m-%d %H:%M:%S")
time.sleep(1)
print("stop:", stop(), flush=True)
lines = _journal(since)
marks = ["stopping:", "teardown: action plugins stopped", "teardown: scene graph", "teardown: lgs-native",
"glassd: stopped", "lgs-shell exited"]
pos = [next((i for i, x in enumerate(lines) if m in x), -1) for m in marks]
check("teardown order: plugins, scene graph, lgs-native + reporter, glassd, exit",
all(p >= 0 for p in pos) and pos == sorted(pos), list(zip(marks, pos)))
check("after stop: no lgs-native, no scene graph, no glassd files",
_main_native() is False and not (_sg_items() or {}).get("items")
and not os.path.exists(GLASSD_JSON) and not os.path.exists(GLASSD_OUT),
f"{_main_native()} {_sg_items()} {os.path.exists(GLASSD_JSON)}")
gl = _bindings()
check("after stop: no lgsLayers / lgsAction global in Steam (G-REMOVE)", gl == {"lgsLayers": "undefined",
"lgsAction": "undefined"}, gl)
# (c) restart backoff and give-up (5 crashes in 120 s)
st = _native_start(_report_window(W, H))
check("native again", st.get("mode") == "native", st.get("mode"))
since = time.strftime("%Y-%m-%d %H:%M:%S")
time.sleep(1)
for i in range(5):
g = (status().get("glassd") or {})
gp = g.get("pid")
if not g.get("running") or not gp:
gone = _wait(lambda: (status().get("glassd") or {}).get("running"), 40, 0.5)
gp = (status().get("glassd") or {}).get("pid")
if not gone:
break
try:
os.kill(gp, sg.SIGKILL)
except OSError:
pass
if i < 4:
_wait(lambda: (lambda g2: g2.get("running") and g2.get("pid") != gp)(status().get("glassd") or {}), 40, 0.5)
gave_up = _wait(lambda: (status().get("mode") or "").startswith("css-only (glassd gave up"), 30, 0.5)
lines = _journal(since)
ex = _journal_times(lines, "glassd: exited with")
stt = _journal_times(lines, "glassd: started pid")
delays = [b - a for a, b in zip(ex, stt[1:] if stt and stt[0] < (ex[0] if ex else 0) else stt)][:4]
check("restart backoff 2, 4, 8, 16 s (±1.5)", len(delays) == 4 and all(abs(d - w) <= 1.5 for d, w in
zip(delays, (2, 4, 8, 16))), delays)
check("gives up after 5 crashes in 120 s: CSS only", bool(gave_up), status().get("mode"))
check("after give-up: lgs-native off, nodes gone, glassd files gone",
_wait(lambda: _main_native() is False and not (_sg_items() or {}).get("items")
and not os.path.exists(GLASSD_JSON), 10, 0.5),
f"{_main_native()} {_sg_items()} {os.path.exists(GLASSD_JSON)}")
def _selftest_freeze(check, args):
"""A freeze of the daemon itself with the real glassd (--no-feed: no camera),
which rewrites glassd-out.json every 2 s (REQ P7->P8, SG-6): after SIGCONT
glassd is not taken for hung (same pid, no restart), systemui gets its
heartbeat first, nodes and lgs-native are back within 1 s."""
import signal as sg
W, H = _main_size()
st = _native_start(_report_window(W, H), glassd=GLASSD_BIN, gargs=["--no-feed"])
check("native mode with the real glassd (--no-feed)", st.get("mode") == "native", st.get("mode"))
check("lgs-native on main", _wait(lambda: _main_native() is True, 20, 0.5))
g0 = (status().get("glassd") or {})
check("glassd writes 'updated' (so its silence is watched)", "updated" in (read_json(GLASSD_OUT) or {}),
sorted((read_json(GLASSD_OUT) or {}).keys())[:12])
since = time.strftime("%Y-%m-%d %H:%M:%S")
pid = status().get("pid")
os.kill(pid, sg.SIGSTOP)
try:
nodes_off = _wait(lambda: (lambda s: s is None or not s.get("items") or s.get("expired"))(_sg_items()), 15, 0.25)
time.sleep(max(0, 15 - 12.5))
finally:
os.kill(pid, sg.SIGCONT)
t1 = time.time()
t_nb = t_sb = None
while time.time() - t1 < 10 and (t_nb is None or t_sb is None):
si = _sg_items() or {}
if t_sb is None and (si.get("items") or 0) > 0 and si.get("expired") is False:
t_sb = round(time.time() - t1, 2)
if t_nb is None and _main_native() is True:
t_nb = round(time.time() - t1, 2)
time.sleep(0.05)
check("freeze: nodes gone (watchdog) while we were stopped", bool(nodes_off))
check("after SIGCONT: nodes back within 1 s", t_sb is not None and t_sb <= 1.0, t_sb)
check("after SIGCONT: lgs-native back within 1.5 s", t_nb is not None and t_nb <= 1.5, t_nb)
time.sleep(8) # past glassd's stale limit (6 s)
g1 = (status().get("glassd") or {})
lines = _journal(since)
check("glassd not taken for hung: same pid, no restart", g1.get("pid") == g0.get("pid")
and g1.get("restarts") == g0.get("restarts") and not any("hung?" in x for x in lines),
(g0.get("pid"), g1.get("pid"), [x for x in lines if "hung" in x][:2]))
check("the stall was noticed and systemui got its heartbeat first",
any("event loop stalled" in x for x in lines) and any("heartbeat sent first" in x for x in lines),
[x for x in lines if "stall" in x][:3])
st = status()
check("still native 8 s later", st.get("mode") == "native" and _main_native() is True, st.get("mode"))
def _selftest_dm2(check, args):
"""glassd.json v3: plates, holes, tints, masks, coverDz, unitM, materialize
and fades, with fakeglassd (and --assume-caps when its caps lack them)."""
W, H = _main_size()
rep = {"seq": 1, "v": 3, "dash": True, "profile": "default", "roomDim": 0.3,
"masks": [{"O": [-0.6, 1.4, -1.2], "U": [0.2, 0, 0], "V": [0, -0.5, 0]}],
"surfaces": [{
"name": "main", "overlayKey": STEAM_MAIN_KEY, "visible": True, "texW": W, "texH": H, "radius": 81,
"material": "window", "mode": "windowless", "shapes": [], "coverDz": 0.001,
"masks": [{"x": -300, "y": 0, "w": 200, "h": H, "dz": 0.03}],
"plates": [{"id": "p8-disc", "x": 300, "y": 200, "w": 180, "h": 180, "r": 90, "material": "liquid"},
{"id": "p8-card", "x": 600, "y": 420, "w": 600, "h": 300, "r": 54, "material": "thick",
"fill": "rgba(0, 0, 0, 0.14)"},
{"id": "p8-dim", "x": 1300, "y": 200, "w": 300, "h": 200, "r": 40, "material": "dim"}],
"layers": [{"id": "p8-play", "x": 320, "y": 820, "w": 420, "h": 120, "r": 60, "dz": 0.0407,
"material": "liquid", "tint": "rgb(48 199 89 / 0.55)",
"hole": {"fill": "rgb(18 20 26 / 0.9)",
"edges": {"top": "rgb(80 52 37)", "left": ["rgb(70 50 40)", "nope" * 20,
[0.3, 0.2, 0.15]], "right": []}}},
{"id": "p8-info", "x": 780, "y": 830, "w": 100, "h": 100, "r": 50, "dz": 0.0271,
"material": "none", "hole": True, "tint": [0.1, 0.45, 1.0, 0.5]},
{"id": "p8-roomdim", "x": 1700, "y": 900, "w": 64, "h": 64, "r": 0, "dz": -0.05,
"material": "dim", "fill": [0, 0, 0, 0.4]}]},
# the keyboard surface (K-G6 platter behind the keys; K-G2 scale), hidden: glassd.json only
{"name": "keyboard", "overlayKey": "valve.steam.gamepadui.keyboard", "visible": False,
"texW": 1500, "texH": 540, "radius": 48, "material": "panel",
"shapes": [{"x": 0, "y": 0, "w": 1500, "h": 540, "r": 48}], "coverDz": -0.027, "scaleFrom": "overlay",
"layers": []}]}
caps = ["plates", "holes", "tint", "masks", "coverDz", "none", "offset", "unitM", "dim", "scaleFrom", "roomDim",
"dimSlab", "holeEdges"]
os.makedirs(TEST_DUMP, exist_ok=True) # fakeglassd writes into it, it does not create it
st = _native_start(rep, caps=caps, gargs=["--dump", TEST_DUMP])
check("native mode with fakeglassd", st.get("mode") == "native", st.get("mode"))
out = read_json(GLASSD_OUT) or {}
print(f"NOTE glassd-out: version {out.get('version')} caps {out.get('caps')} "
f"(assumed for the writer: {caps})", flush=True)
g = _wait(lambda: (lambda d: d if d and any(s.get("plates") for s in d.get("surfaces", [])) else None)(
read_json(GLASSD_JSON)), 10, 0.25) or {}
m = next((s for s in g.get("surfaces", []) if s.get("name") == "main"), {})
check("plates carried, in order", [p.get("id") for p in m.get("plates") or []] == ["p8-disc", "p8-card", "p8-dim"],
m.get("plates"))
check("new plates and slabs materialize", all(x.get("appear") == "materialize"
for x in (m.get("plates") or []) + (m.get("slabs") or [])))
sl = {s.get("id"): s for s in m.get("slabs") or []}
check("slab tint (CSS string) and hole fill carried",
(sl.get("p8-play") or {}).get("tint") == "rgb(48 199 89 / 0.55)"
and ((sl.get("p8-play") or {}).get("hole") or {}).get("fill") == "rgb(18 20 26 / 0.9)", sl.get("p8-play"))
check("hole.edges carried and cleaned (REQ P9->P8: bad sample and empty edge dropped)",
((sl.get("p8-play") or {}).get("hole") or {}).get("edges") == {
"top": "rgb(80 52 37)", "left": ["rgb(70 50 40)", [0.3, 0.2, 0.15]]}, (sl.get("p8-play") or {}).get("hole"))
check("material none slab with hole: true and a colour list tint",
(sl.get("p8-info") or {}).get("material") == "none" and (sl.get("p8-info") or {}).get("hole") is True
and (sl.get("p8-info") or {}).get("tint") == [0.1, 0.45, 1, 0.5], sl.get("p8-info"))
check("surface masks and coverDz", m.get("masks") == [{"x": -300, "y": 0, "w": 200, "h": H, "dz": 0.03}]
and m.get("coverDz") == 0.001, (m.get("masks"), m.get("coverDz")))
check("top-level world masks, unitM = live geometry, reduceMotion, roomDim",
g.get("masks") == [{"O": [-0.6, 1.4, -1.2], "U": [0.2, 0, 0], "V": [0, -0.5, 0]}]
and g.get("unitM") == (status().get("geom") or {}).get("unitM") and g.get("reduceMotion") is False
and g.get("roomDim") == 0.3, {k: g.get(k) for k in ("masks", "unitM", "reduceMotion", "roomDim")})
check("dim slab (room-dim cell) with its fill", (sl.get("p8-roomdim") or {}).get("material") == "dim"
and (sl.get("p8-roomdim") or {}).get("fill") == [0, 0, 0, 0.4], sl.get("p8-roomdim"))
kb = next((s for s in g.get("surfaces", []) if s.get("name") == "keyboard"), {})
check("keyboard surface: coverDz -0.027, scaleFrom overlay, cover shape, hidden",
kb.get("coverDz") == -0.027 and kb.get("scaleFrom") == "overlay" and kb.get("visible") is False
and len(kb.get("shapes") or []) == 1, kb)
out = read_json(GLASSD_OUT) or {}
om = (out.get("surfaces") or {}).get("main") or {}
check("fakeglassd drew the plates (glassd-out plates)", om.get("plates") == ["p8-disc", "p8-card", "p8-dim"]
and (om.get("counts") or {}).get("holes") == 2, {k: om.get(k) for k in ("plates", "counts")})
sp = _wait(lambda: (lambda s: s if s and s.get("items") else None)(_sg_items()), 10, 0.5)
check("scene graph built from the v3 spec", bool(sp), sp)
import shutil
try: # the first scene's textures (procedural: fakeglassd paints no camera image)
for fn in os.listdir(TEST_DUMP):
if fn.endswith(".png") and not fn.startswith("first-"):
shutil.copy(os.path.join(TEST_DUMP, fn), os.path.join(TEST_DUMP, "first-" + fn))
except OSError:
pass
# fades: the disc plate and the play slab leave
time.sleep(1.0)
rep2 = json.loads(json.dumps(rep))
rep2["seq"] = 2
s2 = rep2["surfaces"][0]
s2["plates"] = [p for p in s2["plates"] if p["id"] != "p8-disc"]
s2["layers"] = [x for x in s2["layers"] if x["id"] != "p8-play"]
t0 = time.time()
write_json_atomic(TEST_REPORT, rep2)
def item(d, kind, iid):
mm = next((s for s in (d or {}).get("surfaces", []) if s.get("name") == "main"), {})
return next((x for x in mm.get(kind) or [] if x.get("id") == iid), None)
faded = _wait(lambda: (lambda d: d if (item(d, "plates", "p8-disc") or {}).get("phase") == 0
and (item(d, "slabs", "p8-play") or {}).get("phase") == 0 else None)(read_json(GLASSD_JSON)),
6, 0.02) # the test report is re-read on the 1 s Steam tick; a busy Steam slows it
t_faded = time.time()
check("leaving plate and slab written with phase 0", bool(faded),
f"{round(t_faded - t0, 2)} s after the report changed (--test-report is re-read once a second)"
if faded else str(read_json(GLASSD_JSON))[:300])
gone = _wait(lambda: (lambda d: d if item(d, "plates", "p8-disc") is None and item(d, "slabs", "p8-play") is None
and item(d, "plates", "p8-card") else None)(read_json(GLASSD_JSON)), 4, 0.02)
t_gone = round(time.time() - t_faded, 2)
check("then removed after the 350 ms out-ramp (≤ 0.35 + 0.3 s)", bool(gone) and 0.3 <= t_gone <= 0.65, t_gone)
try:
pngs = sorted(os.listdir(TEST_DUMP))
except OSError:
pngs = []
print(f"NOTE fakeglassd dumps (procedural, no camera): {pngs}"
+ ("" if "--keep-dump" in args else " (deleted after the test; --keep-dump keeps them)"), flush=True)
check("fakeglassd --dump wrote textures", bool(pngs), pngs)
FM_DUMP_JS = ("(async()=>{const d=await OverlayStore.DumpLaserOverlays();const o=d.overlays||{};"
"const k=Object.keys(o).find(n=>/frame\\.menu/.test(n));const p=k&&o[k].scene_graph_panel;"
"return JSON.stringify(p?{fWidth:p.fWidth,fHeight:p.fHeight}:null)})()")
def _selftest_dm3(check, args):
"""Geometry against the SP §1.2 snippet, and the frame menu's panel against
a DumpLaserOverlays of our own: atomic (review R1 m3), i.e. the daemon's own
dump came after ours and the panel did not change between our two dumps.
Other agents restart the daemon (native-session, lgs on): a run during
which the daemon's pid changed is repeated (up to 3 times)."""
for run in range(3):
res = []
pid0 = _wait(lambda: (lambda s: s.get("pid") if s.get("geom") and s.get("geomDumpAt") else None)(status()),
20, 0.5)
_selftest_dm3_once(lambda n, ok, info="": res.append((n, ok, info)))
if pid0 and status().get("pid") == pid0:
break
print(f"NOTE the daemon was restarted during the run (pid {pid0} -> {status().get('pid')}); again", flush=True)
for r in res:
check(*r)
def _selftest_dm3_once(check):
st = status()
g = st.get("geom") or {}
ref = json.loads(_vr_js("systemui", SP12_SNIPPET))
for k, rk, tol in (("S", "S", 6e-5), ("H0", "H0", 6e-5), ("r", "r", 6e-5), ("Hm", "Hw", 6e-5),
("unitM", "unitToM", 6e-5), ("mmPerCssPx", "mmPerCssPx", 6e-5), ("dashDist", "dashDist", 6e-4)):
check(f"geom.{k} = snippet {rk}", _close(g.get(k), ref.get(rk), tol), f"{g.get(k)} vs {ref.get(rk)}")
got = None
for attempt in range(3):
d0 = json.loads(_vr_js("systemui", FM_DUMP_JS, 15))
t0 = time.time()
s1 = _wait(lambda: (lambda x: x if (x.get("geomDumpAt") or 0) > t0 else None)(status()), GEOM_DUMP_S + 6, 0.5)
d1 = json.loads(_vr_js("systemui", FM_DUMP_JS, 15))
if s1 and d0 == d1:
got = (s1, d1, round(s1["geomDumpAt"] - t0, 1), attempt + 1)
break
print(f"NOTE frame menu panel changed during the step ({d0} -> {d1}) or no daemon dump ({bool(s1)}); again",
flush=True)
if got is None:
check("frame menu stable over one daemon dump (3 tries)", False, "panel kept changing or no daemon dump")
elif got[1] is None:
check("frame menu present in DumpLaserOverlays", False, "no frame menu panel")
else:
s1, dump, lag, tries = got
fm = (s1.get("geom") or {}).get("frameMenu") or {}
print(f"NOTE daemon dump {lag} s after ours (try {tries}); our dumps before and after it equal", flush=True)
check("frameMenu.fHeight = DumpLaserOverlays (same step)", _close(fm.get("fHeight"), dump.get("fHeight"), 6e-5),
f"{fm.get('fHeight')} vs {dump.get('fHeight')}")
check("frameMenu.fWidth = DumpLaserOverlays (same step)", _close(fm.get("fWidth"), dump.get("fWidth"), 6e-5),
f"{fm.get('fWidth')} vs {dump.get('fWidth')}")
check("frameMenu.clipCssH = fHeight / mpp / 1.5",
_close(fm.get("clipCssH"), float(dump["fHeight"]) / float(fm.get("mpp") or 1) / 1.5, 0.11),
f"{fm.get('clipCssH')} mpMm {fm.get('mpMm')}")
st = s1
steam = json.loads(_vr_js("systemui", "JSON.stringify((()=>{const sf=FrameStore.frames.find(x=>x.m_mapPages&&"
"[...x.m_mapPages.values()].some(p=>p.m_sSummonOverlayKey==="
"'valve.steam.gamepadui.main'));return sf&&sf.activePage&&"
"sf.activePage.m_sSummonOverlayKey})())"))
page = st.get("page") or {}
check("page.summonKey = FrameStore", page.get("summonKey") == steam, f"{page.get('summonKey')} vs {steam}")
check("page.steam", page.get("steam") == (steam == STEAM_MAIN_KEY))
g = st.get("geom") or {}
rt = _steam_js("(() => { const R = window.__LGS_RT, b = R && R.bridge; if (!b || !b.get) return null;"
" return JSON.stringify({geom: b.get('geom'), page: b.get('page'), daemon: b.get('daemon')}); })()")
if rt is None:
print("NOTE bridge: no __LGS_RT.bridge in Steam (runtime off); bridge delivery not checked", flush=True)
else:
b = json.loads(rt)
check("bridge geom = published geom", (b.get("geom") or {}).get("r") == g.get("r")
and (b.get("geom") or {}).get("S") == g.get("S")
and ((b.get("geom") or {}).get("frameMenu") or {}).get("fWidth") == (g.get("frameMenu") or {}).get("fWidth"),
str(b.get("geom"))[:160])
check("bridge page = published page", (b.get("page") or {}).get("summonKey") == page.get("summonKey"))
d = b.get("daemon") or {}
check("bridge daemon heartbeat fresh", d and time.time() * 1000 - d.get("at", 0) < d.get("ttlMs", 0),
str(d)[:160])
P8_FIX_JS = r"""
(function (on) {
for (const p of g_PopupManager.m_mapPopups.values()) {
if (!/^VR_uid/.test(p.m_strName)) continue;
const d = p.window.document;
let f = d.getElementById('lgs-p8-fix');
if (f) f.remove();
if (!on) return 'removed';
f = d.createElement('div');
f.id = 'lgs-p8-fix';
f.setAttribute('data-lgs-plate', 'liquid');
f.setAttribute('data-lgs-plate-id', 'p8-live');
f.style.cssText = 'position:fixed;left:220px;top:180px;width:160px;height:160px;border-radius:80px;' +
'pointer-events:none;z-index:2147483647;background:rgba(255,255,255,0.08)';
d.body.appendChild(f);
return 'added';
}
return 'no main window';
})
"""
P8_FIX_STATE_JS = ("(() => { for (const p of g_PopupManager.m_mapPopups.values()) if (/^VR_uid/.test(p.m_strName)) {"
" const f = p.window.document.getElementById('lgs-p8-fix'); return JSON.stringify({fix: !!f,"
" ack: !!(f && f.hasAttribute('data-lgs-plate-ack'))}); } return null; })()")
def _selftest_plates(check, args):
"""The real reporter (P6 v3) with fakeglassd: options passed, a plate from a
fixture element goes to glassd.json, is acked (data-lgs-plate-ack), fades
out when the element goes; motion.js is prepended to lgs_sg.js."""
print("start native (reporter, fakeglassd):",
start(glassd=FAKEGLASSD, stay=True, glassd_args=[], native="on"), flush=True)
st = _wait(lambda: (lambda s: s if s.get("mode") == "native" and (s.get("steam") or {}).get("reports") else None)(
status()), 45, 0.5) or status()
check("native with the real reporter reporting", st.get("mode") == "native"
and (st.get("steam") or {}).get("source") == "reporter", (st.get("mode"), (st.get("steam") or {}).get("reporter")))
# the reporter deletes __LGS_LAYERS_OPTS once read: its status() shows what it got
ro = _steam_js("(() => { const L = window.__LGS_LAYERS, s = L && L.status ? L.status() : {};"
" return JSON.stringify({v: s.v, fragments: s.fragments, profile: s.profile, geom: s.geom,"
" ackMode: s.ackMode}); })()")
ro = json.loads(ro) if ro else {}
frags = sorted(os.path.basename(f) for f in json_files(LAYERS_DIR))
got = ro.get("fragments") or [] # a fragment whose flag is off may be left out
check("reporter v3 got theme/layers fragments, profile, geom (= ours) and ackMode", (ro.get("v") or 0) >= 3
and got and set(got) <= set(frags) and frags[0] in got and ro.get("profile") in ("default", "wearer")
and (ro.get("geom") or {}).get("src") in ("opts", "bridge") # both are ours; P6 prefers the live bridge
and _close((ro.get("geom") or {}).get("S"), (status().get("geom") or {}).get("S"), 1e-4)
and ro.get("ackMode") is True, (ro, frags))
mot = _vr_js("systemui", "typeof globalThis.__LGS_MOTION")
check("motion.js prepended to lgs_sg.js (__LGS_MOTION in vr:systemui)", mot == "object", mot)
try:
check("plate fixture added to the main window", _steam_js(f"({P8_FIX_JS})(true)") == "added")
def plate(d):
m = next((s for s in (d or {}).get("surfaces", []) if s.get("name") == "main"), {})
return next((p for p in m.get("plates") or [] if p.get("id") == "p8-live"), None)
t0 = time.time()
p = _wait(lambda: plate(read_json(GLASSD_JSON)), 6, 0.1)
check("its plate reaches glassd.json (materialize)", p and p.get("appear") == "materialize"
and p.get("w", 0) > 200, p)
drawn = _wait(lambda: "p8-live" in (((read_json(GLASSD_OUT) or {}).get("surfaces") or {}).get("main") or {})
.get("plates", []), 5, 0.1)
check("fakeglassd draws it (glassd-out plates)", bool(drawn))
ack = _wait(lambda: (json.loads(_steam_js(P8_FIX_STATE_JS) or "{}") or {}).get("ack"), 6, 0.2)
t_ack = round(time.time() - t0, 2)
check("plate acked: data-lgs-plate-ack on the element (after its 250 ms ramp + 350 ms)", bool(ack)
and t_ack >= 0.6, t_ack)
pa = _wait(lambda: (lambda m: m if any("p8-live" in v for v in m.values()) else None)(
(status().get("steam") or {}).get("plateAcks") or {}), 5, 0.5)
check("status lists the plate ack", bool(pa), pa)
_steam_js(f"({P8_FIX_JS})(false)")
faded = _wait(lambda: (plate(read_json(GLASSD_JSON)) or {}).get("phase") == 0, 4, 0.05)
check("element removed: its plate fades (phase 0)", bool(faded))
gone = _wait(lambda: plate(read_json(GLASSD_JSON)) is None, 3, 0.05)
check("then leaves glassd.json", bool(gone))
finally:
_steam_js(f"({P8_FIX_JS})(false)")
TEST_PLUGIN_PY = '''# lgs_shell.py selftest dm4: a RAM test plugin (kind read; the test removes it)
ACTIONS = {"p8.probe": {"args": {"n": {"type": "int", "min": 0, "max": 99}}, "sources": ["main"], "kind": "read",
"rate": 0}}
async def run(ctx, type, args):
return {"probe": args["n"] + 1, "plugin": ctx.name}
'''
TEST_PLUGIN_NAV_PY = '''# lgs_shell.py selftest dm4: a test plugin that asks for kind nav (must be refused from the RAM folder)
ACTIONS = {"p8.go": {"args": {}, "kind": "nav"}}
async def run(ctx, type, args):
return 1
'''
def _selftest_dm4(check, args):
"""Actions through the binding: echo, unknown type, wrong source, bad args,
rate, all logged; then a plugin file loaded live from the RAM test folder
(TEST_EXT_DIR), called, refused when it asks for kind nav, unloaded when
its file goes (review R1 m8)."""
if _steam_js("typeof window.lgsAction") != "function":
check("lgsAction binding present", False, "typeof window.lgsAction != function")
return
n = os.urandom(3).hex()
def call(cid, obj):
return _steam_js(f"(window.lgsAction(JSON.stringify({json.dumps(dict(obj, id=cid))})), 0)")
call(f"t-echo-{n}", {"type": "echo", "args": {"text": "hi"}, "src": "main"})
time.sleep(0.3)
call(f"t-unknown-{n}", {"type": "nope.nope", "args": {}, "src": "main"})
call(f"t-src-{n}", {"type": "echo.main", "args": {}, "src": "bar"})
call(f"t-args-{n}", {"type": "echo", "args": {"text": "x" * 300}, "src": "main"})
time.sleep(0.3)
_steam_js(f"(window.lgsAction(JSON.stringify({json.dumps({'id': 't-rate1-' + n, 'type': 'echo', 'args': {}, 'src': 'main'})})),"
f" window.lgsAction(JSON.stringify({json.dumps({'id': 't-rate2-' + n, 'type': 'echo', 'args': {}, 'src': 'main'})})), 0)")
time.sleep(2.6)
st = status()
reps = {r.get("id"): r for r in (st.get("actions") or {}).get("replies") or []}
rej = {r.get("id"): r for r in (st.get("actions") or {}).get("rejected") or []}
exp = {f"t-echo-{n}": None, f"t-unknown-{n}": "unknown-type", f"t-src-{n}": "bad-source",
f"t-args-{n}": "bad-args", f"t-rate1-{n}": None, f"t-rate2-{n}": "rate"}
for cid, err in exp.items():
r = reps.get(cid) or {}
if err is None:
check(f"{cid.rsplit('-', 1)[0]} answered ok", r.get("ok") is True, str(r))
else:
check(f"{cid.rsplit('-', 1)[0]} rejected {err} and logged",
r.get("ok") is False and r.get("error") == err and (rej.get(cid) or {}).get("error") == err,
f"{r} / {rej.get(cid)}")
bridge_js = ("(() => { const R = window.__LGS_RT, b = R && R.bridge; return b && b.get ?"
" JSON.stringify(b.get('reply')) : null; })()")
last = _steam_js(bridge_js)
if last is None:
print("NOTE bridge: no __LGS_RT.bridge in Steam (runtime off); replies checked in shell.json only",
flush=True)
else:
lr = json.loads(last) or {}
check("bridge carries the last reply", lr.get("id") == f"t-rate2-{n}" and lr.get("error") == "rate",
str(lr)[:160])
lines = [x for x in _journal("-2min") if n in x]
check("every call logged in the journal", len(lines) >= 6, f"{len(lines)} lines")
# a plugin file, live: the running daemon scans TEST_EXT_DIR (RAM) once a second
plug, nav = os.path.join(TEST_EXT_DIR, "p8probe.py"), os.path.join(TEST_EXT_DIR, "p8nav.py")
os.makedirs(TEST_EXT_DIR, exist_ok=True)
try:
with open(plug, "w", encoding="utf-8") as f:
f.write(TEST_PLUGIN_PY)
with open(nav, "w", encoding="utf-8") as f:
f.write(TEST_PLUGIN_NAV_PY)
a = _wait(lambda: (lambda x: x if "p8.probe" in (x.get("types") or []) and "test:p8nav" in (x.get("failed") or {})
else None)(status().get("actions") or {}), 6, 0.5) or status().get("actions") or {}
check("plugin file loaded live by the running daemon (p8.probe)", "p8.probe" in (a.get("types") or []),
a.get("types"))
check("a RAM test plugin asking for kind nav is refused", "test:p8nav" in (a.get("failed") or {})
and "p8.go" not in (a.get("types") or []), (a.get("failed") or {}).get("test:p8nav"))
call(f"t-plug-{n}", {"type": "p8.probe", "args": {"n": 41}, "src": "main"})
r = _wait(lambda: {x.get("id"): x for x in (status().get("actions") or {}).get("replies") or []}
.get(f"t-plug-{n}"), 5, 0.5) or {}
check("plugin call answered ok", r.get("ok") is True, r)
last = _steam_js(bridge_js)
if last is not None:
lr = json.loads(last) or {}
check("plugin result on the bridge: {probe: 42, plugin: test:p8probe}", lr.get("id") == f"t-plug-{n}"
and (lr.get("result") or {}).get("probe") == 42
and (lr.get("result") or {}).get("plugin") == "test:p8probe", str(lr)[:160])
finally:
for p in (plug, nav):
try:
os.remove(p)
except OSError:
pass
a = _wait(lambda: (lambda x: x if "p8.probe" not in (x.get("types") or []) and "test:p8nav" not in
(x.get("failed") or {}) else None)(status().get("actions") or {}), 6, 0.5)
check("plugin unloaded when its file goes", bool(a), (status().get("actions") or {}).get("types"))
def _selftest_dm5(check, args):
"""A SteamVR page script: injected only into its page, listed, removed when
its file goes; with --stop also removed on daemon stop (then restarted)."""
os.makedirs(os.path.dirname(TEST_SCRIPT), exist_ok=True)
try:
with open(TEST_SCRIPT, "w", encoding="utf-8") as f:
f.write(TEST_SCRIPT_JS)
ver = hashlib.sha1(TEST_SCRIPT_JS.encode()).hexdigest()[:10]
probe = ("JSON.stringify({vrx: !!(window.__LGS_VRX && window.__LGS_VRX.p8echo), "
"attr: document.documentElement.dataset.lgsEcho || null})")
got = _wait(lambda: (lambda d: d if d["vrx"] else None)(json.loads(_vr_js("systemui", probe))), 8)
check("injected into vr:systemui", got and got.get("attr") == ver, str(got))
st = _wait(lambda: (lambda s: s if s.get("p8echo") in ("installed", "ok") else None)(
(status().get("vrScripts") or {}).get("systemui") or {}), 5) or {}
check("status lists it", st.get("p8echo") in ("installed", "ok"), str(st))
others = [t.get("title") for t in lgs_vr.targets() if t.get("title") != "systemui"]
leak = [p for p in others if _vr_js(p, "!!(window.__LGS_VRX && window.__LGS_VRX.p8echo)")]
check("only in matching pages", not leak, str(leak))
os.remove(TEST_SCRIPT)
gone = _wait(lambda: (lambda d: d if not d["vrx"] and not d["attr"] else None)(
json.loads(_vr_js("systemui", probe))), 8)
check("removed when its file goes", bool(gone), str(gone))
if "--stop" in args:
with open(TEST_SCRIPT, "w", encoding="utf-8") as f:
f.write(TEST_SCRIPT_JS)
got = _wait(lambda: (lambda d: d if d["vrx"] else None)(json.loads(_vr_js("systemui", probe))), 8)
check("re-injected", bool(got), str(got))
print("stop:", stop(), flush=True)
after = json.loads(_vr_js("systemui", probe))
check("removed on daemon stop", not after["vrx"] and not after["attr"], str(after))
gl = _bindings()
check("after stop: no lgsLayers / lgsAction global in Steam (G-REMOVE)",
gl == {BINDING: "undefined", ACTION_BINDING: "undefined"}, gl)
os.remove(TEST_SCRIPT)
print("start:", start(native="auto"), flush=True)
finally:
try:
os.remove(TEST_SCRIPT)
except OSError:
pass
def _selftest_grace(check, args, nat):
"""Theme off without lgs off: dormant (status mode 'dormant'), same pid;
native (nat): glassd stopped while dormant and its files gone, not counted
as a crash; theme back: resumed in the same process (native: glassd again,
lgs-native back)."""
if nat:
W, H = _main_size()
write_json_atomic(TEST_REPORT, _report_window(W, H))
print("start native (no --stay):", start(glassd=FAKEGLASSD, stay=False, glassd_args=[], native="on",
test_report=TEST_REPORT), flush=True)
st = _wait(lambda: (lambda s: s if s.get("mode") == "native" else None)(status()), 45, 0.5) or status()
check("native with fakeglassd before", st.get("mode") == "native", st.get("mode"))
check("native unit: lgsLayers binding present", _bindings().get(BINDING) == "function", _bindings())
st0 = status()
pid0 = st0.get("pid")
g0 = st0.get("glassd") or {}
check("unit active before", st0.get("unit") == "active" and not st0.get("dormant"), st0.get("mode"))
was_on = lgs.is_on()
try:
if was_on:
lgs.op("off", quiet=True)
d = _wait(lambda: (lambda s: s if s.get("dormant") else None)(status()), 12, 0.5) or {}
check("theme off: dormant, same pid, status mode 'dormant'", d.get("pid") == pid0
and (d.get("dormant") or {}).get("graceS") and d.get("mode") == "dormant",
f"{d.get('pid')} vs {pid0} {d.get('dormant')} {d.get('mode')}")
if nat:
gone = _wait(lambda: not _alive(g0.get("pid")), 4, 0.25)
check("dormant: glassd process stopped within 4 s (review R1 M2)", bool(gone) and g0.get("pid"),
f"pid {g0.get('pid')} alive {_alive(g0.get('pid'))}")
check("dormant: no glassd.json / glassd-out.json", not os.path.exists(GLASSD_JSON)
and not os.path.exists(GLASSD_OUT), (os.path.exists(GLASSD_JSON), os.path.exists(GLASSD_OUT)))
check("dormant: lgs-native off on main", _wait(lambda: _main_native() is False, 4, 0.5), _main_native())
time.sleep(3)
s2 = status()
check("still running 8 s after the theme went off", s2.get("unit") == "active" and s2.get("pid") == pid0,
f"{s2.get('unit')} {s2.get('pid')}")
check("dormant: no scene graph in systemui", not (_sg_items() or {}).get("items"), _sg_items())
if nat:
check("dormant: glassd still stopped 3 s later",
not (s2.get("glassd") or {}).get("running") and not _alive(g0.get("pid")),
(s2.get("glassd") or {}).get("pid"))
finally:
if was_on:
lgs.op("on", quiet=True)
r = _wait(lambda: (lambda s: s if s.get("unit") == "active" and not s.get("dormant") else None)(status()),
10, 0.5) or {}
check("theme back: resumed (not dormant, same pid)", r.get("pid") == pid0, f"{r.get('pid')} {r.get('dormant')}")
if nat:
n = _wait(lambda: (lambda s: s if s.get("mode") == "native" else None)(status()), 30, 0.5) or status()
g1 = n.get("glassd") or {}
check("theme back: native again with a new glassd", n.get("mode") == "native" and g1.get("pid")
and g1.get("pid") != g0.get("pid"), (n.get("mode"), g0.get("pid"), g1.get("pid")))
check("the dormancy stop was not counted as a crash (restarts unchanged)",
g1.get("restarts") == g0.get("restarts") and not g1.get("givenUp"), (g0.get("restarts"), g1.get("restarts")))
check("lgs-native back on main", _wait(lambda: _main_native() is True, 20, 0.5), _main_native())
back = _wait(lambda: _steam_js("typeof window.lgsAction") == "function", 5)
check("action binding present after resume", bool(back))
rt = _wait(lambda: _steam_js("(() => { const R = window.__LGS_RT, b = R && R.bridge; const d = b && b.get && "
"b.get('daemon'); return d ? Date.now() - d.at < d.ttlMs : null; })()") is True, 8)
if _steam_js("!!(window.__LGS_RT && window.__LGS_RT.bridge)"):
check("bridge daemon heartbeat fresh after resume", bool(rt))
pages = (status().get("steamvrPages") or {}).get("pages") or {}
check("SteamVR pages themed", pages and all(v in ("ok", "applied") for v in pages.values()), pages)
def selftest(args):
"""Lab checks against the running unit. Prints PASS/FAIL lines and a JSON
summary; exit 0 when every check passed."""
name = args[0] if args else ""
results = []
def check(n, ok, info=""):
results.append({"check": n, "pass": bool(ok), "info": info if not ok or info else ""})
print(("PASS " if ok else "FAIL ") + n + (f" [{info}]" if info != "" else ""), flush=True)
if name == "actions":
import shutil
import tempfile
tmp = tempfile.mkdtemp(prefix="lgs-p8-", dir="/tmp/lgs" if os.path.isdir("/tmp/lgs") else None)
try:
p, f, res = shell_ext.selftest(tmp)
finally:
shutil.rmtree(tmp, ignore_errors=True)
for n, ok, info in res:
check(n, ok, "" if ok else info)
elif name == "dm6":
nat, src, reason = resolve_native("auto")
check("auto resolves to CSS only without a gate pass", (not nat) if src == "builtin" else True,
f"{nat} {src} {reason}")
if "--live" in args:
with FileLock("/tmp/lgs/native.lock", 1800), FileLock("/tmp/lgs/lab.lock", 600), FileLock("/tmp/lgs/lab-vr.lock", 240):
print("stop:", stop(), flush=True)
print("start(auto):", start(native="auto"), flush=True)
st = _wait(lambda: (status() if status().get("unit") == "active" and
(status().get("mode") or "").startswith("css-only") else None), 20)
st = st or status()
nat_st = st.get("native") or {}
check("status.mode css-only with the reason", (st.get("mode") or "").startswith("css-only (native=auto")
and nat_st.get("requested") == "auto", f"{st.get('mode')} / {nat_st}")
check("a running unit is kept by start(auto)", start(native="auto").startswith("running"))
elif name == "dm3":
with FileLock("/tmp/lgs/lab-vr.lock", 240): # it reads systemui and runs DumpLaserOverlays there
_selftest_dm3(check, args)
elif name == "dm4":
with FileLock("/tmp/lgs/lab.lock", 600): # it calls lgsAction in Steam's SharedJSContext
_selftest_dm4(check, args)
elif name == "dm5":
# --stop restarts the daemon: native.lock and lab.lock as well (order native -> lab -> lab-vr)
import contextlib
stopping = "--stop" in args
with (FileLock("/tmp/lgs/native.lock", 1800) if stopping else contextlib.nullcontext()), \
(FileLock("/tmp/lgs/lab.lock", 600) if stopping else contextlib.nullcontext()), \
FileLock("/tmp/lgs/lab-vr.lock", 240):
_selftest_dm5(check, args)
elif name == "grace":
# theme off without `lgs off` (lab --stock, P1's selftests): dormant, not gone.
# --native: the same with a native unit (fakeglassd, test report): glassd stopped while dormant
import contextlib
nat = "--native" in args
if nat and not os.access(FAKEGLASSD, os.X_OK):
print(f"BLOCKED: no fakeglassd at {FAKEGLASSD} (python glass.py native-build fake)")
return 3
with (FileLock("/tmp/lgs/native.lock", 1800) if nat else contextlib.nullcontext()), \
FileLock("/tmp/lgs/lab.lock", 600), FileLock("/tmp/lgs/lab-vr.lock", 240):
try:
_selftest_grace(check, args, nat)
finally:
if nat:
print("back to css:", start(native="off"), flush=True)
back = _wait(lambda: (status().get("mode") or "").startswith("css-only"), 30)
check("back to CSS only", back, status().get("mode"))
try:
os.remove(TEST_REPORT)
except OSError:
pass
elif name in ("dm1", "dm2", "plates", "freeze"):
need = GLASSD_BIN if name == "freeze" else FAKEGLASSD
if not os.access(need, os.X_OK):
print(f"BLOCKED: no {os.path.basename(need)} at {need} (python glass.py native-build)")
return 3
# native.lock for the native session; lab-vr.lock too, since systemui's
# scene graph changes (and is frozen) under other agents' vr: steps
import contextlib
# plates edits the Steam main window (a fixture): lab.lock as well
with FileLock("/tmp/lgs/native.lock", 1800), \
(FileLock("/tmp/lgs/lab.lock", 600) if name == "plates" else contextlib.nullcontext()), \
FileLock("/tmp/lgs/lab-vr.lock", 240):
try:
{"dm1": _selftest_dm1, "dm2": _selftest_dm2, "plates": _selftest_plates,
"freeze": _selftest_freeze}[name](check, args)
finally:
# always back to CSS only (lgs on --css), then wait for it
print("back to css:", start(native="off"), flush=True)
back = _wait(lambda: (status().get("mode") or "").startswith("css-only"), 30)
check("back to CSS only", back, status().get("mode"))
time.sleep(2)
left = {"reporter": _steam_js("typeof window.__LGS_LAYERS"),
"motion": _vr_js("systemui", "typeof globalThis.__LGS_MOTION"),
"sg": _vr_js("systemui", "typeof window.__LGS_SG"), "glassd.json": os.path.exists(GLASSD_JSON)}
check("nothing native left (reporter, __LGS_MOTION, __LGS_SG, glassd.json)",
left["motion"] == "undefined" and left["glassd.json"] is False
and (left["sg"] == "undefined" or not (_sg_items() or {}).get("items")), left)
gl = _bindings()
check("CSS-only unit: no lgsLayers global, lgsAction present", gl.get(BINDING) == "undefined"
and gl.get(ACTION_BINDING) == "function", gl)
for p in (TEST_REPORT,):
try:
os.remove(p)
except OSError:
pass
if "--keep-dump" not in args:
import shutil
shutil.rmtree(TEST_DUMP, ignore_errors=True)
else:
print("selftest NAME: actions | dm1 | dm2 [--keep-dump] | dm3 | dm4 | dm5 [--stop] | dm6 [--live] | "
"grace [--native] | plates | freeze")
return 2
ok = all(r["pass"] for r in results)
print(json.dumps({"selftest": name, "passed": sum(r["pass"] for r in results), "failed":
sum(not r["pass"] for r in results), "at": time.strftime("%Y-%m-%d %H:%M:%S")}))
return 0 if ok else 1
def opt_arg(argv, name):
if name in argv and argv.index(name) + 1 < len(argv):
return argv[argv.index(name) + 1]
return None
def main(argv):
cmd = argv[1] if len(argv) > 1 else "status"
if cmd == "daemon":
try:
return asyncio.run(Shell(glassd_arg(argv), stay_arg(argv), glassd_args_arg(argv),
native=bool(native_arg(argv)), native_req=opt_arg(argv, "--req"),
native_src=opt_arg(argv, "--src"),
test_report=opt_arg(argv, "--test-report"),
assume_caps=[c for c in (opt_arg(argv, "--assume-caps") or "").split(",") if c],
pod="--pod" in argv).run())
except KeyboardInterrupt:
return 0
if cmd == "start":
gargs = glassd_args_arg(argv)
glassd = glassd_arg(argv)
nat = native_arg(argv)
# From the command line (lab and tests) the real glassd does not read
# the camera unless asked to: --feed, or explicit --glassd-args.
if gargs is None and "--feed" not in argv and glassd and os.path.basename(glassd) == "glassd":
gargs = ["--no-feed"]
print(start(glassd, stay_arg(argv), gargs, None if nat is None or "--auto" in argv else nat,
test_report=opt_arg(argv, "--test-report"),
assume_caps=[c for c in (opt_arg(argv, "--assume-caps") or "").split(",") if c],
pod="--pod" in argv))
elif cmd == "stop":
print(stop())
elif cmd == "status":
print(json.dumps(status(), indent=1))
elif cmd == "selftest":
return selftest(argv[2:])
else:
print(__doc__)
return 2
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv))
Reference in new issue View Git Blame Copy Permalink