diff --git a/docs/re/autopilot-memory-driven.md b/docs/re/autopilot-memory-driven.md new file mode 100644 index 0000000..e97c7ad --- /dev/null +++ b/docs/re/autopilot-memory-driven.md @@ -0,0 +1,104 @@ +# Memory-driven autopilot โ€” build log and current state + +**Status: ๐ŸŸก PARTIAL โ€” infrastructure works, the craft is not yet flown.** +Started 2026-07-29. This is the honest state, not a plan: what is proven, what +is not, and the one thing that most cheaply unblocks the rest. + +## Goal + +Fly and fight a mission by reading the game's own world state out of guest RAM +and driving the pad from it โ€” the RE payoff being an oracle for the +reimplementation's flight model and AI, and a way to reach missions the save +cannot otherwise reach (unit coverage for +[unit-struct-runtime](structures/unit-struct-runtime.md) is capped at 21/110 +because definitions load **per stage**). + +## What works (verified) + +| Piece | Tool | Evidence | +|---|---|---| +| Live guest-RAM reads at loop rate | `gworld.py` | `/dev/shm` file opened once, `pread` per tick; a whole-RAM scan is ~6 s, a targeted read is microseconds | +| Whole-RAM float scanning | numpy over `SEEK_DATA` extents | 1 270 orthonormal 3ร—3 blocks located in 6.2 s | +| Entity enumeration by unit type | `gworld.py entities` | 116 live instances in Stage 02, typed by unit ID, incl. exactly one `โ€ฆ_Player` | +| Pad control at loop rate | `flight_probe.Pad` | writes command lines straight into the vgamepad FIFO; the `vgamepad` CLI spawns a process per command and its `tap`/`hold` sleep *inside* the server, so neither is usable in a control loop | +| Unattended mission entry | `launch_mission.sh` | title โ†’ LOAD GAME โ†’ slot 01 โ†’ READY ROOM โ†’ TAKE OFF โ†’ in flight, repeatable | +| Hangar loadout | `launch_mission.sh --hangar` | the "Recommended" control is **AUTO SELECT โ€” "Mount most suitable weapons"**, already the default cursor position. At 5 % progress it is a no-op: only two weapons are developed, and they are already mounted | +| Finding moving objects | `findplayer.py` | position triples recovered from motion alone โ€” straight-line, constant-speed filter over K whole-RAM samples | + +## What is NOT solved + +**Identifying our own craft, and typing the other entities.** Both remain open, +and the autopilot cannot work without them. + +1. **The `0x820af030` class is not the live entity.** It has one object per + spawned thing and carries the unit-ID string, so it looked like the entity + list โ€” but over a 29 s in-flight capture **all 384 words of it are + constant** (`whatchanges.py`). It is a static spawn record. The earlier + claim in `unit-struct-runtime.md` that this class is "the spawned entity + instance โ€” live state" is **wrong in the second half**: it is per-spawn, but + it is not live state. The parts of that document that depend on the + *definition* class `0x820af844` are unaffected. +2. **No transform in or one hop from that object**: no orthonormal 3ร—3, and no + unit quaternion, within `0x2000` of it or behind any of its 121 pointers. +3. **Input correlation finds *a* self-object, but not obviously the craft.** + Holding hard-left then hard-right yaw and looking for a position whose turn + axis reverses (`findself.py`, `selfstate.py`) gives clean hits + (`cos โ‰ˆ โˆ’0.99`), but they cluster at `0x40009xxx` in what looks like an + 8-corner box with ยฑ45 000 coordinates โ€” a camera/skybox volume that follows + the player, not the craft. Its speed (โ‰ˆ359/s) is suspiciously close to the + HUD's 350, which supports "follows the player" but is not proof of identity. +4. **Entity typing is unavailable**: no definition pointer within ยฑ0x800 of the + self-position, so the trick of learning one object's layout and applying it + to all the others has nothing to anchor on. Without typing, the 33 418 + moving triples in a firefight cannot be separated into enemies, friendlies + and bullets, so there is nothing to aim at. + +## Dead ends, recorded so they are not re-run + +* **Speed-scan for the player object** (`findspeed.py`, the classic two-state + value scan: coast โ†’ boost โ†’ coast). Sound method, but **`RT` is not the + throttle** โ€” 5 864 floats matched the cruise speed and none rose. The control + actually bound to acceleration was never established, and the run that would + have established it ended in GAME OVER. +* **Comparing orientation matrices 2 s apart.** At a real turn rate that is far + outside the small-angle regime, so the skew part of `AยทBแต€` is not the rotation + vector and the "angular velocity" comes out as ~30 000. Sample incrementally + (6 Hz) and re-check orthonormality on every read โ€” blocks found by a scan get + overwritten between the scan and the read. + +## The next step that unblocks the most + +**Find the game's own target pointer instead of typing entities ourselves.** +The HUD has a lock-on system (a `TARGET` marker and a target-cycle button), so +a global almost certainly holds a pointer to the currently-targeted entity. +Reading that gives an enemy's live object address directly โ€” which yields both +target selection *and* the entity layout (position offset within it), i.e. it +collapses problems 3 and 4 into one. It is also cheap to find: cycle the target +with the pad and watch which pointer-shaped global changes in step. + +## Operational notes + +* An unattended craft **dies** โ€” the ship flies straight into a firefight, and + two long scans were invalidated by a GAME OVER mid-run. Any scan longer than + ~30 s needs either a survivable holding pattern or a fresh mission. +* `Xvfb` and the emulator die on their own every few minutes here, cleanly + (exit 0), cause unidentified. Everything that must not be interrupted is run + as **one background task** that starts the display, the emulator, the + navigation and the measurement together, so nothing has to survive between + tool calls. +* `pgrep` cannot be used for liveness in this container: PID 1 is + `sleep infinity` and never reaps, so dead processes linger as `` and + still match by name. Use `ps -o stat=` and skip `Z`, or `xdpyinfo` for X. +* numpy is not installed system-wide; `pip install --break-system-packages + numpy` puts it in `/sylph-home/.local`, which is only on `sys.path` when + `HOME=/sylph-home`. Scripts run with `HOME=/sylph-home/re` need + `PYTHONPATH=/sylph-home/.local/lib/python3.12/site-packages`. + +## Files + +`gworld.py` (live reader + entity list) ยท `flight_probe.py` (scripted inputs + +sampling, and the `Pad` FIFO client) ยท `flight_analyze.py` ยท `whatchanges.py` +(encoding-agnostic "which words are live") ยท `findplayer.py` ยท `findself.py` ยท +`findrot_global.py` ยท `findspeed.py` ยท `liveents.py` ยท `selfstate.py` (the +whole chain โ†’ JSON) ยท `autopilot2.py` (PD controller; **untested โ€” it has never +had a valid config to run against**) ยท `launch_mission.sh`. diff --git a/docs/re/structures/unit-struct-runtime.md b/docs/re/structures/unit-struct-runtime.md index dfad2e3..57b4836 100644 --- a/docs/re/structures/unit-struct-runtime.md +++ b/docs/re/structures/unit-struct-runtime.md @@ -44,9 +44,16 @@ The two vtables are **two different things**, and telling them apart matters: | vtable | what it is | evidence | |---|---|---| -| `0x820af030` | **spawned entity instance** โ€” live state | 12 objects for 4 IDs; the same ID appears many times (one per box in the scene); irregular spacing | +| `0x820af030` | **spawned entity record** โ€” one per spawned thing, but **not live state** (see below) | 12 objects for 4 IDs; the same ID appears many times (one per box in the scene); irregular spacing | | `0x820af844` | **parsed definition** โ€” the `.tbl` | exactly one object per distinct unit ID; minimum spacing `0x380` | +**Correction (2026-07-29, from the autopilot work):** `0x820af030` was +described here as holding live state. It does not โ€” across a 29 s in-flight +capture **all 384 words of it are constant**. It is one record per spawned +thing, but the flying entity's transform is somewhere else entirely. See +[autopilot-memory-driven](../autopilot-memory-driven.md). Nothing below depends +on it; the definition class `0x820af844` is unaffected. + Only `0x820af844` is used. It is the runtime image of the `.tbl`: * **Within one run** it is byte-identical across two snapshots taken ~12 minutes diff --git a/tools/re-capture/autopilot2.py b/tools/re-capture/autopilot2.py new file mode 100644 index 0000000..9e3bb3e --- /dev/null +++ b/tools/re-capture/autopilot2.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +"""Memory-driven autopilot: fly and fight from the game's own world state. + +The earlier screen-scraping autopilot oscillated because it only ever saw a +2-D arrow on the HUD and had no rate feedback. This one reads the actual +transform of every entity out of guest RAM, so it can do proper PD control: +the derivative term uses the craft's real body angular velocity, recovered from +two consecutive rotation matrices, not a differenced pixel position. + + world state <- /dev/shm/xenia_memory_* (see gworld.py) + control -> the vgamepad FIFO, written directly at loop rate + +Offsets come from flight_analyze.py and are passed in / stored in offsets.json; +nothing here hard-codes a value that was not derived from a capture. +""" +import json +import math +import os +import struct +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gworld # noqa: E402 +from flight_probe import Pad # noqa: E402 + + +# --------------------------------------------------------------- 3-D helpers + +def dot(a, b): + return a[0] * b[0] + a[1] * b[1] + a[2] * b[2] + + +def sub(a, b): + return (a[0] - b[0], a[1] - b[1], a[2] - b[2]) + + +def norm(a): + return math.sqrt(dot(a, a)) + + +def clamp(v, lo=-1.0, hi=1.0): + return max(lo, min(hi, v)) + + +def body_rates(R_prev, R, dt): + """Angular velocity in body axes from two rotation matrices. + + R rows are the craft's axes in world space, so R_prev @ R^T is the + incremental rotation; its skew part is the rotation vector. + """ + if dt <= 0: + return (0.0, 0.0, 0.0) + # M = R_prev * R^T (3x3, row-major tuples) + M = [[sum(R_prev[i * 3 + k] * R[j * 3 + k] for k in range(3)) for j in range(3)] + for i in range(3)] + wx = (M[2][1] - M[1][2]) / 2.0 + wy = (M[0][2] - M[2][0]) / 2.0 + wz = (M[1][0] - M[0][1]) / 2.0 + return (wx / dt, wy / dt, wz / dt) + + +# ------------------------------------------------------------------- reader + +class Flight: + def __init__(self, cfg): + self.cfg = cfg + self.w = gworld.World() + self.pos_off = cfg["pos"] + self.rot_off = cfg["rot"] + self.hp_off = cfg.get("hp") + self.player_name = cfg.get("player", "Player") + self.entities = [] + self.last_scan = 0.0 + + def rescan(self): + self.entities = self.w.refresh() + self.last_scan = time.time() + + def state(self, off): + buf = self.w.read_off(off, gworld.WINDOW) + if len(buf) < gworld.WINDOW: + return None + pos = struct.unpack_from(">3f", buf, self.pos_off) + rot = struct.unpack_from(">9f", buf, self.rot_off) + if not all(map(math.isfinite, pos + rot)): + return None + hp = struct.unpack_from(">f", buf, self.hp_off)[0] if self.hp_off else None + return {"pos": pos, "rot": rot, "hp": hp} + + def player(self): + for va, off, nm in self.entities: + if self.player_name in nm: + s = self.state(off) + if s: + s["name"] = nm + return s + return None + + def hostiles(self): + """ADAN units are the enemy; the disc IDs encode the faction. + + `UN_e*` / `UN_be*` = ADAN, `UN_f*` / `UN_bf*` = TCAF (ours). + """ + out = [] + for va, off, nm in self.entities: + base = nm[3:] + if not (base.startswith("e") or base.startswith("be")): + continue + s = self.state(off) + if s: + s["name"] = nm + s["off"] = off + out.append(s) + return out + + +# ---------------------------------------------------------------- controller + +class Autopilot: + KP_YAW, KD_YAW = 1.6, 0.35 + KP_PITCH, KD_PITCH = 1.6, 0.35 + FIRE_CONE = math.radians(6) + FIRE_RANGE = 4000.0 + + def __init__(self, flight, pad, log=sys.stdout): + self.f = flight + self.pad = pad + self.log = log + self.prev_rot = None + self.prev_t = None + self.target = None + self.firing = False + + def pick_target(self, me): + hs = self.f.hostiles() + if not hs: + return None + # nearest, mildly preferring what is already ahead of us + def score(h): + v = sub(h["pos"], me["pos"]) + d = norm(v) or 1.0 + fwd = me["rot"][6:9] + ahead = dot(v, fwd) / d + return d * (1.0 if ahead > 0 else 2.0) + return min(hs, key=score) + + def step(self, dt): + me = self.f.player() + if not me: + return "no-player" + R = me["rot"] + w = body_rates(self.prev_rot, R, dt) if self.prev_rot else (0, 0, 0) + self.prev_rot = R + + tgt = self.pick_target(me) + if not tgt: + self.pad.axis("LX", 0.0) + self.pad.axis("LY", 0.0) + self.pad.trig("RT", 0.6) + return "no-target" + + v = sub(tgt["pos"], me["pos"]) + dist = norm(v) or 1.0 + # into body axes: rows are right / up / forward + lx = dot(R[0:3], v) + ly = dot(R[3:6], v) + lz = dot(R[6:9], v) + yaw_err = math.atan2(lx, lz if lz > 1e-3 else 1e-3) + pitch_err = math.atan2(ly, lz if lz > 1e-3 else 1e-3) + if lz < 0: # behind us: turn the short way, hard + yaw_err = math.copysign(math.pi / 2, lx if lx else 1.0) + + stick_x = clamp(self.KP_YAW * yaw_err - self.KD_YAW * w[1]) + stick_y = clamp(-(self.KP_PITCH * pitch_err - self.KD_PITCH * w[0])) + self.pad.axis("LX", stick_x) + self.pad.axis("LY", stick_y) + self.pad.trig("RT", 1.0 if dist > 1500 else 0.3) + + aligned = abs(yaw_err) < self.FIRE_CONE and abs(pitch_err) < self.FIRE_CONE + want_fire = aligned and dist < self.FIRE_RANGE + if want_fire != self.firing: + (self.pad.press if want_fire else self.pad.release)(self.f.cfg["fire_btn"]) + self.firing = want_fire + return (f"tgt={tgt['name'][3:20]:<18} d={dist:8.0f} yaw={math.degrees(yaw_err):+6.1f} " + f"pit={math.degrees(pitch_err):+6.1f} stick=({stick_x:+.2f},{stick_y:+.2f}) " + f"fire={int(self.firing)} hp={me['hp']}") + + def run(self, seconds, hz=15.0): + self.f.rescan() + t0 = time.time() + last = t0 + while time.time() - t0 < seconds: + t = time.time() + if t - self.f.last_scan > 3.0: + self.f.rescan() + msg = self.step(t - last) + last = t + print(f"[{t-t0:6.1f}] {msg}", file=self.log, flush=True) + time.sleep(max(0, 1.0 / hz - (time.time() - t))) + self.pad.reset() + + +def main(): + cfg = json.load(open(sys.argv[1])) + secs = float(sys.argv[2]) if len(sys.argv) > 2 else 60.0 + ap = Autopilot(Flight(cfg), Pad()) + ap.run(secs) + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/findplayer.py b/tools/re-capture/findplayer.py new file mode 100644 index 0000000..0749281 --- /dev/null +++ b/tools/re-capture/findplayer.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +"""Find the player craft's live position in guest RAM, from motion alone. + +The spawn records at vtable 0x820af030 turned out to hold no live state (every +word is constant across a 29 s capture), so the flying entity is some other +object. Rather than guess which class, this finds it by what it must *do*: + + a position triple moves in a straight line at constant speed while coasting. + +Method: sample all of RAM K times ~dt apart, keep word indices whose float +value changes every time, then require three consecutive such words to satisfy + * equal step length each interval (constant speed), and + * a constant direction (straight flight), + * with speed in a sane range for a craft. + +Only the wholly-mechanical properties of motion are used; no offset, class or +encoding is assumed. + +Usage: findplayer.py [samples] [interval_s] +""" +import math +import os +import struct +import sys +import time + +import numpy as np + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gmem # noqa: E402 + + +def snapshot(fd, size): + """[(start_off, np.ndarray big-endian f32)] over allocated extents.""" + out = [] + for a, b in gmem.extents(fd, size): + n = (b - a) // 4 * 4 + if n < 64: + continue + buf = os.pread(fd, n, a) + out.append((a, np.frombuffer(buf, dtype=">f4"))) + return out + + +def main(): + K = int(sys.argv[1]) if len(sys.argv) > 1 else 5 + dt = float(sys.argv[2]) if len(sys.argv) > 2 else 0.35 + path = gmem.mem_path() + fd = os.open(path, os.O_RDONLY) + size = os.path.getsize(path) + + print(f"# sampling {K}x every {dt}s", flush=True) + snaps, times = [], [] + for k in range(K): + t = time.time() + snaps.append(snapshot(fd, size)) + times.append(t) + print(f"# sample {k} ({len(snaps[-1])} extents, " + f"{sum(len(a) for _, a in snaps[-1])*4/1e6:.0f} MB) " + f"in {time.time()-t:.2f}s", flush=True) + time.sleep(max(0, dt - (time.time() - t))) + + # extents must line up across samples for a word-wise comparison + shapes = [tuple((o, len(a)) for o, a in s) for s in snaps] + if len(set(shapes)) != 1: + common = set(shapes[0]) + for sh in shapes[1:]: + common &= set(sh) + print(f"# extents shifted between samples; using {len(common)} common ones") + keep = lambda s: [(o, a) for o, a in s if (o, len(a)) in common] + snaps = [keep(s) for s in snaps] + + hits = [] + for e in range(len(snaps[0])): + base = snaps[0][e][0] + arrs = [np.nan_to_num(s[e][1].astype(np.float64), nan=0, posinf=0, neginf=0) + for s in snaps] + # per-interval deltas + d = [arrs[k + 1] - arrs[k] for k in range(K - 1)] + moving = np.ones(len(arrs[0]), dtype=bool) + for dk in d: + moving &= np.abs(dk) > 1e-3 + idx = np.flatnonzero(moving) + if len(idx) == 0: + continue + # candidate triples: i, i+1, i+2 all moving + cand = idx[np.isin(idx + 1, idx) & np.isin(idx + 2, idx)] + for i in cand: + steps = [] + dirs = [] + ok = True + for k in range(K - 1): + v = np.array([d[k][i], d[k][i + 1], d[k][i + 2]]) + n = float(np.linalg.norm(v)) + ddt = times[k + 1] - times[k] + sp = n / ddt + if not (20.0 < sp < 3000.0): + ok = False + break + steps.append(sp) + dirs.append(v / n) + if not ok or len(steps) < 3: + continue + if max(steps) - min(steps) > 0.25 * (sum(steps) / len(steps)): + continue + if min(float(np.dot(dirs[0], dd)) for dd in dirs) < 0.985: + continue + va = gmem.primary_va(base + int(i) * 4) + pos = tuple(float(arrs[0][i + j]) for j in range(3)) + hits.append((va, base + int(i) * 4, sum(steps) / len(steps), pos)) + + print(f"\n# {len(hits)} position-like triples (straight, constant speed)") + hits.sort(key=lambda h: -h[2]) + for va, off, sp, pos in hits[:40]: + print(f" va {va:#010x} speed {sp:8.1f}/s pos " + f"({pos[0]:+11.1f},{pos[1]:+11.1f},{pos[2]:+11.1f})") + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/findrot_global.py b/tools/re-capture/findrot_global.py new file mode 100644 index 0000000..ac516cb --- /dev/null +++ b/tools/re-capture/findrot_global.py @@ -0,0 +1,147 @@ +#!/usr/bin/env python3 +"""Scan ALL of guest RAM for 3x3 orthonormal float blocks (rotation matrices). + +Vectorised with numpy: nine shifted views of each extent give every candidate +9-float window at once, so the whole 250 MB working set tests in seconds. + +A rotation matrix is a very strong signature โ€” unit rows, mutually orthogonal โ€” +so the hits are almost entirely real orientations. Two passes taken a moment +apart, with a known stick input in between, then say which of them is *ours*. + +Usage: findrot_global.py [--track seconds] +""" +import os +import struct +import sys +import time + +import numpy as np + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gmem # noqa: E402 + +FIFO = "/tmp/sylph-vgamepad.fifo" +TOL = 2e-3 + + +def pad(line): + with open(FIFO, "w") as f: + f.write(line + "\n") + + +def scan(fd, size): + """[(file_offset, 9 floats)] for every orthonormal 3x3 block.""" + hits = [] + for a, b in gmem.extents(fd, size): + n = (b - a) // 4 * 4 + if n < 64: + continue + arr = np.frombuffer(os.pread(fd, n, a), dtype=">f4").astype(np.float32) + if arr.size < 16: + continue + m = arr.size - 8 + cols = [arr[i:i + m] for i in range(9)] + with np.errstate(invalid="ignore", over="ignore"): + finite = np.ones(m, dtype=bool) + for c in cols: + finite &= np.isfinite(c) & (np.abs(c) <= 1.001) + # row norms + n0 = cols[0] ** 2 + cols[1] ** 2 + cols[2] ** 2 + n1 = cols[3] ** 2 + cols[4] ** 2 + cols[5] ** 2 + n2 = cols[6] ** 2 + cols[7] ** 2 + cols[8] ** 2 + ok = finite + ok &= np.abs(n0 - 1) < TOL + ok &= np.abs(n1 - 1) < TOL + ok &= np.abs(n2 - 1) < TOL + d01 = cols[0] * cols[3] + cols[1] * cols[4] + cols[2] * cols[5] + d02 = cols[0] * cols[6] + cols[1] * cols[7] + cols[2] * cols[8] + d12 = cols[3] * cols[6] + cols[4] * cols[7] + cols[5] * cols[8] + ok &= np.abs(d01) < TOL + ok &= np.abs(d02) < TOL + ok &= np.abs(d12) < TOL + for i in np.flatnonzero(ok): + hits.append(a + int(i) * 4) + return hits + + +def read_mat(fd, off): + b = os.pread(fd, 36, off) + if len(b) < 36: + return None + return np.array(struct.unpack(">9f", b)) + + +def main(): + fd = os.open(gmem.mem_path(), os.O_RDONLY) + size = os.path.getsize(gmem.mem_path()) + pad("reset") + time.sleep(0.6) + t0 = time.time() + hits = scan(fd, size) + print(f"# {len(hits)} orthonormal 3x3 blocks in RAM ({time.time()-t0:.1f}s)", flush=True) + if not hits: + return + + # which of them rotate when WE yaw? measure change under left vs right + def deltas(lx, secs=2.5, hz=6.0): + """Mean per-step rotation vector while the stick is held. + + Sampled incrementally: at a few degrees per step the skew part of + AยทBแต€ is the rotation vector, which it is not over a 2 s turn. Any block + that stops being orthonormal mid-phase is memory that got reused, not an + orientation, and is dropped. + """ + pad(f"axis LX {lx}") + time.sleep(0.5) + seq = [] + for _ in range(int(secs * hz)): + t = time.time() + seq.append({o: read_mat(fd, o) for o in hits}) + time.sleep(max(0, 1.0 / hz - (time.time() - t))) + pad("axis LX 0") + time.sleep(1.0) + + def ortho(m): + if m is None or not np.all(np.isfinite(m)): + return False + M = m.reshape(3, 3) + return np.max(np.abs(M @ M.T - np.eye(3))) < 5e-3 + + out = {} + for o in hits: + ms = [s[o] for s in seq] + if not all(ortho(m) for m in ms): + continue + ws = [] + for a, b in zip(ms, ms[1:]): + M = a.reshape(3, 3) @ b.reshape(3, 3).T + w = np.array([M[2, 1] - M[1, 2], M[0, 2] - M[2, 0], M[1, 0] - M[0, 1]]) / 2 + if np.linalg.norm(w) < 0.5: # small-angle regime only + ws.append(w) + if len(ws) >= 4: + out[o] = np.mean(ws, axis=0) + return out + + dl = deltas(-0.9) + dr = deltas(+0.9) + rows = [] + for o in hits: + if o not in dl or o not in dr: + continue + a, b = dl[o], dr[o] + na, nb = np.linalg.norm(a), np.linalg.norm(b) + if na < 0.01 or nb < 0.01: + continue + cos = float(a @ b / (na * nb)) + rows.append((cos, o, na, nb)) + rows.sort() + print("\n# orientation blocks that rotate OPPOSITE ways for left vs right stick") + for cos, o, na, nb in rows[:12]: + va = gmem.primary_va(o) + print(f" va {va:#010x} cos={cos:+.3f} |wL|={na:.3f} |wR|={nb:.3f}") + if not rows: + print(" (none)") + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/findself.py b/tools/re-capture/findself.py new file mode 100644 index 0000000..dc6470d --- /dev/null +++ b/tools/re-capture/findself.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +"""Identify which moving object in RAM is the *player* craft, by input correlation. + +Position triples are easy to find (findplayer.py); saying which one is us is the +hard part, and no static property answers it. This does: hold hard-left yaw for +a few seconds, then hard-right, and look for the object whose turn axis reverses +in phase with the stick. Every other craft is AI-flown and uncorrelated with our +input, so the discriminator is causal rather than circumstantial. + +Phase 1 finds candidate triples from two whole-RAM samples; after that only the +candidates' 12 bytes are re-read, so the sampling loop is cheap. + +Usage: findself.py [phase_seconds] [hz] +""" +import math +import os +import struct +import sys +import time + +import numpy as np + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gmem # noqa: E402 + +FIFO = "/tmp/sylph-vgamepad.fifo" + + +def pad(line): + with open(FIFO, "w") as f: + f.write(line + "\n") + + +def snapshot(fd, size): + return [(a, np.frombuffer(os.pread(fd, (b - a) // 4 * 4, a), dtype=">f4")) + for a, b in gmem.extents(fd, size) if (b - a) >= 64] + + +def candidates(fd, size, dt=0.35): + s0 = snapshot(fd, size) + t0 = time.time() + time.sleep(dt) + s1 = snapshot(fd, size) + t1 = time.time() + out = [] + for (o, a), (o2, b) in zip(s0, s1): + if o != o2 or len(a) != len(b): + continue + with np.errstate(invalid="ignore"): + af = np.nan_to_num(a.astype(np.float64), nan=0, posinf=0, neginf=0) + bf = np.nan_to_num(b.astype(np.float64), nan=0, posinf=0, neginf=0) + d = bf - af + idx = np.flatnonzero(np.abs(d) > 1e-3) + cand = idx[np.isin(idx + 1, idx) & np.isin(idx + 2, idx)] + for i in cand: + v = np.array([d[i], d[i + 1], d[i + 2]]) + sp = float(np.linalg.norm(v)) / (t1 - t0) + if 80.0 < sp < 2000.0: + out.append(o + int(i) * 4) + return out + + +def sample(fd, offs): + t = time.time() + pos = np.empty((len(offs), 3)) + for k, o in enumerate(offs): + b = os.pread(fd, 12, o) + if len(b) < 12: + pos[k] = np.nan + continue + pos[k] = struct.unpack(">3f", b) + return t, pos + + +def turn_axis(series): + """Mean normalised cross(v_k, v_k+1) over a phase, per candidate.""" + ts = [t for t, _ in series] + ps = [p for _, p in series] + vs = [] + for k in range(len(ps) - 1): + dt = ts[k + 1] - ts[k] + vs.append((ps[k + 1] - ps[k]) / max(dt, 1e-3)) + ax = np.zeros((ps[0].shape[0], 3)) + n = 0 + for k in range(len(vs) - 1): + c = np.cross(vs[k], vs[k + 1]) + nrm = np.linalg.norm(c, axis=1, keepdims=True) + with np.errstate(invalid="ignore", divide="ignore"): + ax += np.where(nrm > 1e-6, c / nrm, 0.0) + n += 1 + return ax / max(n, 1) + + +def main(): + secs = float(sys.argv[1]) if len(sys.argv) > 1 else 3.0 + hz = float(sys.argv[2]) if len(sys.argv) > 2 else 6.0 + path = gmem.mem_path() + fd = os.open(path, os.O_RDONLY) + size = os.path.getsize(path) + + pad("reset") + time.sleep(0.5) + offs = candidates(fd, size) + print(f"# {len(offs)} moving-triple candidates", flush=True) + if not offs: + sys.exit("nothing is moving โ€” in flight?") + + phases = {} + for name, lx in (("left", -0.9), ("right", 0.9)): + pad(f"axis LX {lx}") + time.sleep(0.6) # let the turn establish + series = [] + n = int(secs * hz) + for _ in range(n): + t0 = time.time() + series.append(sample(fd, offs)) + time.sleep(max(0, 1.0 / hz - (time.time() - t0))) + phases[name] = turn_axis(series) + pad("axis LX 0") + time.sleep(1.2) + print(f"# phase {name} done", flush=True) + pad("reset") + + aL, aR = phases["left"], phases["right"] + nL = np.linalg.norm(aL, axis=1) + nR = np.linalg.norm(aR, axis=1) + with np.errstate(invalid="ignore", divide="ignore"): + cos = np.sum(aL * aR, axis=1) / (nL * nR) + good = np.isfinite(cos) & (nL > 0.5) & (nR > 0.5) + order = np.argsort(np.where(good, cos, 9e9)) + print("\n# objects whose turn axis REVERSES with the stick (most negative first)") + shown = 0 + for k in order: + if not good[k]: + break + print(f" va {gmem.primary_va(offs[k]):#010x} cos(axisL,axisR) = {cos[k]:+.3f}" + f" |L|={nL[k]:.2f} |R|={nR[k]:.2f}") + shown += 1 + if shown >= 12: + break + if not shown: + print(" (none โ€” try longer phases)") + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/findspeed.py b/tools/re-capture/findspeed.py new file mode 100644 index 0000000..457f139 --- /dev/null +++ b/tools/re-capture/findspeed.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""Locate the player's flight object via its speed scalar, then its position. + +The HUD prints the craft's speed, so it is a known quantity we can *change on +demand* โ€” the classic two-state value scan, which is far more reliable than +hunting for a structure by shape: + + 1. coast -> RAM holds the cruise speed somewhere; collect every float โ‰ˆ it; + 2. boost -> the real one rises; everything coincidental is filtered out; + 3. coast -> it must come back down. + +Whatever survives all three is the player's speed. Its object then contains the +position, which is confirmed independently: a position triple near that scalar +must move at exactly the speed the scalar reports. + +Usage: findspeed.py [boost_wait] +""" +import os +import struct +import sys +import time + +import numpy as np + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gmem # noqa: E402 + +FIFO = "/tmp/sylph-vgamepad.fifo" + + +def pad(line): + with open(FIFO, "w") as f: + f.write(line + "\n") + + +def extents_arrays(fd, size): + for a, b in gmem.extents(fd, size): + n = (b - a) // 4 * 4 + if n >= 64: + yield a, np.frombuffer(os.pread(fd, n, a), dtype=">f4") + + +def scan_equal(fd, size, val, tol): + hits = [] + for a, arr in extents_arrays(fd, size): + with np.errstate(invalid="ignore"): + ok = np.isfinite(arr) & (np.abs(arr.astype(np.float64) - val) <= tol) + for i in np.flatnonzero(ok): + hits.append(a + int(i) * 4) + return hits + + +def read_f(fd, off): + b = os.pread(fd, 4, off) + return struct.unpack(">f", b)[0] if len(b) == 4 else float("nan") + + +def main(): + cruise = float(sys.argv[1]) + wait = float(sys.argv[2]) if len(sys.argv) > 2 else 4.0 + fd = os.open(gmem.mem_path(), os.O_RDONLY) + size = os.path.getsize(gmem.mem_path()) + + pad("reset") + time.sleep(2.0) + c0 = scan_equal(fd, size, cruise, 2.0) + print(f"# pass 1 (coast {cruise}): {len(c0)} candidates", flush=True) + + pad("trig RT 1.0") + time.sleep(wait) + c1 = [o for o in c0 if read_f(fd, o) > cruise + 40] + print(f"# pass 2 (boost): {len(c1)} rose above {cruise+40:.0f}", flush=True) + vals = {o: read_f(fd, o) for o in c1} + + pad("reset") + time.sleep(wait + 2.0) + c2 = [o for o in c1 if abs(read_f(fd, o) - cruise) <= 6.0] + print(f"# pass 3 (coast again): {len(c2)} returned to โ‰ˆ{cruise}", flush=True) + for o in c2[:20]: + print(f" va {gmem.primary_va(o):#010x} boost value was {vals[o]:.1f}") + + if not c2: + print("# no survivors โ€” is the craft actually accelerating?") + return + + # position triple in the same object: must move at exactly that speed + print("\n# looking for a position triple near each survivor", flush=True) + RAD = 0x400 + for o in c2[:8]: + lo = max(0, o - RAD) + n = RAD * 2 + t0 = time.time() + a0 = np.frombuffer(os.pread(fd, n, lo), dtype=">f4").astype(np.float64) + time.sleep(0.4) + t1 = time.time() + a1 = np.frombuffer(os.pread(fd, n, lo), dtype=">f4").astype(np.float64) + sp_now = read_f(fd, o) + dt = t1 - t0 + best = [] + for i in range(len(a0) - 2): + d = a1[i:i + 3] - a0[i:i + 3] + if not np.all(np.isfinite(d)): + continue + v = float(np.linalg.norm(d)) / dt + if abs(v - sp_now) <= max(8.0, 0.06 * sp_now): + best.append((lo + i * 4, v, tuple(a0[i:i + 3]))) + print(f" survivor va {gmem.primary_va(o):#010x} (speed {sp_now:.1f}): " + f"{len(best)} matching triples") + for off, v, p in best[:4]: + print(f" pos va {gmem.primary_va(off):#010x} delta-off " + f"{off - o:+#07x} |v|={v:7.1f} ({p[0]:+9.1f},{p[1]:+9.1f},{p[2]:+9.1f})") + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/flight_analyze.py b/tools/re-capture/flight_analyze.py new file mode 100644 index 0000000..fe3bc59 --- /dev/null +++ b/tools/re-capture/flight_analyze.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +"""Derive the player craft's live transform offsets from a flight_probe capture. + +Nothing here is guessed from plausible-looking numbers. Two independent +structural facts do the work: + + 1. a rotation matrix is 9 floats that are orthonormal with det +1 โ€” a + property essentially no other data has; + 2. a position triple's frame-to-frame delta must point along the craft's + own forward axis when it is flying straight, and its magnitude must be the + same for every frame at constant speed. + +Test 2 validates the position candidate *and* the rotation candidate against +each other, so a coincidence has to satisfy both at once. + +Usage: flight_analyze.py [name-substring] +""" +import math +import os +import struct +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gworld # noqa: E402 + + +def load(path): + f = open(path, "rb") + assert f.read(8) == b"SYLPHPRB" + n_inst, window, n_in = struct.unpack("f", buf, o)[0] + + +def vec(buf, o): + return (f32(buf, o), f32(buf, o + 4), f32(buf, o + 8)) + + +def sub(a, b): + return (a[0] - b[0], a[1] - b[1], a[2] - b[2]) + + +def norm(a): + return math.sqrt(sum(c * c for c in a)) + + +def dot(a, b): + return sum(x * y for x, y in zip(a, b)) + + +def analyse(insts, window, frames, want): + idx = [i for i, (_, _, nm) in enumerate(insts) if want in nm] + if not idx: + sys.exit(f"no instance matching {want!r}; have: " + + ", ".join(sorted({nm for _, _, nm in insts}))) + i = idx[0] + va, off, nm = insts[i] + print(f"# player instance {va:#010x} {nm} ({len(frames)} frames)") + + # ---- rotation blocks that hold up across every frame ------------------ + per_frame = [] + for t, inp, lab, bufs in frames: + per_frame.append({o for o, kind, m in gworld.find_rotations(bufs[i]) if kind == "3x3"}) + stable = set.intersection(*per_frame) if per_frame else set() + print(f"# 3x3 orthonormal blocks valid in ALL frames: " + + (", ".join(f"{o:#05x}" for o in sorted(stable)) or "none")) + + # ---- position candidates --------------------------------------------- + # constant-speed straight flight: |delta| equal every frame, direction fixed + idle = [k for k, (t, inp, lab, b) in enumerate(frames) if lab.startswith("idle")] + if len(idle) < 6: + idle = list(range(min(20, len(frames)))) + seg = idle[:20] + cands = [] + for o in range(0, window - 12, 4): + ds, ok = [], True + for a, b in zip(seg, seg[1:]): + if b != a + 1: + continue + dt = frames[b][0] - frames[a][0] + p0, p1 = vec(frames[a][3][i], o), vec(frames[b][3][i], o) + if not all(map(math.isfinite, p0 + p1)): + ok = False + break + d = sub(p1, p0) + if dt <= 0: + ok = False + break + ds.append((norm(d) / dt, d)) + if not ok or len(ds) < 4: + continue + sp = [s for s, _ in ds] + mean = sum(sp) / len(sp) + if mean < 1.0 or mean > 5000.0: # a craft, not a counter + continue + spread = max(sp) - min(sp) + if spread > 0.15 * mean: # constant speed while coasting + continue + # direction must be steady too + dirs = [(d[0] / (norm(d) or 1), d[1] / (norm(d) or 1), d[2] / (norm(d) or 1)) + for _, d in ds] + if min(dot(dirs[0], d) for d in dirs) < 0.99: + continue + cands.append((o, mean, dirs[0])) + + print(f"# position candidates (steady speed + steady heading while coasting): " + f"{len(cands)}") + for o, sp, d in cands[:12]: + print(f" +{o:#05x} speed {sp:8.2f}/s dir ({d[0]:+.3f},{d[1]:+.3f},{d[2]:+.3f})") + + # ---- cross-validate: velocity must lie along a rotation row ----------- + print("\n# cross-check: does the motion direction match a rotation-matrix axis?") + best = [] + for o, sp, d in cands: + for ro in sorted(stable): + m = struct.unpack_from(">9f", frames[seg[0]][3][i], ro) + for r, label in ((m[0:3], "row0/right"), (m[3:6], "row1/up"), (m[6:9], "row2/fwd")): + c = abs(dot(d, r)) + if c > 0.98: + best.append((c, o, ro, label, sp)) + best.sort(reverse=True) + if not best: + print(" none โ€” position and orientation candidates do not corroborate") + for c, o, ro, label, sp in best[:10]: + print(f" pos +{o:#05x} โˆฅ rot +{ro:#05x} {label} |cos|={c:.5f} speed {sp:.1f}") + return stable, cands + + +def main(): + path = sys.argv[1] + want = sys.argv[2] if len(sys.argv) > 2 else "Player" + insts, window, frames = load(path) + analyse(insts, window, frames, want) + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/flight_probe.py b/tools/re-capture/flight_probe.py new file mode 100755 index 0000000..e04ba71 --- /dev/null +++ b/tools/re-capture/flight_probe.py @@ -0,0 +1,107 @@ +#!/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 [seconds] +""" +import os +import struct +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gworld # noqa: E402 + +FIFO = "/tmp/sylph-vgamepad.fifo" + + +class Pad: + """Talk to the vgamepad server directly over its FIFO. + + The `vgamepad` CLI spawns a process per command (~20 ms); a control loop + cannot afford that, and `tap`/`hold` additionally sleep *inside* the server. + Writing lines to the FIFO ourselves keeps a tick under a millisecond. + """ + + def __init__(self): + self.f = open(FIFO, "w", buffering=1) + self.state = {"LX": 0.0, "LY": 0.0, "RX": 0.0, "RY": 0.0, "LT": 0.0, "RT": 0.0} + + def axis(self, name, v): + self.state[name] = v + self.f.write(f"axis {name} {v:.3f}\n") + + def trig(self, name, v): + self.state[name] = v + self.f.write(f"trig {name} {v:.3f}\n") + + def press(self, b): + self.f.write(f"press {b}\n") + + def release(self, b): + self.f.write(f"release {b}\n") + + def reset(self): + self.f.write("reset\n") + for k in self.state: + self.state[k] = 0.0 + + 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(" [hz] [out] record every instance's raw bytes over time + orient report orthonormal 3x3 blocks per instance +""" + +import os +import re +import struct +import sys +import time + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gmem # noqa: E402 + +INST_VTABLE = 0x820AF030 +DEF_VTABLE = 0x820AF844 +WINDOW = 0x600 # bytes of an instance we read; its real size is unknown +NAME_STR_OFF = 0x10 + + +class World: + def __init__(self, path=None): + self.path = path or gmem.mem_path() + self.fd = os.open(self.path, os.O_RDONLY) + self.size = os.path.getsize(self.path) + self.instances = [] # [(va, file_off, name)] + + # ---------------------------------------------------------------- io + def read(self, va, n): + return os.pread(self.fd, n, gmem.va_to_off(va)) + + def read_off(self, off, n): + return os.pread(self.fd, n, off) + + def u32(self, va): + return struct.unpack(">I", self.read(va, 4))[0] + + def cstr(self, va, n=96): + b = os.pread(self.fd, n, gmem.va_to_off(va)).split(b"\0")[0] + try: + return b.decode("ascii") + except UnicodeDecodeError: + return None + + # ------------------------------------------------------------- scan + def scan_vtable(self, vt): + pat = struct.pack(">I", vt) + hits = [] + for a, b in gmem.extents(self.fd, self.size): + blob = os.pread(self.fd, b - a, a) + start = 0 + while True: + i = blob.find(pat, start) + if i < 0: + break + off = a + i + if off % 4 == 0: + hits.append(off) + start = i + 4 + return sorted(set(hits)) + + def name_of(self, off): + """The unit ID an object at `off` names, via its name-record pointer.""" + raw = self.read_off(off + 4, 4) + if len(raw) < 4: + return None + (p,) = struct.unpack(">I", raw) + try: + return self.cstr(p + NAME_STR_OFF) + except ValueError: + return None + + def refresh(self, vt=INST_VTABLE): + """Re-scan for live instances. Costs ~1 s; call it rarely.""" + out = [] + for off in self.scan_vtable(vt): + nm = self.name_of(off) + if nm and nm.startswith("UN_"): + out.append((gmem.primary_va(off), off, nm)) + self.instances = out + return out + + +# --------------------------------------------------------------- geometry + + +def floats(buf, off, n): + return struct.unpack_from(">" + "f" * n, buf, off) + + +def is_rotation(m, tol=2e-3): + """m = 9 floats, row-major. Orthonormal rows + det +1?""" + r = [m[0:3], m[3:6], m[6:9]] + for row in r: + n2 = sum(c * c for c in row) + if abs(n2 - 1.0) > tol: + return False + for i in range(3): + for j in range(i + 1, 3): + if abs(sum(r[i][k] * r[j][k] for k in range(3))) > tol: + return False + det = (r[0][0] * (r[1][1] * r[2][2] - r[1][2] * r[2][1]) + - r[0][1] * (r[1][0] * r[2][2] - r[1][2] * r[2][0]) + + r[0][2] * (r[1][0] * r[2][1] - r[1][1] * r[2][0])) + return abs(det - 1.0) < 1e-2 + + +def find_rotations(buf, stride=4): + """Offsets where 9 consecutive floats form a rotation matrix. + + Also tries the 4-float-per-row (3x4 / 4x4 matrix) stride, which is how a + transform with a translation column is normally stored. + """ + out = [] + for off in range(0, len(buf) - 36, stride): + m = floats(buf, off, 9) + if all(abs(c) <= 1.001 for c in m) and is_rotation(m): + out.append((off, "3x3", m)) + for off in range(0, len(buf) - 48, stride): + rows = [floats(buf, off + 16 * i, 3) for i in range(3)] + m = rows[0] + rows[1] + rows[2] + if all(abs(c) <= 1.001 for c in m) and is_rotation(m): + out.append((off, "4x4", m)) + return out + + +# ------------------------------------------------------------------ cmds + + +def cmd_entities(w, args): + t0 = time.time() + inst = w.refresh() + print(f"# {len(inst)} live instances (scan {time.time()-t0:.2f}s)") + from collections import Counter + c = Counter(n for _, _, n in inst) + for nm, k in c.most_common(): + print(f" {k:4d} {nm}") + + +def cmd_orient(w, args): + inst = w.refresh() + want = args[0] if args else None + for va, off, nm in inst: + if want and want not in nm: + continue + buf = w.read_off(off, WINDOW) + rots = find_rotations(buf) + print(f"\n{va:#010x} {nm} -> {len(rots)} rotation-like block(s)") + for o, kind, m in rots[:6]: + print(f" +{o:#05x} {kind} " + + " ".join(f"{v:+.3f}" for v in m)) + + +def cmd_probe(w, args): + secs = float(args[0]) if args else 6.0 + hz = float(args[1]) if len(args) > 1 else 5.0 + out = args[2] if len(args) > 2 else "probe.bin" + inst = w.refresh() + print(f"# probing {len(inst)} instances for {secs}s at {hz}Hz -> {out}") + with open(out, "wb") as f: + f.write(struct.pack(" LOAD GAME -> slot 01 -> YES -> READY ROOM +# -> TAKE OFF. With `--hangar` it stops in the HANGAR instead so a loadout can +# be picked first. +set -u +MODE="${1:-fly}" +export HOME=/sylph-home/re SDL_AUDIODRIVER=dummy DISPLAY=:98 +SD="$(cd "$(dirname "$0")" && pwd)" +SHOTS=/sylph-home/re/shots + +alive(){ ps -o pid=,stat= -C xenia_canary 2>/dev/null | awk '$2 !~ /^Z/ {print $1}'; } +ensure_display(){ + if ! xdpyinfo -display "$DISPLAY" >/dev/null 2>&1; then + setsid nohup Xvfb "$DISPLAY" -screen 0 1280x720x24 -ac -nolisten tcp \ + +extension GLX +extension RANDR /tmp/xvfb98.log 2>&1 & + for _ in $(seq 1 50); do xdpyinfo -display "$DISPLAY" >/dev/null 2>&1 && break; sleep 0.2; done + setsid nohup env DISPLAY="$DISPLAY" HOME=/sylph-home openbox /tmp/openbox98.log 2>&1 & + sleep 1 + fi + xdpyinfo -display "$DISPLAY" >/dev/null 2>&1 || { echo "DISPLAY UNAVAILABLE"; exit 1; } +} +step(){ vgamepad dpad "$1"; sleep 0.25; vgamepad dpad center; sleep 0.7; } +shot(){ screenshot "$SHOTS/$1" >/dev/null 2>&1; } + +pkill -x xenia_canary 2>/dev/null; sleep 2 +[ -n "$(alive)" ] && { kill -9 $(alive) 2>/dev/null; sleep 2; } +rm -f /dev/shm/xenia_memory_* /dev/shm/xenia_code_cache_* 2>/dev/null +ensure_display + +cd /sylph-home/re +setsid nohup run-canary --audio --apu=sdl --log_mask=13 \ + --logged_profile_slot_0_xuid=E0300000EFBEA3D4 /dev/null 2>&1 & +sleep 5 +"$SD/skip_intro.sh" 600 || { echo "BOOT FAILED"; exit 1; } +sleep 14 # main menu is not input-ready before this + +step down # NEW GAME -> LOAD GAME +vgamepad tap A 250; sleep 8 # save list, slot 01 preselected +vgamepad tap A 250; sleep 4 # "Load game?" -- cursor starts on NO +step up +vgamepad tap A 250 +sleep 28 # -> READY ROOM +shot "lm-readyroom.png" + +if [ "$MODE" = "--hangar" ]; then + step down; step down # BRIEFINGS -> HANGAR + vgamepad tap A 250; sleep 12 + shot "lm-hangar.png" + echo "STOPPED IN HANGAR"; exit 0 +fi + +step up # BRIEFINGS -> TAKE OFF +vgamepad tap A 250 +sleep 75 # launch cinematic + stage load + objective card +shot "lm-objective.png" +vgamepad tap A 250; sleep 6 # dismiss the OBJECTIVE panel +shot "lm-flight.png" +echo "IN FLIGHT (emulator left running)" diff --git a/tools/re-capture/liveents.py b/tools/re-capture/liveents.py new file mode 100644 index 0000000..faba069 --- /dev/null +++ b/tools/re-capture/liveents.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Type the live entities: link each moving position back to a unit definition. + +findplayer.py locates positions by motion alone, but a bare (x,y,z) says nothing +about *what* is moving. Every live entity should reference its parsed `.tbl` +definition (vtable 0x820af844, one object per unit type, addresses known), so +scanning the neighbourhood of each position for a word equal to a definition +address both types the entity and reveals the live object's own layout โ€” the +delta from that pointer to the position triple is the position offset. + +Usage: liveents.py [radius_hex] +""" +import os +import struct +import sys +from collections import Counter + +import numpy as np + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gmem # noqa: E402 +import gworld # noqa: E402 + + +def definitions(w): + """{definition_va: unit_id}""" + out = {} + for off in w.scan_vtable(gworld.DEF_VTABLE): + nm = w.name_of(off) + if nm and nm.startswith("UN_"): + va = gmem.primary_va(off) + if va is not None: + out[va] = nm + return out + + +def main(): + radius = int(sys.argv[1], 16) if len(sys.argv) > 1 else 0x600 + w = gworld.World() + defs = definitions(w) + print(f"# {len(defs)} unit definitions") + by_word = {struct.pack(">I", va): nm for va, nm in defs.items()} + + # sample twice to locate movers, exactly as findplayer.py does + import time + fd, size = w.fd, w.size + + def snap(): + return [(a, np.frombuffer(os.pread(fd, (b - a) // 4 * 4, a), dtype=">f4")) + for a, b in gmem.extents(fd, size) if (b - a) >= 64] + + s0 = snap(); t0 = time.time() + time.sleep(0.4) + s1 = snap(); t1 = time.time() + shapes0 = {(o, len(a)) for o, a in s0} + pairs = [(o, a, b) for (o, a), (o2, b) in zip(s0, s1) + if o == o2 and len(a) == len(b) and (o, len(a)) in shapes0] + + movers = [] + for base, a, b in pairs: + af = np.nan_to_num(a.astype(np.float64), nan=0, posinf=0, neginf=0) + bf = np.nan_to_num(b.astype(np.float64), nan=0, posinf=0, neginf=0) + d = bf - af + m = np.abs(d) > 1e-3 + idx = np.flatnonzero(m) + cand = idx[np.isin(idx + 1, idx) & np.isin(idx + 2, idx)] + for i in cand: + v = np.array([d[i], d[i + 1], d[i + 2]]) + sp = float(np.linalg.norm(v)) / (t1 - t0) + if 20.0 < sp < 3000.0: + movers.append((base + int(i) * 4, tuple(float(af[i + j]) for j in range(3)), sp)) + + print(f"# {len(movers)} moving triples; typing them...") + typed, untyped = [], 0 + seen = set() + for off, pos, sp in movers: + lo = max(0, off - radius) + blob = os.pread(fd, radius * 2, lo) + hit = None + for k in range(0, len(blob) - 3, 4): + nm = by_word.get(blob[k:k + 4]) + if nm: + hit = (lo + k, nm) + break + if hit: + ptr_off, nm = hit + key = (ptr_off, nm) + if key in seen: + continue + seen.add(key) + typed.append((ptr_off, off - ptr_off, nm, pos, sp)) + else: + untyped += 1 + + print(f"# typed {len(typed)}, untyped {untyped}") + deltas = Counter(d for _, d, _, _, _ in typed) + print(f"# most common (def-pointer -> position) deltas: {deltas.most_common(6)}") + for ptr_off, d, nm, pos, sp in sorted(typed, key=lambda x: x[2])[:40]: + va = gmem.primary_va(ptr_off) + print(f" obj~{va:#010x} defptr+{d:#06x} {nm:<40} " + f"({pos[0]:+10.1f},{pos[1]:+10.1f},{pos[2]:+10.1f}) {sp:7.1f}/s") + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/selfstate.py b/tools/re-capture/selfstate.py new file mode 100644 index 0000000..27fa3b3 --- /dev/null +++ b/tools/re-capture/selfstate.py @@ -0,0 +1,208 @@ +#!/usr/bin/env python3 +"""Derive everything the autopilot needs about the live world, and write it to +a JSON config. Each step is checked against an independent property, so a wrong +answer has to survive several unrelated tests. + + 1. moving position triples (motion, whole-RAM diff) + 2. which one is US (turn axis reverses with our stick) + 3. our orientation matrix (a row must track our velocity) + 4. how to type any entity (fixed offset to its definition + pointer, learned from ours) + +Output: {"pos_va":โ€ฆ, "rot_va":โ€ฆ, "fwd_row":โ€ฆ, "def_delta":โ€ฆ} + +Usage: selfstate.py +""" +import json +import math +import os +import struct +import sys +import time + +import numpy as np + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import gmem # noqa: E402 +import gworld # noqa: E402 +from findrot_global import scan as scan_rot # noqa: E402 + +FIFO = "/tmp/sylph-vgamepad.fifo" + + +def pad(line): + with open(FIFO, "w") as f: + f.write(line + "\n") + + +def arrays(fd, size): + for a, b in gmem.extents(fd, size): + n = (b - a) // 4 * 4 + if n >= 64: + yield a, np.frombuffer(os.pread(fd, n, a), dtype=">f4") + + +def moving_triples(fd, size, lo=120.0, hi=1600.0, dt=0.4): + s0 = list(arrays(fd, size)) + t0 = time.time() + time.sleep(dt) + s1 = list(arrays(fd, size)) + t1 = time.time() + out = [] + for (o, a), (o2, b) in zip(s0, s1): + if o != o2 or len(a) != len(b): + continue + with np.errstate(invalid="ignore"): + af = np.nan_to_num(a.astype(np.float64), nan=0, posinf=0, neginf=0) + bf = np.nan_to_num(b.astype(np.float64), nan=0, posinf=0, neginf=0) + d = bf - af + idx = np.flatnonzero(np.abs(d) > 1e-3) + cand = idx[np.isin(idx + 1, idx) & np.isin(idx + 2, idx)] + for i in cand: + v = np.array([d[i], d[i + 1], d[i + 2]]) + sp = float(np.linalg.norm(v)) / (t1 - t0) + if lo < sp < hi: + out.append(o + int(i) * 4) + return out + + +def read3(fd, off): + b = os.pread(fd, 12, off) + return np.array(struct.unpack(">3f", b)) if len(b) == 12 else np.full(3, np.nan) + + +def sample_positions(fd, offs, secs, hz): + seq = [] + n = int(secs * hz) + for _ in range(n): + t = time.time() + seq.append((t, np.array([read3(fd, o) for o in offs]))) + time.sleep(max(0, 1.0 / hz - (time.time() - t))) + return seq + + +def mean_turn_axis(seq): + ts = [t for t, _ in seq] + ps = [p for _, p in seq] + vs = [(ps[k + 1] - ps[k]) / max(ts[k + 1] - ts[k], 1e-3) for k in range(len(ps) - 1)] + acc = np.zeros((ps[0].shape[0], 3)) + n = 0 + for k in range(len(vs) - 1): + c = np.cross(vs[k], vs[k + 1]) + nr = np.linalg.norm(c, axis=1, keepdims=True) + with np.errstate(invalid="ignore", divide="ignore"): + acc += np.where(nr > 1e-9, c / nr, 0.0) + n += 1 + return acc / max(n, 1) + + +def main(): + out_path = sys.argv[1] + fd = os.open(gmem.mem_path(), os.O_RDONLY) + size = os.path.getsize(gmem.mem_path()) + + pad("reset") + time.sleep(1.5) + offs = moving_triples(fd, size) + print(f"# {len(offs)} moving triples", flush=True) + if not offs: + sys.exit("nothing moving") + + # ---- 2. which one is us: turn axis must reverse with the stick ---------- + axes = {} + for name, lx in (("L", -0.95), ("R", 0.95)): + pad(f"axis LX {lx}") + time.sleep(0.6) + seq = sample_positions(fd, offs, 2.5, 6.0) + axes[name] = mean_turn_axis(seq) + pad("axis LX 0") + time.sleep(1.5) + print(f"# phase {name} sampled", flush=True) + pad("reset") + aL, aR = axes["L"], axes["R"] + nL, nR = np.linalg.norm(aL, axis=1), np.linalg.norm(aR, axis=1) + with np.errstate(invalid="ignore", divide="ignore"): + cos = np.sum(aL * aR, axis=1) / (nL * nR) + good = np.isfinite(cos) & (nL > 0.6) & (nR > 0.6) + if not good.any(): + sys.exit("no object reversed with the stick") + k = int(np.argmin(np.where(good, cos, 9e9))) + pos_off = offs[k] + print(f"# self position: va {gmem.primary_va(pos_off):#010x} cos={cos[k]:+.3f}") + + # ---- 3. orientation: a row must track our velocity ---------------------- + time.sleep(1.0) + rots = scan_rot(fd, size) + print(f"# {len(rots)} orthonormal blocks; matching one to our velocity", flush=True) + seq = sample_positions(fd, [pos_off], 2.0, 8.0) + ps = [p[0] for _, p in seq] + ts = [t for t, _ in seq] + vdirs = [] + for a, b, ta, tb in zip(ps, ps[1:], ts, ts[1:]): + v = (b - a) / max(tb - ta, 1e-3) + n = np.linalg.norm(v) + if n > 1e-3: + vdirs.append(v / n) + vdir = np.mean(vdirs, axis=0) + vdir /= np.linalg.norm(vdir) + speed = float(np.mean([np.linalg.norm((b - a) / max(tb - ta, 1e-3)) + for a, b, ta, tb in zip(ps, ps[1:], ts, ts[1:])])) + print(f"# our heading {vdir.round(3)} speed {speed:.1f}/s") + + # A block found by the earlier scan may have been overwritten by the time we + # read it, so re-verify orthonormality here -- otherwise a garbage window + # yields a meaningless (and unbounded) "cosine". + best = None + for ro in rots: + b = os.pread(fd, 36, ro) + if len(b) < 36: + continue + m = np.array(struct.unpack(">9f", b)) + if not np.all(np.isfinite(m)): + continue + M = m.reshape(3, 3) + if np.max(np.abs(M @ M.T - np.eye(3))) > 5e-3: + continue + for r in range(3): + c = float(M[r] @ vdir) + if best is None or abs(c) > abs(best[0]): + best = (c, ro, r) + if best is None: + sys.exit("no valid orientation block") + print(f"# best orientation match: va {gmem.primary_va(best[1]):#010x} " + f"row {best[2]} cos={best[0]:+.4f}") + + # ---- 4. how to type an entity ------------------------------------------ + w = gworld.World() + defs = {} + for doff in w.scan_vtable(gworld.DEF_VTABLE): + nm = w.name_of(doff) + if nm and nm.startswith("UN_"): + va = gmem.primary_va(doff) + if va is not None: + defs[struct.pack(">I", va)] = nm + delta = None + blob = os.pread(fd, 0x1000, max(0, pos_off - 0x800)) + for i in range(0, len(blob) - 3, 4): + nm = defs.get(blob[i:i + 4]) + if nm: + delta = (pos_off - 0x800) + i - pos_off + print(f"# definition pointer at pos{delta:+#07x} -> {nm}") + break + if delta is None: + print("# no definition pointer near our position (entity typing unavailable)") + + cfg = { + "pos_va": gmem.primary_va(pos_off), + "rot_va": gmem.primary_va(best[1]), + "fwd_row": best[2], + "fwd_sign": 1 if best[0] > 0 else -1, + "def_delta": delta, + "cruise_speed": speed, + } + json.dump(cfg, open(out_path, "w"), indent=1) + print("\n" + json.dumps(cfg, indent=1)) + + +if __name__ == "__main__": + main() diff --git a/tools/re-capture/whatchanges.py b/tools/re-capture/whatchanges.py new file mode 100644 index 0000000..d2c754e --- /dev/null +++ b/tools/re-capture/whatchanges.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Encoding-agnostic: which words of an object actually carry live state? + +Rather than assume the orientation is a matrix (it is not) or a quaternion, +classify every 4-byte word of the captured window by how it behaves over the +capture: constant, smoothly varying (a continuous quantity), or jumpy. + +Usage: whatchanges.py [name-substring] [max-report] +""" +import math +import struct +import sys + +sys.path.insert(0, "/home/fabi/RE - Project Sylpheed/sylpheed-reborn/tools/re-capture") +from flight_analyze import load # noqa: E402 + + +def main(): + path = sys.argv[1] + want = sys.argv[2] if len(sys.argv) > 2 else "Player" + top = int(sys.argv[3]) if len(sys.argv) > 3 else 60 + insts, window, frames = load(path) + i = next(k for k, (_, _, nm) in enumerate(insts) if want in nm) + print(f"# {insts[i][2]} @ {insts[i][0]:#010x}, {len(frames)} frames, window {window:#x}") + + nconst = nsmooth = njump = 0 + smooth = [] + for o in range(0, window - 3, 4): + vals = [] + ok = True + for t, inp, lab, bufs in frames: + (v,) = struct.unpack_from(">f", bufs[i], o) + if not math.isfinite(v): + ok = False + break + vals.append(v) + if not ok: + continue + lo, hi = min(vals), max(vals) + if lo == hi: + nconst += 1 + continue + rng = hi - lo + steps = [abs(b - a) for a, b in zip(vals, vals[1:])] + mx = max(steps) + # smooth = no single step is a large fraction of the whole excursion + if mx < 0.25 * rng: + nsmooth += 1 + smooth.append((o, lo, hi, vals)) + else: + njump += 1 + print(f"# constant {nconst} smooth {nsmooth} jumpy {njump}") + print(f"\n# smoothly varying words (candidate continuous state):") + for o, lo, hi, vals in smooth[:top]: + print(f" +{o:#05x} [{lo:12.4g} .. {hi:12.4g}] " + f"start {vals[0]:12.4g} end {vals[-1]:12.4g}") + + # consecutive runs of smooth words -> vectors + offs = [o for o, _, _, _ in smooth] + runs, cur = [], [] + for o in offs: + if cur and o == cur[-1] + 4: + cur.append(o) + else: + if len(cur) >= 3: + runs.append(cur) + cur = [o] + if len(cur) >= 3: + runs.append(cur) + print(f"\n# runs of >=3 consecutive smooth words (vector-shaped):") + for r in runs: + print(f" +{r[0]:#05x} .. +{r[-1]:#05x} ({len(r)} words)") + + +if __name__ == "__main__": + main()