conco/OpenMaidEngine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Model linked INIT record tables

Browse Source
This commit is contained in:
gamer147
2026-07-22 16:57:59 -04:00
parent d6a69ba9b4
commit 5f743275bf
10 changed files with 268 additions and 11020 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
tools/extract_init.py
View File

@@ -20,6 +20,7 @@ Usage: py -3.11 -X utf8 tools/extract_init.py <TABLE> [OUTNAME] [--mode name|num
from __future__ import annotations
import json
import sys
from functools import cache
from pathlib import Path
HERE = Path(__file__).resolve().parent
@@ -89,6 +90,44 @@ def _infer_record_span(string_addrs):
return max(counts, key=lambda delta: (counts[delta], delta))
@cache
def _known_record_tables():
"""Return corpus-observed (base, stride) pairs used by lookup-array-2d.
INIT scripts often populate linked row-major tables while defining an
entity. Treating every such write as `destination - entity_id` invents a
different one-off parallel column for every row. Consumer bytecode gives
us the unambiguous table base and stride instead.
"""
tables = set()
for path in paths.scripts().values():
try:
script = sys4load.load(path)
except sys4load.Sys4Error:
continue
for ins in script.instructions:
if (sys4load.display_label(ins.opcode) == "lookup-array-2d"
and len(ins.args) >= 5
and ins.args[1][0] in (T_GLOBAL_INT, 6)
and ins.args[3][0] == T_IMM
and ins.args[3][1] > 0):
tables.add((ins.args[1][1], ins.args[3][1]))
return tuple(sorted(tables))
def _record_table_cell(destination, record_id):
matches = []
for base, stride in _known_record_tables():
column = destination - (base + record_id * stride)
if 0 <= column < stride:
matches.append((base, stride, column))
if len(matches) > 1:
raise ValueError(
f"ambiguous record-table destination 0x{destination:x} for id {record_id}: {matches}"
)
return matches[0] if matches else None
def extract_name(scr):
string_addrs = [
ins.args[0][1]
@@ -112,17 +151,31 @@ def extract_name(scr):
# boundary: ITINIT begins with 101 consecutive name-only records,
# which the old heuristic collapsed into item zero.
if name_write_base <= addr < name_write_base + record_span:
cur = {"id": addr - name_base, "name": txt, "fields": {}}
cur = {"id": addr - name_base, "name": txt, "fields": {}, "record_fields": {}}
records.append(cur); desc_slot = 0
elif cur is not None:
key = "desc" if desc_slot == 0 else f"desc{desc_slot}"
cur[key] = txt; desc_bases.setdefault(key, addr - cur["id"]); desc_slot += 1
elif ins.opcode == MOV and cur is not None and ins.args and ins.args[0][0] == T_GLOBAL_INT:
cur["fields"][f"0x{ins.args[0][1] - cur['id']:x}"] = _val(ins.args[1])
destination = ins.args[0][1]
cell = _record_table_cell(destination, cur["id"])
if cell is None:
cur["fields"][f"0x{destination - cur['id']:x}"] = _val(ins.args[1])
else:
base, stride, column = cell
cur["record_fields"][f"0x{base:x}/{stride}/{column}"] = _val(ins.args[1])
for record in records:
if not record["record_fields"]:
del record["record_fields"]
record_columns = sorted(
{key for record in records for key in record.get("record_fields", {})},
key=lambda key: tuple(int(part, 0) for part in key.split("/")),
)
return records, {"name_array_base": f"0x{name_base:x}",
"name_write_base": f"0x{name_write_base:x}",
"first_record_id": first_record_id,
"record_span": record_span,
"record_field_columns": record_columns,
"desc_array_bases": {k: f"0x{v:x}" for k, v in sorted(desc_bases.items())}}
@@ -201,17 +254,23 @@ def write_data_index(data_dir: Path) -> None:
"bases remain available in every record; confirmed field meanings live in",
"`vm-map/globals.toml` and the generated `docs/global-reference.md`.",
"",
"| file | mode | records | field bases |",
"|---|---|---:|---:|",
"| file | mode | records | array fields | record columns |",
"|---|---|---:|---:|---:|",
]
for filename, data in tables:
columns = len(data.get("field_columns") or [])
lines.append(f"| `{filename}` | {data['mode']} | {data['record_count']} | {columns} |")
record_columns = len(data.get("record_field_columns") or [])
lines.append(
f"| `{filename}` | {data['mode']} | {data['record_count']} | "
f"{columns} | {record_columns} |"
)
lines += [
"",
"Name-mode tables expose one-based runtime `id` values, the lookup `name_array_base`,",
"the first populated `name_write_base`, and the reserved `record_span`. Fields are keyed",
"by the runtime lookup base used by `lookup-array`, not merely the first written cell.",
"Linked row-major fields are stored separately in `record_fields`, keyed as",
"`base/stride/column` from corpus-observed `lookup-array-2d` consumers.",
"",
"Use `tools/init_table_profile.py <TABLE> --build` to generate value/population and",
"direct-consumer evidence. `STINIT` still requires a bespoke mixed numeric/string parser.",

83
tools/init_table_profile.py
View File

@@ -49,32 +49,58 @@ def value_key(value) -> str:
def profile_columns(data: dict) -> list[dict]:
records = data["records"]
values: dict[int, list] = collections.defaultdict(list)
examples: dict[int, list[dict]] = collections.defaultdict(list)
values: dict[str, list] = collections.defaultdict(list)
examples: dict[str, list[dict]] = collections.defaultdict(list)
identities: dict[str, dict] = {}
for record in records:
for address, value in record.get("fields", {}).items():
base = int(address, 16)
values[base].append(value)
if len(examples[base]) < 5:
examples[base].append({
key = f"0x{base:x}"
identities[key] = {
"key": key, "kind": "parallel-array", "base": key,
"stride": None, "column": None,
}
values[key].append(value)
if len(examples[key]) < 5:
examples[key].append({
"id": record["id"],
"name": record.get("name", ""),
"value": value,
})
for key, value in record.get("record_fields", {}).items():
base_text, stride_text, column_text = key.split("/")
base = int(base_text, 16)
stride = int(stride_text)
column = int(column_text)
normalized_key = f"0x{base:x}/{stride}/{column}"
identities[normalized_key] = {
"key": normalized_key,
"kind": "record-column",
"base": f"0x{base:x}",
"stride": stride,
"column": column,
}
values[normalized_key].append(value)
if len(examples[normalized_key]) < 5:
examples[normalized_key].append({
"id": record["id"],
"name": record.get("name", ""),
"value": value,
})
rows = []
for base, vals in values.items():
for key, vals in values.items():
common = collections.Counter(value_key(value) for value in vals).most_common(6)
numeric = vals and all(isinstance(value, int) for value in vals)
rows.append({
"base": f"0x{base:x}",
**identities[key],
"population": len(vals),
"coverage": len(vals) / len(records) if records else 0.0,
"distinct_values": len({value_key(value) for value in vals}),
"min": min(vals) if numeric else None,
"max": max(vals) if numeric else None,
"common": [{"value": value, "count": count} for value, count in common],
"examples": examples[base],
"examples": examples[key],
"references": 0,
"reader_scripts": [],
"reference_ops": [],
@@ -83,7 +109,9 @@ def profile_columns(data: dict) -> list[dict]:
def add_direct_references(rows: list[dict], source_name: str) -> None:
by_base = {int(row["base"], 16): row for row in rows}
by_base: dict[int, list[dict]] = collections.defaultdict(list)
for row in rows:
by_base[int(row["base"], 16)].append(row)
scripts: dict[int, collections.Counter] = {
base: collections.Counter() for base in by_base
}
@@ -104,20 +132,21 @@ def add_direct_references(rows: list[dict], source_name: str) -> None:
scripts[value][name] += 1
ops[value][f"{sys4load.display_label(ins.opcode)}:arg{arg_index + 1}"] += 1
for base, row in by_base.items():
row["references"] = sum(scripts[base].values())
row["reader_scripts"] = [
{"script": script, "count": count}
for script, count in scripts[base].most_common()
]
row["reference_ops"] = [
{"operation": operation, "count": count}
for operation, count in ops[base].most_common()
]
for base, base_rows in by_base.items():
for row in base_rows:
row["references"] = sum(scripts[base].values())
row["reader_scripts"] = [
{"script": script, "count": count}
for script, count in scripts[base].most_common()
]
row["reference_ops"] = [
{"operation": operation, "count": count}
for operation, count in ops[base].most_common()
]
def render_markdown(data: dict, rows: list[dict], limit: int) -> str:
ranked = sorted(rows, key=lambda row: (-row["population"], -row["references"], row["base"]))
ranked = sorted(rows, key=lambda row: (-row["population"], -row["references"], row["key"]))
shown = ranked[:limit]
lines = [
f"# {data['table']} field profile",
@@ -126,10 +155,10 @@ def render_markdown(data: dict, rows: list[dict], limit: int) -> str:
"> This is evidence for investigation; confirmed names live in `vm-map/globals.toml`.",
"",
f"- records: {data['record_count']}",
f"- populated global-array bases: {len(rows)}",
f"- populated fields: {len(rows)}",
f"- rows shown: {len(shown)} (ranked by record coverage, then consumer references)",
"",
"| base | populated | distinct | range | direct refs | readers | common values | examples |",
"| field | populated | distinct | range | direct refs | readers | common values | examples |",
"|---|---:|---:|---|---:|---|---|---|",
]
for row in shown:
@@ -142,7 +171,7 @@ def render_markdown(data: dict, rows: list[dict], limit: int) -> str:
for entry in row["examples"][:3]
).replace("|", "\\|")
lines.append(
f"| `{row['base']}` | {row['population']}/{data['record_count']} "
f"| `{row['key']}` | {row['population']}/{data['record_count']} "
f"({row['coverage']:.0%}) | {row['distinct_values']} | {value_range} | "
f"{row['references']} | {readers} | {common} | {examples} |"
)
@@ -165,8 +194,12 @@ def main() -> int:
"table": data["table"],
"source": data["source"],
"record_count": data["record_count"],
"field_base_count": len(rows),
"columns": sorted(rows, key=lambda row: int(row["base"], 16)),
"field_column_count": len(rows),
"parallel_array_count": sum(row["kind"] == "parallel-array" for row in rows),
"record_column_count": sum(row["kind"] == "record-column" for row in rows),
"columns": sorted(rows, key=lambda row: (
int(row["base"], 16), row["stride"] or 0, row["column"] or 0
)),
}
markdown = render_markdown(data, rows, args.limit)
print(markdown)

6
tools/test_extract_init.py
View File

@@ -45,6 +45,12 @@ def test_real_name_tables() -> None:
check(by_id[101]["desc"] == "ＨＰ３０回復", "ITINIT item 101 keeps its description")
check(by_id[1]["fields"]["0x8c879"] == 10,
"ITINIT columns use the runtime lookup base")
check(len({key for record in items for key in record["fields"]}) == 13,
"ITINIT has thirteen parallel-array fields")
check(len({key for record in items for key in record.get("record_fields", {})}) == 43,
"ITINIT linked row-major tables expose 43 populated columns")
check(by_id[101]["record_fields"]["0xa5301/3/0"] == 30,
"ITINIT item 101 stores HP recovery in row-major column zero")
if __name__ == "__main__":

12
tools/test_init_table_profile.py
View File

@@ -12,12 +12,14 @@ import init_table_profile as profile
def main() -> int:
fixture = {
"records": [
{"id": 1, "name": "one", "fields": {"0x10": 2, "0x20": 0}},
{"id": 1, "name": "one", "fields": {"0x10": 2, "0x20": 0},
"record_fields": {"0x30/3/0": 9}},
{"id": 3, "name": "three", "fields": {"0x10": 2}},
{"id": 7, "name": "seven", "fields": {"0x10": 5}},
{"id": 7, "name": "seven", "fields": {"0x10": 5},
"record_fields": {"0x30/3/2": 4}},
]
}
rows = {row["base"]: row for row in profile.profile_columns(fixture)}
rows = {row["key"]: row for row in profile.profile_columns(fixture)}
assert rows["0x10"]["population"] == 3
assert rows["0x10"]["coverage"] == 1.0
assert rows["0x10"]["distinct_values"] == 2
@@ -25,6 +27,10 @@ def main() -> int:
assert rows["0x10"]["common"][0] == {"value": "2", "count": 2}
assert rows["0x20"]["population"] == 1
assert rows["0x20"]["examples"][0]["name"] == "one"
assert rows["0x30/3/0"]["kind"] == "record-column"
assert rows["0x30/3/0"]["base"] == "0x30"
assert rows["0x30/3/0"]["stride"] == 3
assert rows["0x30/3/2"]["column"] == 2
print("all init_table_profile checks passed")
return 0