From bbe006b3d65a8c8d461fc31ec55a1ce9d421d1fb Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Thu, 17 Sep 2026 07:43:22 +0800 Subject: [PATCH] feat: add S-52 presentation rule engine --- .../CM95_S52_Presentation引擎阶段报告_v1.md | 50 +++++ src/cm95pbf/builder.py | 41 +++- src/cm95pbf/cli.py | 12 ++ src/cm95pbf/csp.py | 134 +++++++++++++ src/cm95pbf/mapping.py | 10 + src/cm95pbf/presentation.py | 186 ++++++++++++++++++ src/cm95pbf/s52.py | 59 +++++- tools/cm95_s52_presentation_audit.py | 172 ++++++++++++++++ 8 files changed, 645 insertions(+), 19 deletions(-) create mode 100644 docs/报告/CM95_S52_Presentation引擎阶段报告_v1.md create mode 100644 src/cm95pbf/csp.py create mode 100644 src/cm95pbf/presentation.py create mode 100644 tools/cm95_s52_presentation_audit.py diff --git a/docs/报告/CM95_S52_Presentation引擎阶段报告_v1.md b/docs/报告/CM95_S52_Presentation引擎阶段报告_v1.md new file mode 100644 index 0000000..190aed3 --- /dev/null +++ b/docs/报告/CM95_S52_Presentation引擎阶段报告_v1.md @@ -0,0 +1,50 @@ +# CM95 S-52 Presentation 引擎阶段报告 v1 + +日期：2026-09-17 + +## 本阶段完成 + +- 建立 `S-52 INST → Presentation Rule IR` 通用解释器。 +- 按 OpenCPN `StringToRules/SCANFWRD` 容错语义解析历史 Presentation Library 数据。 +- 支持 `SY/LS/LC/AC/AP/TX/TE/CS`，并预留 CSP 生成的 `CA/MP`。 +- OpenCPN 当前 `chartsymbols.xml` 3057 条 Lookup 全量解析：3057/3057，失败 0。 +- S52 Lookup 从 Point 扩展到 Point/Line/Area：Paper/Lines/Plain 或 Symbolized。 +- 新增通用诊断 fallback layer：`s52_points/s52_lines/s52_areas`。 + +## 已落地 CSP + +### DEPARE01 / DEPARE02 + +按 OpenCPN 行为生成区域显示规则；福冈湾 z12/3531/1641 实测： + +- DEPARE/DRGARE：172 个。 +- DEPVS：47。 +- DEPMS：68。 +- DEPMD：33。 +- DEPDW：24。 +- DRGARE 4 个同时得到 `AP(DRGARE01)` 与虚线边界。 + +### DEPCNT02 + +配置默认 safety contour=3m；数据不存在精确 3m contour 时，自动选择下一个更深的实际 contour。 +福冈湾实测： + +- DEPCNT：188 条。 +- 5m 被选择为实际 safety contour。 +- DEPSC（安全等深线，宽 2）：91 条。 +- DEPCN（普通等深线，宽 1）：97 条。 + +## 福冈湾完整 S-52 覆盖 + +同一 z12 tile 共 3293 个输出 Feature，其中 1264 个已经取得 S-52 Lookup/Presentation 属性。 +新增通用 fallback 中 `s52_points=42`、`s52_areas=4`；原有业务 layer 继续保留，不重复制造已有对象。 + +浏览器验证使用同一 PBF，已确认四级深度色、安全等深线、测深数字、底质缩写、地名和更多 S-52 航标实际渲染。 + +## 尚未完成 + +- LIGHTS05/LIGHTS06 的 `CA` 灯弧最终 renderer。 +- SOUNDG02/SOUNDG03 的完整 S-52 sounding symbol 组合。 +- SBDARE 底质长文本/组合格式。 +- OBSTRN/WRECKS/UDWHAZ、RESARE/RESTRN、SLCONS/QUAPOS 等 CSP。 +- 正式 `style.json` compiler；当前 Web 页面仍是诊断 Style。 diff --git a/src/cm95pbf/builder.py b/src/cm95pbf/builder.py index 5d89d73..f919187 100644 --- a/src/cm95pbf/builder.py +++ b/src/cm95pbf/builder.py @@ -272,6 +272,7 @@ class Cm95TileBuilder: profile: str = "delivery", s52_lookups_path: Path | None = None, s52_point_table: str = "Paper", + s52_boundary_table: str = "Plain", ): self.db = db self.dataset_id = int(dataset_id) @@ -282,11 +283,17 @@ class Cm95TileBuilder: self.report_root = Path(report_root) self.profile = str(profile) self.s52 = ( - S52LookupRegistry(Path(s52_lookups_path), point_table=s52_point_table) + S52LookupRegistry( + Path(s52_lookups_path), + point_table=s52_point_table, + boundary_table=s52_boundary_table, + ) if s52_lookups_path is not None else None ) self.s52_point_table = str(s52_point_table) + self.s52_boundary_table = str(s52_boundary_table) + self._s52_context: dict[str, Any] = {} if self.profile not in {"delivery", "engineering"}: raise ValueError("profile must be delivery or engineering") @@ -432,6 +439,7 @@ class Cm95TileBuilder: consumed: set[int] = set() s52_context = self._build_s52_context(selected) + self._s52_context = s52_context nav_outputs = self._compose_navigation_aids( selected, manifest, @@ -1696,10 +1704,21 @@ class Cm95TileBuilder: "conflict": name_conflict, } - def _build_s52_context(self, rows: list[SourceFeature]) -> dict[str, set[Any]]: + def _build_s52_context(self, rows: list[SourceFeature]) -> dict[str, Any]: floating_relations: set[Any] = set() floating_points: set[Any] = set() + contour_values: list[float] = [] + safety_contour = ( + self.s52.settings.safety_contour_m if self.s52 is not None else None + ) for row in rows: + if row.object_class == "DEPCNT" and safety_contour is not None: + try: + value = float(row.attrs.get("VALDCO")) + except (TypeError, ValueError): + value = -1.0 + if value >= safety_contour: + contour_values.append(value) if row.geometry_type != "POINT": continue if not (row.object_class.startswith("BOY") or row.object_class in {"LITFLT", "LITVES"}): @@ -1713,10 +1732,11 @@ class Cm95TileBuilder: return { "floating_relations": floating_relations, "floating_points": floating_points, + "next_safe_contour_m": min(contour_values) if contour_values else None, } @staticmethod - def _topmar_is_floating(row: SourceFeature, context: dict[str, set[Any]]) -> bool: + def _topmar_is_floating(row: SourceFeature, context: dict[str, Any]) -> bool: relation = row.attrs.get("_cm93_relation20_raw") if relation is not None and (row.cell_id, str(relation)) in context["floating_relations"]: return True @@ -1732,6 +1752,7 @@ class Cm95TileBuilder: row.object_class, row.geometry_type, row.attrs, + next_safe_contour_m=self._s52_context.get("next_safe_contour_m"), ) def _append_s52_fallback_output( @@ -1740,11 +1761,9 @@ class Cm95TileBuilder: manifest, row: SourceFeature, trace_rows, - s52_context: dict[str, set[Any]], + s52_context: dict[str, Any], ) -> bool: - if self.s52 is None or row.geometry_type != "POINT": - return False - if row.object_class not in {"BOYLAT", "TOPMAR"}: + if self.s52 is None or row.geometry_type not in {"POINT", "LINE", "AREA"}: return False s52_props = self._s52_properties(row) @@ -1757,12 +1776,16 @@ class Cm95TileBuilder: s52_props["s52_csp_resolved"] = "TOPMAR01" s52_props.pop("s52_csp", None) resolved_csp = "TOPMAR01" - if not s52_props.get("s52_symbol_code") and not s52_props.get("s52_csp"): + if not s52_props.get("s52_lookup_rcid"): return False source_id = self.source_id(row) fid = manifest.mvt_id(source_id, ROLE_PRIMARY) - target = "navigation_marks" + target = { + "POINT": "s52_points", + "LINE": "s52_lines", + "AREA": "s52_areas", + }[row.geometry_type] props = { "canonical_object_type": f"s52_{row.object_class.lower()}", **s52_props, diff --git a/src/cm95pbf/cli.py b/src/cm95pbf/cli.py index 7552b0f..d3cb016 100644 --- a/src/cm95pbf/cli.py +++ b/src/cm95pbf/cli.py @@ -63,6 +63,11 @@ def build_parser() -> argparse.ArgumentParser: choices=["Simplified", "Paper"], default="Paper", ) + tile.add_argument( + "--s52-boundary-table", + choices=["Plain", "Symbolized"], + default="Plain", + ) sample = sub.add_parser( "build-sample-aoi", @@ -87,6 +92,11 @@ def build_parser() -> argparse.ArgumentParser: choices=["Simplified", "Paper"], default="Paper", ) + sample.add_argument( + "--s52-boundary-table", + choices=["Plain", "Symbolized"], + default="Plain", + ) prod_plan = sub.add_parser( @@ -310,6 +320,7 @@ def main() -> None: profile=args.profile, s52_lookups_path=(Path(args.s52_lookups) if args.s52_lookups else None), s52_point_table=args.s52_point_table, + s52_boundary_table=args.s52_boundary_table, ) result = builder.build_tile( TileJob(args.z, args.x, args.y) @@ -344,6 +355,7 @@ def main() -> None: profile=profile, s52_lookups_path=(Path(args.s52_lookups) if args.s52_lookups else None), s52_point_table=args.s52_point_table, + s52_boundary_table=args.s52_boundary_table, ) for z, x, y in jobs: result = builder.build_tile(TileJob(z, x, y)) diff --git a/src/cm95pbf/csp.py b/src/cm95pbf/csp.py new file mode 100644 index 0000000..4c5a773 --- /dev/null +++ b/src/cm95pbf/csp.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .presentation import parse_instruction_string, static_rule_hints + + +@dataclass(frozen=True) +class S52MarinerSettings: + shallow_contour_m: float = 2.0 + safety_contour_m: float = 3.0 + safety_depth_m: float = 3.0 + deep_contour_m: float = 6.0 + two_shades: bool = False + + +def _float(attrs: dict[str, Any], key: str, default: float) -> float: + try: + return float(attrs.get(key, default)) + except (TypeError, ValueError): + return default + + +def resolve_depare01_instruction( + object_class: str, + attrs: dict[str, Any], + settings: S52MarinerSettings, +) -> str: + drval1_found = attrs.get("DRVAL1") not in {None, ""} + drval1 = _float(attrs, "DRVAL1", -1.0) + drval2 = _float(attrs, "DRVAL2", drval1 + 0.01) + if drval2 <= drval1: + drval2 = drval1 + 0.01 + + color = "DEPIT" + if drval1 >= 0.0 and drval2 > 0.0: + color = "DEPVS" + + if settings.two_shades: + if ( + drval1 >= settings.safety_contour_m + and drval2 > settings.safety_contour_m + ): + color = "DEPDW" + else: + if ( + drval1 >= settings.shallow_contour_m + and drval2 > settings.shallow_contour_m + ): + color = "DEPMS" + if ( + drval1 >= settings.safety_contour_m + and drval2 > settings.safety_contour_m + ): + color = "DEPMD" + if ( + drval1 >= settings.deep_contour_m + and drval2 > settings.deep_contour_m + ): + color = "DEPDW" + + if object_class == "DRGARE" and not drval1_found: + color = "DEPMD" + + instruction = f"AC({color})" + if object_class == "DRGARE": + instruction += ";AP(DRGARE01);LS(DASH,1,CHGRF)" + return instruction + + +def resolve_depcnt02_instruction( + object_class: str, + attrs: dict[str, Any], + settings: S52MarinerSettings, + *, + next_safe_contour_m: float | None = None, +) -> tuple[str, bool]: + safe = False + if object_class == "DEPARE": + drval1 = _float(attrs, "DRVAL1", 0.0) + drval2 = _float(attrs, "DRVAL2", drval1) + if drval1 <= settings.safety_contour_m <= drval2: + safe = True + elif next_safe_contour_m is not None: + safe = abs(drval1 - next_safe_contour_m) < 1e-4 + else: + valdco = _float(attrs, "VALDCO", 0.0) + if abs(valdco - settings.safety_contour_m) < 1e-4: + safe = True + elif next_safe_contour_m is not None: + safe = abs(valdco - next_safe_contour_m) < 1e-4 + + try: + quapos = int(attrs.get("QUAPOS") or 0) + except (TypeError, ValueError): + quapos = 0 + if 2 <= quapos < 10: + return ("LS(DASH,2,DEPSC)" if safe else "LS(DASH,1,DEPCN)"), safe + return ("LS(SOLD,2,DEPSC)" if safe else "LS(SOLD,1,DEPCN)"), safe + + +def resolve_csp_hints( + csp_name: str, + object_class: str, + attrs: dict[str, Any], + settings: S52MarinerSettings, + *, + next_safe_contour_m: float | None = None, +) -> dict[str, Any]: + name = str(csp_name or "")[:8] + instruction: str | None = None + extra: dict[str, Any] = {} + + if name in {"DEPARE01", "DEPARE02"}: + instruction = resolve_depare01_instruction(object_class, attrs, settings) + elif name == "DEPCNT02": + instruction, safe = resolve_depcnt02_instruction( + object_class, + attrs, + settings, + next_safe_contour_m=next_safe_contour_m, + ) + extra["s52_is_safety_contour"] = int(safe) + else: + return {} + + hints = static_rule_hints(instruction) + hints.update(extra) + hints["s52_csp_resolved"] = name + hints["s52_csp_rule_opcodes"] = ",".join( + rule.opcode for rule in parse_instruction_string(instruction) + ) + return hints diff --git a/src/cm95pbf/mapping.py b/src/cm95pbf/mapping.py index 5b051e3..e7317a6 100644 --- a/src/cm95pbf/mapping.py +++ b/src/cm95pbf/mapping.py @@ -34,7 +34,17 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset( "place_name_en", "s52_csp", "s52_csp_resolved", + "s52_csp_rule_opcodes", + "s52_is_safety_contour", "s52_display_category", + "s52_area_color", + "s52_area_pattern_code", + "s52_has_text", + "s52_line_color", + "s52_line_complex_code", + "s52_line_style", + "s52_line_width", + "s52_rule_opcodes", "s52_light_arc_color", "s52_light_arc_radius_mm", "s52_light_mode", diff --git a/src/cm95pbf/presentation.py b/src/cm95pbf/presentation.py new file mode 100644 index 0000000..b5599ff --- /dev/null +++ b/src/cm95pbf/presentation.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass +from typing import Any, Iterable + + +SUPPORTED_OPCODES = frozenset( + {"SY", "LS", "LC", "AC", "AP", "TX", "TE", "CS", "CA", "MP"} +) + + +class PresentationParseError(ValueError): + pass + + +@dataclass(frozen=True) +class PresentationRule: + opcode: str + args_raw: str + args: tuple[str, ...] + raw: str + sequence: int + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + data["args"] = list(self.args) + return data + + +def _split_args(text: str) -> tuple[str, ...]: + if not text.strip(): + return () + out: list[str] = [] + buf: list[str] = [] + quote: str | None = None + depth = 0 + i = 0 + while i < len(text): + ch = text[i] + if quote is not None: + buf.append(ch) + if ch == quote: + quote = None + elif ch == "\\" and i + 1 < len(text): + i += 1 + buf.append(text[i]) + elif ch in {"'", '"'}: + quote = ch + buf.append(ch) + elif ch == "(": + depth += 1 + buf.append(ch) + elif ch == ")": + depth = max(0, depth - 1) + buf.append(ch) + elif ch == "," and depth == 0: + out.append("".join(buf).strip()) + buf.clear() + else: + buf.append(ch) + i += 1 + out.append("".join(buf).strip()) + return tuple(out) + + +def parse_instruction_string(instruction: str) -> list[PresentationRule]: + """Parse S-52 INST using OpenCPN StringToRules/SCANFWRD semantics. + + OpenCPN intentionally scans each command to the next semicolon/record end + instead of balancing parentheses. Some production Presentation Library + records rely on that tolerance, so this parser does the same. + """ + text = str(instruction or "").replace("\x1f", ";").strip() + rules: list[PresentationRule] = [] + i = 0 + while i < len(text): + while i < len(text) and (text[i].isspace() or text[i] == ";"): + i += 1 + if i >= len(text): + break + if i + 2 >= len(text): + raise PresentationParseError(f"truncated opcode at {i}: {text!r}") + opcode = text[i : i + 2] + if opcode not in SUPPORTED_OPCODES: + raise PresentationParseError( + f"unsupported opcode {opcode!r} at {i}: {text!r}" + ) + start = i + i += 2 + while i < len(text) and text[i].isspace(): + i += 1 + if i >= len(text) or text[i] != "(": + raise PresentationParseError( + f"missing '(' after {opcode} at {i}: {text!r}" + ) + i += 1 + args_start = i + while i < len(text) and text[i] != ";": + i += 1 + payload = text[args_start:i].strip() + if payload.endswith(")"): + payload = payload[:-1].rstrip() + raw = text[start:i].strip() + rules.append( + PresentationRule( + opcode=opcode, + args_raw=payload, + args=_split_args(payload), + raw=raw, + sequence=len(rules), + ) + ) + return rules + +def compile_lookup_ir(row: dict[str, Any]) -> dict[str, Any]: + rules = parse_instruction_string(str(row.get("instruction") or "")) + return { + "id": row.get("id"), + "rcid": row.get("rcid"), + "object_class": row.get("object_class"), + "geometry_type": row.get("geometry_type"), + "table_name": row.get("table_name"), + "display_priority": row.get("display_priority"), + "display_category": row.get("display_category"), + "radar_priority": row.get("radar_priority"), + "attribute_conditions_raw": list(row.get("attribute_conditions_raw") or []), + "instruction": row.get("instruction") or "", + "rules": [rule.to_dict() for rule in rules], + } + + +def referenced_resource(rule: PresentationRule) -> tuple[str, str] | None: + if not rule.args: + return None + kind_by_opcode = { + "SY": "symbol", + "LC": "line_style", + "AP": "pattern", + "CS": "conditional", + "AC": "color", + } + kind = kind_by_opcode.get(rule.opcode) + if kind is None: + return None + value = rule.args[0].strip().strip("'\"") + if rule.opcode in {"SY", "LC", "AP", "CS"}: + value = value[:8] + return (kind, value) if value else None + + +def flatten_rules(rows: Iterable[dict[str, Any]]) -> list[PresentationRule]: + out: list[PresentationRule] = [] + for row in rows: + out.extend(parse_instruction_string(str(row.get("instruction") or ""))) + return out + + +def static_rule_hints(instruction: str) -> dict[str, Any]: + """Materialize compact scalar hints; full rule IR remains lookup-side.""" + rules = parse_instruction_string(instruction) + out: dict[str, Any] = {} + opcodes = [rule.opcode for rule in rules] + if opcodes: + out["s52_rule_opcodes"] = ",".join(opcodes) + + for rule in rules: + args = rule.args + if rule.opcode == "SY" and args and "s52_symbol_code" not in out: + out["s52_symbol_code"] = args[0].strip().strip("'\"")[:8] + elif rule.opcode == "AC" and args and "s52_area_color" not in out: + out["s52_area_color"] = args[0].strip().strip("'\"") + elif rule.opcode == "AP" and args and "s52_area_pattern_code" not in out: + out["s52_area_pattern_code"] = args[0].strip().strip("'\"")[:8] + elif rule.opcode == "LC" and args and "s52_line_complex_code" not in out: + out["s52_line_complex_code"] = args[0].strip().strip("'\"")[:8] + elif rule.opcode == "LS" and len(args) >= 3 and "s52_line_style" not in out: + out["s52_line_style"] = args[0].strip().strip("'\"") + out["s52_line_width"] = args[1].strip().strip("'\"") + out["s52_line_color"] = args[2].strip().strip("'\"") + elif rule.opcode in {"TX", "TE"}: + out["s52_has_text"] = 1 + elif rule.opcode == "CS" and args: + name = args[0].strip().strip("'\"")[:8] + existing = out.get("s52_csp") + out["s52_csp"] = f"{existing},{name}" if existing else name + return out diff --git a/src/cm95pbf/s52.py b/src/cm95pbf/s52.py index c75dc7d..6d244a3 100644 --- a/src/cm95pbf/s52.py +++ b/src/cm95pbf/s52.py @@ -2,6 +2,9 @@ from __future__ import annotations from typing import Any, Iterable +from .presentation import static_rule_hints +from .csp import S52MarinerSettings, resolve_csp_hints + CM93_COLMAR_TO_S57_COLOUR: dict[int, str] = { 1: "4", # green @@ -284,15 +287,26 @@ def resolve_lights05(attrs: dict[str, Any]) -> dict[str, Any]: class S52LookupRegistry: """Read extracted S-52 lookups once and expose PBF-ready properties.""" - def __init__(self, path, point_table: str = "Paper"): + def __init__( + self, + path, + point_table: str = "Paper", + boundary_table: str = "Plain", + ): import json from pathlib import Path self.path = Path(path) self.point_table = str(point_table) + self.boundary_table = str(boundary_table) + self.settings = S52MarinerSettings() rows = json.loads(self.path.read_text(encoding="utf-8")) self.by_class: dict[str, list[dict[str, Any]]] = {} for row in rows: + row = dict(row) + row["_static_hints"] = static_rule_hints( + str(row.get("instruction") or "") + ) object_class = str(row.get("object_class") or "") self.by_class.setdefault(object_class, []).append(row) @@ -303,13 +317,19 @@ class S52LookupRegistry: attrs: dict[str, Any], ) -> dict[str, Any] | None: geometry = str(geometry_type).upper() - if geometry != "POINT": + table_geometry = { + "POINT": (self.point_table, "Point"), + "LINE": ("Lines", "Line"), + "AREA": (self.boundary_table, "Area"), + }.get(geometry) + if table_geometry is None: return None + table_name, lookup_geometry = table_geometry return find_best_lookup( self.by_class.get(str(object_class), []), attrs, - table_name=self.point_table, - geometry_type="Point", + table_name=table_name, + geometry_type=lookup_geometry, ) def properties( @@ -317,6 +337,8 @@ class S52LookupRegistry: object_class: str, geometry_type: str, attrs: dict[str, Any], + *, + next_safe_contour_m: float | None = None, ) -> dict[str, Any]: row = self.resolve(object_class, geometry_type, attrs) if row is None: @@ -328,12 +350,29 @@ class S52LookupRegistry: "s52_display_priority": str(row.get("display_priority") or ""), "s52_table_name": str(row.get("table_name") or ""), } - symbol = primary_symbol_code(row) - if symbol: - out["s52_symbol_code"] = symbol - procedures = conditional_procedures(row) - if procedures: - out["s52_csp"] = ",".join(procedures) + out.update(dict(row.get("_static_hints") or {})) + procedures = [ + token.strip() + for token in str(out.get("s52_csp") or "").split(",") + if token.strip() + ] + unresolved: list[str] = [] + for procedure in procedures: + resolved = resolve_csp_hints( + procedure, + str(object_class), + attrs, + self.settings, + next_safe_contour_m=next_safe_contour_m, + ) + if resolved: + out.update(resolved) + else: + unresolved.append(procedure) + if unresolved: + out["s52_csp"] = ",".join(unresolved) + else: + out.pop("s52_csp", None) return {key: value for key, value in out.items() if value != ""} diff --git a/tools/cm95_s52_presentation_audit.py b/tools/cm95_s52_presentation_audit.py new file mode 100644 index 0000000..a1e4a87 --- /dev/null +++ b/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.*?)\{\"########\"", + 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())