Sylpheed/tools/re-capture/frame_vs_audio_clock.py

#!/usr/bin/env python3
"""Does this container's FRAME clock run at the same rate as its AUDIO clock?

The corpus reads "the game presents at 27.6 fps" and "the splash dwells run 8.5 %
long" off wall-clock windows on one container. A guest running ~92 % of real time
produces identical numbers, and three trials sharing a container cannot separate
them (ui-keyframe-time-unit.md, corrected 2026-08-30).

The AUDIO clock is already bounded: BGM_103's loop cycle is 22.03 M / 22.68 M bits
of a stream whose declared rate makes it 62.34 / 63.29 media-seconds, against a
measured 61.87 s wall -- ratio 0.985, where a uniform 8.5 % slowdown predicts
1.085. ⚠️ But audio on a GPU-less box can hold real time while RENDERING lags, so
that bounds the audio clock only.

This measures a UI ANIMATION's period in the same run as the audio rate. If the
frame clock ran slow while audio did not, the animation's wall-clock period would
exceed its declared period while the audio rate stayed nominal.

🔴 CONTROL FIRST, and the run is void without it: the same detector must recover
the TITLE plate's period, which this corpus has measured four times at
2.12-2.34 s and twice by mid-crossings at 2.530 / 2.540 s. An estimator that
cannot find a known period cannot be trusted on an unknown one.

Phase 2 also answers a question sylpheed-port has open: whether a FOCUSED MENU
BUTTON glows at all. Eleven focus records declare a 120-unit cycle; only the
plate is authored to animate.

    frame_vs_audio_clock.py LOG OUTDIR [title_s] [menu_s]
"""
import os, re, subprocess, sys, time
import numpy as np
from PIL import Image

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from ring_row import ring_row

LOG, OUT = sys.argv[1], sys.argv[2]
TITLE_S = float(sys.argv[3]) if len(sys.argv) > 3 else 60.0
MENU_S = float(sys.argv[4]) if len(sys.argv) > 4 else 90.0
W, H = 1280, 720
PAD = os.path.join(os.path.dirname(os.path.abspath(__file__)), "pad.py")
RO = re.compile(rb"XmaContext(?:Fake)? (\d+): Looped Data: (\d+) < (\d+) \(Start: (\d+)\)")
T0 = time.time()
os.makedirs(OUT, exist_ok=True)


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


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


def deliveries(vk):
    try:
        return len(re.findall(
            (r"RE-INPUT\] XamInputGetKeystrokeEx -> user=\d+ vk=%s flags=0001" % vk).encode(),
            open(LOG, "rb").read()))
    except FileNotFoundError:
        return 0


def press(btn, vk):
    for _ in range(5):
        before = deliveries(vk)
        subprocess.run([sys.executable, PAD, "tap", btn, "0.5"], check=False)
        for _ in range(20):
            time.sleep(0.25)
            if deliveries(vk) > before:
                return True
    return False


def sample(seconds, signal, tag):
    """Sample `signal(frame)` at ~10 Hz for `seconds`. Returns (t[], v[])."""
    p = _open(); n = W * H * 3
    ts, vs = [], []
    t_end = time.time() + seconds
    seg = time.time()
    while time.time() < t_end:
        if time.time() - seg > 25:
            p.kill(); p = _open(); seg = time.time()
        b = p.stdout.read(n)
        if len(b) < n:
            p.kill(); p = _open(); seg = time.time(); continue
        a = np.frombuffer(b, np.uint8).reshape(H, W, 3).astype(int)
        ts.append(time.time() - T0); vs.append(signal(a))
    p.kill()
    np.savetxt(f"{OUT}/{tag}.tsv", np.column_stack([ts, vs]), fmt="%.4f", delimiter="\t")
    return np.array(ts), np.array(vs, dtype=float)


def period(ts, vs, lo=0.6, hi=8.0):
    """Dominant period by autocorrelation on a uniform resample."""
    if len(ts) < 40:
        return None, 0.0
    dt = np.median(np.diff(ts))
    if not np.isfinite(dt) or dt <= 0:
        return None, 0.0
    grid = np.arange(ts[0], ts[-1], dt)
    y = np.interp(grid, ts, vs)
    y = y - y.mean()
    if y.std() == 0:
        return None, 0.0
    ac = np.correlate(y, y, "full")[len(y) - 1:]
    ac /= ac[0]
    k0, k1 = max(1, int(lo / dt)), min(len(ac) - 1, int(hi / dt))
    if k1 <= k0:
        return None, 0.0
    k = k0 + int(np.argmax(ac[k0:k1]))
    return k * dt, float(ac[k])


def audio_rate(window):
    """bits/s of read_offset progress over the last `window` seconds of log."""
    try:
        data = open(LOG, "rb").read()
    except FileNotFoundError:
        return {}
    per = {}
    for m in RO.finditer(data):
        per.setdefault(int(m.group(1)), []).append(int(m.group(2)))
    return {c: (v[0], v[-1], len(v)) for c, v in per.items()}


# ---- phase 1: the TITLE plate, as the control -------------------------------
print(f"[{time.time()-T0:7.1f}s] waiting for the title", flush=True)
p = _open(); n = W * H * 3
while True:
    b = p.stdout.read(n)
    if len(b) < n:
        p.kill(); p = _open(); continue
    a = np.frombuffer(b, np.uint8).reshape(H, W, 3).astype(int)
    if 500 <= glyph(a) <= 2500:
        break
    if time.time() - T0 > 900:
        p.kill(); sys.exit("never reached the title")
p.kill()
print(f"[{time.time()-T0:7.1f}s] TITLE — sampling the plate for {TITLE_S:.0f}s", flush=True)
a0 = audio_rate(0)
t1, v1 = sample(TITLE_S, glyph, "title-plate")
per1, ac1 = period(t1, v1)
print(f"   plate period = {per1:.3f} s (autocorr {ac1:.2f}) from {len(t1)} samples"
      if per1 else "   plate period: NOT FOUND", flush=True)

# ---- phase 2: the MENU ------------------------------------------------------
print(f"[{time.time()-T0:7.1f}s] pressing A for the menu", flush=True)
press("A", "5800")
time.sleep(12)
p = _open()
b = p.stdout.read(n); p.kill()
a = np.frombuffer(b, np.uint8).reshape(H, W, 3).astype(int) if len(b) == n else None
y = ring_row(Image.fromarray(a.astype(np.uint8))) if a is not None else None
print(f"[{time.time()-T0:7.1f}s] on the menu, ring y = {y}", flush=True)
if y is None:
    sys.exit("no ring row on the menu — cannot place the glow window")
lo, hi = int(y) - 26, int(y) + 26
ab = audio_rate(0)
t2, v2 = sample(MENU_S, lambda f: float(f[lo:hi, 480:900].mean()), "menu-focus")
per2, ac2 = period(t2, v2)
ae = audio_rate(0)

print("\n================ RESULT ================")
print(f"CONTROL  title plate period : {per1:.3f} s  (autocorr {ac1:.2f})"
      if per1 else "CONTROL  title plate period : NOT FOUND")
print(f"         corpus mid-crossings: 2.530 / 2.540 s")
print(f"MEASURE  menu focus period  : {per2:.3f} s  (autocorr {ac2:.2f})"
      if per2 else "MEASURE  menu focus period  : NOT FOUND — no periodic glow detected")
print("\nAUDIO read_offset progress during the run (bits):")
for c in sorted(ae):
    s, e, k = ae[c]
    print(f"   ctx{c}: {s:,} -> {e:,}  ({k} samples)")
print("FRAME VS AUDIO CLOCK RUN DONE", flush=True)