Expose semantic INIT column names

This commit is contained in:
gamer147
2026-07-22 23:15:33 -04:00
parent b536e91fbc
commit 9efa56c210
11 changed files with 251 additions and 37 deletions

View File

@@ -280,6 +280,40 @@ def join_messages(records: list[dict], message_scr) -> dict:
}
@cache
def _global_registry() -> dict:
path = paths.BUILD / "globals.json"
try:
return json.loads(path.read_text(encoding="utf8")).get("globals", {})
except (OSError, json.JSONDecodeError):
return {}
def field_semantics(records: list[dict]) -> dict[str, str]:
"""Map raw extracted field keys to canonical semantic names when available."""
keys = {
key
for record in records
for key in (*record.get("fields", {}), *record.get("record_fields", {}))
}
registry = _global_registry()
semantics = {}
for key in sorted(keys, key=lambda value: tuple(
int(part, 0) for part in value.split("/")
)):
parts = key.split("/")
entry = registry.get(f"0x{int(parts[0], 16):x}", {})
name = entry.get("name")
if not name:
continue
if len(parts) == 3:
column = parts[2]
column_name = entry.get("columns", {}).get(column, f"column_{column}")
name = f"{name}.{column_name}"
semantics[key] = name
return semantics
def write_data_index(data_dir: Path) -> None:
"""Regenerate the disposable build/data index from current table JSONs."""
tables = []
@@ -323,6 +357,8 @@ def write_data_index(data_dir: Path) -> None:
"Where a matching `*MES` dispatcher exists, `message` preserves its player-facing",
"title, description, furigana, and bytecode dispatch offset separately from the",
"short description stored by the INIT script.",
"Top-level `field_semantics` maps raw array/row-column keys to canonical machine-readable",
"names from `vm-map/globals.toml`; raw keys remain intact as bytecode provenance.",
"",
"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.",
@@ -350,9 +386,11 @@ def main() -> int:
)
cols = sorted({c for r in recs for c in r.get("fields", {})}, key=lambda h: int(h, 16))
semantics = field_semantics(recs)
out = {"table": name, "source": scr.path.name, "magic": scr.magic, "mode": mode,
"record_count": len(recs), **meta,
"field_columns": cols if mode != "footer" else None, "records": recs}
"field_columns": cols if mode != "footer" else None,
"field_semantics": semantics, "records": recs}
outpath = paths.BUILD / "data" / f"{outname}.json"
outpath.parent.mkdir(parents=True, exist_ok=True)
outpath.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")

View File

@@ -49,6 +49,20 @@ def lint(entries: dict[int, dict], all_addrs: set[int]) -> tuple[list[str], list
errors.append(f"{tag}: bad confidence {conf!r}")
if src == "auto-shape" and conf == "high":
errors.append(f"{tag}: auto-shape source may not claim high confidence")
columns = e.get("columns", {})
if not isinstance(columns, dict):
errors.append(f"{tag}: columns must be a table")
else:
for column, name in columns.items():
try:
column_index = int(column)
except (TypeError, ValueError):
errors.append(f"{tag}: bad column index {column!r}")
continue
if column_index < 0:
errors.append(f"{tag}: negative column index {column_index}")
if not isinstance(name, str) or not name:
errors.append(f"{tag}: column {column_index} has no semantic name")
for dep in e.get("depends_on", []):
if _parse_addr(dep) not in all_addrs:
errors.append(f"{tag}: depends_on missing address {dep}")
@@ -96,6 +110,10 @@ def merge(curated: dict[int, dict], auto: dict) -> dict[int, dict]:
"source": e.get("source", "inference"), "confidence": e.get("confidence", "low"),
"depends_on": [f"0x{_parse_addr(d):x}" for d in e.get("depends_on", [])],
"provenance": "curated"}
if e.get("columns"):
out[addr]["columns"] = {
str(column): name for column, name in e["columns"].items()
}
return out
@@ -121,6 +139,14 @@ def emit_reference_md(merged: dict[int, dict]) -> str:
e = merged[addr]
name = e["name"] or ""
usage = (e["usage"] or "").replace("|", "\\|").replace("\n", " ")
if columns := e.get("columns"):
mapping = ", ".join(
f"{column}={column_name}"
for column, column_name in sorted(
columns.items(), key=lambda item: int(item[0])
)
)
usage += f" Columns: {mapping}."
L.append(f"| `{e['address']}` | {name} | {e['confidence']} | {e['source']} | {usage} |")
L.append("")
return "\n".join(L) + "\n"

View File

