Sylpheed/tools/re-capture/f1_hold_capture.py

#!/usr/bin/env python3
"""F1 -- hold a direction on the SETTLED main menu, and read the cursor's
position PER FRAME off the draw log, not a coarse screen-diff.

Why this instrument and not another pass of nav_repeat_and_b.py's screen-diff:
that detector grabs frames at ~4-5 fps (`ffmpeg -r 4`), and f1-no-repeat-was-
the-harness.md's 2026-09-12 update found C_PAD_RINGBUF carries analog-axis-
shaped fields, not a keystroke queue -- meaning the file driver's continuous
GetState() was always capable of driving a real repeat, and a coarse detector
could plausibly alias a fast one down to "one spike". The draw log has no such
ceiling: every submitted quad, every frame, at whatever rate the guest
presents.

Reuses nav_repeat_and_b.py's proven boot-to-menu gate (glyph counting over a
live x11grab pipe) verbatim in spirit -- that gate is the part of this
apparatus already known to work -- and replaces its *measurement* half.

    f1_hold_capture.py OUTDIR [hold_secs]
"""
import os
import subprocess
import sys
import time

import numpy as np

OUT = sys.argv[1]
HOLD_S = float(sys.argv[2]) if len(sys.argv) > 2 else 2.5
os.makedirs(OUT, exist_ok=True)
SD = os.path.dirname(os.path.abspath(__file__))
PAD = os.path.join(SD, "pad.py")
W, H = 1280, 720
NEED, CEIL, HOLD_N = 500, 2500, 12
MENU_LO, MENU_HI, MENU_HOLD = 250, 420, 6
WAIT_S = 520

env = dict(os.environ)
env["HOME"] = "/sylph-home/re"
env["SDL_AUDIODRIVER"] = "dummy"
env["DISPLAY"] = ":98"
env["XENIA_PAD_FILE"] = os.path.join(OUT, "pad.txt")


def pad(state):
    tmp = env["XENIA_PAD_FILE"] + ".tmp"
    with open(tmp, "w") as f:
        f.write(state)
    os.replace(tmp, env["XENIA_PAD_FILE"])


def tap(button, secs=0.5):
    # NOT a subprocess to pad.py: that spawns with os.environ, not this
    # script's local `env` dict, so it would write /tmp/xenia_pad.txt while
    # Canary watches OUT/pad.txt -- an unobserved press that looks identical
    # to a dead pad. Cost one full 520s boot the first time this ran.
    pad(f"press={button}")
    time.sleep(secs)
    pad("")


def _open():
    return subprocess.Popen(
        ["ffmpeg", "-loglevel", "error", "-f", "x11grab", "-draw_mouse", "0",
         "-video_size", f"{W}x{H}", "-i", ":98", "-r", "4",
         "-f", "rawvideo", "-pix_fmt", "rgb24", "-"],
        stdout=subprocess.PIPE, bufsize=W * H * 3 * 2)


def grab(p, n):
    buf = p.stdout.read(n)
    if len(buf) < n:
        return None
    return np.frombuffer(buf, np.uint8).reshape(H, W, 3).astype(float)


def glyph(a):
    r, g, b = a[:, :, 0], a[:, :, 1], a[:, :, 2]
    return int(((g > 130) & (g - r > 45) & (g - b > 45)).sum())


def alive():
    out = subprocess.run(["ps", "-o", "pid=,stat=", "-C", "xenia_canary"],
                          capture_output=True, text=True).stdout
    return [ln.split()[0] for ln in out.splitlines() if "Z" not in ln.split()[1]] if out.strip() else []


def xdotool(*args):
    return subprocess.run(["xdotool", *args], capture_output=True, text=True)


def arm_f10():
    r = xdotool("search", "--name", "Xenia-canary")
    wins = [w for w in r.stdout.split() if w]
    if not wins:
        print("FATAL: no Xenia window to arm F10", flush=True)
        return False
    win = wins[-1]
    xdotool("windowactivate", win)
    xdotool("windowfocus", win)
    xdotool("key", "--window", win, "F10")
    xdotool("key", "F10")
    print(f"armed F10 (win={win})", flush=True)
    return True


