Sylpheed/tools/re-capture/boot_timeline_probe.py

"""Time the whole boot, and measure the PRESENTATION RATE per screen.

Two jobs, one oracle session, because they need each other.

1. **The boot timeline the port asked for.** Every screen-to-screen transition
   from launch to the main menu, wall-clock, with the black holds marked. The
   port paces its boot off `ScreenView.settle_time()` = a group's `rest.t`, and
   `rest.t` is NOT when a screen settles (docs/re/REFUTED.md) — so every screen's
   dwell is currently wrong by an unknown amount.

2. **Frames, not seconds.** Two pages of the corpus measure the same declared
   120 keyframe units during a static hold and disagree by 2 %: settle→plate is
   2.135 s (28.10 fps implied) and one focus-ring revolution is 2.177 s (27.56
   implied). Either the presentation rate differed between those sessions, or one
   interval is not 120 units. The two were taken on DIFFERENT SCREENS, so this
   measures the rate on each — with the game's own frame counter, not a guess
   about what changes between grabs.

`--log_ui_draws=true --ui_draw_capture_frames=N` makes Canary log
`[UI-CAP] capture armed` and then `[UI-CAP] done: D draws over F frames`. Timing
between those two lines in its own stdout gives frames/second directly, and it
re-arms (the log index is `{:02d}`), so one session can measure several screens.

⚠️ The instrument can perturb what it measures — writing a draw log costs the
emulator something. Control built in: the ring period is measured both DURING a
capture and OUTSIDE one, and a rate that is an artefact of logging would move it.

    boot_timeline_probe.py --control
    boot_timeline_probe.py --run SECONDS OUT.tsv CANARY_STDOUT [shots_dir]
"""
import os
import subprocess
import sys
import time

import numpy as np
from PIL import Image

SD = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SD)
import title_timing_probe as T  # noqa: E402  (same crop, same ZNCC, same controls)

REPO = os.path.dirname(os.path.dirname(SD))
CAP = os.path.join(REPO, "docs", "re", "captures")

# The boot shows two splashes before the movie; both are committed captures.
T.REFS["splash_pub"] = "title-builds/live-splash-publisher.png"
T.REFS["splash_dev"] = "title-builds/live-splash-developer.png"
T._R.clear()

RATE = 8
CAPTURE_FRAMES = int(os.environ.get("UICAP_FRAMES", "300"))


def control():
    """Every control title_timing_probe has, plus the two splashes."""
    T.CONTROLS.extend([
        (os.path.join(CAP, "title-builds/live-splash-publisher.png"), "splash_pub"),
        (os.path.join(CAP, "title-builds/live-splash-developer.png"), "splash_dev"),
    ])
    return T.control()


def _tail(path, seen):
    """New lines appended to the emulator's stdout since the last call."""
    try:
        with open(path, "rb") as f:
            f.seek(seen)
            b = f.read()
        return b.decode("utf-8", "replace"), seen + len(b)
    except OSError:
        return "", seen


def arm_capture(log_path, seen, timeout=90.0):
    """Press F10, then time the emulator's own armed->done lines.

    Timed between the two LOG lines, not from the keypress: the arm latency is
    then excluded rather than folded into the rate.
    """
    win = subprocess.run(["xdotool", "search", "--name", "Xenia-canary"],
                         capture_output=True, text=True).stdout.split()
    if not win:
        return None, seen
    w = win[-1]
    subprocess.run(["xdotool", "windowactivate", w], capture_output=True)
    subprocess.run(["xdotool", "key", "--window", w, "F10"], capture_output=True)
    subprocess.run(["xdotool", "key", "F10"], capture_output=True)
    t_armed = t_done = None
    frames = draws = None
    deadline = time.time() + timeout
    while time.time() < deadline:
        chunk, seen = _tail(log_path, seen)
        for line in chunk.splitlines():
            if "[UI-CAP] capture armed" in line and t_armed is None:
                t_armed = time.time()
            elif "[UI-CAP] done" in line and t_armed is not None:
                t_done = time.time()
                # "[UI-CAP] done: 1526 draws over 300 frames"
                parts = line.replace(":", " ").split()
                try:
                    draws = int(parts[parts.index("done") + 1])
                    frames = int(parts[parts.index("over") + 1])
                except (ValueError, IndexError):
                    pass
        if t_done:
            break
        time.sleep(0.02)
    if not (t_armed and t_done and frames):
        return None, seen
    dt = t_done - t_armed
    return {"frames": frames, "draws": draws, "seconds": dt, "fps": frames / dt}, seen


