CM95/tools/cm95_s52_presentation_audit.py

#!/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())