def main():
    pad("")
    # Not cosmetic: the X root keeps a DEAD session's last frame, so a fresh
    # launch's first grabs can read a stale window from the previous run and
    # falsely classify it as "title" before the new process has a window at
    # all -- skip_intro.sh blanks the root for exactly this reason. Cost one
    # full 520s run here (glyph matched at 2.6s, long before any real window
    # could exist).
    subprocess.run(["xsetroot", "-solid", "black"], env=env, check=False)
    canary_log = open(os.path.join(OUT, "canary.stdout"), "w")
    xuid = os.environ.get("SYLPH_XUID", "")
    if not xuid:
        content = "/sylph-home/re/.local/share/Xenia/content"
        entries = os.listdir(content) if os.path.isdir(content) else []
        xuid = entries[0] if entries else ""
    if not xuid:
        print("FATAL: no profile signed in and none found under content/ -- "
              "run: run-canary --create_profile_if_none=Tag, wait ~5s, kill it",
              flush=True)
        return
    print(f"signing in profile {xuid}", flush=True)
    proc = subprocess.Popen(
        ["run-canary", f"--logged_profile_slot_0_xuid={xuid}"],
        cwd=OUT, env=env, stdout=canary_log, stderr=subprocess.STDOUT)
    print(f"canary pid={proc.pid}, waiting for window", flush=True)

    T0 = time.time()
    p, n = _open(), W * H * 3
    seg = time.time()
    phase, streak = "wait", 0
    result = {}
    while True:
        el = time.time() - T0
        if el > WAIT_S:
            print(f"TIMEOUT in phase {phase} at {el:.1f}s", flush=True)
            break
        if time.time() - seg > 30:
            p.kill(); p = _open(); seg = time.time()
        a = grab(p, n)
        if a is None:
            p.kill(); p = _open(); seg = time.time()
            continue
        c = glyph(a)
        if phase == "wait":
            streak = streak + 1 if NEED <= c <= CEIL else 0
            if streak >= HOLD_N:
                print(f"[{el:7.1f}s] TITLE (glyph {c})", flush=True)
                tap("A", 0.5)
                phase, streak = "tomenu", 0
        elif phase == "tomenu":
            streak = streak + 1 if MENU_LO <= c <= MENU_HI else 0
            if streak >= MENU_HOLD:
                print(f"[{el:7.1f}s] MENU (glyph {c}) -- settling 2s then arming", flush=True)
                time.sleep(2.0)
                if not arm_f10():
                    result["error"] = "f10 arm failed"
                    break
                pre_hold_ts = time.time()
                print(f"[{time.time()-T0:7.1f}s] HOLDING DOWN for {HOLD_S}s", flush=True)
                pad("press=DOWN")
                time.sleep(HOLD_S)
                pad("")
                release_ts = time.time()
                print(f"[{time.time()-T0:7.1f}s] RELEASED, waiting 2s tail", flush=True)
                time.sleep(2.0)
                result["hold_started_wall"] = pre_hold_ts
                result["hold_released_wall"] = release_ts
                phase = "done"
                break
    p.kill()
    print(f"[{time.time()-T0:7.1f}s] killing emulator", flush=True)
    proc.terminate()
    try:
        proc.wait(timeout=10)
    except subprocess.TimeoutExpired:
        proc.kill()
    canary_log.close()
    with open(os.path.join(OUT, "result.txt"), "w") as f:
        for k, v in result.items():
            f.write(f"{k}\t{v}\n")
        f.write(f"phase_at_exit\t{phase}\n")
    logs = [f for f in os.listdir(OUT) if f.startswith("xenia_re_ui_draws_")]
    print(f"done. phase={phase}, draw logs: {logs}", flush=True)


if __name__ == "__main__":
    main()