#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json from pathlib import Path import re import xml.etree.ElementTree as ET from typing import Any EXPECTED_COUNTS = {"SY": 1018, "AP": 30, "LC": 59} EXPECTED_TOTAL = 1107 EXPECTED_ACTIVE_TOTAL = 977 def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def _text(node: ET.Element, tag: str) -> str: child = node.find(tag) return (child.text or "").strip() if child is not None else "" def _resource_definitions(root: ET.Element) -> dict[str, set[str]]: return { "SY": {_text(n, "name") for n in root.findall("./symbols/symbol") if _text(n, "name")}, "AP": {_text(n, "name") for n in root.findall("./patterns/pattern") if _text(n, "name")}, "LC": {_text(n, "name") for n in root.findall("./line-styles/line-style") if _text(n, "name")}, } def _lookup_refs(root: ET.Element) -> dict[str, set[str]]: out = {"SY": set(), "AP": set(), "LC": set()} for node in root.findall("./lookups/lookup"): instruction = _text(node, "instruction") for opcode in out: for match in re.finditer(r"\b" + opcode + r"\(([^,;)]+)", instruction): out[opcode].add(match.group(1).strip()) return out def _active_cpp_text(path: Path) -> str: text = path.read_text(encoding="utf-8", errors="replace") lines: list[str] = [] disabled = 0 for line in text.splitlines(): stripped = line.strip() if re.match(r"#if\s+0\b", stripped): disabled += 1 continue if disabled: if re.match(r"#(?:if|ifdef|ifndef)\b", stripped): disabled += 1 elif re.match(r"#endif\b", stripped): disabled -= 1 continue lines.append(line) text = "\n".join(lines) text = re.sub(r"/\*.*?\*/", "", text, flags=re.S) return re.sub(r"//.*", "", text) def _cpp_refs(path: Path) -> dict[str, set[str]]: text = _active_cpp_text(path) out = {"SY": set(), "AP": set(), "LC": set()} for opcode in out: for match in re.finditer(r"\b" + opcode + r"\(([^,;)]+)", text): name = match.group(1).strip().strip('"') if "%" not in name: out[opcode].add(name) return out def _load_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def _built_resources(args: argparse.Namespace) -> tuple[dict[str, set[str]], dict[str, str]]: sprite = _load_json(args.sprite_index) symbols = _load_json(args.symbol_metadata) bitmap_patterns = _load_json(args.bitmap_pattern_metadata) vector_patterns = _load_json(args.vector_pattern_metadata) line_metadata = _load_json(args.line_metadata) built_sy = {name for name in symbols if name in sprite} built_ap = {name for name in bitmap_patterns if name in sprite} for name, meta in vector_patterns.items(): key = str((meta or {}).get("sprite_key") or "") if key and key in sprite: built_ap.add(name) built_lc: set[str] = set() for name, meta in line_metadata.items(): key = str((meta or {}).get("sprite_key") or "") if key and key in sprite: built_lc.add(name) hashes = { "symbol_metadata_sha256": sha256_file(args.symbol_metadata), "bitmap_pattern_metadata_sha256": sha256_file(args.bitmap_pattern_metadata), "vector_pattern_metadata_sha256": sha256_file(args.vector_pattern_metadata), "line_metadata_sha256": sha256_file(args.line_metadata), "sprite_index_sha256": sha256_file(args.sprite_index), } return {"SY": built_sy, "AP": built_ap, "LC": built_lc}, hashes def main() -> int: ap = argparse.ArgumentParser(description="Audit the frozen CM95/OpenCPN S-52 resource contract.") ap.add_argument("--chartsymbols", required=True, type=Path) ap.add_argument("--opencpn-csp-source", required=True, type=Path) ap.add_argument("--symbol-metadata", required=True, type=Path) ap.add_argument("--bitmap-pattern-metadata", required=True, type=Path) ap.add_argument("--vector-pattern-metadata", required=True, type=Path) ap.add_argument("--line-metadata", required=True, type=Path) ap.add_argument("--sprite-index", required=True, type=Path) ap.add_argument("--reference-commit", required=True) ap.add_argument("--output", required=True, type=Path) args = ap.parse_args() root = ET.parse(args.chartsymbols).getroot() defined = _resource_definitions(root) lookup = _lookup_refs(root) cpp = _cpp_refs(args.opencpn_csp_source) active = {kind: (lookup[kind] | cpp[kind]) & defined[kind] for kind in defined} active["SY"] |= { name for name in defined["SY"] if re.fullmatch(r"SOUND[GS](?:[0-5][0-9]|A1|B1|C2)", name) } built, artifact_hashes = _built_resources(args) defined_missing = {kind: sorted(defined[kind] - built[kind]) for kind in defined} active_missing = {kind: sorted(active[kind] - built[kind]) for kind in defined} unresolved = { kind: sorted((lookup[kind] | cpp[kind]) - defined[kind]) for kind in defined } counts = {kind: len(defined[kind]) for kind in defined} active_counts = {kind: len(active[kind]) for kind in active} failures: list[str] = [] if counts != EXPECTED_COUNTS or sum(counts.values()) != EXPECTED_TOTAL: failures.append("reference_resource_count_mismatch") if sum(active_counts.values()) != EXPECTED_ACTIVE_TOTAL: failures.append("reference_active_resource_count_mismatch") if any(defined_missing.values()): failures.append("defined_resource_missing_from_bundle") if any(active_missing.values()): failures.append("active_resource_missing_from_bundle") result = { "schema": "navsea.cm95.s52-resource-contract/v1", "status": "PASS" if not failures else "FAIL", "reference": { "project": "OpenCPN", "commit": args.reference_commit, "chartsymbols_sha256": sha256_file(args.chartsymbols), "csp_source_sha256": sha256_file(args.opencpn_csp_source), }, "defined_resource_counts": {**counts, "total": sum(counts.values())}, "active_resource_counts": {**active_counts, "total": sum(active_counts.values())}, "built_resource_counts": { **{kind: len(built[kind]) for kind in built}, "total": sum(len(built[kind]) for kind in built), }, "defined_missing_from_bundle": defined_missing, "active_missing_from_bundle": active_missing, "upstream_undefined_references": unresolved, "undefined_reference_fallback": { "SY": "QUESMRK1", "LC": "QUESMRK1 symbol fallback", "AP": "QUESMRK1 pattern fallback", "source": "OpenCPN S52PLIB StringToRules", }, "artifact_hashes": artifact_hashes, "failures": failures, } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if not failures else 2 if __name__ == "__main__": raise SystemExit(main())