"""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)