Join item and skill message semantics
This commit is contained in:
@@ -26,6 +26,7 @@ from pathlib import Path
|
||||
HERE = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(HERE))
|
||||
import paths
|
||||
import extract_message_table
|
||||
import sys4load
|
||||
|
||||
SET_STRING = 0x192
|
||||
@@ -37,6 +38,11 @@ T_GLOBAL_INT = 3
|
||||
T_GLOBAL_STRING = 5
|
||||
T_IMM = 0
|
||||
|
||||
MESSAGE_TABLES = {
|
||||
"ITINIT": "ITMES",
|
||||
"SKINIT": "SKMES",
|
||||
}
|
||||
|
||||
|
||||
def resolve(name: str) -> Path:
|
||||
for cand in (paths.GAME_DIR / f"{name}.BIN", paths.DATA1 / f"{name}.BIN"):
|
||||
@@ -252,6 +258,28 @@ def extract_footer(scr):
|
||||
return records, {}
|
||||
|
||||
|
||||
def join_messages(records: list[dict], message_scr) -> dict:
|
||||
"""Join a message-dispatch script to INIT records by runtime id."""
|
||||
messages, message_meta = extract_message_table.extract_messages(message_scr)
|
||||
by_id = {message["id"]: message for message in messages}
|
||||
joined = 0
|
||||
for record in records:
|
||||
if message := by_id.get(record["id"]):
|
||||
record["message"] = {
|
||||
key: value for key, value in message.items() if key != "id"
|
||||
}
|
||||
joined += 1
|
||||
init_ids = {record["id"] for record in records}
|
||||
message_ids = set(by_id)
|
||||
return {
|
||||
"source": message_scr.path.name,
|
||||
**message_meta,
|
||||
"joined_count": joined,
|
||||
"init_ids_without_message": sorted(init_ids - message_ids),
|
||||
"message_ids_without_init": sorted(message_ids - init_ids),
|
||||
}
|
||||
|
||||
|
||||
def write_data_index(data_dir: Path) -> None:
|
||||
"""Regenerate the disposable build/data index from current table JSONs."""
|
||||
tables = []
|
||||
@@ -272,15 +300,18 @@ 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 | array fields | record columns |",
|
||||
"|---|---|---:|---:|---:|",
|
||||
"| file | mode | records | messages | array fields | record columns |",
|
||||
"|---|---|---:|---:|---:|---:|",
|
||||
]
|
||||
for filename, data in tables:
|
||||
columns = len(data.get("field_columns") or [])
|
||||
record_columns = len(data.get("record_field_columns") or [])
|
||||
message_count = data.get("message_table", {}).get(
|
||||
"joined_count", data.get("message_count", 0)
|
||||
)
|
||||
lines.append(
|
||||
f"| `{filename}` | {data['mode']} | {data['record_count']} | "
|
||||
f"{columns} | {record_columns} |"
|
||||
f"{message_count} | {columns} | {record_columns} |"
|
||||
)
|
||||
lines += [
|
||||
"",
|
||||
@@ -289,6 +320,9 @@ def write_data_index(data_dir: Path) -> None:
|
||||
"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.",
|
||||
"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.",
|
||||
"",
|
||||
"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.",
|
||||
@@ -309,6 +343,11 @@ def main() -> int:
|
||||
mode = mode_arg or detect_mode(scr)
|
||||
extractor = {"name": extract_name, "numeric": extract_numeric, "footer": extract_footer}[mode]
|
||||
recs, meta = extractor(scr)
|
||||
if mode == "name" and name in MESSAGE_TABLES:
|
||||
message_name = MESSAGE_TABLES[name]
|
||||
meta["message_table"] = join_messages(
|
||||
recs, sys4load.load(extract_message_table.resolve(message_name))
|
||||
)
|
||||
|
||||
cols = sorted({c for r in recs for c in r.get("fields", {})}, key=lambda h: int(h, 16))
|
||||
out = {"table": name, "source": scr.path.name, "magic": scr.magic, "mode": mode,
|
||||
|
||||
175
tools/extract_message_table.py
Normal file
175
tools/extract_message_table.py
Normal file
@@ -0,0 +1,175 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Extract an ID-dispatched SYS4 message table.
|
||||
|
||||
ITMES and SKMES are long chains of:
|
||||
|
||||
eq <temporary>, <selected id global>, <record id>
|
||||
jcc ...
|
||||
show-text ...
|
||||
...
|
||||
jmp <shared exit>
|
||||
|
||||
This tool discovers the selector global from that repeated shape, reconstructs
|
||||
the displayed lines (including the surface text of furigana spans), and emits a
|
||||
reusable ID-to-message JSON table.
|
||||
|
||||
Usage:
|
||||
py -3.11 -X utf8 tools/extract_message_table.py ITMES
|
||||
py -3.11 -X utf8 tools/extract_message_table.py SKMES [OUTNAME]
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import collections
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(HERE))
|
||||
import paths
|
||||
import sys4load
|
||||
|
||||
|
||||
T_IMM = 0
|
||||
T_INLINE_STRING = 2
|
||||
T_GLOBAL_INT = 3
|
||||
|
||||
|
||||
def resolve(name: str) -> Path:
|
||||
normalized = name.upper().removesuffix(".BIN")
|
||||
scripts = paths.scripts()
|
||||
try:
|
||||
return scripts[f"{normalized}.BIN"]
|
||||
except KeyError:
|
||||
raise SystemExit(f"not found: {normalized}.BIN") from None
|
||||
|
||||
|
||||
def _dispatch_comparison(ins):
|
||||
"""Return (selector global, record id) for an eq global/int-immediate guard."""
|
||||
if sys4load.display_label(ins.opcode) != "eq" or len(ins.args) != 3:
|
||||
return None
|
||||
left, right = ins.args[1:]
|
||||
if left[0] == T_GLOBAL_INT and right[0] == T_IMM:
|
||||
return left[1], right[1]
|
||||
if right[0] == T_GLOBAL_INT and left[0] == T_IMM:
|
||||
return right[1], left[1]
|
||||
return None
|
||||
|
||||
|
||||
def discover_selector(scr) -> tuple[int, int]:
|
||||
"""Return the most frequently ID-compared global and its guard count."""
|
||||
counts = collections.Counter(
|
||||
selector
|
||||
for ins in scr.instructions
|
||||
if (pair := _dispatch_comparison(ins)) is not None
|
||||
for selector in [pair[0]]
|
||||
)
|
||||
if not counts:
|
||||
raise ValueError(f"{scr.path.name}: no global/immediate eq dispatch guards")
|
||||
return counts.most_common(1)[0]
|
||||
|
||||
|
||||
def _strings(ins, scr) -> list[str]:
|
||||
return [
|
||||
scr.strings[value][0]
|
||||
for arg_type, value in ins.args
|
||||
if arg_type == T_INLINE_STRING and value in scr.strings
|
||||
]
|
||||
|
||||
|
||||
def _message_body(scr, start: int, stop: int) -> dict | None:
|
||||
fragments: list[str] = []
|
||||
lines: list[str] = []
|
||||
furigana: list[dict] = []
|
||||
for ins in scr.instructions[start:stop]:
|
||||
operation = sys4load.display_label(ins.opcode)
|
||||
if operation == "jmp":
|
||||
break
|
||||
values = _strings(ins, scr)
|
||||
if operation == "show-text":
|
||||
fragments.extend(values)
|
||||
elif operation == "display-furigana" and values:
|
||||
surface = values[0]
|
||||
fragments.append(surface)
|
||||
annotation = {"line": len(lines), "text": surface}
|
||||
if len(values) > 1:
|
||||
annotation["reading"] = values[1]
|
||||
furigana.append(annotation)
|
||||
elif operation == "end-text-line":
|
||||
lines.append("".join(fragments))
|
||||
fragments = []
|
||||
if fragments:
|
||||
lines.append("".join(fragments))
|
||||
if not lines:
|
||||
return None
|
||||
message = {
|
||||
"title": lines[0],
|
||||
"description": "\n".join(lines[1:]),
|
||||
}
|
||||
if furigana:
|
||||
message["furigana"] = furigana
|
||||
return message
|
||||
|
||||
|
||||
def extract_messages(scr, selector: int | None = None) -> tuple[list[dict], dict]:
|
||||
"""Extract ordered message records and dispatch metadata from a script."""
|
||||
discovered_selector, comparison_count = discover_selector(scr)
|
||||
selector = discovered_selector if selector is None else selector
|
||||
guards = [
|
||||
(index, pair[1])
|
||||
for index, ins in enumerate(scr.instructions)
|
||||
if (pair := _dispatch_comparison(ins)) is not None and pair[0] == selector
|
||||
]
|
||||
records: list[dict] = []
|
||||
seen: set[int] = set()
|
||||
for guard_index, (instruction_index, record_id) in enumerate(guards):
|
||||
stop = guards[guard_index + 1][0] if guard_index + 1 < len(guards) else len(scr.instructions)
|
||||
message = _message_body(scr, instruction_index + 1, stop)
|
||||
if message is None:
|
||||
continue
|
||||
if record_id in seen:
|
||||
raise ValueError(f"{scr.path.name}: duplicate message id {record_id}")
|
||||
seen.add(record_id)
|
||||
records.append({
|
||||
"id": record_id,
|
||||
"dispatch_offset": f"0x{scr.instructions[instruction_index].offset:x}",
|
||||
**message,
|
||||
})
|
||||
return records, {
|
||||
"selector_global": f"0x{selector:x}",
|
||||
"dispatch_guard_count": len(guards),
|
||||
"message_count": len(records),
|
||||
"selector_discovery_count": comparison_count,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) < 2:
|
||||
raise SystemExit(__doc__)
|
||||
name = sys.argv[1].upper().removesuffix(".BIN")
|
||||
outname = sys.argv[2] if len(sys.argv) > 2 else name
|
||||
scr = sys4load.load(resolve(name))
|
||||
records, meta = extract_messages(scr)
|
||||
output = {
|
||||
"table": name,
|
||||
"source": scr.path.name,
|
||||
"magic": scr.magic,
|
||||
"mode": "message-dispatch",
|
||||
"record_count": len(records),
|
||||
**meta,
|
||||
"records": records,
|
||||
}
|
||||
outpath = paths.BUILD / "data" / f"{outname}.json"
|
||||
outpath.parent.mkdir(parents=True, exist_ok=True)
|
||||
outpath.write_text(
|
||||
json.dumps(output, ensure_ascii=False, indent=2), encoding="utf8"
|
||||
)
|
||||
print(
|
||||
f"{name}: selector={meta['selector_global']}, {len(records)} messages "
|
||||
f"-> build/data/{outname}.json"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -108,6 +108,29 @@ def profile_columns(data: dict) -> list[dict]:
|
||||
return rows
|
||||
|
||||
|
||||
def profile_messages(data: dict) -> dict:
|
||||
"""Summarize the joined player-facing message evidence."""
|
||||
records = data["records"]
|
||||
with_message = [record for record in records if "message" in record]
|
||||
with_furigana = [
|
||||
record for record in with_message if record["message"].get("furigana")
|
||||
]
|
||||
return {
|
||||
"population": len(with_message),
|
||||
"coverage": len(with_message) / len(records) if records else 0.0,
|
||||
"furigana_records": len(with_furigana),
|
||||
"examples": [
|
||||
{
|
||||
"id": record["id"],
|
||||
"name": record.get("name", ""),
|
||||
"title": record["message"]["title"],
|
||||
"description": record["message"]["description"],
|
||||
}
|
||||
for record in with_message[:5]
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def add_direct_references(rows: list[dict], source_name: str) -> None:
|
||||
by_base: dict[int, list[dict]] = collections.defaultdict(list)
|
||||
for row in rows:
|
||||
@@ -148,6 +171,7 @@ def add_direct_references(rows: list[dict], source_name: str) -> None:
|
||||
def render_markdown(data: dict, rows: list[dict], limit: int) -> str:
|
||||
ranked = sorted(rows, key=lambda row: (-row["population"], -row["references"], row["key"]))
|
||||
shown = ranked[:limit]
|
||||
message_profile = profile_messages(data)
|
||||
lines = [
|
||||
f"# {data['table']} field profile",
|
||||
"",
|
||||
@@ -156,6 +180,9 @@ def render_markdown(data: dict, rows: list[dict], limit: int) -> str:
|
||||
"",
|
||||
f"- records: {data['record_count']}",
|
||||
f"- populated fields: {len(rows)}",
|
||||
f"- player-facing messages: {message_profile['population']}/{data['record_count']} "
|
||||
f"({message_profile['coverage']:.0%})",
|
||||
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 |",
|
||||
@@ -189,6 +216,7 @@ def main() -> int:
|
||||
name = args.table.upper().removesuffix(".JSON").removesuffix(".BIN")
|
||||
data = load_table(name)
|
||||
rows = profile_columns(data)
|
||||
messages = profile_messages(data)
|
||||
add_direct_references(rows, data["source"])
|
||||
output = {
|
||||
"table": data["table"],
|
||||
@@ -197,6 +225,7 @@ def main() -> int:
|
||||
"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),
|
||||
"message_profile": messages,
|
||||
"columns": sorted(rows, key=lambda row: (
|
||||
int(row["base"], 16), row["stride"] or 0, row["column"] or 0
|
||||
)),
|
||||
|
||||
@@ -10,6 +10,7 @@ from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
import extract_init
|
||||
import extract_message_table
|
||||
import paths
|
||||
import sys4load
|
||||
|
||||
@@ -65,9 +66,75 @@ def test_static_negative_write() -> None:
|
||||
"INIT subtraction writes preserve negative values")
|
||||
|
||||
|
||||
def test_real_message_tables() -> None:
|
||||
scripts = paths.scripts()
|
||||
expected = {
|
||||
"ITMES.BIN": (0x8C877, 287),
|
||||
"SKMES.BIN": (0xA6E59, 131),
|
||||
}
|
||||
for name, (selector, count) in expected.items():
|
||||
records, meta = extract_message_table.extract_messages(
|
||||
sys4load.load(scripts[name])
|
||||
)
|
||||
check(meta["selector_global"] == f"0x{selector:x}",
|
||||
f"{name}: discovers selector global 0x{selector:x}")
|
||||
check(len(records) == count, f"{name}: extracts {count} messages")
|
||||
check(len(records) == meta["dispatch_guard_count"],
|
||||
f"{name}: every dispatch guard yields a message")
|
||||
check(len({record['id'] for record in records}) == len(records),
|
||||
f"{name}: message ids are unique")
|
||||
|
||||
item_messages, _ = extract_message_table.extract_messages(
|
||||
sys4load.load(scripts["ITMES.BIN"])
|
||||
)
|
||||
items = {record["id"]: record for record in item_messages}
|
||||
check(items[1]["title"] == "【重要:銅の鍵】 LEVEL-E",
|
||||
"ITMES item 1 keeps its display title")
|
||||
check(items[1]["description"] == " 銅の扉を開閉することが可能",
|
||||
"ITMES item 1 keeps its player-facing behavior")
|
||||
check("濃緑色" in items[32]["title"],
|
||||
"ITMES reconstructs furigana surface text inside a title")
|
||||
check(items[32]["furigana"][0]["reading"] == "のうりょくしょく",
|
||||
"ITMES preserves furigana readings")
|
||||
|
||||
skill_messages, _ = extract_message_table.extract_messages(
|
||||
sys4load.load(scripts["SKMES.BIN"])
|
||||
)
|
||||
skills = {record["id"]: record for record in skill_messages}
|
||||
check(skills[1]["title"] == "【移動スキル:飛行】",
|
||||
"SKMES skill 1 keeps its display title")
|
||||
check(skills[1]["description"] == " 床のない地形を移動可能になる",
|
||||
"SKMES skill 1 keeps its player-facing behavior")
|
||||
|
||||
|
||||
def test_message_join() -> None:
|
||||
scripts = paths.scripts()
|
||||
expected = {
|
||||
"IT": (287, " 銅の扉を開閉することが可能"),
|
||||
"SK": (131, " 床のない地形を移動可能になる"),
|
||||
}
|
||||
joined = {}
|
||||
for prefix, (count, _) in expected.items():
|
||||
records, _ = extract_init.extract_name(
|
||||
sys4load.load(scripts[f"{prefix}INIT.BIN"])
|
||||
)
|
||||
meta = extract_init.join_messages(
|
||||
records, sys4load.load(scripts[f"{prefix}MES.BIN"])
|
||||
)
|
||||
check(meta["joined_count"] == count,
|
||||
f"{prefix}INIT joins all {count} {prefix}MES messages")
|
||||
check(not meta["init_ids_without_message"] and not meta["message_ids_without_init"],
|
||||
f"{prefix}INIT and {prefix}MES ids match exactly")
|
||||
joined[prefix] = {record["id"]: record for record in records}
|
||||
check(joined["IT"][1]["message"]["description"] == expected["IT"][1],
|
||||
"INIT/MES join uses the shared runtime id")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_real_name_tables()
|
||||
test_static_negative_write()
|
||||
test_real_message_tables()
|
||||
test_message_join()
|
||||
if FAILS:
|
||||
raise SystemExit(f"{len(FAILS)} failed checks")
|
||||
print("all extract_init checks passed")
|
||||
|
||||
@@ -13,7 +13,9 @@ def main() -> int:
|
||||
fixture = {
|
||||
"records": [
|
||||
{"id": 1, "name": "one", "fields": {"0x10": 2, "0x20": 0},
|
||||
"record_fields": {"0x30/3/0": 9}},
|
||||
"record_fields": {"0x30/3/0": 9},
|
||||
"message": {"title": "One", "description": "First",
|
||||
"furigana": [{"line": 0, "text": "One", "reading": "one"}]}},
|
||||
{"id": 3, "name": "three", "fields": {"0x10": 2}},
|
||||
{"id": 7, "name": "seven", "fields": {"0x10": 5},
|
||||
"record_fields": {"0x30/3/2": 4}},
|
||||
@@ -31,6 +33,11 @@ def main() -> int:
|
||||
assert rows["0x30/3/0"]["base"] == "0x30"
|
||||
assert rows["0x30/3/0"]["stride"] == 3
|
||||
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"
|
||||
print("all init_table_profile checks passed")
|
||||
return 0
|
||||
|
||||
|
||||
Reference in New Issue
Block a user