def run(limit, out_path, log_path, shots_dir):
    os.makedirs(shots_dir, exist_ok=True)
    n = T.W * T.H * 3
    p = T.open_stream()
    t0 = time.time()
    seg = t0
    frames = 0
    seen = 0
    ev = []
    rates = {}
    state = "boot"
    last_label = None
    last_gray = None
    prev_mean = -1.0
    fh = open(out_path, "w")
    fh.write("#t\tglyph\tmean\tmotion\t" + "\t".join(T.REFS) + "\tlabel\n")

    def mark(name, t=None):
        t = time.time() - t0 if t is None else t
        ev.append((name, t))
        print(f"EVENT {name} t={t:.3f}", flush=True)
        return t

    while time.time() - t0 < limit and state != "done":
        now = time.time()
        # Restart only while still waiting; never across a measured interval.
        if state == "boot" and now - seg > 30:
            p.kill(); p = T.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 = T.open_stream(); seg = time.time(); continue
        t = time.time() - t0
        rgb = np.frombuffer(buf, np.uint8).reshape(T.H, T.W, 3)
        g = T.gray_of(rgb)
        gl = T.glyph(rgb)
        sc = T.scores(g)
        lb = T.label(sc)
        surf = T.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
        fh.write(f"{t:.3f}\t{gl}\t{mn:.3f}\t{mo:.3f}\t"
                 + "\t".join(f"{sc[k]:+.4f}" for k in T.REFS) + f"\t{lb}\n")

        # Every label change and every entry/exit from pure black is a boot event.
        blk = "black" if mn < 1.0 else lb
        if blk != last_label:
            mark(f"screen:{blk}", t)
            last_label = blk
            if blk in ("splash_pub", "splash_dev", "title_noplate", "menu"):
                Image.fromarray(rgb).save(os.path.join(shots_dir, f"boot-{blk}.png"))

        if state == "boot":
            if lb in ("title_noplate", "title_plate") and 0 <= mo < 2.0 and gl >= 100:
                mark("title_settled", t); state = "title"
        elif state == "title":
            if gl >= T.PLATE_GLYPH:
                mark("plate", t); state = "plate_hold"; hold_from = t
        elif state == "plate_hold":
            if t - ev[-1][1] > 3.0:
                fh.flush()
                r, seen = arm_capture(log_path, seen)
                rates["title"] = r
                mark(f"rate_title={r and round(r['fps'], 3)}")
                state = "press"
        elif state == "press":
            tp = T.tap("A") - t0
            ev.append(("pressA", tp)); print(f"EVENT pressA t={tp:.3f}", flush=True)
            state = "toMenu"
        elif state == "toMenu":
            if lb == "menu":
                mark("menu", t); state = "menuSettle"; menu_at = t
        elif state == "menuSettle":
            if t - ev[-1][1] > 6.0:
                fh.write(f"#ring_free_start\t{t:.3f}\n")
                state = "ringFree"; ring_from = t
        elif state == "ringFree":
            # 12 s of ring with NOTHING else running -- the outside-capture control
            if t - ring_from > 12.0:
                fh.write(f"#ring_free_end\t{t:.3f}\n")
                fh.flush()
                r, seen = arm_capture(log_path, seen)
                rates["menu"] = r
                mark(f"rate_menu={r and round(r['fps'], 3)}")
                fh.write(f"#ring_capture_end\t{time.time()-t0:.3f}\n")
                state = "ringAfter"; after_from = time.time() - t0
        elif state == "ringAfter":
            if t - after_from > 12.0:
                state = "done"

    p.kill()
    dt = time.time() - t0
    fh.write(f"#summary\tframes={frames}\telapsed={dt:.1f}\tfps={frames/dt:.2f}\trequested={RATE}\n")
    for k, r in rates.items():
        if r:
            fh.write(f"#rate\t{k}\tframes={r['frames']}\tdraws={r['draws']}"
                     f"\tseconds={r['seconds']:.3f}\tfps={r['fps']:.4f}\n")
        else:
            fh.write(f"#rate\t{k}\tFAILED\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 = {frames/dt:.2f} fps (requested {RATE})")
    for k, r in rates.items():
        print(f"  presentation rate on {k}: "
              + (f"{r['fps']:.4f} fps ({r['frames']} frames in {r['seconds']:.3f} s, "
                 f"{r['draws']} draws)" if r else "FAILED"))
    return 0


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--control":
        sys.exit(control())
    if len(sys.argv) > 4 and sys.argv[1] == "--run":
        sys.exit(run(float(sys.argv[2]), sys.argv[3], sys.argv[4],
                     sys.argv[5] if len(sys.argv) > 5 else "/sylph-home/re/shots/boot-timeline"))
    print(__doc__)
    sys.exit(2)