tools: a title timing probe that costs 8.7 ms a frame, not 1503
The four durations withdrawn yesterday were produced by a classifier costing
1503 ms/frame draining an 8 fps x11grab at 0.64 fps -- a backlog, which
preserves ordering and destroys durations. This is the instrument for retaking
them.
What makes it cheap: every committed capture aligns at exactly dy=0 dx=0
(five-screens-acceptance), so the +/-8 px offset search screen_match does at
full resolution is 25 ZNCCs buying nothing on this path. Decimate 4x, do one
ZNCC per reference. Measured 8.7 ms per frame including the glyph count -- 173x.
Controls, run before the measurement and not after it:
* 9/9 content controls, including the two committed movie frames that are the
class this oracle exists to reject;
* 4/4 on the plate detector itself, which is a threshold on the green-glyph
counter and so needs its own control (no-plate title 159, plate title 753,
movie frames 0).
And three things learned from run 1, folded back in:
* do NOT restart the stream once the measurement is under way. Run 1's restart
landed 0.25 s after the (A) press and its stale frames straddled exactly the
interval being timed;
* press INLINE, not through pad.py's subprocess -- an interpreter start plus
the 0.25 s hold sat between the press and the timestamp;
* count the longest run of byte-identical surface means and report it. That is
the freeze signature, and it is how run 2 showed the 26.626 hold is the
guest rather than the capture path.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014voBspJ6kFncNErZJuZcLw
This commit is contained in:
sylph-decoder
368
tools/re-capture/title_timing_probe.py
Executable file
368
tools/re-capture/title_timing_probe.py
Executable file
@@ -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)
|
||||
Reference in New Issue
Block a user