frame-autopass: automatic colour/IR passthrough for the Steam Frame

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

tools/autopass.cpp

@@ -0,0 +1,310 @@
+// autopass: install, inspect and support autopassd.
+//
+//   autopass install          install the systemd user unit that starts autopassd
+//                           with SteamVR and stops it with SteamVR; start it
+//   autopass uninstall        stop and remove the unit, switch back to colour
+//   autopass status           autopassd's status and whether the unit is active
+//   autopass update           download and install the latest release
+//   autopass report           write ~/frame-autopass-report.txt for a bug report
+//   autopass version          print the version
+//   autopass guard [reset]    show or clear the crash guard (XRService restarts
+//                           soon after autopassd switches)
+//   autopass state [SECONDS]  passive: camera config and colour light value
+//                           from shared memory, no SteamVR calls
+//   autopass get              read the camera config (private interface)
+//   autopass set rgb|mono [--quiet]
+//                           switch the camera source (used by autopassd). Exit
+//                           codes: 0 done or already so, 1 error, 3 SteamVR's
+//                           interface changed, 4 camera not enabled, 5 no
+//                           Arcturus colour module.
+//
+// `get` and `set` connect to SteamVR as a background client for a few
+// milliseconds; that never starts SteamVR and does not wake the headset.
+
+#include "app_paths.hpp"
+#include "passthrough_state.hpp"
+#include "private_camera.hpp"
+
+#include <openvr.h>
+
+#include <chrono>
+#include <cstdio>
+#include <cstdlib>
+#include <string>
+#include <sys/wait.h>
+#include <thread>
+#include <unistd.h>
+
+namespace {
+
+constexpr const char* kUnit = "frame-autopass.service";
+constexpr const char* kInstallScript = "https://github.com/bod09/frame-autopass/releases/latest/download/install.sh";
+
+int usage() {
+    std::fprintf(stderr,
+        "usage: autopass install | uninstall | status | update | report | version |\n"
+        "                guard [reset] | state [SECONDS] | get | set rgb|mono [--quiet]\n");
+    return 2;
+}
+
+std::string unit_path() {
+    const char* xdg = std::getenv("XDG_CONFIG_HOME");
+    const char* home = std::getenv("HOME");
+    const std::string base = xdg && *xdg == '/' ? xdg : std::string(home ? home : "/tmp") + "/.config";
+    return base + "/systemd/user/" + kUnit;
+}
+
+// Runs `systemctl --user ARGS...`, returns its exit code.
+int systemctl(std::initializer_list<const char*> args, bool quiet = false) {
+    std::string cmd = "systemctl --user";
+    for (const char* a : args) cmd += std::string(" ") + a;
+    if (quiet) cmd += " >/dev/null 2>&1";
+    const int rc = std::system(cmd.c_str());
+    return WIFEXITED(rc) ? WEXITSTATUS(rc) : 1;
+}
+
+std::string unit_text() {
+    return std::string(
+        "# Installed by `autopass install`. Starts autopassd with SteamVR and stops it\n"
+        "# with SteamVR. Remove with `autopass uninstall`.\n"
+        "[Unit]\n"
+        "Description=Adaptive passthrough (colour in good light, IR in the dark)\n"
+        "After=steamvr.service\n"
+        "BindsTo=steamvr.service\n"
+        "\n"
+        "[Service]\n"
+        "Type=simple\n"
+        "ExecStart=") + autopass::install_dir() + "/autopassd\n"
+        "Restart=on-failure\n"
+        "RestartSec=5\n"
+        "Slice=session.slice\n"
+        "Nice=10\n"
+        "\n"
+        "[Install]\n"
+        "WantedBy=steamvr.service\n";
+}
+
+int install() {
+    const std::string dir = autopass::install_dir();
+    if (::access((dir + "/autopassd").c_str(), X_OK) != 0 || ::access((dir + "/autopass").c_str(), X_OK) != 0) {
+        std::fprintf(stderr, "autopassd and autopass must be in %s first\n", dir.c_str());
+        return 1;
+    }
+    if (!autopass::colour_module_present()) {
+        std::fprintf(stderr,
+            "The Arcturus Vision colour module was not found (no arcimx616 camera in\n"
+            "/sys/class/video4linux). frame-autopass switches between that module and the\n"
+            "built-in IR cameras, so it needs the module attached. Not installing.\n");
+        return 1;
+    }
+    const std::string path = unit_path();
+    if (!autopass::make_dirs(path.substr(0, path.rfind('/'))) || !autopass::write_file_atomic(path, unit_text())) {
+        std::fprintf(stderr, "cannot write %s\n", path.c_str());
+        return 1;
+    }
+    if (systemctl({"daemon-reload"}) != 0 || systemctl({"enable", "--now", kUnit}) != 0) {
+        std::fprintf(stderr, "systemctl failed; see `systemctl --user status %s`\n", kUnit);
+        return 1;
+    }
+    std::printf("installed %s: autopassd now starts and stops with SteamVR\n", path.c_str());
+    return 0;
+}
+
+int switch_source(bool want_rgb, bool quiet);
+
+int uninstall() {
+    systemctl({"disable", "--now", kUnit}, true);
+    std::remove(unit_path().c_str());
+    systemctl({"daemon-reload"}, true);
+    std::printf("removed %s\n", kUnit);
+    // Leave the headset showing colour, the stock behaviour with the module.
+    switch_source(true, true);
+    return 0;
+}
+
+int status() {
+    std::string text;
+    std::printf("unit: %s, %s\n", systemctl({"is-enabled", "--quiet", kUnit}, true) == 0 ? "installed" : "not installed",
+                systemctl({"is-active", "--quiet", kUnit}, true) == 0 ? "running" : "not running");
+    std::printf("version: %s\n", autopass::version());
+    if (autopass::read_file(autopass::status_path(), &text)) {
+        std::printf("status: %s", text.c_str());
+        if (text.find("\"blocked\": \"steamvr-changed\"") != std::string::npos)
+            std::printf("\nSteamVR has changed the camera interface frame-autopass uses, so it has\n"
+                        "stopped switching (passthrough itself works normally). Run `autopass update`;\n"
+                        "if that does not help, a new release is needed.\n");
+        else if (text.find("\"blocked\": \"no-colour-module\"") != std::string::npos)
+            std::printf("\nThe Arcturus Vision colour module is not attached, so there is nothing to\n"
+                        "switch to; it resumes when the module is back.\n");
+        else if (text.find("\"blocked\": \"crash-guard\"") != std::string::npos)
+            std::printf("\nXRService restarted twice soon after a switch, so switching is paused as a\n"
+                        "precaution. `autopass guard reset`, then restart SteamVR, to resume.\n");
+    }
+    if (autopass::read_file(autopass::crash_guard_path(), &text) && !text.empty())
+        std::printf("crash guard entries (switching stops at 2):\n%s", text.c_str());
+    return 0;
+}
+
+int guard(bool reset) {
+    std::string text;
+    const bool present = autopass::read_file(autopass::crash_guard_path(), &text) && !text.empty();
+    if (reset) {
+        std::remove(autopass::crash_guard_path().c_str());
+        std::printf("crash guard cleared; restart autopassd (or SteamVR) to resume switching\n");
+        return 0;
+    }
+    std::printf(present ? "crash guard entries:\n%s" : "crash guard: no entries\n", text.c_str());
+    return 0;
+}
+
+int state(double seconds) {
+    const autopass::PassthroughState st;
+    if (st.path().empty()) {
+        std::fprintf(stderr, "passthrough state file not found in /dev/shm\n");
+        return 1;
+    }
+    const auto end = std::chrono::steady_clock::now() + std::chrono::duration<double>(seconds);
+    do {
+        const auto s = st.read();
+        if (!s) {
+            std::fprintf(stderr, "cannot read %s\n", st.path().c_str());
+            return 1;
+        }
+        std::printf("config: %s", s->config.describe().c_str());
+        if (s->colour) std::printf(" | colour light %.3f", s->colour->light);
+        std::printf(" | newest mono %.3f colour %.3f\n", s->mono_timestamp, s->colour_timestamp);
+        if (seconds > 0) std::this_thread::sleep_for(std::chrono::milliseconds(500));
+    } while (std::chrono::steady_clock::now() < end);
+    return 0;
+}
+
+int update() {
+    std::printf("fetching %s\n", kInstallScript);
+    const std::string cmd = std::string("curl -fsSL ") + kInstallScript + " | bash";
+    const int rc = std::system(cmd.c_str());
+    return WIFEXITED(rc) ? WEXITSTATUS(rc) : 1;
+}
+
+// Appends a command's output to the report.
+void section(std::string* out, const char* title, const std::string& cmd) {
+    *out += std::string("\n== ") + title + " ==\n";
+    if (FILE* p = ::popen((cmd + " 2>&1").c_str(), "r")) {
+        char buf[4096];
+        std::size_t n;
+        while ((n = std::fread(buf, 1, sizeof buf, p)) > 0) out->append(buf, n);
+        ::pclose(p);
+    }
+}
+
+int report() {
+    const char* home = std::getenv("HOME");
+    const std::string path = std::string(home ? home : "/tmp") + "/frame-autopass-report.txt";
+    std::string out = std::string("frame-autopass report, version ") + autopass::version() + "\n";
+    out += std::string("Arcturus colour module: ") + (autopass::colour_module_present() ? "found" : "NOT found") + "\n";
+    section(&out, "date", "date");
+    section(&out, "SteamOS", "grep -E '^(VERSION_ID|BUILD_ID)=' /etc/os-release");
+    section(&out, "SteamVR", "cat /opt/steamvr/bin/version.txt; grep -m1 -E 'cv: version [0-9]' ~/.local/share/Steam/logs/vrserver.txt");
+    section(&out, "unit", std::string("systemctl --user status --no-pager ") + kUnit + " | head -5");
+    section(&out, "status", "cat " + autopass::status_path());
+    section(&out, "config", "cat " + autopass::conf_path() + " 2>/dev/null || echo '(defaults)'");
+    section(&out, "crash guard", "cat " + autopass::crash_guard_path() + " 2>/dev/null || echo '(empty)'");
+    section(&out, "autopassd log (last 300 lines)", "tail -n 300 " + autopass::log_path());
+    section(&out, "XRService (last 150 relevant lines)",
+            "grep -hE 'IREmitters|Passthrough cameras|UserPresence|IMUFallback\\] (En|Dis)|Transition to IMU' "
+            "~/.local/share/Steam/logs/xrservice.txt | cut -c1-140 | tail -n 150");
+    if (!autopass::write_file_atomic(path, out)) {
+        std::fprintf(stderr, "cannot write %s\n", path.c_str());
+        return 1;
+    }
+    std::printf("wrote %s\nIt contains your SteamOS and SteamVR versions and recent logs (no personal\n"
+                "data that I know of; have a look before sharing it). Attach it to a GitHub issue.\n",
+                path.c_str());
+    return 0;
+}
+
+struct Session {
+    bool ok = false;
+    Session() {
+        vr::EVRInitError err = vr::VRInitError_None;
+        vr::VR_Init(&err, vr::VRApplication_Background);
+        ok = err == vr::VRInitError_None;
+        if (!ok) std::fprintf(stderr, "SteamVR unavailable: %s\n", vr::VR_GetVRInitErrorAsEnglishDescription(err));
+    }
+    ~Session() {
+        if (ok) vr::VR_Shutdown();
+    }
+};
+
+int get_config() {
+    Session session;
+    if (!session.ok) return 1;
+    autopass::PrivateCamera camera;
+    if (!camera.ok()) {
+        std::fprintf(stderr, "%s\n", camera.error().c_str());
+        return 3;
+    }
+    const auto c = camera.get();
+    if (!c) {
+        std::fprintf(stderr, "%s\n", camera.error().c_str());
+        return 1;
+    }
+    std::printf("%s\n", c->describe().c_str());
+    return 0;
+}
+
+int switch_source(bool want_rgb, bool quiet) {
+    if (!autopass::colour_module_present()) {
+        if (!quiet) std::fprintf(stderr, "Arcturus colour module not found; not changing anything\n");
+        return 5;
+    }
+    Session session;
+    if (!session.ok) return 1;
+    autopass::PrivateCamera camera;
+    if (!camera.ok()) {
+        std::fprintf(stderr, "%s\n", camera.error().c_str());
+        return 3;
+    }
+    auto c = camera.get();
+    if (!c) {
+        std::fprintf(stderr, "%s\n", camera.error().c_str());
+        return 1;
+    }
+    if (!c->enabled) {
+        if (!quiet) std::fprintf(stderr, "passthrough camera is not enabled; not changing anything\n");
+        return 4;
+    }
+    if (c->rgb == static_cast<std::uint8_t>(want_rgb)) {
+        if (!quiet) std::printf("already %s\n", want_rgb ? "colour" : "IR");
+        return 0;
+    }
+    c->rgb = want_rgb ? 1 : 0;
+    if (!camera.set(*c)) {
+        std::fprintf(stderr, "%s\n", camera.error().c_str());
+        return 1;
+    }
+    if (!quiet) std::printf("switched to %s\n", want_rgb ? "colour" : "IR");
+    return 0;
+}
+
+} // namespace
+
+int main(int argc, char** argv) {
+    if (argc < 2) return usage();
+    const std::string cmd = argv[1];
+    if (cmd == "install" && argc == 2) return install();
+    if (cmd == "uninstall" && argc == 2) return uninstall();
+    if (cmd == "status" && argc == 2) return status();
+    if (cmd == "update" && argc == 2) return update();
+    if (cmd == "report" && argc == 2) return report();
+    if ((cmd == "version" || cmd == "--version") && argc == 2) return std::printf("%s\n", autopass::version()) < 0;
+    if (cmd == "guard" && argc == 2) return guard(false);
+    if (cmd == "guard" && argc == 3 && std::string(argv[2]) == "reset") return guard(true);
+    if (cmd == "state" && argc <= 3) return state(argc == 3 ? std::atof(argv[2]) : 0);
+    if (cmd == "get" && argc == 2) return get_config();
+    if (cmd == "set" && (argc == 3 || (argc == 4 && std::string(argv[3]) == "--quiet"))) {
+        const std::string which = argv[2];
+        if (which != "rgb" && which != "mono") return usage();
+        return switch_source(which == "rgb", argc == 4);
+    }
+    return usage();
+}