@@ -19,6 +19,7 @@ from __future__ import annotations
import argparse
import collections
import json
import re
import sys
from pathlib import Path
@@ -49,6 +50,7 @@ def value_key(value) -> str:
def profile_columns(data: dict) -> list[dict]:
records = data["records"]
field_semantics = data.get("field_semantics", {})
values: dict[str, list] = collections.defaultdict(list)
examples: dict[str, list[dict]] = collections.defaultdict(list)
identities: dict[str, dict] = {}
@@ -59,14 +61,18 @@ def profile_columns(data: dict) -> list[dict]:
identities[key] = {
"key": key, "kind": "parallel-array", "base": key,
"stride": None, "column": None,
"semantic_name": field_semantics.get(key),
}
values[key].append(value)
if len(examples[key]) < 5:
examples[key].append({
example = {
"id": record["id"],
"name": record.get("name", ""),
"value": value,
})
}
if message := record.get("message"):
example["message_description"] = message.get("description", "")
examples[key].append(example)
for key, value in record.get("record_fields", {}).items():
base_text, stride_text, column_text = key.split("/")
base = int(base_text, 16)
@@ -79,14 +85,18 @@ def profile_columns(data: dict) -> list[dict]:
"base": f"0x{base:x}",
"stride": stride,
"column": column,
"semantic_name": field_semantics.get(normalized_key),
}
values[normalized_key].append(value)
if len(examples[normalized_key]) < 5:
examples[normalized_key].append({
example = {
"id": record["id"],
"name": record.get("name", ""),
"value": value,
})
}
if message := record.get("message"):
example["message_description"] = message.get("description", "")
examples[normalized_key].append(example)
rows = []
for key, vals in values.items():
@@ -131,6 +141,48 @@ def profile_messages(data: dict) -> dict:
}
def find_message_matches(data: dict, pattern: str) -> list[dict]:
"""Return records whose name/title/description matches a regular expression."""
regex = re.compile(pattern, re.IGNORECASE)
return [
record
for record in data["records"]
if regex.search("\n".join([
record.get("name", ""),
record.get("message", {}).get("title", ""),
record.get("message", {}).get("description", ""),
]))
]
def render_message_matches(data: dict, pattern: str) -> str:
"""Render message hits beside every populated INIT field for correlation."""
matches = find_message_matches(data, pattern)
escaped_pattern = pattern.replace("`", "\\`")
lines = [
f"# {data['table']} message matches",
"",
f"- query: `{escaped_pattern}`",
f"- matches: {len(matches)}",
"",
"| id | name | player-facing description | populated fields |",
"|---:|---|---|---|",
]
for record in matches:
fields = {**record.get("fields", {}), **record.get("record_fields", {})}
rendered_fields = ", ".join(
f"`{data.get('field_semantics', {}).get(key, key)}` (`{key}`)={value}"
for key, value in sorted(fields.items())
)
name = record.get("name", "").replace("|", "\\|")
description = record.get("message", {}).get("description", "").replace("|", "\\|")
lines.append(
f"| {record['id']} | {name} | {description} | {rendered_fields} |"
)
lines.append("")
return "\n".join(lines)
def add_direct_references(rows: list[dict], source_name: str) -> None:
by_base: dict[int, list[dict]] = collections.defaultdict(list)
for row in rows:
@@ -185,8 +237,8 @@ def render_markdown(data: dict, rows: list[dict], limit: int) -> str:
f"- messages with furigana spans: {message_profile['furigana_records']}",
f"- rows shown: {len(shown)} (ranked by record coverage, then consumer references)",
"",
"| field | populated | distinct | range | direct refs | readers | common values | examples |",
"|---|---:|---:|---|---:|---|---|---|",
"| field | meaning | populated | distinct | range | direct refs | readers | common values | examples |",
"|---|---|---:|---:|---|---:|---|---|---|",
]
for row in shown:
value_range = "" if row["min"] is None else f"{row['min']}..{row['max']}"
@@ -198,7 +250,8 @@ def render_markdown(data: dict, rows: list[dict], limit: int) -> str:
for entry in row["examples"][:3]
).replace("|", "\\|")
lines.append(
f"| `{row['key']}` | {row['population']}/{data['record_count']} "
f"| `{row['key']}` | {row.get('semantic_name') or ''} | "
f"{row['population']}/{data['record_count']} "
f"({row['coverage']:.0%}) | {row['distinct_values']} | {value_range} | "
f"{row['references']} | {readers} | {common} | {examples} |"
)
@@ -211,6 +264,11 @@ def main() -> int:
parser.add_argument("table", help="extracted table name, e.g. ITINIT")
parser.add_argument("--build", action="store_true", help="write JSON and Markdown profiles")
parser.add_argument("--limit", type=int, default=40, help="Markdown/console row limit")
parser.add_argument(
"--message-query",
metavar="REGEX",
help="show matching names/player-facing messages beside all populated fields",
)
args = parser.parse_args()
name = args.table.upper().removesuffix(".JSON").removesuffix(".BIN")
@@ -231,6 +289,8 @@ def main() -> int:
)),
}
markdown = render_markdown(data, rows, args.limit)
if args.message_query:
print(render_message_matches(data, args.message_query))
print(markdown)
if args.build:
stem = paths.BUILD / "data" / f"{name}-field-profile"

