feat: separate S-52 mariner runtime presentation

tools/cm95_s52_runtime_contract_check.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+from collections import Counter
+import json
+from pathlib import Path
+from typing import Any
+
+import mapbox_vector_tile
+
+
+DYNAMIC_FINAL_FIELDS = {
+    "s52_is_safety_contour",
+    "s52_sounding_prefix",
+    "s52_sounding_is_shallow",
+    "s52_isolated_danger",
+    "s52_display_base_promoted",
+}
+
+
+def decode_features(path: Path) -> list[tuple[str, dict[str, Any]]]:
+    decoded = mapbox_vector_tile.decode(path.read_bytes())
+    features: list[tuple[str, dict[str, Any]]] = []
+    for layer_name, layer in decoded.items():
+        for feature in layer.get("features", []):
+            features.append((layer_name, dict(feature.get("properties") or {})))
+    return features
+
+
+def audit_runtime(features: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]:
+    runtime = [
+        (layer, props)
+        for layer, props in features
+        if props.get("s52_mariner_dependent") == 1
+    ]
+    failures: list[str] = []
+    if not runtime:
+        failures.append("没有发现 s52_mariner_dependent=1 Feature")
+
+    csp_counts = Counter(str(props.get("s52_runtime_csp") or "") for _, props in runtime)
+    leaked = Counter()
+    for _, props in runtime:
+        csp = str(props.get("s52_runtime_csp") or "")
+        if not csp:
+            failures.append("runtime Feature 缺少 s52_runtime_csp")
+        if csp and csp not in str(props.get("s52_csp") or "").split(","):
+            failures.append(f"{csp} 未保留在 s52_csp")
+        for key in DYNAMIC_FINAL_FIELDS:
+            if key in props:
+                leaked[key] += 1
+        if csp in {"DEPARE01", "DEPARE02"}:
+            if "depth_min_m" not in props:
+                failures.append(f"{csp} 缺少 depth_min_m")
+            if "s52_area_color" in props:
+                leaked["DEPARE:s52_area_color"] += 1
+        elif csp == "DEPCNT02" and "contour_depth_m" not in props:
+            failures.append("DEPCNT02 缺少 contour_depth_m")
+        elif csp == "SOUNDG02" and "s52_sounding_depth_m" not in props:
+            failures.append("SOUNDG02 缺少 s52_sounding_depth_m")
+
+    if leaked:
+        failures.append(f"发现 runtime 最终状态泄漏: {dict(leaked)}")
+
+    return {
+        "feature_count": len(features),
+        "mariner_dependent_count": len(runtime),
+        "runtime_csp_counts": dict(csp_counts),
+        "layer_counts": dict(Counter(layer for layer, _ in runtime)),
+        "fact_counts": {
+            key: sum(1 for _, props in features if key in props)
+            for key in (
+                "depth_min_m",
+                "depth_max_m",
+                "contour_depth_m",
+                "s52_sounding_depth_m",
+                "s52_hazard_depth_m",
+                "s52_surrounding_max_drval1_m",
+                "s52_surrounding_min_drval2_m",
+            )
+        },
+        "dynamic_final_leaks": dict(leaked),
+        "failures": failures,
+        "status": "PASS" if not failures else "FAIL",
+    }
+
+
+def audit_reference(features: list[tuple[str, dict[str, Any]]]) -> dict[str, Any]:
+    failures: list[str] = []
+    runtime_markers = sum(
+        1 for _, props in features if props.get("s52_mariner_dependent") == 1
+    )
+    if runtime_markers:
+        failures.append(f"reference PBF 出现 {runtime_markers} 个 runtime marker")
+    return {
+        "feature_count": len(features),
+        "runtime_marker_count": runtime_markers,
+        "fixed_result_counts": {
+            "safety_contour": sum(
+                1 for _, props in features if props.get("s52_is_safety_contour") == 1
+            ),
+            "isolated_danger": sum(
+                1 for _, props in features if props.get("s52_isolated_danger") == 1
+            ),
+            "display_base_promoted": sum(
+                1 for _, props in features if props.get("s52_display_base_promoted") == 1
+            ),
+            "sounding_prefix": sum(
+                1 for _, props in features if "s52_sounding_prefix" in props
+            ),
+        },
+        "presentation_sequence_counts": {
+            key: sum(1 for _, props in features if key in props)
+            for key in (
+                "s52_symbol_count",
+                "s52_line_count",
+                "s52_symbol_rotation_deg",
+                "s52_symbol_rule_sequence",
+                "s52_priority_override",
+            )
+        },
+        "failures": failures,
+        "status": "PASS" if not failures else "FAIL",
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="检查 CM95 S-52 runtime/reference Delivery PBF 边界"
+    )
+    parser.add_argument("--runtime-pbf", type=Path, required=True)
+    parser.add_argument("--reference-pbf", type=Path, required=True)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+
+    runtime = audit_runtime(decode_features(args.runtime_pbf))
+    reference = audit_reference(decode_features(args.reference_pbf))
+    result = {
+        "schema": "cm95-s52-runtime-contract-check/v1",
+        "runtime_pbf": str(args.runtime_pbf),
+        "reference_pbf": str(args.reference_pbf),
+        "runtime": runtime,
+        "reference": reference,
+        "status": "PASS"
+        if runtime["status"] == reference["status"] == "PASS"
+        else "FAIL",
+    }
+    text = json.dumps(result, ensure_ascii=False, indent=2)
+    print(text)
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(text + "\n", encoding="utf-8")
+    return 0 if result["status"] == "PASS" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())