Close EventEffects semantic descriptor frontier

tools/py/build_event_effect_semantic_catalog.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+def normalize_world_scalar_key(label: str) -> str:
+    parts: list[str] = []
+    current = []
+    for ch in label:
+        if ch.isalnum():
+            current.append(ch.lower())
+        elif current:
+            parts.append("".join(current))
+            current = []
+    if current:
+        parts.append("".join(current))
+    return "world." + "_".join(parts)
+
+
+def classify(row: dict[str, object]) -> dict[str, object]:
+    descriptor_id = int(row["descriptor_id"])
+    label = str(row["label"])
+    signature_byte_0x63 = int(row["signature_byte_0x63"])
+    signature_byte_0x64 = int(row["signature_byte_0x64"])
+
+    parameter_family = "recovered_effect_table_descriptor"
+    runtime_key = None
+    runtime_status = "evidence_blocked"
+    executable_in_runtime = False
+
+    if 4 <= descriptor_id <= 7:
+        parameter_family = "scenario_outcome_shell_action"
+        runtime_status = "shell_owned"
+    elif descriptor_id in {10, 11, 12, 23}:
+        parameter_family = "company_confiscation_variant"
+        runtime_status = "variant_or_scope_blocked"
+    elif 17 <= descriptor_id <= 21:
+        parameter_family = "company_or_territory_destruction_variant"
+        runtime_status = "shell_owned"
+    elif descriptor_id == 24:
+        parameter_family = "control_transfer_shell_action"
+        runtime_status = "shell_owned"
+    elif descriptor_id in {55, 58}:
+        parameter_family = "company_finance_shell_scalar"
+        runtime_status = "shell_owned"
+    elif descriptor_id in {56, 57}:
+        parameter_family = "company_governance_scalar"
+        runtime_status = "executable"
+        executable_in_runtime = True
+    elif 59 <= descriptor_id <= 104:
+        parameter_family = "world_scalar_override"
+        runtime_key = normalize_world_scalar_key(label)
+        runtime_status = "executable"
+        executable_in_runtime = True
+    elif 105 <= descriptor_id <= 176:
+        parameter_family = "cargo_price_scalar"
+    elif 177 <= descriptor_id <= 240:
+        parameter_family = "cargo_production_scalar"
+    elif 241 <= descriptor_id <= 351 or 457 <= descriptor_id <= 474:
+        parameter_family = "locomotive_availability_scalar"
+    elif 352 <= descriptor_id <= 451 or 475 <= descriptor_id <= 502:
+        parameter_family = "locomotive_cost_scalar"
+    elif descriptor_id == 453:
+        parameter_family = "territory_access_cost_scalar"
+        runtime_status = "executable"
+        executable_in_runtime = True
+    elif descriptor_id == 454:
+        parameter_family = "world_flag_toggle"
+        runtime_key = "world.all_steam_locos_available"
+        runtime_status = "executable"
+        executable_in_runtime = True
+    elif descriptor_id == 455:
+        parameter_family = "world_flag_toggle"
+        runtime_key = "world.all_diesel_locos_available"
+        runtime_status = "executable"
+        executable_in_runtime = True
+    elif descriptor_id == 456:
+        parameter_family = "world_flag_toggle"
+        runtime_key = "world.all_electric_locos_available"
+        runtime_status = "executable"
+        executable_in_runtime = True
+    elif 503 <= descriptor_id <= 519:
+        parameter_family = "world_building_spawn"
+        runtime_status = "shell_owned"
+    elif signature_byte_0x63 == 0 and signature_byte_0x64 == 0x8F:
+        parameter_family = "runtime_variable_scalar"
+    elif "Earthquake" in label or "Storm" in label:
+        parameter_family = "world_disaster_scalar"
+
+    return {
+        "descriptor_id": descriptor_id,
+        "label": label,
+        "target_mask_bits": int(row["target_mask_bits"]),
+        "parameter_family": parameter_family,
+        "runtime_key": runtime_key,
+        "runtime_status": runtime_status,
+        "executable_in_runtime": executable_in_runtime,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Build a checked-in semantic catalog over the raw RT3 EventEffects table."
+    )
+    parser.add_argument("raw_table", type=Path)
+    parser.add_argument("out", type=Path)
+    args = parser.parse_args()
+
+    raw_artifact = json.loads(args.raw_table.read_text(encoding="utf-8"))
+    descriptors = [classify(row) for row in raw_artifact["descriptors"]]
+    artifact = {
+        "descriptor_count": len(descriptors),
+        "raw_table_binary_sha256": raw_artifact.get("binary_sha256"),
+        "semantic_catalog_version": 1,
+        "descriptors": descriptors,
+    }
+    args.out.write_text(json.dumps(artifact, indent=2) + "\n", encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()