tools/disasm.py

@@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+"""Annotated arm64 disassembly of functions in a stripped shared object.
+
+Used to learn the signatures of SteamVR's private interfaces from the
+headset's own binaries. Resolves adrp+add/ldr pairs to strings or data,
+PLT stubs to imported symbol names, and RELATIVE relocations in data.
+
+Usage: disasm.py LIB ADDR [ADDR ...] [--max N] [--until ADDR]
+       disasm.py --xref LIB TARGET [TARGET ...]
+Addresses are hex. Disassembles linearly from ADDR until a `ret` that is
+not skipped over by an earlier forward branch, or N instructions.
+"""
+
+import re
+import struct
+import subprocess
+import sys
+
+import capstone
+from capstone import arm64_const as A
+
+
+class Elf:
+    def __init__(self, path):
+        self.path = path
+        self.data = open(path, "rb").read()
+        self.sections = []
+        out = subprocess.run(["readelf", "-SW", path], capture_output=True, text=True).stdout
+        for m in re.finditer(r"\]\s+(\S+)\s+\S+\s+([0-9a-f]+)\s+([0-9a-f]+)\s+([0-9a-f]+)", out):
+            name, addr, off, size = m.group(1), int(m.group(2), 16), int(m.group(3), 16), int(m.group(4), 16)
+            self.sections.append((name, addr, off, size))
+        self.relative = {}
+        self.symbolic = {}
+        out = subprocess.run(["readelf", "-rW", path], capture_output=True, text=True).stdout
+        for line in out.splitlines():
+            m = re.match(r"\s*([0-9a-f]+)\s+[0-9a-f]+\s+(R_AARCH64_\w+)\s+(\S+)?\s*(.*)", line)
+            if not m:
+                continue
+            where, kind = int(m.group(1), 16), m.group(2)
+            if kind == "R_AARCH64_RELATIVE":
+                self.relative[where] = int(m.group(3), 16)
+            elif kind in ("R_AARCH64_JUMP_SLOT", "R_AARCH64_GLOB_DAT", "R_AARCH64_ABS64"):
+                rest = (m.group(4) or "").strip()
+                sym = rest.split("+")[0].strip() if rest else ""
+                self.symbolic[where] = sym.split("@")[0]
+        self.plt = self._map_plt()
+
+    def section_of(self, addr):
+        for name, a, off, size in self.sections:
+            if a and a <= addr < a + size:
+                return name, a, off
+        return None
+
+    def read(self, addr, n):
+        s = self.section_of(addr)
+        if not s or s[0] == ".bss":
+            return None
+        return self.data[addr - s[1] + s[2]: addr - s[1] + s[2] + n]
+
+    def cstring(self, addr, limit=160):
+        raw = self.read(addr, limit)
+        if not raw:
+            return None
+        end = raw.find(b"\0")
+        if end <= 0:
+            return None
+        text = raw[:end]
+        if all(32 <= c < 127 or c in (9, 10) for c in text):
+            return text.decode()
+        return None
+
+    def u64(self, addr):
+        raw = self.read(addr, 8)
+        return struct.unpack("<Q", raw)[0] if raw and len(raw) == 8 else None
+
+    def _map_plt(self):
+        plt = {}
+        s = [x for x in self.sections if x[0] == ".plt"]
+        if not s:
+            return plt
+        _, addr, off, size = s[0]
+        md = capstone.Cs(capstone.CS_ARCH_ARM64, capstone.CS_MODE_ARM)
+        md.detail = True
+        page = None
+        for ins in md.disasm(self.data[off:off + size], addr):
+            if ins.id == A.ARM64_INS_ADRP:
+                page = ins.operands[1].imm
+            elif ins.id == A.ARM64_INS_LDR and page is not None and len(ins.operands) > 1:
+                got = page + ins.operands[1].mem.disp
+                sym = self.symbolic.get(got)
+                if sym:
+                    plt[ins.address - 4] = sym
+                page = None
+        return plt
+
+    def describe(self, addr):
+        """Best-effort label for an address used as data."""
+        if addr in self.plt:
+            return "plt:" + self.plt[addr]
+        s = self.cstring(addr)
+        if s:
+            return repr(s)
+        if addr in self.relative:
+            target = self.relative[addr]
+            s = self.cstring(target)
+            return f"-> {target:#x}" + (f" {s!r}" if s else "")
+        if addr in self.symbolic:
+            return "got:" + self.symbolic[addr]
+        sec = self.section_of(addr)
+        return sec[0] if sec else None
+
+
+def disassemble(elf, start, max_ins, until=None):
+    md = capstone.Cs(capstone.CS_ARCH_ARM64, capstone.CS_MODE_ARM)
+    md.detail = True
+    code = elf.read(start, max_ins * 4)
+    regs_page = {}
+    furthest = start
+    for ins in md.disasm(code, start):
+        note = ""
+        ops = ins.operands
+        if ins.id == A.ARM64_INS_ADRP:
+            regs_page[ops[0].reg] = ops[1].imm
+        elif ins.id == A.ARM64_INS_ADD and len(ops) == 3 and ops[1].reg in regs_page and ops[2].type == A.ARM64_OP_IMM:
+            target = regs_page.pop(ops[1].reg) + ops[2].imm
+            note = f"{target:#x} {elf.describe(target) or ''}"
+        elif ins.id in (A.ARM64_INS_LDR, A.ARM64_INS_STR) and len(ops) > 1 and ops[1].type == A.ARM64_OP_MEM \
+                and ops[1].mem.base in regs_page:
+            target = regs_page[ops[1].mem.base] + ops[1].mem.disp
+            note = f"[{target:#x}] {elf.describe(target) or ''}"
+        elif ins.id in (A.ARM64_INS_BL, A.ARM64_INS_B) and ops and ops[0].type == A.ARM64_OP_IMM:
+            target = ops[0].imm
+            label = elf.plt.get(target)
+            note = f"-> {label}" if label else ""
+            if ins.id == A.ARM64_INS_B and target > furthest:
+                furthest = target
+        elif ins.group(capstone.CS_GRP_JUMP) and ops and ops[-1].type == A.ARM64_OP_IMM:
+            if ops[-1].imm > furthest:
+                furthest = ops[-1].imm
+        print(f"  {ins.address:#x}: {ins.mnemonic:8} {ins.op_str:40} {note}".rstrip())
+        if until is not None:
+            if ins.address + 4 >= until:
+                break
+            continue
+        if ins.id == A.ARM64_INS_RET and ins.address >= furthest:
+            break
+        if ins.id == A.ARM64_INS_B and ops and ops[0].type == A.ARM64_OP_IMM and ins.address >= furthest \
+                and ops[0].imm < ins.address:
+            break
+
+
+def xrefs(elf, target):
+    """Addresses in .text whose adrp+add (or adrp+ldr) pair forms `target`."""
+    name, addr, off, size = [s for s in elf.sections if s[0] == ".text"][0]
+    words = struct.unpack_from(f"<{size // 4}I", elf.data, off)
+    hits = []
+    page_of = {}
+    for i, w in enumerate(words):
+        pc = addr + i * 4
+        if (w & 0x9F000000) == 0x90000000:  # adrp
+            rd = w & 0x1F
+            immlo = (w >> 29) & 3
+            immhi = (w >> 5) & 0x7FFFF
+            imm = (immhi << 2) | immlo
+            if imm & (1 << 20):
+                imm -= 1 << 21
+            page_of[rd] = ((pc & ~0xFFF) + (imm << 12), pc)
+        elif (w & 0xFF800000) == 0x91000000:  # add (immediate, 64-bit, no shift)
+            rn = (w >> 5) & 0x1F
+            if rn in page_of and pc - page_of[rn][1] < 64:
+                if page_of[rn][0] + ((w >> 10) & 0xFFF) == target:
+                    hits.append(pc)
+    return hits
+
+
+def main():
+    args = sys.argv[1:]
+    if args and args[0] == "--xref":
+        elf = Elf(args[1])
+        for a in args[2:]:
+            print(a, [hex(h) for h in xrefs(elf, int(a, 16))])
+        return
+    max_ins = 400
+    until = None
+    if "--until" in args:
+        i = args.index("--until")
+        until = int(args[i + 1], 16)
+        del args[i:i + 2]
+    if "--max" in args:
+        i = args.index("--max")
+        max_ins = int(args[i + 1])
+        del args[i:i + 2]
+    elf = Elf(args[0])
+    for a in args[1:]:
+        start = int(a, 16)
+        print(f"== {start:#x}")
+        disassemble(elf, start, max_ins if until is None else (until - start) // 4 + 1, until)
+
+
+if __name__ == "__main__":
+    main()

tools/shm_analyse.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""Offline analysis of a shm_capture.py recording (runs on the PC).
+
+Ground truth is the colour light value (g4) read from the
+VR_CameraPassthroughState file, available only while RGB is selected.
+Every 32-bit word of every other captured buffer is correlated with it
+during the RGB phase; the best candidates are then printed across the
+whole run so we can see whether they still follow the lights while mono
+is selected.
+
+Usage: shm_analyse.py CAPTURE [--top N] [--min-r R]
+"""
+
+import argparse
+import struct
+import zlib
+from collections import defaultdict
+
+import numpy as np
+
+STATE_SIZE = 0x6200
+GAMMA = bytes.fromhex("2fbae83e")
+
+
+def load(path):
+    series = defaultdict(list)
+    with open(path, "rb") as f:
+        data = f.read()
+    pos = 0
+    while pos < len(data):
+        (n,) = struct.unpack_from("<I", data, pos)
+        pos += 4
+        name = data[pos:pos + n].decode()
+        pos += n
+        t, raw_len, z_len = struct.unpack_from("<dII", data, pos)
+        pos += 16
+        buf = zlib.decompress(data[pos:pos + z_len])
+        pos += z_len
+        if len(buf) == raw_len:
+            series[name].append((t, buf))
+    return series
+
+
+def colour_light(buf):
+    best = None
+    for pos in range(0x20E8 + 0xD8, STATE_SIZE - 0x18, 4):
+        if buf[pos:pos + 4] == GAMMA:
+            ts = struct.unpack_from("<d", buf, pos - 0xD8 + 0x10)[0]
+            light = struct.unpack_from("<f", buf, pos + 0x14)[0]
+            if best is None or ts > best[0]:
+                best = (ts, light)
+    return best
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("capture")
+    ap.add_argument("--top", type=int, default=25)
+    ap.add_argument("--min-r", type=float, default=0.8)
+    args = ap.parse_args()
+
+    series = load(args.capture)
+    state_name = next(n for n, s in series.items() if len(s[0][1]) == STATE_SIZE)
+    state = series.pop(state_name)
+
+    # Ground truth: g4 while RGB is selected and the colour record is fresh.
+    truth_t, truth_v, mono_t = [], [], []
+    last_ts = None
+    for t, buf in state:
+        rgb = buf[4]
+        if not rgb:
+            mono_t.append(t)
+            continue
+        cl = colour_light(buf)
+        if cl and cl[0] != last_ts:
+            truth_t.append(t)
+            truth_v.append(cl[1])
+            last_ts = cl[0]
+    truth_t = np.array(truth_t)
+    truth_v = np.array(truth_v)
+    print(f"state file {state_name}: {len(state)} snapshots, {len(truth_t)} fresh RGB light samples, "
+          f"mono from {min(mono_t, default=float('nan')):.1f} to {max(mono_t, default=float('nan')):.1f} s")
+    print("light over time (RGB phase):", " ".join(f"{t:.0f}s={v:.2f}" for t, v in zip(truth_t[::8], truth_v[::8])))
+
+    candidates = []
+    for name, snaps in series.items():
+        times = np.array([t for t, _ in snaps])
+        rgb_idx = [i for i, t in enumerate(times) if truth_t.size and truth_t[0] <= t <= truth_t[-1]
+                   and not any(abs(t - m) < 0.5 for m in mono_t[:1] + mono_t[-1:])
+                   and not (mono_t and mono_t[0] - 1 <= t <= mono_t[-1] + 1)]
+        if len(rgb_idx) < 6:
+            continue
+        size = len(snaps[0][1]) // 4 * 4
+        mat = np.frombuffer(b"".join(snaps[i][1][:size] for i in rgb_idx), dtype="<f4").reshape(len(rgb_idx), -1)
+        ref = np.interp(times[rgb_idx], truth_t, truth_v)
+        with np.errstate(all="ignore"):
+            finite = np.isfinite(mat).all(axis=0)
+            m = np.where(finite, mat, 0).astype(np.float64)
+            m -= m.mean(axis=0)
+            r0 = ref - ref.mean()
+            denom = np.sqrt((m ** 2).sum(axis=0) * (r0 ** 2).sum())
+            r = np.where((denom > 0) & finite, (m * r0[:, None]).sum(axis=0) / denom, 0)
+        for w in np.argsort(-np.abs(r))[:args.top]:
+            if abs(r[w]) >= args.min_r:
+                candidates.append((abs(r[w]), r[w], name, int(w) * 4))
+
+    candidates.sort(reverse=True)
+    print(f"\n{len(candidates)} words with |r| >= {args.min_r} against the RGB light value")
+    for absr, r, name, off in candidates[:args.top]:
+        snaps = series[name]
+        vals = [struct.unpack_from("<f", b, off)[0] for _, b in snaps]
+        rgb_vals = [v for (t, _), v in zip(snaps, vals) if not (mono_t and mono_t[0] <= t <= mono_t[-1])]
+        mono_vals = [v for (t, _), v in zip(snaps, vals) if mono_t and mono_t[0] <= t <= mono_t[-1]]
+        print(f"r={r:+.3f} {name.split('/')[-1]}+{off:#x} rgb range {min(rgb_vals):.4g}..{max(rgb_vals):.4g}"
+              f" | mono range {min(mono_vals, default=float('nan')):.4g}..{max(mono_vals, default=float('nan')):.4g}")
+    return candidates
+
+
+if __name__ == "__main__":
+    main()

tools/shm_capture.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Passively capture every shared-memory file a process maps.
+
+Reads tmpfs files only: no SteamVR calls, no locks, no devices. Used to
+find which of XRService's shared buffers carry colour-camera statistics
+while mono passthrough is displayed.
+
+Each snapshot is appended to the output as a record:
+    u32 name_len, name, f64 t, u32 raw_len, u32 z_len, zlib(data)
+Small files are read every --fast seconds, large ones every --slow.
+
+Usage: shm_capture.py [--pid PID | --process NAME] [--seconds N]
+                      [--fast S] [--slow S] [--large BYTES] [--also PATH] --out PATH
+"""
+
+import argparse
+import os
+import re
+import struct
+import subprocess
+import time
+import zlib
+
+
+def find_pid(pattern):
+    out = subprocess.run(["pgrep", "-f", pattern], capture_output=True, text=True).stdout.split()
+    pids = [int(p) for p in out if int(p) != os.getpid()]
+    if not pids:
+        raise SystemExit(f"no process matches {pattern!r}")
+    return min(pids)
+
+
+def mapped_shm(pid):
+    paths = set()
+    with open(f"/proc/{pid}/maps") as f:
+        for line in f:
+            m = re.search(r"(/dev/shm/\S+)", line)
+            if m:
+                paths.add(m.group(1))
+    for fd in os.listdir(f"/proc/{pid}/fd"):
+        try:
+            target = os.readlink(f"/proc/{pid}/fd/{fd}")
+        except OSError:
+            continue
+        if target.startswith("/dev/shm/"):
+            paths.add(target)
+    return sorted(p for p in paths if os.path.isfile(p))
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--pid", type=int)
+    ap.add_argument("--process", default="XRService --documentsRoot")
+    ap.add_argument("--seconds", type=float, default=90)
+    ap.add_argument("--fast", type=float, default=0.25)
+    ap.add_argument("--slow", type=float, default=2.0)
+    ap.add_argument("--large", type=int, default=1 << 20)
+    ap.add_argument("--also", action="append", default=[],
+                    help="extra file to capture on the fast schedule (repeatable)")
+    ap.add_argument("--out", required=True)
+    args = ap.parse_args()
+
+    pid = args.pid or find_pid(args.process)
+    paths = sorted(set(mapped_shm(pid)) | set(args.also))
+    sizes = {p: os.path.getsize(p) for p in paths}
+    print(f"pid {pid}: {len(paths)} shm files, {sum(sizes.values()) / 1e6:.1f} MB total")
+
+    t0 = time.monotonic()
+    next_slow = 0.0
+    with open(args.out, "wb") as out:
+        while True:
+            now = time.monotonic() - t0
+            if now >= args.seconds:
+                break
+            do_slow = now >= next_slow
+            if do_slow:
+                next_slow = now + args.slow
+            for p in paths:
+                if sizes[p] > args.large and not do_slow:
+                    continue
+                try:
+                    with open(p, "rb") as f:
+                        data = f.read()
+                except OSError:
+                    continue
+                z = zlib.compress(data, 1)
+                name = p.encode()
+                out.write(struct.pack("<I", len(name)) + name + struct.pack("<dII", now, len(data), len(z)) + z)
+            spent = time.monotonic() - t0 - now
+            time.sleep(max(0.0, args.fast - spent))
+    print(f"wrote {os.path.getsize(args.out) / 1e6:.1f} MB")
+
+
+if __name__ == "__main__":
+    main()

tools/shm_records.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Record every new per-frame record of both passthrough streams.
+
+Passive: only reads SteamVR's VR_CameraPassthroughState tmpfs file (the
+25088-byte /dev/shm segment); no SteamVR calls, no devices. For each
+stream and camera, whenever the newest record's timestamp changes, one
+line is written: wall time, stream (mono/colour), camera, record index,
+timestamp and the record's raw bytes as hex. Used to look for per-frame
+fields that follow the IR emitters (FINDINGS.md 38).
+
+Usage: shm_records.py --seconds N [--hz N] --out PATH
+"""
+
+import argparse
+import glob
+import os
+import struct
+import sys
+import time
+
+STATE_SIZE = 0x6200
+STREAMS = {"mono": 0x14, "colour": 0x20E8}
+CAMS, RECS, REC_SIZE, CAM_SIZE = 2, 16, 0x104, 0x1040
+
+
+def find_state_file():
+    m = [p for p in glob.glob("/dev/shm/u1000-Shm_*") if os.path.getsize(p) == STATE_SIZE]
+    if len(m) != 1:
+        sys.exit(f"expected one {STATE_SIZE}-byte shm file, found {m}")
+    return m[0]
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--seconds", type=float, required=True)
+    ap.add_argument("--hz", type=float, default=50)
+    ap.add_argument("--out", required=True)
+    a = ap.parse_args()
+    path = find_state_file()
+    last = {}
+    end = time.time() + a.seconds
+    with open(path, "rb") as f, open(a.out, "w") as out:
+        while time.time() < end:
+            f.seek(0)
+            buf = f.read(STATE_SIZE)
+            now = time.time()
+            out.write(f"{now:.3f} cfg {buf[2:7].hex()}\n") if last.get("cfg") != buf[2:7] else None
+            last["cfg"] = buf[2:7]
+            for name, base in STREAMS.items():
+                for cam in range(CAMS):
+                    best = None
+                    for rec in range(RECS):
+                        off = base + 0x38 + cam * CAM_SIZE + rec * REC_SIZE
+                        ts = struct.unpack_from("<d", buf, off + 0x10)[0]
+                        if ts == ts and (best is None or ts > best[0]):
+                            best = (ts, rec, off)
+                    if best and last.get((name, cam)) != best[0]:
+                        last[(name, cam)] = best[0]
+                        ts, rec, off = best
+                        out.write(f"{now:.3f} {name} {cam} {rec} {ts:.6f} {buf[off:off + REC_SIZE].hex()}\n")
+            time.sleep(1 / a.hz)
+
+
+if __name__ == "__main__":
+    main()

tools/shm_sampler.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""Passively sample SteamVR's VR_CameraPassthroughState shared memory.
+
+Only reads a tmpfs file. It makes no SteamVR calls, takes no locks and
+opens no devices, so it cannot disturb passthrough or tracking. Readings
+may occasionally tear (the writer holds a mutex we deliberately ignore);
+that is acceptable for exploration.
+
+Per-frame records are found by their gamma marker (1/2.2 as float32).
+For each record whose frame counter changed, one CSV row is written with
+the raw fields around the marker, so we can see which ones track light.
+
+With --raw PATH, every snapshot is also appended to PATH as
+[float64 time][STATE_SIZE bytes] so it can be re-parsed offline.
+
+Usage: shm_sampler.py [--seconds N] [--hz N] [--out PATH] [--raw PATH] [--shm PATH]
+"""
+
+import argparse
+import glob
+import os
+import struct
+import sys
+import time
+
+GAMMA = struct.pack("<f", 1 / 2.2)
+STATE_SIZE = 0x6200
+RECORD_BEFORE_GAMMA = 0xD8   # record start is this far before the marker
+FIELDS_AFTER_GAMMA = 12      # float32 fields logged after the marker
+
+
+def find_state_file():
+    matches = [p for p in glob.glob("/dev/shm/u1000-Shm_*") if os.path.getsize(p) == STATE_SIZE]
+    if len(matches) != 1:
+        sys.exit(f"expected exactly one {STATE_SIZE}-byte shm file, found {matches}")
+    return matches[0]
+
+
+def records(buf):
+    pos = buf.find(GAMMA)
+    while pos != -1:
+        start = pos - RECORD_BEFORE_GAMMA
+        if start >= 0:
+            yield start, pos
+        pos = buf.find(GAMMA, pos + 4)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--seconds", type=float, default=120)
+    ap.add_argument("--hz", type=float, default=20)
+    ap.add_argument("--out", default="shm_samples.csv")
+    ap.add_argument("--raw")
+    ap.add_argument("--shm")
+    args = ap.parse_args()
+    path = args.shm or find_state_file()
+
+    header = ["t", "offset", "stream", "id", "frame", "ts0", "ts1", "fx", "fy", "cx", "cy"]
+    header += [f"g{i}" for i in range(FIELDS_AFTER_GAMMA)] + ["owner", "seq", "config"]
+    last_frame = {}
+    t0 = time.monotonic()
+    raw = open(args.raw, "wb") if args.raw else None
+    with open(args.out, "w") as out, open(path, "rb") as f:
+        out.write(",".join(header) + "\n")
+        while time.monotonic() - t0 < args.seconds:
+            f.seek(0)
+            buf = f.read(STATE_SIZE)
+            now = time.monotonic() - t0
+            if raw:
+                raw.write(struct.pack("<d", now) + buf)
+            config = buf[2:7].hex()
+            for start, g in records(buf):
+                ident, frame = struct.unpack_from("<II", buf, start)
+                owner, seq = struct.unpack_from("<II", buf, g + 4 + FIELDS_AFTER_GAMMA * 4)
+                key = start
+                if last_frame.get(key) == (frame, seq):
+                    continue
+                last_frame[key] = (frame, seq)
+                ts0, ts1 = struct.unpack_from("<dd", buf, start + 0x10)
+                fx, fy, cx, cy = struct.unpack_from("<4f", buf, start + 0x20)
+                fields = struct.unpack_from(f"<{FIELDS_AFTER_GAMMA}f", buf, g + 4)
+                stream = 1 if start >= 0x20E8 else 0
+                row = [f"{now:.3f}", hex(start), str(stream), str(ident), str(frame), f"{ts0:.4f}", f"{ts1:.4f}",
+                       f"{fx:.1f}", f"{fy:.1f}", f"{cx:.1f}", f"{cy:.1f}"]
+                row += [f"{v:.6g}" for v in fields] + [str(owner), str(seq), config]
+                out.write(",".join(row) + "\n")
+            time.sleep(1 / args.hz)
+    if raw:
+        raw.close()
+
+
+if __name__ == "__main__":
+    main()