feat: enforce S-52 semantic conservation audits

This commit is contained in:
OpenAI Codex
2026-09-19 16:42:03 +08:00
parent 8ac774dff1
commit 06dc10abd9
8 changed files with 1324 additions and 0 deletions

View File

@@ -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": []
}
}
}

View File

@@ -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。

View File

@@ -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": {

View File

@@ -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

View File

@@ -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,

View File

@@ -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()

View File

@@ -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()

View File

@@ -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,