feat: add S-52 presentation rule engine

tools/cm95_s52_presentation_audit.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+from collections import Counter, defaultdict
+import json
+from pathlib import Path
+import re
+import xml.etree.ElementTree as ET
+
+from cm95pbf.presentation import (
+    compile_lookup_ir,
+    parse_instruction_string,
+    referenced_resource,
+)
+
+
+def _text(node: ET.Element, tag: str) -> str:
+    child = node.find(tag)
+    return (child.text or "").strip() if child is not None else ""
+
+
+def _lookup_rows(root: ET.Element) -> list[dict]:
+    out = []
+    parent = root.find("lookups")
+    if parent is None:
+        return out
+    for node in parent.findall("lookup"):
+        out.append({
+            "id": node.attrib.get("id"),
+            "rcid": node.attrib.get("RCID"),
+            "object_class": node.attrib.get("name", ""),
+            "geometry_type": _text(node, "type"),
+            "display_priority": _text(node, "disp-prio"),
+            "radar_priority": _text(node, "radar-prio"),
+            "table_name": _text(node, "table-name"),
+            "display_category": _text(node, "display-cat"),
+            "attribute_conditions_raw": [
+                (x.text or "").strip()
+                for x in node.findall("attrib-code")
+                if (x.text or "").strip()
+            ],
+            "instruction": _text(node, "instruction"),
+        })
+    return out
+
+
+def _resource_names(root: ET.Element) -> dict[str, set[str]]:
+    symbols = {
+        _text(node, "name")
+        for node in root.find("symbols").findall("symbol")
+        if _text(node, "name")
+    } if root.find("symbols") is not None else set()
+    lines = {
+        _text(node, "name")
+        for node in root.find("line-styles").findall("line-style")
+        if _text(node, "name")
+    } if root.find("line-styles") is not None else set()
+    patterns = {
+        _text(node, "name")
+        for node in root.find("patterns").findall("pattern")
+        if _text(node, "name")
+    } if root.find("patterns") is not None else set()
+    colors: set[str] = set()
+    color_tables = root.find("color-tables")
+    if color_tables is not None:
+        for table in color_tables.findall("color-table"):
+            for node in table.findall("color"):
+                name = node.attrib.get("name", "").strip()
+                if name:
+                    colors.add(name)
+    return {
+        "symbol": symbols,
+        "line_style": lines,
+        "pattern": patterns,
+        "color": colors,
+    }
+
+
+def _opencpn_csp_names(path: Path | None) -> set[str]:
+    if path is None or not path.is_file():
+        return set()
+    text = path.read_text(encoding="utf-8", errors="replace")
+    block_match = re.search(
+        r"Cond\s+condTable\[\]\s*=\s*\{(?P<body>.*?)\{\"########\"",
+        text,
+        re.S,
+    )
+    if not block_match:
+        return set()
+    return set(re.findall(r'\{\"([A-Z0-9#]{8})\"\s*,', block_match.group("body")))
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Audit S-52 lookup -> Presentation Rule IR coverage.")
+    parser.add_argument("--chartsymbols", required=True, type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    parser.add_argument("--opencpn-csp-source", type=Path)
+    args = parser.parse_args()
+    root = ET.parse(args.chartsymbols).getroot()
+    rows = _lookup_rows(root)
+    resources = _resource_names(root)
+    csp_names = _opencpn_csp_names(args.opencpn_csp_source)
+
+    compiled: list[dict] = []
+    failures: list[dict] = []
+    opcode_counts: Counter[str] = Counter()
+    references: dict[str, Counter[str]] = defaultdict(Counter)
+
+    for row in rows:
+        try:
+            ir = compile_lookup_ir(row)
+            compiled.append(ir)
+            for rule in parse_instruction_string(row["instruction"]):
+                opcode_counts[rule.opcode] += 1
+                ref = referenced_resource(rule)
+                if ref:
+                    references[ref[0]][ref[1]] += 1
+        except Exception as exc:
+            failures.append({
+                "object_class": row.get("object_class"),
+                "rcid": row.get("rcid"),
+                "instruction": row.get("instruction"),
+                "error": f"{type(exc).__name__}: {exc}",
+            })
+
+    missing: dict[str, list[str]] = {}
+    for kind in ("symbol", "line_style", "pattern", "color"):
+        missing[kind] = sorted(set(references[kind]) - resources[kind])
+    if csp_names:
+        missing["conditional"] = sorted(set(references["conditional"]) - csp_names)
+    else:
+        missing["conditional"] = []
+    generated_examples = {
+        "CA": ";CA(OUTLW,4,LITRD,2,180.0,220.0,10.0,25.0)",
+        "MP": "MP()",
+    }
+    generated_parse = {
+        name: [rule.to_dict() for rule in parse_instruction_string(text)]
+        for name, text in generated_examples.items()
+    }
+
+    report = {
+        "schema": "navsea.cm95.s52-presentation-audit/v1",
+        "lookup_count": len(rows),
+        "compiled_lookup_count": len(compiled),
+        "parse_failure_count": len(failures),
+        "opcode_counts": dict(sorted(opcode_counts.items())),
+        "resource_reference_counts": {
+            kind: len(counter) for kind, counter in sorted(references.items())
+        },
+        "missing_resources": missing,
+        "opencpn_csp_registry_count": len(csp_names),
+        "generated_opcode_parse": generated_parse,
+    }
+
+    args.output.mkdir(parents=True, exist_ok=True)
+    (args.output / "lookup_rule_ir.json").write_text(
+        json.dumps(compiled, ensure_ascii=False, indent=2), encoding="utf-8"
+    )
+    (args.output / "audit.json").write_text(
+        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
+    )
+    (args.output / "parse_failures.json").write_text(
+        json.dumps(failures, ensure_ascii=False, indent=2), encoding="utf-8"
+    )
+    print(json.dumps(report, ensure_ascii=False, indent=2))
+    return 0 if not failures else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())