diff --git a/tools/re-capture/title_timing_probe.py b/tools/re-capture/title_timing_probe.py new file mode 100755 index 00000000..af098ee5 --- /dev/null +++ b/tools/re-capture/title_timing_probe.py @@ -0,0 +1,368 @@ +#!/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)