Sylpheed/tools/re-capture/f3_sting_probe.py

#!/usr/bin/env python3
"""F3, the sting half -- watch for a NEW XMA stream during the title's
build-in, with NO input, aligned against a continuous glyph time series so
"when did the plate reach full alpha" is measured, not assumed from a
threshold crossing.

Why continuous, not a threshold trigger: this container's own boot gate
(nav_repeat_and_b.py, f1_hold_capture.py) waits for the glyph count to HOLD
in [500,2500] for 12 samples before calling it "TITLE" -- which could
already be past the build-in's interesting part. This script starts
recording both streams (glyph count, XMA-PARAM arrivals) from the moment
Canary's window exists, so the whole rise from 0 can be read back, not just
the plateau.

Positive control, per R4: BGM cues 1102/1103 are already known to play on
the title (f3-title-plays-bgm-102-and-103.md) via this exact probe
mechanism (menu-audio-cues.md). If this run logs zero XMA-PARAM lines at
all, the probe found nothing INCLUDING the thing it's supposed to find, and
the run is void -- not a negative about a sting.

    f3_sting_probe.py OUTDIR [duration_s]
"""
import os
import re
import subprocess
import sys
import time

import numpy as np

OUT = sys.argv[1]
DURATION = float(sys.argv[2]) if len(sys.argv) > 2 else 200.0
os.makedirs(OUT, exist_ok=True)
W, H = 1280, 720

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 glyph(a):
    r, g, b = a[:, :, 0], a[:, :, 1], a[:, :, 2]
    return int(((g > 130) & (g - r > 45) & (g - b > 45)).sum())


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


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 main():
    with open(env["XENIA_PAD_FILE"], "w"):
        pass
    subprocess.run(["xsetroot", "-solid", "black"], env=env, check=False)

    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 -- run: run-canary "
              "--create_profile_if_none=Tag, wait ~5s, kill it", flush=True)
        return

    canary_log_path = os.path.join(OUT, "canary.stdout")
    canary_log = open(canary_log_path, "w")
    proc = subprocess.Popen(
        ["run-canary", f"--logged_profile_slot_0_xuid={xuid}",
         "--xma_param_probe=true", "--log_level=2"],
        cwd=OUT, env=env, stdout=canary_log, stderr=subprocess.STDOUT)
    print(f"canary pid={proc.pid}, xma_param_probe=true, waiting for window",
          flush=True)

    T0 = time.time()
    while not subprocess.run(
            ["xdotool", "search", "--name", "Xenia-canary"],
            capture_output=True, text=True).stdout.strip():
        if time.time() - T0 > 60:
            print("FATAL: no window after 60s", flush=True)
            return
        time.sleep(1)
    print(f"[{time.time()-T0:6.1f}s] window exists, recording", flush=True)

    glyph_out = open(os.path.join(OUT, "glyph-timeseries.tsv"), "w")
    glyph_out.write("# t_s\tglyph\n")
    xma_seen = set()
    xma_out = open(os.path.join(OUT, "xma-param-arrivals.tsv"), "w")
    xma_out.write("# t_s\tline\n")
    XMA_RE = re.compile(rb"XMA-PARAM.*")

    p, n = _open(), W * H * 3
    seg = time.time()
    log_pos = 0
    while time.time() - T0 < DURATION:
        el = time.time() - T0
        if time.time() - seg > 30:
            p.kill(); p = _open(); seg = time.time()
        a = grab(p, n)
        if a is not None:
            g = glyph(a)
            glyph_out.write(f"{el:.2f}\t{g}\n")
            glyph_out.flush()
        else:
            p.kill(); p = _open(); seg = time.time()
        # Drain any new XMA-PARAM lines that arrived since last check --
        # stamped on ARRIVAL (Xenia's own log lines carry no timestamp),
        # same technique xma_readoff_trace.py already uses.
        try:
            with open(canary_log_path, "rb") as f:
                f.seek(log_pos)
                chunk = f.read()
                log_pos = f.tell()
        except FileNotFoundError:
            chunk = b""
        for line in chunk.splitlines():
            if XMA_RE.search(line):
                key = line
                if key not in xma_seen:
                    xma_seen.add(key)
                    xma_out.write(f"{el:.2f}\t{line.decode('utf-8','replace')}\n")
                    xma_out.flush()
                    print(f"[{el:7.1f}s] NEW {line.decode('utf-8','replace')}",
                          flush=True)
    p.kill()
    glyph_out.close()
    xma_out.close()
    print(f"[{time.time()-T0:7.1f}s] killing emulator, "
          f"{len(xma_seen)} distinct XMA-PARAM lines seen", flush=True)
    proc.terminate()
    try:
        proc.wait(timeout=10)
    except subprocess.TimeoutExpired:
        proc.kill()
    canary_log.close()


if __name__ == "__main__":
    main()