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

#!/usr/bin/env python3
"""Measure the craft's speed law and check it against its definition.

The unit definition gives `Acceleration`, `Deceleration`, `MinimumVelocity`,
`CruisingVelocity` and `MaximumVelocity` (harvested 2026-08-13), but not how the
game applies them. This holds each throttle input and samples the player's own
position at ~10 Hz, so speed and its rate of change are measured rather than
assumed.

Why not `ctrl_probe.py`: it re-scans for the player when it starts, and that scan
misses in roughly half the samples on a busy stage (entities move while it walks
memory). This locks onto the player object ONCE and then only re-reads its
position triple, which is why it survives a firefight.

Usage: speed_law.py <out.csv> [hold_s]
"""
import glob, os, struct, subprocess, sys, time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import gworld, entities2


def require_live_emulator():
    """Refuse to measure a corpse.

    A killed Canary leaves its `/dev/shm/xenia_memory_*` image behind AND leaves its
    last frame on the X display, so a dead emulator looks alive from both of the
    obvious angles: the screenshot shows a mission in progress and the memory image
    still parses. The scans then report `0 moving triples` / "player entity not
    found", which reads like a tooling bug rather than a dead game. Note `pgrep -x
    xenia_canary` is NOT enough — zombies match it — so the state letter is checked.
    """
    out = subprocess.run(["ps", "-o", "pid=,stat=", "-C", "xenia_canary"],
                         capture_output=True, text=True).stdout
    live = [l for l in out.splitlines() if l.split()[1:2] and not l.split()[1].startswith("Z")]
    if not live:
        stale = glob.glob("/dev/shm/xenia_memory_*")
        sys.exit(f"no live xenia_canary (zombies only); {len(stale)} stale shm image(s) "
                 f"— the screen and the memory file are both leftovers, relaunch first")

def find_player(retries=8):
    """Lock onto the player object that is actually FLYING.

    Verifies the emulator is alive first — see `require_live_emulator`.

    A mission holds more than one `*_Player` object — `entities2 list` shows two —
    and at least one of them never moves. Taking the first match locks onto that
    one, and then every probe reads a speed of exactly 0 while the game is visibly
    flying (and the HUD keeps reading 350). So sample each candidate twice and keep
    the one that displaces."""
    require_live_emulator()
    for _ in range(retries):
        w = gworld.World()
        defs = entities2.definitions(w)
        movers = entities2.moving(w.fd, w.size)
        cands = [(off, nm) for off, nm, _, _ in entities2.typed(w.fd, defs, movers, 0x130)
                 if nm.endswith("_Player")]
        best = None
        for off, nm in cands:
            a = pos_at(w.fd, off)
            time.sleep(0.5)
            b = pos_at(w.fd, off)
            d = sum((b[i] - a[i]) ** 2 for i in range(3)) ** 0.5
            if best is None or d > best[0]:
                best = (d, off, nm)
        # A craft can be briefly slow (launch, a hard turn), so do not demand a big
        # step — just prefer the candidate that moves at all. The failure this
        # guards against is the STATIC duplicate, which never moves by any amount.
        if best and best[0] > 0.05:
            return w, best[1], best[2]
        print(f"# {len(cands)} player candidates, best displacement "
              f"{best[0] if best else 0:.3f} — retrying", flush=True)
        time.sleep(2)
    return None, None, None

def pos_at(fd, off):
    b = os.pread(fd, 12, off)
    return struct.unpack(">3f", b)

def pad(*args):
    subprocess.run(["vgamepad", *args], capture_output=True)

def sample(fd, off, seconds, hz=20):
    """Raw (t, x, y, z) samples. Speed is NOT computed per sample: the guest
    updates the position at its own rate, so a 10 Hz difference alternates
    between 0 and a double step — the first run of this probe read 0, 1519, 1985,
    0, 2681 … for a craft flying smoothly. Differentiate over a window instead."""
    out, t0 = [], time.time()
    while time.time() - t0 < seconds:
        p = pos_at(fd, off)
        out.append((round(time.time() - t0, 3), p[0], p[1], p[2]))
        time.sleep(1.0 / hz)
    return out

def main():
    out_csv = sys.argv[1]
    hold = float(sys.argv[2]) if len(sys.argv) > 2 else 8.0
    w, off, nm = find_player()
    if not w:
        sys.exit("player entity not found after retries")
    print(f"# locked on {nm} at file offset {off:#x}")
    rows = []
    # RT/LT are ANALOGUE TRIGGERS, not buttons: `vgamepad trig RT 1.0` holds full
    # throttle, `trig RT 0.0` releases. Using `hold RT` (the button verb) is a
    # silent no-op — the first run of this probe measured only the drift of a
    # craft nobody was flying.
    for label, keys in (("idle", [("RT", 0.0), ("LT", 0.0)]),
                        ("RT", [("RT", 1.0), ("LT", 0.0)]),
                        ("coast", [("RT", 0.0), ("LT", 0.0)]),
                        ("LT", [("LT", 1.0), ("RT", 0.0)]),
                        ("coast2", [("RT", 0.0), ("LT", 0.0)])):
        for axis, v in keys:
            pad("trig", axis, str(v))
        seq = sample(w.fd, off, hold)
        for t, x, y, z in seq:
            rows.append((label, t, x, y, z))
        # windowed speed: displacement over the phase's middle second
        mid = [s for s in seq if seq[-1][0] * 0.4 <= s[0] <= seq[-1][0] * 0.9]
        if len(mid) > 2:
            d = sum((mid[-1][i] - mid[0][i]) ** 2 for i in (1, 2, 3)) ** 0.5
            print(f"# {label:<8} windowed speed {d / (mid[-1][0] - mid[0][0]):8.1f}/s")
    pad("trig", "RT", "0.0"); pad("trig", "LT", "0.0")
    with open(out_csv, "w") as f:
        f.write("phase,t,x,y,z\n")
        for r in rows:
            f.write(f"{r[0]},{r[1]},{r[2]},{r[3]},{r[4]}\n")
    print(f"# wrote {out_csv} ({len(rows)} samples)")

if __name__ == "__main__":
    main()