#!/usr/bin/env python3 """Time the boot title: when the PRESS (A) plate arrives, and what a press costs. WHY THIS EXISTS. Four durations published on 2026-08-29 were withdrawn the same day because `screen_match.classify_array` costs 1503 ms/frame and a probe calling it per frame drained an 8 fps x11grab at 0.64 fps. A backlog PRESERVES ORDERING and DESTROYS DURATIONS, so every "latency" it produced was really the queue depth. See docs/re/menu-idle-and-b-2026-08-29.md. So this probe is built the other way round: * per-frame work is a few MILLISECONDS, not 1.5 s. The cost in screen_match is the +/-8 px offset search over a full-res surface (25 znccs); every committed capture aligns at exactly dy=0 dx=0 (five screens, +/-2 px search, five-screens-acceptance.md), so this classifier decimates 4x and does ONE zncc per reference. --control checks that shortcut against the same fixtures screen_match uses, INCLUDING the movie-frame negatives. * the stream is torn down and restarted every RESTART_S, because a long-lived x11grab degrades and then freezes on a stale frame (fast_title_probe.py). * an INDEPENDENT one-shot grab every CHECK_S is compared with the stream's own latest frame. A stalled stream cannot pass that, and the check is logged so a negative result can be audited rather than believed. * the loop's real sample rate is reported. If frames/elapsed is not close to the requested rate, the durations in the log are NOT trustworthy and the probe says so in its own summary. Every frame is written to a TSV; the durations are computed offline from it, so nothing here depends on the probe having classified in real time. title_timing_probe.py --control title_timing_probe.py --run SECONDS OUT.tsv """ import os import subprocess import sys import time import numpy as np from PIL import Image REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) CAP = os.path.join(REPO, "docs", "re", "captures") SD = os.path.dirname(os.path.abspath(__file__)) W, H = 1280, 720 SURFACE_TOP = 45 # xenia window chrome; the game surface is 1279x675 DS = 4 # decimation for both live frames and references RATE = 8 # requested frames/s RESTART_S = 30 # a long-lived x11grab freezes on a stale frame CHECK_S = 20 # independent one-shot grab, cross-checked against the stream THRESH = 0.70 DISPLAY = os.environ.get("DISPLAY", ":98") REFS = { # the interactive title WITH the plate -- what the run is waiting to see arrive "title_plate": "title-builds/live-title-press-a.png", # the same screen BEFORE the plate. This is the reference the plate delay is # measured from, and it is a committed capture, not a render of ours. "title_noplate": "title-builds/live-title-build4-no-plate.png", "menu": "title-builds/live-main-menu.png", } def surface(a): h, w = a.shape if h == H and w == W: return a[SURFACE_TOP:, :1279] return a def load_gray(p): return np.asarray(Image.open(p).convert("L"), dtype=np.float32) _R = {} def refs(): if not _R: for k, v in REFS.items(): r = surface(load_gray(os.path.join(CAP, v)))[::DS, ::DS] _R[k] = (r - r.mean()) / (np.sqrt((r * r).sum() - r.size * r.mean() ** 2) or 1.0) return _R def scores(gray): """ZNCC of a frame against every reference, decimated, NO offset search.""" img = surface(gray)[::DS, ::DS] out = {} for k, rn in refs().items(): h = min(img.shape[0], rn.shape[0]) w = min(img.shape[1], rn.shape[1]) x = img[:h, :w] y = rn[:h, :w] xc = x - x.mean() d = np.sqrt((xc * xc).sum()) out[k] = float((xc * y).sum() / d) if d else 0.0 return out def label(sc): k = max(sc, key=sc.get) return k if sc[k] >= THRESH else "other" def glyph(rgb): """Byte-identical to is_title.py's counter.""" r = rgb[:, :, 0].astype(np.int16) g = rgb[:, :, 1].astype(np.int16) b = rgb[:, :, 2].astype(np.int16) return int(((g > 130) & (g - r > 45) & (g - b > 45)).sum()) def gray_of(rgb): return (0.299 * rgb[:, :, 0] + 0.587 * rgb[:, :, 1] + 0.114 * rgb[:, :, 2]).astype(np.float32) # ---------------------------------------------------------------- control CONTROLS = [ (os.path.join(CAP, "title-builds/live-title-press-a.png"), "title_plate"), (os.path.join(CAP, "title-screen-oracle.png"), "title_plate"), (os.path.join(CAP, "title-builds/live-title-build4-no-plate.png"), "title_noplate"), (os.path.join(CAP, "title-builds/live-main-menu.png"), "menu"), (os.path.join(CAP, "main-menu-oracle.png"), "menu"), (os.path.join(CAP, "main-menu-reached.png"), "menu"), # the class this oracle exists to reject (os.path.join(CAP, "instrument-controls/movie-frame-attract-a.png"), "other"), (os.path.join(CAP, "instrument-controls/movie-frame-attract-b.png"), "other"), (os.path.join(CAP, "difficulty-screen.png"), "other"), ] # The plate detector is a THRESHOLD on the glyph counter, so it needs its own # control: the committed no-plate title reads ~159 and plate titles 753..1493. GLYPH_CONTROLS = [ (os.path.join(CAP, "title-builds/live-title-build4-no-plate.png"), "lo"), (os.path.join(CAP, "title-builds/live-title-press-a.png"), "hi"), (os.path.join(CAP, "instrument-controls/movie-frame-attract-a.png"), "lo"), (os.path.join(CAP, "instrument-controls/movie-frame-attract-b.png"), "lo"), ] PLATE_GLYPH = 400 def control(): bad = 0 print("--- content classifier (decimated, no offset search) ---") for p, exp in CONTROLS: if not os.path.exists(p): print(f" SKIP (missing) {os.path.basename(p)}") continue t = time.time() sc = scores(load_gray(p)) got = label(sc) ms = (time.time() - t) * 1000 ok = got == exp bad += 0 if ok else 1 print(f" {'ok ' if ok else 'FAIL'} {os.path.basename(p):<36} -> {got:<13} " f"(exp {exp:<13}) " + " ".join(f"{k}={v:+.3f}" for k, v in sc.items()) + f" [{ms:.1f} ms]") print(f"\n--- plate detector (glyph >= {PLATE_GLYPH}) ---") for p, exp in GLYPH_CONTROLS: if not os.path.exists(p): print(f" SKIP (missing) {os.path.basename(p)}") continue n = glyph(np.asarray(Image.open(p).convert("RGB"))) got = "hi" if n >= PLATE_GLYPH else "lo" ok = got == exp bad += 0 if ok else 1 print(f" {'ok ' if ok else 'FAIL'} {os.path.basename(p):<36} glyph={n:<6} -> {got} (exp {exp})") print(f"\n{'CONTROL PASSED' if not bad else f'CONTROL FAILED ({bad})'}") return 1 if bad else 0 # ---------------------------------------------------------------- live run def open_stream(): return subprocess.Popen( ["ffmpeg", "-loglevel", "error", "-f", "x11grab", "-draw_mouse", "0", "-video_size", f"{W}x{H}", "-i", DISPLAY, "-r", str(RATE), "-f", "rawvideo", "-pix_fmt", "rgb24", "-"], stdout=subprocess.PIPE, bufsize=W * H * 3 * 2) def oneshot(): """An INDEPENDENT grab, through a fresh short-lived process.""" p = subprocess.run( ["ffmpeg", "-loglevel", "error", "-f", "x11grab", "-draw_mouse", "0", "-video_size", f"{W}x{H}", "-i", DISPLAY, "-frames:v", "1", "-f", "rawvideo", "-pix_fmt", "rgb24", "-"], stdout=subprocess.PIPE, timeout=20) b = p.stdout if len(b) < W * H * 3: return None return np.frombuffer(b[:W * H * 3], np.uint8).reshape(H, W, 3) PAD = os.environ.get("XENIA_PAD_FILE", "/tmp/xenia_pad.txt") def _pad_write(state): tmp = PAD + ".tmp" with open(tmp, "w") as f: f.write(state) os.replace(tmp, PAD) def tap(button, secs=0.25): """Press INLINE and return the moment the press landed. pad.py through subprocess.run costs a python start plus the hold before the caller can timestamp anything, so run 1's press times were ~0.3 s late with no way to tell how late. Same file, same rename-into-place, no interpreter. """ _pad_write(f"press={button}") t = time.time() time.sleep(secs) _pad_write("") return t def run(limit, out_path, shots_dir): os.makedirs(shots_dir, exist_ok=True) n = W * H * 3 p = open_stream() t0 = time.time() seg = t0 chk = t0 frames = 0 saved = set() ev = [] # (name, t) -- ordering only; durations come from the TSV state = "wait" # wait -> title -> plate -> pressedA -> menu -> pressedB -> done last_gray = None prev_mean = -1.0 same = 0 longest_same = 0 fh = open(out_path, "w") fh.write("#t\tglyph\tmean\tmotion\ttitle_plate\ttitle_noplate\tmenu\tlabel\n") def mark(name): t = time.time() - t0 ev.append((name, t)) print(f"EVENT {name} t={t:.3f}", flush=True) return t title_seen_at = None while time.time() - t0 < limit and state != "done": now = time.time() # 🔴 Do NOT restart once the measurement is under way. Run 1 restarted # 0.25 s after the (A) press and then reported 14 byte-identical frames # over 1.5 s -- a stale stream straddling exactly the interval being # timed, which is how a press latency gets inflated by 1.5 s. The # degradation the restart guards against is a minutes-scale drift # (fast_title_probe.py); the whole measuring window is under 30 s, so # freezing the stream for it is strictly safer than restarting inside it. if state == "wait" and now - seg > RESTART_S: p.kill() p = open_stream() seg = now fh.write(f"#restart\t{now - t0:.3f}\n") buf = p.stdout.read(n) if len(buf) < n: p.kill() p = open_stream() seg = time.time() continue t = time.time() - t0 rgb = np.frombuffer(buf, np.uint8).reshape(H, W, 3) g = gray_of(rgb) gl = glyph(rgb) sc = scores(g) lb = label(sc) surf = surface(g) mn = float(surf.mean()) mo = float(np.abs(surf[::8, ::8] - last_gray).mean()) if last_gray is not None else -1.0 last_gray = surf[::8, ::8].copy() frames += 1 if abs(mn - prev_mean) < 1e-6: same += 1 longest_same = max(longest_same, same) else: same = 0 prev_mean = mn fh.write(f"{t:.3f}\t{gl}\t{mn:.3f}\t{mo:.3f}\t{sc['title_plate']:+.4f}\t" f"{sc['title_noplate']:+.4f}\t{sc['menu']:+.4f}\t{lb}\n") # --- independent cross-check that the stream is not stale if time.time() - chk > CHECK_S: chk = time.time() o = oneshot() if o is None: fh.write(f"#check\t{t:.3f}\tONESHOT_FAILED\n") else: om = float(surface(gray_of(o)).mean()) fh.write(f"#check\t{t:.3f}\tstream={mn:.3f}\toneshot={om:.3f}\t" f"delta={abs(om - mn):.3f}\n") fh.flush() # --- the drive. DO NOT press during a movie: a run that taps through # the intro reaches a title that accepts nothing (skip_intro.sh). if state == "wait": if lb in ("title_noplate", "title_plate") and 0 <= mo < 2.0: title_seen_at = mark("title_static") if gl >= PLATE_GLYPH: mark("plate_already") # would mean the plate is not late state = "plate" else: state = "title" Image.fromarray(rgb).save(os.path.join(shots_dir, "t0-title.png")) elif state == "title": if gl >= PLATE_GLYPH: mark("plate") Image.fromarray(rgb).save(os.path.join(shots_dir, "t1-plate.png")) state = "plate" plate_at = t elif state == "plate": if t - ev[-1][1] > 5.0: tp = tap("A") - t0 ev.append(("pressA", tp)) print(f"EVENT pressA t={tp:.3f}", flush=True) state = "pressedA" elif state == "pressedA": if lb == "menu": mark("menu") Image.fromarray(rgb).save(os.path.join(shots_dir, "t2-menu.png")) state = "menu" elif state == "menu": if t - ev[-1][1] > 8.0: tp = tap("B") - t0 ev.append(("pressB", tp)) print(f"EVENT pressB t={tp:.3f}", flush=True) state = "pressedB" elif state == "pressedB": if lb in ("title_plate", "title_noplate"): mark("back_title") Image.fromarray(rgb).save(os.path.join(shots_dir, "t3-back-title.png")) state = "done" p.kill() dt = time.time() - t0 fps = frames / dt if dt else 0 fh.write(f"#summary\tframes={frames}\telapsed={dt:.1f}\tfps={fps:.2f}\trequested={RATE}" f"\tlongest_identical_run={longest_same}\n") for name, t in ev: fh.write(f"#event\t{name}\t{t:.3f}\n") fh.close() print(f"\n{frames} frames in {dt:.1f}s = {fps:.2f} fps (requested {RATE})") print(f"longest run of byte-identical surface means: {longest_same} frames " f"({longest_same / RATE:.2f} s at the requested rate)") if fps < RATE * 0.75: print("🔴 SAMPLE RATE FELL BELOW 75% OF REQUESTED — durations in this log " "are NOT trustworthy (this is the backlog failure mode).") for name, t in ev: print(f" {name:<14} {t:8.3f}") return 0 if __name__ == "__main__": if len(sys.argv) > 1 and sys.argv[1] == "--control": sys.exit(control()) if len(sys.argv) > 3 and sys.argv[1] == "--run": sys.exit(run(float(sys.argv[2]), sys.argv[3], sys.argv[4] if len(sys.argv) > 4 else "/sylph-home/re/shots/title-timing")) print(__doc__) sys.exit(2)