From 06dc10abd9b46d2a69c3a75cc2f8a74478ce632e Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Sat, 19 Sep 2026 16:42:03 +0800 Subject: [PATCH] feat: enforce S-52 semantic conservation audits --- ...2_runtime_capabilities_web_compare_v1.json | 70 +++ ...M95_S52_语义守恒与运行时能力审计标准_v1.md | 326 +++++++++++++ src/cm95pbf/cli.py | 6 + src/cm95pbf/presentation_audit.py | 114 +++++ src/cm95pbf/release.py | 112 +++++ ...m95_s52_presentation_conservation_audit.py | 447 ++++++++++++++++++ tools/cm95_s52_runtime_capability_audit.py | 212 +++++++++ tools/cm95_s52_style_compiler.py | 37 ++ 8 files changed, 1324 insertions(+) create mode 100644 config/s52/cm95_s52_runtime_capabilities_web_compare_v1.json create mode 100644 docs/CM95_S52_语义守恒与运行时能力审计标准_v1.md create mode 100644 src/cm95pbf/presentation_audit.py create mode 100644 tools/cm95_s52_presentation_conservation_audit.py create mode 100644 tools/cm95_s52_runtime_capability_audit.py diff --git a/config/s52/cm95_s52_runtime_capabilities_web_compare_v1.json b/config/s52/cm95_s52_runtime_capabilities_web_compare_v1.json new file mode 100644 index 0000000..720ae86 --- /dev/null +++ b/config/s52/cm95_s52_runtime_capabilities_web_compare_v1.json @@ -0,0 +1,70 @@ +{ + "schema": "navsea.cm95.s52-runtime-capability-registry/v1", + "target": "kyushu_compare_web_debug", + "description": "Kyushu side-by-side browser target. IMPLEMENTED means the declared runtime kind has concrete target-side code evidence; PREVIEW_ONLY and NOT_IMPLEMENTED are non-compliant for strict portrayal parity.", + "capabilities": { + "RUNTIME_CA": { + "status": "IMPLEMENTED", + "fidelity": "S52_SCREEN_MM", + "evidence": [ + { + "path": "web/kyushu_compare.html", + "contains": [ + "left-ca-overlay", + "drawLightArcFeature", + "s52_arc_radius_mm", + "s52_arc_sector_radius_mm" + ] + } + ] + }, + "RUNTIME_DEPARE": { + "status": "PREVIEW_ONLY", + "fidelity": "DEFAULT_MARINER_SETTINGS_PREVIEW", + "evidence": [ + { + "path": "web/kyushu_compare.html", + "contains": [ + "runtime-preview-depare", + "runtimeDepthColorExpression" + ] + } + ] + }, + "RUNTIME_DEPCNT": { + "status": "PREVIEW_ONLY", + "fidelity": "DEFAULT_MARINER_SETTINGS_PREVIEW", + "evidence": [ + { + "path": "web/kyushu_compare.html", + "contains": [ + "runtime-preview-depcnt", + "updateRuntimeSafetyContour" + ] + } + ] + }, + "RUNTIME_SOUNDG": { + "status": "PREVIEW_ONLY", + "fidelity": "TEXT_VALUE_ONLY_NOT_S52_GLYPH_COMPOSITION", + "evidence": [ + { + "path": "web/kyushu_compare.html", + "contains": [ + "runtime-preview-soundg" + ] + } + ] + }, + "RUNTIME_OBSTRN": { + "status": "NOT_IMPLEMENTED", + "fidelity": "NONE", + "evidence": [] + }, + "RUNTIME_WRECKS": { + "status": "NOT_IMPLEMENTED", + "fidelity": "NONE", + "evidence": [] + } + } +} diff --git a/docs/CM95_S52_语义守恒与运行时能力审计标准_v1.md b/docs/CM95_S52_语义守恒与运行时能力审计标准_v1.md new file mode 100644 index 0000000..6246226 --- /dev/null +++ b/docs/CM95_S52_语义守恒与运行时能力审计标准_v1.md @@ -0,0 +1,326 @@ +# CM95 S-52 语义守恒与运行时能力审计标准 v1 + +日期：2026-09-19 + +## 1. 目的 + +本标准用于防止 CM95 Source → S-52 Presentation → Delivery PBF → Runtime Renderer 的任意一层发生“输出看起来结构正确，但语义已经被降维或丢失”的情况。 + +本标准的核心原则是： + +> 审计不能只验证输出自身是否自洽，必须验证输入语义是否完整守恒到输出，并继续验证输出中的每一种 runtime instruction 是否存在真实消费者。 + +尤其禁止把以下不同概念混为一谈： + +- 物理实体数量； +- presentation 数量； +- drawing instruction 数量； +- S-52 resource 数量； +- runtime renderer capability。 + +这些维度必须分别审计。 +## 2. 审计层级 + +### 2.1 Source Entity 层 + +验证原始数据库中的物理实体、relation、component 关系。 + +例如一个航标 relation group 可以包含： + +- 1 个 TOWERS； +- 2 个 LIGHTS。 + +该关系的物理实体 cardinality 可以是 1，但 LIGHTS presentation cardinality 是 2。 + +### 2.2 S-52 Presentation 层 + +每一条会产生独立 portrayal 语义的 Source record，必须独立解析出 presentation。 + +不得因为多个 Source record 最终附着在同一物理实体上，就只保留其中一个 presentation。 + +### 2.3 Delivery PBF 层 + +Delivery Feature 可以合并物理实体，但必须完整保存所有需要的 presentation facts。 + +物理实体去重不得造成 presentation 去重。 + +### 2.4 Runtime Capability 层 + +任何没有静态物化、需要 Runtime 执行的 instruction/CSP，必须有明确 target renderer capability。 + +“标记为 runtime”不等于“已经实现 renderer”。 + +### 2.5 Visual / Runtime Evidence 层 + +对需要实际屏幕行为的 instruction，还必须有真实 runtime evidence。HTTP 200、DOM 存在、字段存在均不能单独替代真实绘制验证。 +## 3. 基本守恒不变量 + +### 3.1 Physical Entity Invariant + +若一个 relation group 表示一个物理航标，则允许： + +`N source components → 1 navigation_marks Feature` + +但必须有明确、可审计的 component trace。 + +### 3.2 Presentation Cardinality Invariant + +若同一 relation group 中存在 N 条独立 LIGHTS，且这些记录均解析为 LIGHTS05，则： + +`N source LIGHTS → N LIGHTS presentations` + +不得变成： + +`N source LIGHTS → 1 selected LIGHTS presentation` + +### 3.3 Presentation Signature Invariant + +只比较数量不够。 + +必须为每条 presentation 构造 canonical signature，至少覆盖： + +- source_record_index； +- s52_csp_resolved； +- display priority/category； +- light mode； +- light symbol； +- arc outline/color/width； +- arc start/end； +- arc radius / sector radius； +- sector start/end/sweep； +- nominal range； +- rotation / obscured； +- light label / text literal。 + +对于同一个稳定输出 Feature ID： + +`expected signature multiset == actual signature multiset` + +A+B 变成 A+A 即使 count 都为 2，也必须 FAIL。 +### 3.4 Tile Occurrence Invariant + +同一个稳定 Feature ID 出现在不同 XYZ tile 时，其 presentation signature set 必须一致。 + +若同一个 Feature 在不同 tile 中出现不同的 LIGHTS presentation 集合，视为 tile occurrence drift，必须 FAIL。 + +### 3.5 Runtime Capability Invariant + +对于 release 实际出现的每一种 runtime kind，只要 feature_count > 0，就必须检查目标 renderer capability。 + +严格 portrayal parity 下： + +`required runtime kind → capability status IMPLEMENTED → implementation evidence exists` + +以下状态均不得视为 strict PASS： + +- PREVIEW_ONLY； +- NOT_IMPLEMENTED； +- capability entry missing； +- implementation evidence 缺失或 marker 不存在。 + +### 3.6 Resource Invariant + +SY/AP/LC 等资源存在性审计与 Runtime instruction capability 审计是两个独立维度。 + +Resource Contract PASS 不得被解释为 Runtime Portrayal PASS。 +## 4. 审计范围必须绑定 Production Plan + +Source→PBF 审计必须绑定具体 Production Plan： + +- dataset_id； +- plan_id； +- AOI bbox； +- zoom； +- scale ladder； +- FID manifest； +- lookup IR； +- PBF directory。 + +不得把 AOI 内但不属于该 plan scale ladder 的 Source record 计入 expected set。 + +审计 artifact 必须记录 plan_id 和 PBF directory。Finalize 必须验证 artifact 的 plan_id / PBF directory 与当前 release 一致，禁止拿其他 release 的 PASS 报告复用。 +## 5. Multi-LIGHTS 专项规则 + +### 5.1 实体模型 + +一座灯塔/灯标仍只产生一个主 `navigation_marks` Feature。 + +### 5.2 Presentation 模型 + +当 relation group 包含两条或以上 LIGHTS 时，主 Feature 必须有： + +`s52_light_presentations` + +数组中每条记录对应一条 Source LIGHTS，并保存已经完成 S-52 Lookup/CSP 解析的 presentation facts。 + +### 5.3 向后兼容 + +主 Feature 原有单组标量 LIGHTS 属性允许继续存在，但只作为主 presentation 的兼容字段。 + +Runtime 看到 `s52_light_presentations` 后，必须以数组为完整 truth，不得只消费主标量。 + +### 5.4 CA 特殊规则 + +`CA(...)` 是 screen-space drawing instruction。 + +arc radius 与 sector leg radius 使用屏幕毫米语义，不得预烘焙成地理圆。 + +Renderer 必须执行： + +- outline； +- colored arc； +- sector start/end； +- sector legs； +- bearing / screen coordinate 转换。 +## 6. Runtime Kind 标准 + +当前至少审计以下 runtime kinds： + +- RUNTIME_CA； +- RUNTIME_DEPARE； +- RUNTIME_DEPCNT； +- RUNTIME_SOUNDG； +- RUNTIME_OBSTRN； +- RUNTIME_WRECKS。 + +后续新增任何 runtime CSP/opcode 时，必须同步加入： + +1. usage scanner； +2. capability registry； +3. capability audit； +4. 对应 runtime/visual regression。 + +禁止出现“数据里新增 runtime kind，但 capability audit 不认识，因此被静默忽略”的情况。 +## 7. Release Gate + +### 7.1 Data Release Gate + +Builder Contract 0.4.11 及以后，finalize 必须要求独立的 Source→PBF presentation conservation audit artifact。 + +缺少 artifact、schema 不匹配、plan_id 不匹配、PBF directory 不匹配或 audit status != PASS，release finalize 必须 FAIL。 + +### 7.2 Target Portrayal Gate + +当声称某个具体 target 已达到 S-52 runtime portrayal 完整性或视觉 parity 时，必须额外提供该 target 的 runtime capability audit。 + +只要 release 中实际出现的 runtime kind 有任意一个不是 IMPLEMENTED，则不得声称该 target runtime portrayal 完整。 + +Data Release PASS 与 Target Portrayal PASS 必须分开表述。 +## 8. 禁止的审计方式 + +以下方式不能单独作为完整性证明： + +- 只比较 input/output Feature count； +- 只比较 PBF/report/decoded Feature count； +- 只扫描已经损失后的 PBF； +- 只确认字段名存在； +- 只确认 Resource Contract PASS； +- 只确认 style compiler 认识 RUNTIME_CA； +- 只确认 HTTP 200； +- 只确认 DOM / layer / source 已加载； +- 用截图主观观察代替可重复的 signature/capability gate。 + +这些检查仍然有价值，但只能作为各自层级的局部审计。 +## 9. 当前代码实现 + +### 9.1 Presentation Conservation + +实现： + +- `src/cm95pbf/presentation_audit.py` +- `tools/cm95_s52_presentation_conservation_audit.py` + +该审计器直接读取 NAS Source，不依赖 builder report 生成 expected set；按 Production Plan scale ladder 限定范围；使用 FID manifest 将 Source base entity 映射到稳定 PBF Feature ID；再解码 PBF，比较 multi-LIGHTS canonical signature set。 + +当前会检查： + +- expected/actual multi-light group 数； +- expected/actual presentation record 数； +- signature mismatch； +- missing/unexpected group； +- tile occurrence drift； +- Source LIGHTS05 resolution failure； +- PBF presentation JSON/structure failure。 + +### 9.2 Runtime Capability + +实现： + +- `tools/cm95_s52_runtime_capability_audit.py` +- `config/s52/cm95_s52_runtime_capabilities_web_compare_v1.json` + +Style usage audit 已扩展 runtime CSP coverage，覆盖 DEPARE/DEPCNT/SOUNDG/OBSTRN/WRECKS，并继续覆盖 CA。 + +Capability audit 会验证 capability status 和声明的 implementation evidence marker。 + +### 9.3 Finalize Gate + +`finalize-production` 新增： + +- `--presentation-conservation-audit` +- `--runtime-capability-audit` + +Builder Contract 0.4.11+ 的 presentation conservation audit 为硬门槛。 + +Runtime capability audit 在提供时为硬门槛；没有提供时只能说明该 release 的 target runtime capability 尚未审计，不得据此声称 portrayal parity。 +## 10. 已验证回归 + +双 sector LIGHTS 正向 regression： + +- Source expected multi-light groups：1 +- PBF actual multi-light groups：1 +- Source expected presentations：2 +- PBF actual presentations：2 +- signature mismatch：0 +- occurrence drift：0 +- Presentation Conservation：PASS + +同一 AOI 对修复前旧 release 做负向 regression： + +- Source expected multi-light groups：1 +- PBF actual multi-light groups：0 +- Source expected presentations：2 +- PBF actual presentations：0 +- missing group：1 +- Presentation Conservation：FAIL（missing_multi_light_groups） + +该负向回归证明审计器能直接抓到本次历史“multi-LIGHTS 被 composite 吃掉”的真实缺陷。 + +CA-only target capability： + +- RUNTIME_CA observed：4 tile occurrences +- capability：IMPLEMENTED +- implementation evidence：PASS +- CA capability audit：PASS + +完整 Web debug runtime capability： + +- RUNTIME_CA：IMPLEMENTED +- RUNTIME_DEPARE：PREVIEW_ONLY +- RUNTIME_DEPCNT：PREVIEW_ONLY +- RUNTIME_SOUNDG：PREVIEW_ONLY（当 release 实际出现时） +- RUNTIME_OBSTRN：NOT_IMPLEMENTED +- RUNTIME_WRECKS：NOT_IMPLEMENTED +- strict full runtime capability audit：FAIL + +该 FAIL 是正确结果，用于阻止把“部分 runtime preview”错误描述成“完整 S-52 runtime portrayal”。 +## 11. 后续扩展原则 + +本标准不只用于 LIGHTS。 + +任何存在以下结构的转换都必须采用相同方法： + +`多个 Source semantic records → 一个 Delivery physical Feature` + +例如组合航标、复杂危险物、复合文本、多个关联 portrayal record 等。 + +必须分别定义： + +- physical entity invariant； +- presentation cardinality； +- canonical signature； +- runtime kind； +- target capability； +- release gate。 + +新规则如果没有对应 expected/actual 守恒审计，不应进入正式 Production Contract。 diff --git a/src/cm95pbf/cli.py b/src/cm95pbf/cli.py index c452947..db068b5 100644 --- a/src/cm95pbf/cli.py +++ b/src/cm95pbf/cli.py @@ -191,6 +191,8 @@ def build_parser() -> argparse.ArgumentParser: prod_final.add_argument("--style", default="config/style/style.navsea-delivery-full-semantic-unified-20260808-cm95-v1.json") prod_final.add_argument("--sprite-json", default="config/sprite/semantic-sprite-20260808.json") prod_final.add_argument("--gap-disposition", default="config/mappings/cm95_gap_disposition_v1.yaml") + prod_final.add_argument("--presentation-conservation-audit", default=None, help="Source→PBF presentation conservation audit JSON；builder 0.4.11+ 正式 release 必填且必须 PASS") + prod_final.add_argument("--runtime-capability-audit", default=None, help="目标 renderer runtime capability audit JSON；提供后必须 PASS") prod_final.add_argument("--skip-style-contract", action="store_true", help="跳过旧 NavSea style/sprite contract；S-52 runtime release 仍执行 runtime contract gate") return p @@ -298,6 +300,8 @@ def main() -> None: None if args.skip_style_contract else (Path(args.style) if args.style else None), None if args.skip_style_contract else (Path(args.sprite_json) if args.sprite_json else None), Path(args.gap_disposition) if args.gap_disposition else None, + Path(args.presentation_conservation_audit) if args.presentation_conservation_audit else None, + Path(args.runtime_capability_audit) if args.runtime_capability_audit else None, ) print(json.dumps({ "status": result["status"], @@ -319,6 +323,8 @@ def main() -> None: "s52_resource_contract": result.get("s52_resource_contract"), "s52_runtime_contract": result.get("s52_runtime_contract"), "s52_light_presentation_contract": result.get("s52_light_presentation_contract"), + "s52_presentation_conservation_contract": result.get("s52_presentation_conservation_contract"), + "s52_runtime_capability_contract": result.get("s52_runtime_capability_contract"), "quantization_fallbacks": result.get("quantization_fallbacks"), "quantization_omissions": result.get("quantization_omissions"), "nav_name_conflict_audit": { diff --git a/src/cm95pbf/presentation_audit.py b/src/cm95pbf/presentation_audit.py new file mode 100644 index 0000000..c040527 --- /dev/null +++ b/src/cm95pbf/presentation_audit.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +from decimal import Decimal, InvalidOperation +import hashlib +import json +from typing import Any + + +LIGHT_PRESENTATION_SIGNATURE_FIELDS = ( + "source_record_index", + "s52_csp_resolved", + "s52_display_priority", + "s52_display_category", + "s52_light_mode", + "s52_light_symbol_code", + "s52_arc_outline_color", + "s52_arc_outline_width", + "s52_arc_color", + "s52_arc_width", + "s52_arc_start_deg", + "s52_arc_end_deg", + "s52_arc_radius_mm", + "s52_arc_sector_radius_mm", + "s52_sector_start_deg", + "s52_sector_end_deg", + "s52_sector_sweep_deg", + "s52_light_nominal_range_nm", + "s52_light_rotation_deg", + "s52_light_obscured", + "s52_light_label_text", + "s52_text_literal", +) + +_NUMERIC_FIELDS = { + "source_record_index", + "s52_arc_outline_width", + "s52_arc_width", + "s52_arc_start_deg", + "s52_arc_end_deg", + "s52_arc_radius_mm", + "s52_arc_sector_radius_mm", + "s52_sector_start_deg", + "s52_sector_end_deg", + "s52_sector_sweep_deg", + "s52_light_nominal_range_nm", + "s52_light_rotation_deg", +} + + +def _normalize_number(value: Any) -> str | None: + if value in (None, ""): + return None + try: + number = Decimal(str(value)) + except (InvalidOperation, ValueError): + return str(value) + if not number.is_finite(): + return str(value) + number = number.normalize() + if number == number.to_integral(): + return str(number.quantize(Decimal("1"))) + text = format(number, "f") + return text.rstrip("0").rstrip(".") if "." in text else text + + +def canonical_light_presentation( + presentation: dict[str, Any], +) -> dict[str, Any]: + canonical: dict[str, Any] = {} + for field in LIGHT_PRESENTATION_SIGNATURE_FIELDS: + value = presentation.get(field) + if field in _NUMERIC_FIELDS: + value = _normalize_number(value) + elif field == "s52_light_obscured": + if value in (None, ""): + value = None + elif isinstance(value, str): + value = value.strip().lower() in {"1", "true", "yes"} + else: + value = bool(value) + elif value is not None: + value = str(value) + canonical[field] = value + return canonical + + +def light_presentation_signature( + presentation: dict[str, Any], +) -> str: + canonical = canonical_light_presentation(presentation) + payload = json.dumps( + canonical, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def parse_light_presentations(value: Any) -> list[dict[str, Any]]: + if value in (None, ""): + return [] + if isinstance(value, str): + value = json.loads(value) + if not isinstance(value, list): + raise ValueError("s52_light_presentations must decode to a list") + rows: list[dict[str, Any]] = [] + for index, item in enumerate(value): + if not isinstance(item, dict): + raise ValueError( + f"s52_light_presentations[{index}] must be an object" + ) + rows.append(dict(item)) + return rows diff --git a/src/cm95pbf/release.py b/src/cm95pbf/release.py index 3646e4c..824c297 100644 --- a/src/cm95pbf/release.py +++ b/src/cm95pbf/release.py @@ -4,6 +4,7 @@ from collections import Counter from datetime import datetime, timezone import hashlib import json +import re from pathlib import Path from typing import Any @@ -16,10 +17,95 @@ from .style_contract import audit_pbf_style_and_sprite RELEASE_SCHEMA = "navsea.cm95.release-manifest/v1" +def _builder_contract_at_least(value: Any, target: tuple[int, int, int]) -> bool: + match = re.match(r"^cm95pbf-builder/(\d+)\.(\d+)\.(\d+)", str(value or "")) + if not match: + return False + return tuple(int(part) for part in match.groups()) >= target + + def _utc_now(): return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") +def _external_audit_contract( + path: Path | None, + *, + required: bool, + expected_schema: str, + expected_plan_id: str | None = None, + expected_pbf_dir: Path | None = None, +) -> dict[str, Any]: + result: dict[str, Any] = { + "required": required, + "status": "NOT_AUDITED", + "path": None, + "sha256": None, + "schema": None, + "audit_status": None, + "failures": [], + } + if path is None: + if required: + result["status"] = "FAIL" + result["failures"].append("required_audit_missing") + return result + + path = Path(path) + result["path"] = str(path) + if not path.is_file(): + result["status"] = "FAIL" + result["failures"].append("audit_file_missing") + return result + + result["sha256"] = sha256_file(path) + try: + obj = json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: + result["status"] = "FAIL" + result["failures"].append( + f"audit_parse_error:{type(exc).__name__}" + ) + return result + + result["schema"] = obj.get("schema") + result["audit_status"] = obj.get("status") + if obj.get("schema") != expected_schema: + result["failures"].append("audit_schema_mismatch") + if obj.get("status") != "PASS": + result["failures"].append("audit_not_pass") + if expected_plan_id is not None and obj.get("plan_id") != expected_plan_id: + result["failures"].append("audit_plan_id_mismatch") + if expected_pbf_dir is not None: + raw_pbf_dir = obj.get("pbf_dir") + if not raw_pbf_dir: + result["failures"].append("audit_pbf_dir_missing") + else: + try: + actual = Path(raw_pbf_dir).resolve() + expected = Path(expected_pbf_dir).resolve() + except Exception: + result["failures"].append("audit_pbf_dir_invalid") + else: + if actual != expected: + result["failures"].append("audit_pbf_dir_mismatch") + + for key in ( + "expected_multi_light_groups", + "actual_multi_light_groups", + "expected_presentation_records", + "actual_presentation_records", + "signature_mismatch_count", + "occurrence_drift_count", + "required_runtime_kinds", + "missing_runtime_kinds", + "noncompliant_runtime_kinds", + ): + if key in obj: + result[key] = obj.get(key) + + result["status"] = "PASS" if not result["failures"] else "FAIL" + return result def _collect_nav_name_conflict_expectations( @@ -266,6 +352,8 @@ def finalize_release( style_path: Path | None = None, sprite_json_path: Path | None = None, gap_disposition_path: Path | None = None, + presentation_conservation_audit_path: Path | None = None, + runtime_capability_audit_path: Path | None = None, ) -> dict[str, Any]: plan = load_plan(plan_path) release_dir = Path(release_root) / str(plan["release_id"]) @@ -537,6 +625,24 @@ def finalize_release( "failure_examples": light_presentation_failures[:100], } + presentation_conservation_required = _builder_contract_at_least( + plan.get("builder_contract_version"), (0, 4, 11) + ) + presentation_conservation_contract = _external_audit_contract( + presentation_conservation_audit_path, + required=presentation_conservation_required, + expected_schema="navsea.cm95.s52-presentation-conservation-audit/v1", + expected_plan_id=str(plan.get("plan_id") or ""), + expected_pbf_dir=pbf_root, + ) + runtime_capability_contract = _external_audit_contract( + runtime_capability_audit_path, + required=False, + expected_schema="navsea.cm95.s52-runtime-capability-audit/v1", + expected_plan_id=str(plan.get("plan_id") or ""), + expected_pbf_dir=None, + ) + s52_runtime_contract = { "status": ("PASS" if not s52_runtime_failures else "FAIL") if s52_runtime_enabled else "NOT_APPLICABLE", "enabled": s52_runtime_enabled, @@ -677,6 +783,10 @@ def finalize_release( failures.append("s52_runtime_contract") if light_presentation_required and s52_light_presentation_contract.get("status") != "PASS": failures.append("s52_light_presentation_contract") + if presentation_conservation_required and presentation_conservation_contract.get("status") != "PASS": + failures.append("s52_presentation_conservation_contract") + if runtime_capability_audit_path is not None and runtime_capability_contract.get("status") != "PASS": + failures.append("s52_runtime_capability_contract") result = { "schema": RELEASE_SCHEMA, @@ -699,6 +809,8 @@ def finalize_release( "s52_resource_contract": s52_resource_contract, "s52_runtime_contract": s52_runtime_contract, "s52_light_presentation_contract": s52_light_presentation_contract, + "s52_presentation_conservation_contract": presentation_conservation_contract, + "s52_runtime_capability_contract": runtime_capability_contract, "style_sha256": style_sha, "sprite_json_sha256": sprite_sha, "gap_disposition_sha256": gap_disposition_sha, diff --git a/tools/cm95_s52_presentation_conservation_audit.py b/tools/cm95_s52_presentation_conservation_audit.py new file mode 100644 index 0000000..619c79f --- /dev/null +++ b/tools/cm95_s52_presentation_conservation_audit.py @@ -0,0 +1,447 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +from collections import defaultdict +import json +from pathlib import Path +from typing import Any + +import mapbox_vector_tile + +from cm93db.config import load_config +from cm93db.database import connect +from cm95pbf.identity import ( + FidManifest, + ROLE_PRIMARY, + feature_source_key, +) +from cm95pbf.presentation_audit import ( + LIGHT_PRESENTATION_SIGNATURE_FIELDS, + canonical_light_presentation, + light_presentation_signature, + parse_light_presentations, +) +from cm95pbf.s52 import S52LookupRegistry + + +SCHEMA = "navsea.cm95.s52-presentation-conservation-audit/v1" +NAV_COMPONENT_CLASSES = ( + "LIGHTS", + "BCNLAT", + "BCNSPP", + "BOYSPP", + "RTPBCN", + "TOWERS", + "_extgn", +) +BASE_PRECEDENCE = { + "BCNLAT": 0, + "BCNSPP": 1, + "BOYSPP": 2, + "RTPBCN": 3, + "TOWERS": 4, + "_extgn": 5, +} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Audit multi-LIGHTS presentation conservation directly from " + "CM95 source rows to decoded delivery PBF." + ) + ) + parser.add_argument("--dataset-id", type=int, required=True) + parser.add_argument("--plan", required=True) + parser.add_argument("--aoi-bbox", required=True, help="west,south,east,north") + parser.add_argument("--manifest", required=True) + parser.add_argument("--lookups", required=True) + parser.add_argument("--pbf-dir", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--point-table", default="Paper") + parser.add_argument("--boundary-table", default="Plain") + parser.add_argument("--mariner-mode", default="runtime") + return parser.parse_args() + + +def parse_bbox(raw: str) -> tuple[float, float, float, float]: + values = tuple(float(item.strip()) for item in raw.split(",")) + if len(values) != 4: + raise SystemExit("--aoi-bbox must be west,south,east,north") + west, south, east, north = values + if west >= east or south >= north: + raise SystemExit("invalid --aoi-bbox") + return west, south, east, north + + +def json_attrs(value: Any) -> dict[str, Any]: + if isinstance(value, dict): + return dict(value) + if isinstance(value, (bytes, bytearray, memoryview)): + value = bytes(value).decode("utf-8") + try: + obj = json.loads(value or "{}") + except Exception: + return {} + return dict(obj) if isinstance(obj, dict) else {} + + +def source_expected( + dataset_id: int, + bbox: tuple[float, float, float, float], + allowed_scales: set[int], + manifest_path: Path, + registry: S52LookupRegistry, +) -> tuple[dict[int, dict[str, Any]], list[dict[str, Any]]]: + west, south, east, north = bbox + cfg = load_config() + with connect(cfg) as conn: + with conn.cursor() as cur: + placeholders = ",".join(["%s"] * len(NAV_COMPONENT_CLASSES)) + cur.execute( + f""" + SELECT + f.id, + f.cell_id, + c.source_relative_path, + f.source_record_index, + f.object_class, + f.geometry_type, + f.native_scale, + f.attributes_json, + f.bbox_min_lon, + f.bbox_min_lat + FROM chart_feature f + JOIN chart_cell c ON c.id=f.cell_id + WHERE f.dataset_id=%s + AND f.geometry_type='POINT' + AND f.object_class IN ({placeholders}) + AND f.native_scale IN ({",".join(["%s"] * len(allowed_scales))}) + AND f.bbox_max_lon >= %s + AND f.bbox_min_lon <= %s + AND f.bbox_max_lat >= %s + AND f.bbox_min_lat <= %s + ORDER BY f.cell_id, f.source_record_index + """, + ( + dataset_id, + *NAV_COMPONENT_CLASSES, + *sorted(allowed_scales), + west, + east, + south, + north, + ), + ) + rows = [dict(row) for row in cur.fetchall()] + + groups: dict[tuple[int, str], list[dict[str, Any]]] = defaultdict(list) + for row in rows: + row["attrs"] = json_attrs(row.pop("attributes_json", {})) + relation = row["attrs"].get("_cm93_relation20_raw") + if relation is None: + continue + groups[(int(row["cell_id"]), str(relation))].append(row) + + expected: dict[int, dict[str, Any]] = {} + source_failures: list[dict[str, Any]] = [] + with FidManifest(manifest_path) as manifest: + for (cell_id, relation), members in groups.items(): + lights = [row for row in members if row["object_class"] == "LIGHTS"] + bases = [ + row + for row in members + if row["object_class"] in BASE_PRECEDENCE + ] + if len(lights) < 2 or not bases: + continue + base = sorted( + bases, + key=lambda row: ( + BASE_PRECEDENCE.get(row["object_class"], 99), + int(row["source_record_index"]), + ), + )[0] + source_key = feature_source_key( + dataset_id, + str(base["source_relative_path"]), + int(base["source_record_index"]), + str(base["geometry_type"]), + ) + try: + fid = manifest.mvt_id(source_key, ROLE_PRIMARY) + except Exception as exc: + source_failures.append( + { + "cell_id": cell_id, + "relation20": relation, + "base_source_record_index": base["source_record_index"], + "failure": f"manifest:{type(exc).__name__}:{exc}", + } + ) + continue + + presentations: list[dict[str, Any]] = [] + for light in sorted( + lights, key=lambda row: int(row["source_record_index"]) + ): + props = registry.properties( + str(light["object_class"]), + str(light["geometry_type"]), + dict(light["attrs"]), + ) + presentation = { + key: value + for key, value in props.items() + if str(key).startswith("s52_") + } + presentation["source_record_index"] = int( + light["source_record_index"] + ) + if presentation.get("s52_csp_resolved") != "LIGHTS05": + source_failures.append( + { + "feature_id": fid, + "cell_id": cell_id, + "relation20": relation, + "source_record_index": light["source_record_index"], + "failure": "source LIGHTS did not resolve to LIGHTS05", + } + ) + presentations.append(presentation) + + canonical = [ + canonical_light_presentation(item) + for item in presentations + ] + signatures = [ + light_presentation_signature(item) + for item in presentations + ] + expected[fid] = { + "feature_id": fid, + "cell_id": cell_id, + "relation20": relation, + "base_source_record_index": int(base["source_record_index"]), + "base_object_class": str(base["object_class"]), + "native_scale": int(base["native_scale"]), + "presentation_count": len(presentations), + "signatures": sorted(signatures), + "presentations": sorted( + canonical, + key=lambda item: str(item.get("source_record_index")), + ), + } + return expected, source_failures + + +def pbf_actual( + pbf_dir: Path, +) -> tuple[dict[int, dict[str, Any]], list[dict[str, Any]]]: + actual: dict[int, dict[str, Any]] = {} + failures: list[dict[str, Any]] = [] + + for path in sorted(pbf_dir.rglob("*.pbf")): + rel = path.relative_to(pbf_dir).as_posix() + decoded = mapbox_vector_tile.decode(path.read_bytes()) + layer = decoded.get("navigation_marks") or {} + for feature in layer.get("features") or []: + fid = feature.get("id") + props = feature.get("properties") or {} + raw = props.get("s52_light_presentations") + if fid is None or raw in (None, ""): + continue + try: + presentations = parse_light_presentations(raw) + except Exception as exc: + failures.append( + { + "pbf": rel, + "feature_id": fid, + "failure": ( + "invalid s52_light_presentations:" + f"{type(exc).__name__}:{exc}" + ), + } + ) + continue + + canonical = [ + canonical_light_presentation(item) + for item in presentations + ] + signatures = sorted( + light_presentation_signature(item) + for item in presentations + ) + occurrence = { + "pbf": rel, + "presentation_count": len(presentations), + "signatures": signatures, + "presentations": sorted( + canonical, + key=lambda item: str(item.get("source_record_index")), + ), + } + fid_int = int(fid) + state = actual.setdefault( + fid_int, + { + "feature_id": fid_int, + "presentation_count": len(presentations), + "signatures": signatures, + "presentations": occurrence["presentations"], + "occurrence_count": 0, + "occurrence_examples": [], + "occurrence_drift": [], + }, + ) + state["occurrence_count"] += 1 + if len(state["occurrence_examples"]) < 8: + state["occurrence_examples"].append(rel) + if state["signatures"] != signatures: + state["occurrence_drift"].append(occurrence) + return actual, failures + + +def compare_contract( + expected: dict[int, dict[str, Any]], + actual: dict[int, dict[str, Any]], + source_failures: list[dict[str, Any]], + pbf_failures: list[dict[str, Any]], +) -> dict[str, Any]: + missing = sorted(set(expected) - set(actual)) + unexpected = sorted(set(actual) - set(expected)) + signature_mismatches: list[dict[str, Any]] = [] + occurrence_drift: list[dict[str, Any]] = [] + + for fid in sorted(set(expected) & set(actual)): + exp = expected[fid] + got = actual[fid] + if exp["signatures"] != got["signatures"]: + signature_mismatches.append( + { + "feature_id": fid, + "cell_id": exp["cell_id"], + "relation20": exp["relation20"], + "expected_count": exp["presentation_count"], + "actual_count": got["presentation_count"], + "expected_signatures": exp["signatures"], + "actual_signatures": got["signatures"], + "expected_presentations": exp["presentations"], + "actual_presentations": got["presentations"], + } + ) + if got.get("occurrence_drift"): + occurrence_drift.append( + { + "feature_id": fid, + "canonical_signatures": got["signatures"], + "drift_examples": got["occurrence_drift"][:8], + } + ) + + failures: list[str] = [] + if source_failures: + failures.append("source_resolution_failures") + if pbf_failures: + failures.append("pbf_parse_failures") + if missing: + failures.append("missing_multi_light_groups") + if unexpected: + failures.append("unexpected_multi_light_groups") + if signature_mismatches: + failures.append("presentation_signature_mismatch") + if occurrence_drift: + failures.append("tile_occurrence_presentation_drift") + + return { + "status": "PASS" if not failures else "FAIL", + "expected_multi_light_groups": len(expected), + "actual_multi_light_groups": len(actual), + "expected_presentation_records": sum( + int(item["presentation_count"]) for item in expected.values() + ), + "actual_presentation_records": sum( + int(item["presentation_count"]) for item in actual.values() + ), + "matched_group_count": ( + len(set(expected) & set(actual)) - len(signature_mismatches) + ), + "missing_group_count": len(missing), + "unexpected_group_count": len(unexpected), + "signature_mismatch_count": len(signature_mismatches), + "occurrence_drift_count": len(occurrence_drift), + "source_resolution_failure_count": len(source_failures), + "pbf_parse_failure_count": len(pbf_failures), + "failures": failures, + "missing_groups": [ + expected[fid] for fid in missing[:100] + ], + "unexpected_groups": [ + actual[fid] for fid in unexpected[:100] + ], + "signature_mismatches": signature_mismatches[:100], + "occurrence_drift": occurrence_drift[:100], + "source_resolution_failures": source_failures[:100], + "pbf_parse_failures": pbf_failures[:100], + } + + +def main() -> None: + args = parse_args() + bbox = parse_bbox(args.aoi_bbox) + plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) + allowed_scales: set[int] = set() + for item in (plan.get("source_stats") or {}).values(): + for scale in item.get("scale_ladder") or []: + allowed_scales.add(int(scale)) + if not allowed_scales: + raise SystemExit("plan contains no source scale ladder") + registry = S52LookupRegistry( + Path(args.lookups), + point_table=args.point_table, + boundary_table=args.boundary_table, + mariner_mode=args.mariner_mode, + ) + expected, source_failures = source_expected( + args.dataset_id, + bbox, + allowed_scales, + Path(args.manifest), + registry, + ) + actual, pbf_failures = pbf_actual(Path(args.pbf_dir)) + contract = compare_contract( + expected, + actual, + source_failures, + pbf_failures, + ) + result = { + "schema": SCHEMA, + "dataset_id": args.dataset_id, + "plan": str(Path(args.plan)), + "plan_id": plan.get("plan_id"), + "allowed_native_scales": sorted(allowed_scales), + "aoi_bbox_wgs84": list(bbox), + "manifest": str(Path(args.manifest)), + "lookups": str(Path(args.lookups)), + "pbf_dir": str(Path(args.pbf_dir)), + "signature_fields": list(LIGHT_PRESENTATION_SIGNATURE_FIELDS), + **contract, + } + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text( + json.dumps(result, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + if result["status"] != "PASS": + raise SystemExit(2) + + +if __name__ == "__main__": + main() diff --git a/tools/cm95_s52_runtime_capability_audit.py b/tools/cm95_s52_runtime_capability_audit.py new file mode 100644 index 0000000..f716223 --- /dev/null +++ b/tools/cm95_s52_runtime_capability_audit.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + + +SCHEMA = "navsea.cm95.s52-runtime-capability-audit/v1" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Audit runtime portrayal kinds required by a release-scoped " + "S-52 style usage audit against a concrete target capability registry." + ) + ) + parser.add_argument("--plan", required=True) + parser.add_argument("--style-audit", required=True) + parser.add_argument("--capabilities", required=True) + parser.add_argument("--output", required=True) + parser.add_argument( + "--required-kind", + action="append", + default=[], + help=( + "Optional explicit runtime kind. When omitted, every non-zero " + "runtime_coverage kind in the style audit is required." + ), + ) + return parser.parse_args() + + +def load_json(path: Path) -> dict[str, Any]: + obj = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(obj, dict): + raise SystemExit(f"JSON root must be object: {path}") + return obj + + +def verify_evidence( + project_root: Path, + kind: str, + capability: dict[str, Any], +) -> list[dict[str, Any]]: + failures: list[dict[str, Any]] = [] + evidence = capability.get("evidence") or [] + if not evidence: + failures.append( + { + "kind": kind, + "failure": "implemented_capability_has_no_evidence", + } + ) + return failures + + for item in evidence: + if not isinstance(item, dict): + failures.append( + { + "kind": kind, + "failure": "evidence_entry_not_object", + } + ) + continue + raw_path = item.get("path") + if not raw_path: + failures.append( + { + "kind": kind, + "failure": "evidence_path_missing", + } + ) + continue + path = (project_root / str(raw_path)).resolve() + if not path.is_file(): + failures.append( + { + "kind": kind, + "path": str(raw_path), + "failure": "evidence_file_missing", + } + ) + continue + try: + text = path.read_text(encoding="utf-8") + except Exception as exc: + failures.append( + { + "kind": kind, + "path": str(raw_path), + "failure": f"evidence_read_error:{type(exc).__name__}", + } + ) + continue + for marker in item.get("contains") or []: + if str(marker) not in text: + failures.append( + { + "kind": kind, + "path": str(raw_path), + "marker": str(marker), + "failure": "evidence_marker_missing", + } + ) + return failures + + +def main() -> None: + args = parse_args() + plan_path = Path(args.plan) + style_audit_path = Path(args.style_audit) + capability_path = Path(args.capabilities) + + plan = load_json(plan_path) + style_audit = load_json(style_audit_path) + registry = load_json(capability_path) + + if registry.get("schema") != ( + "navsea.cm95.s52-runtime-capability-registry/v1" + ): + raise SystemExit("runtime capability registry schema mismatch") + + runtime_coverage = style_audit.get("runtime_coverage") or {} + observed = { + str(kind): int((row or {}).get("feature_count") or 0) + for kind, row in runtime_coverage.items() + if int((row or {}).get("feature_count") or 0) > 0 + } + if args.required_kind: + required = sorted(set(str(item) for item in args.required_kind)) + else: + required = sorted(observed) + + capabilities = registry.get("capabilities") or {} + missing: list[str] = [] + noncompliant: list[dict[str, Any]] = [] + evidence_failures: list[dict[str, Any]] = [] + checked: list[dict[str, Any]] = [] + + project_root = Path(__file__).resolve().parents[1] + + for kind in required: + capability = capabilities.get(kind) + if not isinstance(capability, dict): + missing.append(kind) + continue + status = str(capability.get("status") or "") + checked.append( + { + "kind": kind, + "feature_count": int(observed.get(kind, 0)), + "status": status, + "fidelity": capability.get("fidelity"), + } + ) + if status != "IMPLEMENTED": + noncompliant.append( + { + "kind": kind, + "status": status, + "fidelity": capability.get("fidelity"), + } + ) + continue + evidence_failures.extend( + verify_evidence(project_root, kind, capability) + ) + + failures: list[str] = [] + if style_audit.get("hard_gate_failures"): + failures.append("style_audit_hard_gate_failures") + if missing: + failures.append("runtime_capability_missing") + if noncompliant: + failures.append("runtime_capability_not_implemented") + if evidence_failures: + failures.append("runtime_capability_evidence_failure") + + result = { + "schema": SCHEMA, + "status": "PASS" if not failures else "FAIL", + "plan": str(plan_path), + "plan_id": plan.get("plan_id"), + "style_audit": str(style_audit_path), + "capability_registry": str(capability_path), + "target": registry.get("target"), + "observed_runtime_kinds": observed, + "required_runtime_kinds": required, + "checked_runtime_kinds": checked, + "missing_runtime_kinds": missing, + "noncompliant_runtime_kinds": noncompliant, + "evidence_failure_count": len(evidence_failures), + "evidence_failures": evidence_failures[:100], + "failures": failures, + } + + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text( + json.dumps(result, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + if result["status"] != "PASS": + raise SystemExit(2) + + +if __name__ == "__main__": + main() diff --git a/tools/cm95_s52_style_compiler.py b/tools/cm95_s52_style_compiler.py index af08155..97bdee5 100755 --- a/tools/cm95_s52_style_compiler.py +++ b/tools/cm95_s52_style_compiler.py @@ -215,6 +215,7 @@ def _scan_pbf_usage(path: Path) -> dict[str, Any]: key: Counter() for key in ("AC", "AP", "LS", "LC", "SY", "LIGHT_SY", "TXTE", "LIGHT_TX", "RUNTIME_SOUNDG", "RUNTIME_CA") } + runtime_csp_counts: Counter[str] = Counter() geometry_ignored: Counter[tuple[str, str, str]] = Counter() layers: dict[str, Any] = {} for layer_name, layer in decoded.items(): @@ -227,6 +228,10 @@ def _scan_pbf_usage(path: Path) -> dict[str, Any]: priority = str(props.get("s52_display_priority") or "") category = str(props.get("s52_display_category") or "") placement = (layer_name, priority, category) + for procedure in str(props.get("s52_runtime_csp") or "").split(","): + procedure = procedure.strip() + if procedure: + runtime_csp_counts[procedure] += 1 geometry_type = str((feature.get("geometry") or {}).get("type") or "") area_geometry = geometry_type in {"Polygon", "MultiPolygon"} line_geometry = geometry_type in {"LineString", "MultiLineString", "Polygon", "MultiPolygon"} @@ -297,6 +302,7 @@ def _scan_pbf_usage(path: Path) -> dict[str, Any]: "pbf": str(path), "layers": layers, "global_values": {k: sorted(v) for k, v in global_values.items() if v}, + "runtime_csp_counts": dict(sorted(runtime_csp_counts.items())), "geometry_ignored": [ {"kind": kind, "source_layer": layer, "geometry_type": geometry, "feature_count": count} for (kind, layer, geometry), count in sorted(geometry_ignored.items()) @@ -317,6 +323,7 @@ def _merge_pbf_usages(usages: list[dict[str, Any]], label: str) -> dict[str, Any layers: dict[str, dict[str, Any]] = {} geometry_ignored: Counter[tuple[str, str, str]] = Counter() render_kind_counts: dict[str, Counter[tuple[str, str, str]]] = {} + runtime_csp_counts: Counter[str] = Counter() for usage in usages: for field, values in (usage.get("global_values") or {}).items(): @@ -342,6 +349,13 @@ def _merge_pbf_usages(usages: list[dict[str, Any]], label: str) -> dict[str, Any str(value) for value in values ) + runtime_csp_counts.update( + { + str(key): int(value) + for key, value in (usage.get("runtime_csp_counts") or {}).items() + } + ) + for row in usage.get("geometry_ignored") or []: key = ( str(row.get("kind") or ""), @@ -384,6 +398,7 @@ def _merge_pbf_usages(usages: list[dict[str, Any]], label: str) -> dict[str, Any "global_values": { key: sorted(values) for key, values in global_values.items() if values }, + "runtime_csp_counts": dict(sorted(runtime_csp_counts.items())), "geometry_ignored": [ { "kind": kind, @@ -1224,6 +1239,28 @@ def _compile_style( "placements": rows, } + runtime_csp_kind = { + "DEPARE01": "RUNTIME_DEPARE", + "DEPARE02": "RUNTIME_DEPARE", + "DEPCNT02": "RUNTIME_DEPCNT", + "SOUNDG02": "RUNTIME_SOUNDG", + "OBSTRN04": "RUNTIME_OBSTRN", + "WRECKS02": "RUNTIME_WRECKS", + } + grouped_csp: dict[str, Counter[str]] = {} + for csp, count in (usage.get("runtime_csp_counts") or {}).items(): + kind = runtime_csp_kind.get(str(csp)) + if not kind: + continue + grouped_csp.setdefault(kind, Counter())[str(csp)] += int(count) + for kind, counts in grouped_csp.items(): + entry = runtime_coverage.setdefault( + kind, + {"feature_count": 0, "placement_count": 0, "placements": []}, + ) + entry["feature_count"] = sum(counts.values()) + entry["csp_counts"] = dict(sorted(counts.items())) + audit = { "schema": "navsea.cm95.s52-style-compiler-audit/v1", "hard_gate_failures": hard_gate_failures,