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

#!/usr/bin/env python3
"""Find `REMAINING OB` in RAM by value-scanning against the HUD, automatically.

The method is `ob_scan.py`'s, and the two traps it records are what this
automates away, because both were paid twice by hand:

* **Confirm the HUD on BOTH sides of the scan.** The scan itself takes ~0.9 s,
  but the counter climbs 004 -> 008 -> 012 within the first minutes of Stage 02,
  and a scan taken across a step selects nothing. A pass whose before- and
  after-readings disagree is thrown away and retried rather than used.
* **Verify across a transition you did NOT select on.** The first filter only
  narrows; the SECOND one is the evidence. Two hand attempts each managed one
  filter and then lost the counter — once because the pilot stopped killing
  anything, once because the mission ended.

The HUD is read by `ob_read.py`, which returns `?` rather than a wrong digit, so
an unreadable frame (explosion, GAME OVER) stalls the hunt instead of poisoning
it.

Usage: ob_hunt.py <out-dir> [transitions] [timeout_s]
"""
import json
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import frozen    # noqa: E402
import gmem      # noqa: E402
import ob_read   # noqa: E402
import ob_scan   # noqa: E402


def hud(shot):
    subprocess.run(["screenshot", shot], stdout=subprocess.DEVNULL,
                   stderr=subprocess.DEVNULL)
    txt, _ = ob_read.read(shot)
    return int(txt) if txt.isdigit() else None


def main():
    out = sys.argv[1]
    want = int(sys.argv[2]) if len(sys.argv) > 2 else 2
    deadline = time.time() + (float(sys.argv[3]) if len(sys.argv) > 3 else 900)
    os.makedirs(out, exist_ok=True)
    mem = gmem.mem_path()
    log = lambda *a: print(*a, flush=True)

    # --- the scan, with the HUD confirmed on both sides -----------------------
    cands = None
    while cands is None and time.time() < deadline:
        v0 = hud(f"{out}/scan-before.png")
        if v0 is None:
            time.sleep(3)
            continue
        offs = ob_scan.scan(mem, v0)
        v1 = hud(f"{out}/scan-after.png")
        if v1 != v0:
            log(f"scan at {v0} discarded: HUD moved to {v1} across it")
            continue
        cands = offs
        log(f"scanned at {v0:03d}: {len(cands)} candidates")
    if cands is None:
        log("never got a clean scan")
        return 1

    # --- filter on each following transition ---------------------------------
    prev, done, results = v0, 0, []
    stuck = 0
    while done < want and time.time() < deadline:
        time.sleep(5)
        v = hud(f"{out}/poll.png")
        if v is None or v == prev:
            # "the counter never moved" is a claim about the GAME; a frozen
            # guest makes it a claim about a dead world instead, and the run
            # then burns its whole timeout saying nothing. Measured once: the
            # world stopped ~10 s into flight and everything downstream --
            # screen_id, the liveness check, the CPU -- stayed happy.
            stuck += 1
            if stuck % 12 == 0 and frozen.frozen(6.0)[0]:
                log("GUEST FROZEN — the world stopped advancing; abandoning")
                break
            continue
        stuck = 0
        # Confirm the new value before spending a filter on it.
        time.sleep(1.5)
        if hud(f"{out}/poll2.png") != v:
            continue
        vals = ob_scan.read_at(mem, cands)
        cands = [o for o, x in vals.items() if x == v]
        done += 1
        kind = "SELECTED ON" if done == 1 else "verification (unselected)"
        log(f"{prev:03d} -> {v:03d}  [{kind}]  {len(cands)} survive")
        results.append({"from": prev, "to": v, "survivors": len(cands)})
        prev = v
        if not cands:
            log("no survivors — the scan value or a reading was wrong")
            break

    payload = {"scan_value": v0, "transitions": results,
               "survivors": [{"off": o, "va": gmem.primary_va(o)} for o in sorted(cands)]}
    json.dump(payload, open(f"{out}/hunt.json", "w"), indent=1)
    # Cap the listing. On a clean run this is a handful of addresses; on an
    # aborted one (frozen guest, no transition) it is the whole scan, and 21 482
    # lines of noise buried the one line that mattered.
    for o in sorted(cands)[:40]:
        log(f"  va {gmem.primary_va(o):#x}")
    if len(cands) > 40:
        log(f"  ... and {len(cands) - 40} more (see hunt.json)")
    log(f"wrote {out}/hunt.json")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())