Syplheed-Reborn/tools/re-capture/flight_probe.py

#!/usr/bin/env python3
"""Drive a scripted input sequence in flight while sampling guest RAM, so the
player craft's live transform can be *derived* rather than guessed.

The point is correlation: each sample is tagged with the stick/trigger state
that produced it, so afterwards we can ask "which floats move only while the
yaw axis is deflected?" and "which triple integrates to the velocity?".

Writes a .npz-ish flat binary: a header of (va, offset, name) per instance,
then per frame a timestamp, the input vector, and every instance's raw window.

Usage: flight_probe.py <out.bin> [seconds]
"""
import os
import struct
import sys
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import gworld  # noqa: E402

# (the old vgamepad FIFO constant is gone with the server it named)


class Pad:
    """Drive Canary's `--hid=file` pad by rewriting its state file.

    🔴 REWRITTEN 2026-08-23, and this is the whole reason `pilot.py` never fired.
    This class used to write lines to `/tmp/sylph-vgamepad.fifo`, "the vgamepad
    server's FIFO". That server does not exist any more — the uinput pad was
    replaced by the `--hid=file` driver, because a uinput device is not
    namespaced and every scripted press leaked to the HOST's desktop. The FIFO is
    now an ordinary 91-byte file that **nothing reads**, so every axis, trigger
    and button this class emitted went nowhere, silently, and the craft flew on
    its own for the whole of every "autopilot" run.

    The corpus already carries this exact trap for the MENU path
    (`canary-scripted-input-traps.md`: "pad.py, NOT vgamepad … every call here
    failed silently"). The shell scripts were fixed then; this class was not, and
    every flight tool imports it — `pilot.py`, `autopilot3.py`, `aim_probe.py`,
    `flight_probe.py` itself.

    Measured symptom, on a run confirmed to be animating throughout: full left
    and full right stick for four seconds each, and the ship's **velocity
    direction did not move by 0.01°** — heading change 0.00° in every phase while
    it travelled 350–735 units. With this rewrite the same probe is what checks
    the fix.

    Format is the driver's own (`file_input_driver.h`): `key=value` pairs, with
    `press=A,START`, `lt`/`rt` 0..255 and `lx`/`ly`/`rx`/`ry` -32768..32767.
    Written through a temp file and renamed, so a poll can never see a partial
    state; the driver re-parses on any nanosecond-mtime or size change.
    """

    PATH = os.environ.get("XENIA_PAD_FILE", "/tmp/xenia_pad.txt")

    def __init__(self):
        self.state = {"LX": 0.0, "LY": 0.0, "RX": 0.0, "RY": 0.0,
                      "LT": 0.0, "RT": 0.0}
        self.buttons = set()
        self._write()

    def _write(self):
        parts = []
        if self.buttons:
            parts.append("press=" + ",".join(sorted(self.buttons)))
        for k, key in (("LT", "lt"), ("RT", "rt")):
            v = int(round(max(0.0, min(1.0, self.state[k])) * 255))
            if v:
                parts.append(f"{key}={v}")
        for k, key in (("LX", "lx"), ("LY", "ly"), ("RX", "rx"), ("RY", "ry")):
            v = int(round(max(-1.0, min(1.0, self.state[k])) * 32767))
            if v:
                parts.append(f"{key}={v}")
        tmp = self.PATH + ".tmp"
        with open(tmp, "w") as f:
            f.write(" ".join(parts))
        os.replace(tmp, self.PATH)

    def axis(self, name, v):
        self.state[name] = v
        self._write()

    def trig(self, name, v):
        self.state[name] = v
        self._write()

    def press(self, b):
        self.buttons.add(b.upper())
        self._write()

    def release(self, b):
        self.buttons.discard(b.upper())
        self._write()

    def reset(self):
        for k in self.state:
            self.state[k] = 0.0
        self.buttons.clear()
        self._write()

    def vector(self):
        return [self.state[k] for k in ("LX", "LY", "RX", "RY", "LT", "RT")]



# (duration_s, description, action)
SCRIPT = [
    (3.0, "idle", lambda p: p.reset()),
    (4.0, "pitch-up", lambda p: p.axis("LY", -0.9)),
    (2.0, "idle2", lambda p: p.reset()),
    (4.0, "yaw-right", lambda p: p.axis("LX", 0.9)),
    (2.0, "idle3", lambda p: p.reset()),
    (4.0, "yaw-left", lambda p: p.axis("LX", -0.9)),
    (2.0, "idle4", lambda p: p.reset()),
    (5.0, "boost", lambda p: p.trig("RT", 1.0)),
    (3.0, "idle5", lambda p: p.reset()),
]


def main():
    out = sys.argv[1]
    w = gworld.World()
    inst = w.refresh()
    print(f"# {len(inst)} instances", flush=True)
    if not inst:
        sys.exit("no instances — not in a mission?")

    pad = Pad()
    hz = 10.0
    with open(out, "wb") as f:
        f.write(b"SYLPHPRB")
        f.write(struct.pack("<III", len(inst), gworld.WINDOW, 6))
        for va, off, nm in inst:
            f.write(struct.pack("<IQ", va, off) + nm.encode()[:63].ljust(64, b"\0"))
        t_start = time.time()
        for dur, label, act in SCRIPT:
            act(pad)
            print(f"# {label} ({dur}s)", flush=True)
            n = int(dur * hz)
            for _ in range(n):
                t = time.time()
                f.write(struct.pack("<d", t - t_start))
                f.write(struct.pack("<6f", *pad.vector()))
                f.write(struct.pack("<32s", label.encode()[:32]))
                for va, off, nm in inst:
                    f.write(w.read_off(off, gworld.WINDOW).ljust(gworld.WINDOW, b"\0"))
                time.sleep(max(0, 1.0 / hz - (time.time() - t)))
        pad.reset()
    print(f"# wrote {out} ({os.path.getsize(out)/1e6:.1f} MB)", flush=True)


if __name__ == "__main__":
    main()