re: memory-driven autopilot -- infrastructure, and an honest account of what is not solved
Reads the live world out of guest RAM and drives the pad from it. Working: loop-rate memory reads, whole-RAM float scanning with numpy (1270 orthonormal 3x3 blocks in 6.2 s), entity enumeration by unit type (116 live instances in Stage 02), pad control written straight into the vgamepad FIFO (the 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, and the Hangar loadout -- the "Recommended" control is AUTO SELECT, which at 5 % progress is a no-op because only two weapons are developed and both are already mounted. Not working, and the reason the craft is not yet flown: the class 0x820af030 is NOT the live entity. It has one object per spawned thing and carries the unit-ID string, which is why it looked like the entity list, but every one of its 384 words is constant across a 29 s in-flight capture. No transform lives in it or one pointer hop from it. Input correlation (hard left yaw vs hard right, looking for a turn axis that reverses) does find self-like objects at cos = -0.99, but they cluster in what looks like a camera volume rather than the craft, and with no definition pointer near them the trick of learning one entity's layout and applying it to the rest has nothing to anchor on -- so the 33418 moving triples in a firefight cannot be split into enemies, friendlies and bullets, and there is nothing to aim at. Two dead ends are recorded so they are not repeated: RT is not the throttle (the two-state speed scan therefore found nothing), and comparing orientation matrices 2 s apart is outside the small-angle regime, which is what produced "angular velocities" of 30000. Also corrects the claim in unit-struct-runtime.md that 0x820af030 holds live state. The definition class 0x820af844 and every value derived from it are unaffected. autopilot2.py (a PD controller using body angular velocity from consecutive rotation matrices) is committed but has never had a valid config to run against, and is marked as untested.
This commit is contained in:
104
docs/re/autopilot-memory-driven.md
Normal file
104
docs/re/autopilot-memory-driven.md
Normal file
@@ -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 `<defunct>` 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`.
|
||||
@@ -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
|
||||
|
||||
212
tools/re-capture/autopilot2.py
Normal file
212
tools/re-capture/autopilot2.py
Normal file
@@ -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()
|
||||
120
tools/re-capture/findplayer.py
Normal file
120
tools/re-capture/findplayer.py
Normal file
@@ -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()
|
||||
147
tools/re-capture/findrot_global.py
Normal file
147
tools/re-capture/findrot_global.py
Normal file
@@ -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()
|
||||
146
tools/re-capture/findself.py
Normal file
146
tools/re-capture/findself.py
Normal file
@@ -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()
|
||||
115
tools/re-capture/findspeed.py
Normal file
115
tools/re-capture/findspeed.py
Normal file
@@ -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 <cruise> [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()
|
||||
157
tools/re-capture/flight_analyze.py
Normal file
157
tools/re-capture/flight_analyze.py
Normal file
@@ -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 <probe.bin> [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("<III", f.read(12))
|
||||
insts = []
|
||||
for _ in range(n_inst):
|
||||
va, off = struct.unpack("<IQ", f.read(12))
|
||||
nm = f.read(64).split(b"\0")[0].decode()
|
||||
insts.append((va, off, nm))
|
||||
frames = []
|
||||
rec = 8 + 4 * n_in + 32 + n_inst * window
|
||||
while True:
|
||||
b = f.read(rec)
|
||||
if len(b) < rec:
|
||||
break
|
||||
t = struct.unpack_from("<d", b, 0)[0]
|
||||
inp = struct.unpack_from("<%df" % n_in, b, 8)
|
||||
label = struct.unpack_from("<32s", b, 8 + 4 * n_in)[0].split(b"\0")[0].decode()
|
||||
base = 8 + 4 * n_in + 32
|
||||
bufs = [b[base + i * window: base + (i + 1) * window] for i in range(n_inst)]
|
||||
frames.append((t, inp, label, bufs))
|
||||
return insts, window, frames
|
||||
|
||||
|
||||
def f32(buf, o):
|
||||
return struct.unpack_from(">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()
|
||||
107
tools/re-capture/flight_probe.py
Executable file
107
tools/re-capture/flight_probe.py
Executable file
@@ -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 <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
|
||||
|
||||
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("<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()
|
||||
203
tools/re-capture/gworld.py
Executable file
203
tools/re-capture/gworld.py
Executable file
@@ -0,0 +1,203 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Live world-state reader: the running game's entities, straight out of guest RAM.
|
||||
|
||||
Builds on gmem.py (Canary backs the guest address space with
|
||||
/dev/shm/xenia_memory_*, so it is an ordinary file). The difference here is that
|
||||
this is meant to run *in a loop* while the game plays, not against a snapshot:
|
||||
the memory file is opened once and re-read with pread, and the expensive
|
||||
whole-RAM vtable scan is done rarely and cached.
|
||||
|
||||
Two classes matter (see docs/re/structures/unit-struct-runtime.md):
|
||||
0x820af844 the parsed `.tbl` definition — one per unit type
|
||||
0x820af030 the spawned entity instance — one per thing in the scene
|
||||
|
||||
This module finds instances and reads their live transform. The transform
|
||||
offsets are NOT guessed: `find_transform` scans an instance for a 3x3 block of
|
||||
floats that is orthonormal to 1e-3 with det = +1, which a rotation matrix is and
|
||||
essentially nothing else is.
|
||||
|
||||
Sub-commands:
|
||||
entities list live instances (name, address)
|
||||
probe <seconds> [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("<II", len(inst), WINDOW))
|
||||
for va, off, nm in inst:
|
||||
nb = nm.encode()[:63].ljust(64, b"\0")
|
||||
f.write(struct.pack("<II", va, off) + nb)
|
||||
n = int(secs * hz)
|
||||
for i in range(n):
|
||||
t = time.time()
|
||||
f.write(struct.pack("<d", t))
|
||||
for va, off, nm in inst:
|
||||
f.write(w.read_off(off, WINDOW).ljust(WINDOW, b"\0"))
|
||||
time.sleep(max(0, 1.0 / hz - (time.time() - t)))
|
||||
print(f"# wrote {n} frames")
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
sys.exit(__doc__)
|
||||
w = World()
|
||||
cmd, args = sys.argv[1], sys.argv[2:]
|
||||
{"entities": cmd_entities, "orient": cmd_orient, "probe": cmd_probe}[cmd](w, args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
62
tools/re-capture/launch_mission.sh
Executable file
62
tools/re-capture/launch_mission.sh
Executable file
@@ -0,0 +1,62 @@
|
||||
#!/usr/bin/env bash
|
||||
# Boot Canary and drive it into the Stage 02 mission from save slot 01, then
|
||||
# LEAVE IT RUNNING (unlike grab_tutorial.sh, which snapshots and exits) so a
|
||||
# live control loop can attach to /dev/shm.
|
||||
#
|
||||
# Nav is the verified route: title -> 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 </dev/null >/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 </dev/null >/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 >/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)"
|
||||
105
tools/re-capture/liveents.py
Normal file
105
tools/re-capture/liveents.py
Normal file
@@ -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()
|
||||
208
tools/re-capture/selfstate.py
Normal file
208
tools/re-capture/selfstate.py
Normal file
@@ -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 <out.json>
|
||||
"""
|
||||
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()
|
||||
76
tools/re-capture/whatchanges.py
Normal file
76
tools/re-capture/whatchanges.py
Normal file
@@ -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 <probe.bin> [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()
|
||||
Reference in New Issue
Block a user