Sylpheed/tools/re-capture/title_timing_probe.py

#!/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)