Sylpheed/tools/re-capture/squadron_state.py

#!/usr/bin/env python3
"""Read per-SQUADRON liveness straight out of the running ScriptPhase.

Why this exists: the phase-1 clear condition polls three named squadrons
(ADN110/111/112), but nothing mapped a live craft back to its roster squadron,
so "38 enemies died" could not say whether the *right* ones died
(mission-phase-membership.md). Chasing craft->squadron was the wrong angle --
the script VM already keeps exactly that table.

From the built-in disassembly (isl-builtins.md):

    [phase+324]            runtime unit array
    [[phase+324]+4]        -> array of per-unit record pointers
    record[idx*4]          indexed by the .ssb SYMBOL TABLE 2 index
    record +4              live object   (NULL = absent)
    record +16             state         (2 = active; 1/3/4 = gone/dead/invalid)
    record +128 / +132     HP / max HP

So a squadron's state is one indexed read, no attribution needed.

Finding the ScriptPhase without a debugger:
  1. the loaded .ssb code is in guest memory -- search for a distinctive run of
     its bytes;
  2. `[phase+232]` is the code base, so scan for a word equal to that address;
  3. the candidate phase is that word's address - 232;
  4. validate: `[cand+244]` and `[cand+324]` must both be plausible pointers.

Usage: squadron_state.py <Stage02.ssb> [names...]
"""
import struct
import sys

sys.path.insert(0, __file__.rsplit('/', 1)[0])
import gmem
import isl

# Bound for the pointer scan; observed ScriptMission addresses: 0xBC7A2A20 (x3),
# 0xBC79C960. Widened generously either side -- a miss falls back to the full
# sweep rather than failing.
PTR_WINDOW = (0xBC000000, 0xBD000000)

HDR_LEN = 20            # version, +4, code offset, symtab1, symtab2 -- distinctive


def _fd_extents(f, size):
    # gmem.extents takes a file DESCRIPTOR, not a file object.
    return gmem.extents(f.fileno(), size)


def _find(f, size, needle, window=None):
    """Search allocated extents for `needle`.

    `window` is an optional (lo_va, hi_va) guest-address bound. The full sweep
    covers ~371 MB and is the leading suspect for the in-mission freeze
    (mission-freeze-resume-spin.md); every ScriptMission observed so far has sat
    in 0xBC79xxxx-0xBC7Axxxx, so bounding the POINTER scan cuts it by ~10x while
    still finding the object. The header scan stays unbounded -- the .ssb has
    been loaded at two different addresses across runs, so it cannot be bounded
    on this evidence.
    """
    out = []
    for start, end in _fd_extents(f, size):
        if window is not None:
            vs = gmem.va_to_off(window[0]), gmem.va_to_off(window[1])
            if vs[0] is not None and vs[1] is not None:
                if end <= vs[0] or start >= vs[1]:
                    continue
                start = max(start, vs[0]); end = min(end, vs[1])
        f.seek(start)
        remaining, base, prev = end - start, start, b''
        while remaining > 0:
            chunk = f.read(min(1 << 22, remaining))
            if not chunk:
                break
            buf = prev + chunk
            i = buf.find(needle)
            while i >= 0:
                out.append(base - len(prev) + i)
                i = buf.find(needle, i + 1)
            prev = chunk[-len(needle):]
            base += len(chunk)
            remaining -= len(chunk)
    return out


def u32(f, va):
    off = gmem.va_to_off(va)
    if off is None:
        return None
    f.seek(off)
    b = f.read(4)
    return int.from_bytes(b, 'big') if len(b) == 4 else None


def find_mission(f, size, ssb):
    """Locate the live ScriptMission for this .ssb.

    1. find the .ssb HEADER in guest memory -- 20 bytes of version + offsets,
       distinctive enough that it hits once;
    2. code base = file base + the header's code offset;
    3. `[ScriptMission+24]` is that code base, so scan for a word equal to it;
    4. validate with `[+44]`, which must equal file base + symtab1 offset + 4 --
       an exact arithmetic check, not a heuristic.
    """
    hdr = ssb[:HDR_LEN]
    code_off = struct.unpack_from('>I', ssb, 0x08)[0]
    sym1_off = struct.unpack_from('>I', ssb, 0x0C)[0]
    for off in _find(f, size, hdr):
        for filebase in gmem.off_to_vas(off):
            code_base = filebase + code_off
            want_44 = filebase + sym1_off + 4
            for poff in _find(f, size, struct.pack('>I', code_base & 0xFFFFFFFF),
                              window=PTR_WINDOW):
                if poff % 4:
                    continue
                for pva in gmem.off_to_vas(poff):
                    m = pva - 24
                    if u32(f, m + 44) == want_44:
                        return m, filebase
    # nothing in the window -- fall back to the full sweep rather than fail
    for off in _find(f, size, hdr):
        for filebase in gmem.off_to_vas(off):
            code_base = filebase + code_off
            want_44 = filebase + sym1_off + 4
            for poff in _find(f, size, struct.pack('>I', code_base & 0xFFFFFFFF)):
                if poff % 4:
                    continue
                for pva in gmem.off_to_vas(poff):
                    m = pva - 24
                    if u32(f, m + 44) == want_44:
                        return m, filebase
    return None, None


def read_states(f, mission, sym2, names):
    phase = u32(f, mission + 4)
    arr = u32(f, phase + 324) if phase else None
    base = u32(f, arr + 4) if arr else None
    idx = {n: i for i, (_t, n) in sym2.items()}
    out = {'phase_ordinal': u32(f, mission + 40),
           'phase_obj': phase, 'finished': u32(f, phase + 196) if phase else None,
           'units': {}}
    active = 0
    if base:
        for i in sorted(sym2):
            rec = u32(f, base + i * 4)
            if not rec:
                continue
            if u32(f, rec + 16) == 2:
                active += 1
        for n in names:
            i = idx.get(n)
            rec = u32(f, base + i * 4) if i is not None else None
            out['units'][n] = None if not rec else {
                'idx': i, 'obj': bool(u32(f, rec + 4)), 'state': u32(f, rec + 16)}
    out['active_records'] = active
    return out


if __name__ == '__main__':
    ssb = isl.load(sys.argv[1])
    sym2 = isl.symbols(ssb, 2)
    names = sys.argv[2:] or ['ADN110', 'ADN111', 'ADN112']
    import os
    path = gmem.mem_path()
    with open(path, 'rb', buffering=0) as f:
        size = os.path.getsize(path)
        m, fb = find_mission(f, size, ssb)
        if m is None:
            print('ScriptMission not located'); sys.exit(1)
        print('file base 0x%08X, ScriptMission 0x%08X' % (fb, m))
        r = read_states(f, m, sym2, names)
        print('  phase ordinal   = %s   (the REAL counter; the +236 mirror reads 0 in phase 1)'
              % r['phase_ordinal'])
        print('  ScriptPhase     = 0x%08X  finished=%s' % (r['phase_obj'] or 0, r['finished']))
        print('  records active  = %d of %d' % (r['active_records'], len(sym2)))
        for n, v in r['units'].items():
            print('  %-10s %s' % (n, v))