509 lines
22 KiB
Python
509 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""vm0 — headless AGE bytecode interpreter (Phase A0 prototype).
|
||
|
||
Executes one script's bytecode to validate the *execution model* before any C#/Godot work.
|
||
Reuses tools/sys4load.py for all parsing/decoding. Effectful ops and call-script are STUBBED;
|
||
show-text is captured. Named-op semantics come from Kelebek's table; the classified markers are
|
||
treated as no-ops (this run TESTS that assumption).
|
||
|
||
Purpose (see docs/phase-a-slice-plan.md):
|
||
* `--test` run the RECOVER unit test (pointer / 2D-array / loop / control-flow correctness).
|
||
* <file> run a script; print captured show-text + an opcode-coverage/stub report.
|
||
|
||
Memory model:
|
||
* one flat GLOBAL bank G (dict addr->int); globals are raw offsets into one space.
|
||
* per-call local frame with sparse typed banks (int/float/string/ptr).
|
||
* a `-ptr` variable retains an address plus its local/global storage domain. lookup-array/2d
|
||
with a ptr dst stores that reference (take-reference); reads/writes dereference the matching
|
||
bank. RECOVER exercises global references; HISTORY's copied coordinate arrays exercise local ones.
|
||
* jcc(cond, tA, tB): cond truthy -> goto tA else tB; 0xFFFFFFFF = fall through (from RECOVER+SCJUMP).
|
||
* call/ret (0x8F/0x05) are intra-script subroutine calls (shared frame); call-script (0x03) is
|
||
the inter-script one and is stubbed.
|
||
"""
|
||
from __future__ import annotations
|
||
import sys
|
||
import collections
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(HERE))
|
||
import paths
|
||
import sys4load
|
||
from age_opcodes import OPCODES
|
||
|
||
NOJUMP = 0xFFFFFFFF
|
||
|
||
# operand type tags
|
||
T_IMM, T_STR = 0x0, 0x2
|
||
T_GINT, T_GFLOAT, T_GSTR, T_GPTR = 0x3, 0x4, 0x5, 0x6
|
||
T_LINT, T_LFLOAT, T_LSTR, T_LPTR = 0x9, 0xA, 0xB, 0xC
|
||
|
||
# opcodes treated as no-ops in v1 (classified markers; see age_opcodes_himegari.py)
|
||
# 0x71 = label-definition pseudo-op (count == T1 table size) — structural, no runtime effect.
|
||
MARKERS = {0x71, 0x1f4, 0x1f5, 0x1d5, 0x1bc, 0x1bf, 0x21b, 0x1d2, 0x258}
|
||
|
||
# loop-guard: halt a run once any single show-text line has been emitted this many times.
|
||
# State-gated scenes run with zero initial state can spin a loop that re-emits the same block
|
||
# forever; capping re-emission terminates them cleanly and flags them LOOPED (vs. blind step limit).
|
||
EMIT_CAP = 2
|
||
|
||
|
||
class Frame:
|
||
def __init__(self):
|
||
self.i = collections.defaultdict(int) # local-int
|
||
self.f = collections.defaultdict(int) # local-float (stored raw)
|
||
self.s = collections.defaultdict(str) # local-string
|
||
self.p = collections.defaultdict(lambda: ("g", 0)) # local-ptr: (bank, address)
|
||
|
||
|
||
class VM:
|
||
def __init__(self, scr: sys4load.Sys4Script, verbose=False, emit_cap=EMIT_CAP, record_trace=False):
|
||
self.scr = scr
|
||
self.verbose = verbose
|
||
self.record_trace = record_trace
|
||
self.trace = [] # executed code offsets (only if record_trace)
|
||
self.code = scr.instructions
|
||
self.by_off = {ins.offset: idx for idx, ins in enumerate(self.code)}
|
||
self.G = collections.defaultdict(int) # global-int bank (flat address space)
|
||
self.Gs = collections.defaultdict(str) # global-string bank
|
||
self.fr = Frame()
|
||
self.callstack = [] # return indices for call/ret
|
||
self.text = [] # captured show-text as (str_offset, text)
|
||
self.settex = [] # set-texture calls: (code_offset, resId, slot)
|
||
self.emit_seen = collections.Counter() # per-offset emit count (loop-guard)
|
||
self.emit_cap = emit_cap
|
||
self.halt_reason = None # 'exit' | 'LOOP:...' | 'STEP-LIMIT' | 'ret-underflow'
|
||
self.log = collections.Counter() # stub/unknown opcode hits
|
||
self.exec_count = collections.Counter() # opcode coverage
|
||
self.steps = 0
|
||
|
||
# ---- operand resolution --------------------------------------------------
|
||
def _string(self, off):
|
||
e = self.scr.strings.get(off)
|
||
if e is not None:
|
||
return e[0]
|
||
txt, _ = sys4load._decode_string(self.scr.dwords, off)
|
||
return txt if txt is not None else ""
|
||
|
||
def read(self, op):
|
||
t, v = op
|
||
if t == T_IMM: return v
|
||
if t == T_STR: return self._string(v)
|
||
if t == T_GINT: return self.G[v]
|
||
if t == T_GFLOAT: return self.G[v]
|
||
if t == T_GSTR: return self.Gs[v]
|
||
if t == T_GPTR: return self.G[self.G[v]]
|
||
if t == T_LINT: return self.fr.i[v]
|
||
if t == T_LFLOAT: return self.fr.f[v]
|
||
if t == T_LSTR: return self.fr.s[v]
|
||
if t == T_LPTR: return self.read_cell(self.fr.p[v]) # deref ptr
|
||
self.log[f"read?t{t:#x}"] += 1
|
||
return v
|
||
|
||
def write(self, op, val):
|
||
t, v = op
|
||
if t == T_GINT or t == T_GFLOAT: self.G[v] = val
|
||
elif t == T_GSTR: self.Gs[v] = val
|
||
elif t == T_GPTR: self.G[self.G[v]] = val
|
||
elif t == T_LINT: self.fr.i[v] = val
|
||
elif t == T_LFLOAT: self.fr.f[v] = val
|
||
elif t == T_LSTR: self.fr.s[v] = val
|
||
elif t == T_LPTR: self.write_cell(self.fr.p[v], val)
|
||
else: self.log[f"write?t{t:#x}"] += 1
|
||
|
||
def read_cell(self, address):
|
||
bank, addr = address
|
||
if bank == "i": return self.fr.i[addr]
|
||
if bank == "f": return self.fr.f[addr]
|
||
return self.G[addr]
|
||
|
||
def write_cell(self, address, value):
|
||
bank, addr = address
|
||
if bank == "i": self.fr.i[addr] = value
|
||
elif bank == "f": self.fr.f[addr] = value
|
||
else: self.G[addr] = value
|
||
|
||
def base_addr(self, op):
|
||
"""The base ADDRESS an operand names, for array lookups."""
|
||
t, v = op
|
||
if t in (T_IMM, T_GINT, T_GFLOAT, T_GSTR, T_GPTR): return ("g", v)
|
||
if t == T_LINT: return ("i", v)
|
||
if t == T_LFLOAT: return ("f", v)
|
||
if t == T_LSTR: return ("s", v)
|
||
if t == T_LPTR: return self.fr.p[v]
|
||
return ("g", v)
|
||
|
||
def lookup_store(self, dst, addr):
|
||
"""lookup result: ptr dst gets the reference (address); non-ptr gets the element value."""
|
||
t, v = dst
|
||
if t == T_LPTR: self.fr.p[v] = addr
|
||
elif t == T_GPTR: self.G[v] = addr[1]
|
||
else: self.write(dst, self.read_cell(addr))
|
||
|
||
# ---- execution -----------------------------------------------------------
|
||
def run(self, entry_off=0, max_steps=2_000_000):
|
||
pc = self.by_off.get(entry_off, 0)
|
||
while 0 <= pc < len(self.code):
|
||
if self.steps >= max_steps:
|
||
self.log["STEP-LIMIT"] += 1
|
||
self.halt_reason = self.halt_reason or "STEP-LIMIT"
|
||
break
|
||
self.steps += 1
|
||
ins = self.code[pc]
|
||
op = ins.opcode
|
||
self.exec_count[op] += 1
|
||
if self.record_trace:
|
||
self.trace.append(ins.offset)
|
||
nxt = self.step(ins, pc)
|
||
if nxt is None: # halt
|
||
break
|
||
pc = nxt
|
||
else:
|
||
self.halt_reason = self.halt_reason or "pc-out-of-range"
|
||
return self
|
||
|
||
def step(self, ins, pc):
|
||
op, a = ins.opcode, ins.args
|
||
lbl = OPCODES.get(op, ("?", 0))[0]
|
||
|
||
# arithmetic / bitwise: dst = a1 <op> a2
|
||
alu = {"add": lambda x, y: x + y, "sub": lambda x, y: x - y,
|
||
"mul": lambda x, y: x * y, "div": lambda x, y: (x // y if y else 0),
|
||
"mod": lambda x, y: (x % y if y else 0), "and": lambda x, y: x & y,
|
||
"or": lambda x, y: x | y, "sar": lambda x, y: x >> (y & 31),
|
||
"shl": lambda x, y: x << (y & 31)}
|
||
cmp = {"eq": lambda x, y: int(x == y), "ne": lambda x, y: int(x != y),
|
||
"lt": lambda x, y: int(x < y), "lte": lambda x, y: int(x <= y),
|
||
"gr": lambda x, y: int(x > y), "gre": lambda x, y: int(x >= y)}
|
||
|
||
if lbl in alu:
|
||
self.write(a[0], alu[lbl](self.read(a[1]), self.read(a[2]))); return pc + 1
|
||
if lbl in cmp:
|
||
self.write(a[0], cmp[lbl](self.read(a[1]), self.read(a[2]))); return pc + 1
|
||
if lbl == "mov":
|
||
self.write(a[0], self.read(a[1])); return pc + 1
|
||
if lbl == "set-string":
|
||
self.write(a[0], self.read(a[1])); return pc + 1
|
||
if lbl == "lookup-array": # dst = base[idx]
|
||
bank, base = self.base_addr(a[1])
|
||
addr = (bank, base + self.read(a[2]))
|
||
self.lookup_store(a[0], addr); return pc + 1
|
||
if lbl == "lookup-array-2d": # dst = base[i*stride + col]
|
||
bank, base = self.base_addr(a[1])
|
||
addr = (bank, base + self.read(a[2]) * self.read(a[3]) + self.read(a[4]))
|
||
self.lookup_store(a[0], addr); return pc + 1
|
||
if lbl == "bit-set":
|
||
bit = self.read(a[1])
|
||
if not 0 <= bit < 32:
|
||
self.halt_reason = f"bit-index-out-of-range:{bit}"
|
||
return None
|
||
self.write(a[0], self.read(a[0]) | (1 << bit)); return pc + 1
|
||
if lbl == "bit-reset":
|
||
bit = self.read(a[1])
|
||
if not 0 <= bit < 32:
|
||
self.halt_reason = f"bit-index-out-of-range:{bit}"
|
||
return None
|
||
self.write(a[0], self.read(a[0]) & ~(1 << bit)); return pc + 1
|
||
if lbl == "check-bit": # p1 = (p2 >> p3) & 1
|
||
self.write(a[0], (self.read(a[1]) >> (self.read(a[2]) & 31)) & 1); return pc + 1
|
||
if lbl == "copy-to-global": # best-effort: p1 = p2 (single cell)
|
||
self.write(a[0], self.read(a[1])); return pc + 1
|
||
|
||
# control flow
|
||
if lbl == "jmp":
|
||
return self.by_off.get(a[0][1], pc + 1)
|
||
if lbl == "call": # intra-script subroutine
|
||
self.callstack.append(pc + 1)
|
||
return self.by_off.get(a[0][1], pc + 1)
|
||
if lbl == "ret":
|
||
if self.callstack:
|
||
return self.callstack.pop()
|
||
self.halt_reason = "ret-underflow"
|
||
return None
|
||
if lbl == "jcc": # (cond, tA, tB); 0xFFFFFFFF=fallthrough
|
||
tgt = a[1][1] if self.read(a[0]) else a[2][1]
|
||
return pc + 1 if tgt == NOJUMP else self.by_off.get(tgt, pc + 1)
|
||
if lbl in ("exit", "exit-script"):
|
||
self.halt_reason = "exit"
|
||
return None
|
||
if lbl == "call-script": # STUB (inter-script)
|
||
self.log[f"call-script({self.read(a[0]) if a else '?'})"] += 1; return pc + 1
|
||
|
||
# ADV text: capture the string operand as (offset, text); loop-guard on re-emission
|
||
if lbl == "show-text":
|
||
for t, v in a:
|
||
if t == T_STR:
|
||
self.emit_seen[v] += 1
|
||
if self.emit_seen[v] > self.emit_cap:
|
||
self.halt_reason = f"LOOP:line@{v:#x}×{self.emit_seen[v]}"
|
||
return None
|
||
self.text.append((v, self._string(v)))
|
||
return pc + 1
|
||
if lbl in ("end-text-line", "wait-for-input", "set-font", "comment",
|
||
"display-furigana", "dev_ukn"):
|
||
return pc + 1
|
||
|
||
if lbl == "set-texture": # 0x1f9 (resId, slot, flag) — trace the load
|
||
resid = self.read(a[0]) if a else None
|
||
slot = self.read(a[1]) if len(a) > 1 else None
|
||
self.settex.append((ins.offset, resid, slot, len(self.trace))) # +trace index
|
||
return pc + 1
|
||
|
||
if op in MARKERS: # classified no-op markers
|
||
return pc + 1
|
||
|
||
# everything else (effectful draw/audio/ui/input, unnamed) -> stub + continue
|
||
self.log[f"stub:{lbl if not lbl.startswith('u00') else hex(op)}"] += 1
|
||
return pc + 1
|
||
|
||
|
||
def run_test():
|
||
"""RECOVER unit test — pointer/2D-array/loop/control-flow correctness."""
|
||
scr = sys4load.load(paths.DATA1 / "RECOVER.BIN")
|
||
vm = VM(scr)
|
||
unit = 0
|
||
vm.G[0x152616] = unit
|
||
A, B = 0x4e11b, 0x4e085 # block-1 tables (stride 14 / 3)
|
||
C, E, F, FL = 0x52383, 0x52f3b, 0x5295f, 0xaacb4 # block-2 tables (stride 30) + flags
|
||
# block 1 source: A[unit*14 + 11..13]
|
||
for k in range(3):
|
||
vm.G[A + unit * 14 + (11 + k)] = 100 + k
|
||
# block 2: slot 5 active, slot 6 zero-C (skip), slot 7 flag-off (skip)
|
||
vm.G[C + unit * 30 + 5] = 7; vm.G[FL + 5] = 1; vm.G[E + unit * 30 + 5] = 42
|
||
vm.G[C + unit * 30 + 6] = 0; vm.G[FL + 6] = 1; vm.G[E + unit * 30 + 6] = 99
|
||
vm.G[C + unit * 30 + 7] = 3; vm.G[FL + 7] = 0; vm.G[E + unit * 30 + 7] = 88
|
||
vm.run()
|
||
|
||
checks = [
|
||
("block1 B[0]=A[11]", vm.G[B + unit * 3 + 0], 100),
|
||
("block1 B[1]=A[12]", vm.G[B + unit * 3 + 1], 101),
|
||
("block1 B[2]=A[13]", vm.G[B + unit * 3 + 2], 102),
|
||
("block2 s5 C:=E", vm.G[C + unit * 30 + 5], 42),
|
||
("block2 s5 F:=-1", vm.G[F + unit * 30 + 5], -1),
|
||
("block2 s6 C skip", vm.G[C + unit * 30 + 6], 0), # C stayed 0 (guard)
|
||
("block2 s7 C skip", vm.G[C + unit * 30 + 7], 3), # flag off -> untouched
|
||
]
|
||
ok = True
|
||
for name, got, want in checks:
|
||
status = "OK " if got == want else "FAIL"
|
||
if got != want:
|
||
ok = False
|
||
print(f" [{status}] {name}: got {got}, want {want}")
|
||
print(f" steps={vm.steps} call-script stubs={sum(v for k,v in vm.log.items() if k.startswith('call-script'))}")
|
||
print("RECOVER unit test:", "PASS" if ok else "FAIL")
|
||
return 0 if ok else 1
|
||
|
||
|
||
def run_file(path):
|
||
scr = sys4load.load(path)
|
||
vm = VM(scr).run()
|
||
print(f"== {Path(path).name}: {vm.steps} steps, {len(vm.text)} show-text lines captured "
|
||
f"(halt: {vm.halt_reason}) ==")
|
||
for i, (off, t) in enumerate(vm.text[:20]):
|
||
print(f" [{i}] {t}")
|
||
if len(vm.text) > 20:
|
||
print(f" ... (+{len(vm.text)-20} more)")
|
||
stubs = [(k, v) for k, v in vm.log.most_common() if k.startswith("stub:")]
|
||
if stubs:
|
||
print(" top stubbed effectful/unknown ops:", ", ".join(f"{k[5:]}×{v}" for k, v in stubs[:10]))
|
||
ncall = sum(v for k, v in vm.log.items() if k.startswith("call-script"))
|
||
print(f" call-script stubs: {ncall} distinct opcodes executed: {len(vm.exec_count)}")
|
||
return 0
|
||
|
||
|
||
# ---- dialogue oracle ---------------------------------------------------------
|
||
import json
|
||
import re
|
||
|
||
SCENE_RE = re.compile(r"^S[CP]\d{4}\.BIN$")
|
||
|
||
|
||
def load_oracle(path=None):
|
||
"""file (UPPER) -> ordered list of (str_offset, text): the static show-text lines
|
||
extract_phase2 dumped. This is the independent oracle the VM is validated against."""
|
||
path = path or (paths.BUILD / "text" / "dialogue.jsonl")
|
||
by = collections.defaultdict(list)
|
||
with open(path, encoding="utf-8") as f:
|
||
for line in f:
|
||
d = json.loads(line)
|
||
by[d["file"].upper()].append((int(d["off"], 16), d["text"]))
|
||
return by
|
||
|
||
|
||
def subsequence_status(emitted_offs, static_offs):
|
||
"""Classify the emitted show-text offset stream against the static ordered set.
|
||
|
||
Emitted may repeat offsets (loop re-emission); we validate the distinct stream in
|
||
first-seen order as an in-order subsequence of the static lines. Returns (status,
|
||
detail): OK | STRAY (offset never in static) | ORDER (in static but out of order)."""
|
||
static_list = list(static_offs)
|
||
static_set = set(static_list)
|
||
seen, distinct = set(), []
|
||
for o in emitted_offs:
|
||
if o not in seen:
|
||
seen.add(o)
|
||
distinct.append(o)
|
||
strays = [o for o in distinct if o not in static_set]
|
||
if strays:
|
||
return "STRAY", strays
|
||
j = 0
|
||
for o in distinct:
|
||
while j < len(static_list) and static_list[j] != o:
|
||
j += 1
|
||
if j >= len(static_list):
|
||
return "ORDER", [o]
|
||
j += 1
|
||
return "OK", []
|
||
|
||
|
||
def verdict(status, halt, n_emit, n_static):
|
||
"""Fold subsequence status + halt reason into one scene verdict."""
|
||
if n_static == 0:
|
||
return "NO-DIALOGUE" # not an ADV scene (no static show-text) — skip in scoring
|
||
if n_emit == 0:
|
||
return "EMPTY" # scene has dialogue but VM emitted none — investigate
|
||
if status != "OK":
|
||
return status # STRAY / ORDER — a real divergence
|
||
return "CLEAN" if halt == "exit" else f"OK/{halt}" # OK subsequence; did it exit cleanly?
|
||
|
||
|
||
def run_scene(name, path, oracle):
|
||
"""Run one scene, return a result dict comparing emitted vs static show-text."""
|
||
scr = sys4load.load(path)
|
||
vm = VM(scr).run()
|
||
emitted = [off for off, _ in vm.text]
|
||
static = oracle.get(name.upper(), [])
|
||
static_offs = [o for o, _ in static]
|
||
status, detail = subsequence_status(emitted, static_offs)
|
||
n_emit = len({o for o in emitted})
|
||
return {"name": name, "vm": vm, "emitted": emitted, "static": static,
|
||
"n_emit_distinct": n_emit, "n_static": len(static_offs),
|
||
"status": status, "detail": detail,
|
||
"verdict": verdict(status, vm.halt_reason, n_emit, len(static_offs))}
|
||
|
||
|
||
def run_one_scene(name):
|
||
"""Detailed single-scene oracle diff (for investigating one script)."""
|
||
oracle = load_oracle()
|
||
scripts = paths.scripts()
|
||
name = name.upper()
|
||
if not name.endswith(".BIN"):
|
||
name += ".BIN"
|
||
if name not in scripts:
|
||
print(f"no such script: {name}")
|
||
return 1
|
||
r = run_scene(name, scripts[name], oracle)
|
||
vm = r["vm"]
|
||
print(f"== {name}: verdict {r['verdict']} ==")
|
||
print(f" emitted {len(r['emitted'])} lines ({r['n_emit_distinct']} distinct), "
|
||
f"static {r['n_static']}, steps {vm.steps}, halt {vm.halt_reason}")
|
||
if r["status"] == "STRAY":
|
||
stat_set = {o for o, _ in r["static"]}
|
||
print(f" STRAY offsets (emitted, not in static): "
|
||
f"{', '.join(hex(o) for o in r['detail'])}")
|
||
for off, txt in vm.text:
|
||
if off in set(r["detail"]):
|
||
print(f" @ {off:#x}: {txt!r}")
|
||
elif r["status"] == "ORDER":
|
||
print(f" first out-of-order offset: {hex(r['detail'][0])}")
|
||
return 0
|
||
|
||
|
||
def run_sweep(limit=None):
|
||
"""Run every SC####/SP#### scene against the dialogue oracle; print a coverage table."""
|
||
oracle = load_oracle()
|
||
scripts = paths.scripts()
|
||
names = sorted(n for n in scripts if SCENE_RE.match(n))
|
||
if limit:
|
||
names = names[:limit]
|
||
buckets = collections.Counter()
|
||
rows = []
|
||
for name in names:
|
||
r = run_scene(name, scripts[name], oracle)
|
||
buckets[r["verdict"]] += 1
|
||
rows.append(r)
|
||
|
||
# non-clean scenes get listed for follow-up
|
||
problem = [r for r in rows if r["verdict"] not in ("CLEAN", "NO-DIALOGUE")]
|
||
if problem:
|
||
print("non-clean scenes:")
|
||
for r in sorted(problem, key=lambda r: r["verdict"]):
|
||
d = ""
|
||
if r["status"] in ("STRAY", "ORDER"):
|
||
d = " " + " ".join(hex(o) for o in r["detail"][:4])
|
||
print(f" {r['name']:<14} {r['verdict']:<16} "
|
||
f"emit {r['n_emit_distinct']:>4}/{r['n_static']:<4} "
|
||
f"steps {r['vm'].steps:>7} halt {r['vm'].halt_reason}{d}")
|
||
print()
|
||
scene_total = sum(v for k, v in buckets.items() if k != "NO-DIALOGUE")
|
||
valid = sum(v for k, v in buckets.items()
|
||
if k == "CLEAN" or k.startswith("OK/"))
|
||
print(f"scenes with dialogue: {scene_total} (skipped {buckets['NO-DIALOGUE']} with no static show-text)")
|
||
print("verdict breakdown:", dict(sorted(buckets.items())))
|
||
print(f"DIALOGUE-VALID (clean in-order subsequence, no garbage): "
|
||
f"{valid}/{scene_total} = {valid/scene_total*100:.1f}%")
|
||
return 0
|
||
|
||
|
||
def run_trace(out_path):
|
||
"""Dump per-scene emitted show-text offsets + halt + steps for the differential test."""
|
||
oracle = load_oracle()
|
||
scripts = paths.scripts()
|
||
names = sorted(n for n in scripts if SCENE_RE.match(n))
|
||
trace = {}
|
||
for name in names:
|
||
r = run_scene(name, scripts[name], oracle)
|
||
vm = r["vm"]
|
||
trace[name] = {"offsets": [off for off, _ in vm.text],
|
||
"halt": vm.halt_reason, "steps": vm.steps}
|
||
Path(out_path).write_text(json.dumps(trace, ensure_ascii=False), encoding="utf-8")
|
||
print(f"trace: {len(trace)} scenes -> {out_path}")
|
||
return 0
|
||
|
||
|
||
def run_settex(name):
|
||
"""Execute a scene and dump its set-texture(resId) trace in execution order.
|
||
|
||
This is the VM side of the asset-resolution scope-selector correlation
|
||
(docs/asset-resolution-re.md): each entry is {i, off, resId, slot}, and aligning this
|
||
ordered resId sequence with the game's Frida load order pins every load to a bytecode
|
||
offset -> localizes where the active CG package/scope switches.
|
||
"""
|
||
scripts = paths.scripts()
|
||
key = name.upper() if name.upper().endswith(".BIN") else name.upper() + ".BIN"
|
||
if key not in scripts:
|
||
raise SystemExit(f"scene not found: {name}")
|
||
vm = VM(sys4load.load(scripts[key]), record_trace=True)
|
||
vm.run()
|
||
out = paths.BUILD / f"settex-{key.removesuffix('.BIN')}.json"
|
||
rows = [{"i": i, "off": f"0x{off:x}", "resId": rid, "slot": slot, "trace_i": ti}
|
||
for i, (off, rid, slot, ti) in enumerate(vm.settex)]
|
||
out.write_text(json.dumps({"scene": key, "halt": vm.halt_reason, "steps": vm.steps,
|
||
"count": len(rows), "settex": rows,
|
||
"trace": [f"0x{o:x}" for o in vm.trace]},
|
||
ensure_ascii=False), encoding="utf-8")
|
||
print(f"{key}: {len(rows)} set-texture calls (halt={vm.halt_reason}, steps={vm.steps}) "
|
||
f"-> {out.relative_to(paths.REPO)}")
|
||
for r in rows[:20]:
|
||
print(f" #{r['i']:<3} {r['off']:>8} resId={r['resId']} (0x{r['resId']:x}) slot={r['slot']}")
|
||
return 0
|
||
|
||
|
||
def main(argv=None):
|
||
argv = argv if argv is not None else sys.argv[1:]
|
||
if not argv or argv[0] == "--test":
|
||
return run_test()
|
||
if argv[0] == "--sweep":
|
||
return run_sweep(limit=int(argv[1]) if len(argv) > 1 else None)
|
||
if argv[0] == "--scene":
|
||
return run_one_scene(argv[1])
|
||
if argv[0] == "--settex":
|
||
return run_settex(argv[1])
|
||
if argv[0] == "--trace":
|
||
return run_trace(argv[1])
|
||
return run_file(argv[0])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|