View File

@@ -130,11 +130,22 @@ def test_message_join() -> None:
"INIT/MES join uses the shared runtime id")
def test_field_semantics() -> None:
scripts = paths.scripts()
items, _ = extract_init.extract_name(sys4load.load(scripts["ITINIT.BIN"]))
semantics = extract_init.field_semantics(items)
check(semantics["0x8c879"] == "item_sort_key",
"parallel INIT fields expose canonical semantic names")
check(semantics["0x9f541/14/8"] == "item_stat_modifiers.critical_chance",
"row-table columns expose canonical semantic names")
if __name__ == "__main__":
test_real_name_tables()
test_static_negative_write()
test_real_message_tables()
test_message_join()
test_field_semantics()
if FAILS:
raise SystemExit(f"{len(FAILS)} failed checks")
print("all extract_init checks passed")

View File

@@ -22,10 +22,13 @@ def test_load_and_lint():
def test_lint_catches_bad_vocab():
bad = {0x1: {"_addr": 0x1, "name": "x", "category": "bogus",
"source": "auto-shape", "confidence": "high"}}
"source": "auto-shape", "confidence": "high",
"columns": {"not-an-index": "x", "-1": ""}}}
errors, _ = G.lint(bad, {0x1})
check(any("category" in e for e in errors), "lint flags bad category")
check(any("confidence" in e for e in errors), "lint flags auto-shape claiming high confidence")
check(any("column index" in e for e in errors), "lint flags nonnumeric column indices")
check(any("negative column" in e for e in errors), "lint flags negative column indices")
def test_merge_precedence():
curated, _ = G.load_toml(paths.VM_MAP / "globals.toml")
@@ -34,6 +37,8 @@ def test_merge_precedence():
check(merged[0xa57]["name"] == "lily_form_a", "curated 0xa57 name wins over auto label")
check(merged[0xa57]["category"] == "story-flag", "curated 0xa57 category overrides auto string-table")
check(merged[0xa57]["provenance"] == "curated", "0xa57 marked curated")
check(merged[0x9f541]["columns"]["8"] == "critical_chance",
"curated row-table column semantics survive the merge")
# an address only in the auto map falls through as provenance=auto
auto_only = next((a for a in auto.get("globals", {})
if int(a, 16) not in curated and auto["globals"][a].get("label")), None)

View File

@@ -11,6 +11,11 @@ import init_table_profile as profile
def main() -> int:
fixture = {
"table": "TEST",
"field_semantics": {
"0x10": "test_parallel",
"0x30/3/0": "test_record.zero",
},
"records": [
{"id": 1, "name": "one", "fields": {"0x10": 2, "0x20": 0},
"record_fields": {"0x30/3/0": 9},
@@ -27,17 +32,25 @@ def main() -> int:
assert rows["0x10"]["distinct_values"] == 2
assert rows["0x10"]["min"] == 2 and rows["0x10"]["max"] == 5
assert rows["0x10"]["common"][0] == {"value": "2", "count": 2}
assert rows["0x10"]["semantic_name"] == "test_parallel"
assert rows["0x10"]["examples"][0]["message_description"] == "First"
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/0"]["semantic_name"] == "test_record.zero"
assert rows["0x30/3/2"]["column"] == 2
messages = profile.profile_messages(fixture)
assert messages["population"] == 1
assert messages["coverage"] == 1 / 3
assert messages["furigana_records"] == 1
assert messages["examples"][0]["description"] == "First"
matches = profile.find_message_matches(fixture, "first|three")
assert [record["id"] for record in matches] == [1, 3]
rendered = profile.render_message_matches(fixture, "First")
assert "| 1 | one | First |" in rendered
assert "`test_record.zero` (`0x30/3/0`)=9" in rendered
print("all init_table_profile checks passed")
return 0