feat: complete S-52 resource contract

This commit is contained in:
OpenAI Codex
2026-09-19 08:32:51 +08:00
parent b9ece0eaaa
commit 9dd6282817
11 changed files with 962 additions and 42 deletions

View File

@@ -19,18 +19,21 @@
"maxzoom": 12
},
"resources": {
"colors": "build/s52_poc_patterns_v1/s52_colors.json",
"symbol_metadata": "build/s52_poc_patterns_v1/s52_symbol_metadata.json",
"bitmap_pattern_metadata": "build/s52_poc_patterns_v1/s52_pattern_metadata.json",
"vector_pattern_metadata": "build/s52_ap_v3/s52-sprite-lc-ap-pattern-metadata.json",
"line_metadata": "build/s52_lc_v1/s52-sprite-lc-line-metadata.json",
"sprite_index": "build/s52_ap_v3/s52-sprite-lc-ap.json",
"lookup_ir": "build/s52_presentation_ir/lookup_rule_ir.json"
"colors": "build/s52_resources_v2/base/s52_colors.json",
"symbol_metadata": "build/s52_resources_v2/base/s52_symbol_metadata.json",
"bitmap_pattern_metadata": "build/s52_resources_v2/base/s52_pattern_metadata.json",
"vector_pattern_metadata": "build/s52_resources_v2/ap/s52-sprite-ap-pattern-metadata.json",
"line_metadata": "build/s52_resources_v2/final/s52-sprite-full-line-metadata.json",
"sprite_index": "build/s52_resources_v2/final/s52-sprite-full.json",
"lookup_ir": "build/s52_presentation_ir_v2/lookup_rule_ir.json",
"resource_contract": "build/s52_resources_v2/resource_contract.json"
},
"source_layers": [
{"name": "anchor_caution_hazard_area", "capabilities": ["fill", "line", "symbol", "text"]},
{"name": "anchor_caution_hazard_point", "capabilities": ["symbol", "text"]},
{"name": "anchorage_area", "capabilities": ["fill", "line", "symbol", "text"]},
{"name": "anchorage_outline", "capabilities": ["line", "symbol", "text"]},
{"name": "anchorage_point", "capabilities": ["symbol", "text"]},
{"name": "baseline_area", "capabilities": ["fill", "line", "symbol", "text"]},
{"name": "baseline_line", "capabilities": ["line", "symbol", "text"]},
{"name": "bridge_structure", "capabilities": ["fill", "line", "symbol", "text"]},

View File

@@ -1,2 +1,2 @@
__version__ = "0.4.8-s52-mariner-runtime-boundary"
BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.8-s52-mariner-runtime-boundary"
__version__ = "0.4.10-s52-resource-contract"
BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.10-s52-resource-contract"

View File

@@ -135,6 +135,11 @@ def build_parser() -> argparse.ArgumentParser:
prod_plan.add_argument("--manifest", default=str(DEFAULT_MANIFEST))
prod_plan.add_argument("--output", default="build/production/cm95_production_plan_v2.json")
prod_plan.add_argument("--s52-lookups", default=None, help="S-52 lookup_rule_ir.json;提供后冻结 SHA256 并进入 production builder")
prod_plan.add_argument(
"--s52-resource-contract",
default=None,
help="S-52 resource_contract.json;S-52 正式生产必须为 PASS,并冻结 SHA256",
)
prod_plan.add_argument("--s52-point-table", choices=["Simplified", "Paper"], default="Paper")
prod_plan.add_argument("--s52-boundary-table", choices=["Plain", "Symbolized"], default="Plain")
prod_plan.add_argument("--s52-mariner-mode", choices=["runtime", "reference"], default="runtime")
@@ -232,6 +237,11 @@ def main() -> None:
zooms=parse_zooms(args.zooms),
profile=args.profile,
s52_lookups_path=(Path(args.s52_lookups) if args.s52_lookups else None),
s52_resource_contract_path=(
Path(args.s52_resource_contract)
if args.s52_resource_contract
else None
),
s52_point_table=args.s52_point_table,
s52_boundary_table=args.s52_boundary_table,
s52_mariner_mode=args.s52_mariner_mode,
@@ -306,6 +316,7 @@ def main() -> None:
"contract_unmatched_count": (result.get("style_sprite_contract") or {}).get("contract_unmatched_count"),
"sprite_missing_count": (result.get("style_sprite_contract") or {}).get("sprite_missing_count"),
},
"s52_resource_contract": result.get("s52_resource_contract"),
"s52_runtime_contract": result.get("s52_runtime_contract"),
"quantization_fallbacks": result.get("quantization_fallbacks"),
"quantization_omissions": result.get("quantization_omissions"),

View File

@@ -636,7 +636,10 @@ def resolve_obstrn04_instruction(
instruction = "SY(DANGER51)"
sounding = True
else:
if watlev in {1, 2}:
if watlev == -9:
instruction = "SY(DANGER01)"
sounding = True
elif watlev in {1, 2}:
instruction = "SY(LNDARE01)"
elif watlev == 3:
instruction = "SY(DANGER52)"
@@ -834,7 +837,7 @@ def resolve_slcons03_instruction(
instructions.append("AP(CROSSX01)")
if quapos_present and 2 <= quapos < 10:
instructions.append("LC(LOWACC01)")
instructions.append("LC(LOWACC21)")
return ";".join(instructions)
condtn = _int_attr(attrs, "CONDTN", 0)

View File

@@ -35,7 +35,7 @@ from .policy import (
PRODUCTION_PLAN_SCHEMA_V1 = "navsea.cm95.production-plan/v1"
PRODUCTION_PLAN_SCHEMA = "navsea.cm95.production-plan/v2"
SUPPORTED_PRODUCTION_PLAN_SCHEMAS = {PRODUCTION_PLAN_SCHEMA_V1, PRODUCTION_PLAN_SCHEMA}
PRODUCTION_PIPELINE_VERSION = "cm95pbf-production/2.1-s52-aoi-complete"
PRODUCTION_PIPELINE_VERSION = "cm95pbf-production/2.2-s52-resource-contract"
PRODUCTION_RUN_SCHEMA = "navsea.cm95.production-run/v1"
TILE_ENUMERATION_MODES = {"auto", "coverage", "aoi-complete"}
@@ -277,6 +277,7 @@ def create_production_plan(
profile: str = "delivery",
*,
s52_lookups_path: Path | None = None,
s52_resource_contract_path: Path | None = None,
s52_point_table: str = "Paper",
s52_boundary_table: str = "Plain",
s52_mariner_mode: str = "runtime",
@@ -302,6 +303,35 @@ def create_production_plan(
s52_path = Path(s52_lookups_path) if s52_lookups_path is not None else None
if s52_path is not None and not s52_path.exists():
raise RuntimeError(f"S-52 lookup file not found: {s52_path}")
resource_contract_path = (
Path(s52_resource_contract_path)
if s52_resource_contract_path is not None
else None
)
if s52_path is not None and resource_contract_path is None:
raise RuntimeError(
"S-52 production requires --s52-resource-contract; "
"resource availability must be frozen before production"
)
resource_contract_obj: dict[str, Any] | None = None
if resource_contract_path is not None:
if not resource_contract_path.exists():
raise RuntimeError(
f"S-52 resource contract not found: {resource_contract_path}"
)
resource_contract_obj = json.loads(
resource_contract_path.read_text(encoding="utf-8")
)
if resource_contract_obj.get("schema") != "navsea.cm95.s52-resource-contract/v1":
raise RuntimeError(
f"unsupported S-52 resource contract schema: "
f"{resource_contract_obj.get('schema')}"
)
if resource_contract_obj.get("status") != "PASS":
raise RuntimeError(
"S-52 resource contract is not PASS: "
f"{resource_contract_obj.get('failures')}"
)
aoi_bbox = None
aoi_box_m = None
@@ -417,6 +447,31 @@ def create_production_plan(
s52_contract = {
"enabled": s52_path is not None,
"lookups_sha256": sha256_file(s52_path) if s52_path is not None else None,
"resource_contract_sha256": (
sha256_file(resource_contract_path)
if resource_contract_path is not None
else None
),
"resource_contract_status": (
resource_contract_obj.get("status")
if resource_contract_obj is not None
else None
),
"resource_contract_reference": (
resource_contract_obj.get("reference")
if resource_contract_obj is not None
else None
),
"resource_defined_counts": (
resource_contract_obj.get("defined_resource_counts")
if resource_contract_obj is not None
else None
),
"resource_active_counts": (
resource_contract_obj.get("active_resource_counts")
if resource_contract_obj is not None
else None
),
"point_table": s52_point_table if s52_path is not None else None,
"boundary_table": s52_boundary_table if s52_path is not None else None,
"mariner_mode": s52_mariner_mode if s52_path is not None else None,
@@ -452,6 +507,9 @@ def create_production_plan(
"policy_path": str(policy_path),
"manifest_path": str(manifest_path),
"s52_lookups_path": str(s52_path) if s52_path is not None else None,
"s52_resource_contract_path": (
str(resource_contract_path) if resource_contract_path is not None else None
),
"source_stats": source_stats,
"job_count": len(job_rows),
}
@@ -637,6 +695,25 @@ def _validate_plan_inputs(db: DbConfig, plan: dict[str, Any]):
raise RuntimeError(f"S-52 lookup file missing: {s52_path}")
if sha256_file(s52_path) != str(s52.get("lookups_sha256") or ""):
raise RuntimeError("S-52 lookup file changed after plan creation")
resource_path_raw = plan.get("s52_resource_contract_path")
if s52.get("resource_contract_sha256"):
if not resource_path_raw:
raise RuntimeError("S-52 enabled plan missing s52_resource_contract_path")
resource_path = Path(resource_path_raw)
if not resource_path.exists():
raise RuntimeError(f"S-52 resource contract missing: {resource_path}")
if sha256_file(resource_path) != str(
s52.get("resource_contract_sha256") or ""
):
raise RuntimeError(
"S-52 resource contract changed after plan creation"
)
resource_obj = json.loads(resource_path.read_text(encoding="utf-8"))
if resource_obj.get("status") != "PASS":
raise RuntimeError(
"S-52 resource contract no longer PASS: "
f"{resource_obj.get('failures')}"
)
with _connect(db) as conn:
fp = dataset_fingerprint(conn, int(plan["dataset_id"]))
if fp != plan["dataset_fingerprint"]:

View File

@@ -496,7 +496,58 @@ def finalize_release(
else "FAIL"
)
s52_resource_contract = {
"status": "NOT_APPLICABLE",
"path": plan.get("s52_resource_contract_path"),
"sha256": None,
"expected_sha256": (plan.get("s52") or {}).get("resource_contract_sha256"),
"defined_resource_counts": None,
"active_resource_counts": None,
"failures": [],
}
expected_resource_sha = str(
(plan.get("s52") or {}).get("resource_contract_sha256") or ""
)
if expected_resource_sha:
resource_path_raw = plan.get("s52_resource_contract_path")
resource_failures: list[str] = []
resource_obj: dict[str, Any] = {}
if not resource_path_raw:
resource_failures.append("missing_resource_contract_path")
else:
resource_path = Path(resource_path_raw)
if not resource_path.exists():
resource_failures.append("resource_contract_file_missing")
else:
actual_sha = sha256_file(resource_path)
s52_resource_contract["sha256"] = actual_sha
if actual_sha != expected_resource_sha:
resource_failures.append("resource_contract_sha256_mismatch")
try:
resource_obj = json.loads(
resource_path.read_text(encoding="utf-8")
)
except Exception as exc:
resource_failures.append(
f"resource_contract_parse_error:{type(exc).__name__}"
)
else:
if resource_obj.get("status") != "PASS":
resource_failures.append("resource_contract_not_pass")
s52_resource_contract["defined_resource_counts"] = (
resource_obj.get("defined_resource_counts")
)
s52_resource_contract["active_resource_counts"] = (
resource_obj.get("active_resource_counts")
)
s52_resource_contract["failures"] = resource_failures
s52_resource_contract["status"] = (
"PASS" if not resource_failures else "FAIL"
)
failures = []
if expected_resource_sha and s52_resource_contract["status"] != "PASS":
failures.append("s52_resource_contract")
if tile_enumeration_mode == "aoi-complete":
if len(report_files) != int(plan["job_count"]):
failures.append("aoi_complete_report_count_mismatch")
@@ -537,6 +588,7 @@ def finalize_release(
"tile_footprint_contract": tile_footprint_contract,
"s52": plan.get("s52"),
"aoi": plan.get("aoi"),
"s52_resource_contract": s52_resource_contract,
"s52_runtime_contract": s52_runtime_contract,
"style_sha256": style_sha,
"sprite_json_sha256": sprite_sha,

View File

@@ -81,6 +81,8 @@ def _line_style_svg(
nominal_width = max(1.0, math.floor((dpi / 25.4) / 5.0))
current: tuple[float, float] | None = None
elements: list[str] = []
polygon_mode = False
polygon_circles: list[tuple[tuple[float, float], float, str]] = []
def cv(p: tuple[float, float]) -> tuple[float, float]:
return ((p[0] - ox) * unit_to_px, (p[1] - oy) * unit_to_px)
@@ -125,13 +127,30 @@ def _line_style_svg(
continue
cx, cy = cv(current)
color = _esc(pen_colors.get(pen_code, next(iter(pen_colors.values()))))
stroke = max(1.0, pen_width * nominal_width)
elements.append(
f'<circle cx="{cx:.3f}" cy="{cy:.3f}" r="{radius:.3f}" fill="none" '
f'stroke="{color}" stroke-width="{stroke:.3f}" />'
)
elif cmd in {"ST", "PM", "FP"}:
raise RuntimeError(f"unsupported HPGL {cmd} in line style {name}")
if polygon_mode:
polygon_circles.append(((cx, cy), radius, color))
else:
stroke = max(1.0, pen_width * nominal_width)
elements.append(
f'<circle cx="{cx:.3f}" cy="{cy:.3f}" r="{radius:.3f}" fill="none" '
f'stroke="{color}" stroke-width="{stroke:.3f}" />'
)
elif cmd == "ST":
# Symbol transparency command; PLNRTE03 uses ST0 before a filled circle.
continue
elif cmd == "PM":
if arg == "0":
polygon_mode = True
polygon_circles = []
elif arg == "2":
polygon_mode = False
elif cmd == "FP":
for (cx, cy), radius, color in polygon_circles:
elements.append(
f'<circle cx="{cx:.3f}" cy="{cy:.3f}" r="{radius:.3f}" fill="{color}" />'
)
polygon_circles = []
polygon_mode = False
else:
raise RuntimeError(f"unsupported HPGL {cmd} in line style {name}")

View File

@@ -4,10 +4,13 @@ from __future__ import annotations
import argparse
from collections import Counter
import json
import math
from pathlib import Path
import re
import shutil
import struct
import subprocess
import tempfile
import xml.etree.ElementTree as ET
from typing import Any
@@ -54,6 +57,25 @@ def _parse_bitmap(node: ET.Element) -> dict[str, int] | None:
}
def _parse_vector(node: ET.Element) -> dict[str, Any] | None:
vector = node.find("vector")
if vector is None:
return None
origin = vector.find("origin")
pivot = vector.find("pivot")
if origin is None or pivot is None:
return None
return {
"width": int(vector.attrib.get("width", "0")),
"height": int(vector.attrib.get("height", "0")),
"origin_x": float(origin.attrib["x"]),
"origin_y": float(origin.attrib["y"]),
"pivot_x": float(pivot.attrib["x"]),
"pivot_y": float(pivot.attrib["y"]),
"hpgl": _text(vector, "HPGL"),
}
def _parse_symbols(root: ET.Element) -> dict[str, dict[str, Any]]:
out: dict[str, dict[str, Any]] = {}
parent = root.find("symbols")
@@ -63,13 +85,25 @@ def _parse_symbols(root: ET.Element) -> dict[str, dict[str, Any]]:
name = _text(node, "name")
if not name:
continue
out[name] = {
candidate = {
"name": name,
"description": _text(node, "description"),
"definition": _text(node, "definition") or None,
"color_ref": _text(node, "color-ref") or None,
"bitmap": _parse_bitmap(node),
"vector": _parse_vector(node),
}
current = out.get(name)
if current is None:
out[name] = candidate
continue
# OpenCPN chartsymbols.xml contains duplicate vector/raster definitions
# for several sounding symbols. Prefer a real bitmap when available so
# the MapLibre sprite matches the raster symbol library exactly.
if current.get("bitmap") is None and candidate.get("bitmap") is not None:
out[name] = candidate
elif current.get("bitmap") is None and current.get("vector") is None:
out[name] = candidate
return out
@@ -151,6 +185,140 @@ def _parse_color_tables(root: ET.Element) -> dict[str, dict[str, Any]]:
return out
def _pen_colors(color_ref: str | None, colors: dict[str, Any]) -> dict[str, str]:
compact = "".join(str(color_ref or "").split())
out: dict[str, str] = {}
for i in range(0, len(compact), 6):
token = compact[i : i + 6]
if len(token) != 6:
continue
rgb = colors.get(token[1:])
if isinstance(rgb, dict):
out[token[0]] = (
f"#{int(rgb['r']):02x}{int(rgb['g']):02x}{int(rgb['b']):02x}"
)
return out or {"A": "#000000"}
def _hpgl_points(text: str) -> list[tuple[float, float]]:
nums = [float(x) for x in text.split(",") if x.strip()]
return [(nums[i], nums[i + 1]) for i in range(0, len(nums) - 1, 2)]
def _vector_symbol_svg(
symbol: dict[str, Any],
colors: dict[str, Any],
*,
dpi: float = 96.0,
) -> tuple[str, dict[str, float]]:
vector = symbol.get("vector")
if not isinstance(vector, dict):
raise RuntimeError(f"symbol {symbol.get('name')} has no vector definition")
unit = (dpi / 25.4) / 100.0
pen_colors = _pen_colors(symbol.get("color_ref"), colors)
pen = next(iter(pen_colors))
pen_width = 1.0
current: tuple[float, float] | None = None
primitives: list[tuple[str, Any]] = []
minx = miny = float("inf")
maxx = maxy = float("-inf")
def include(x: float, y: float) -> None:
nonlocal minx, miny, maxx, maxy
minx = min(minx, x)
miny = min(miny, y)
maxx = max(maxx, x)
maxy = max(maxy, y)
for raw in str(vector.get("hpgl") or "").split(";"):
raw = raw.strip()
if not raw:
continue
cmd, arg = raw[:2], raw[2:]
if cmd == "SP" and arg:
pen = arg[0]
elif cmd == "SW":
try:
pen_width = max(1.0, float(arg))
except ValueError:
pen_width = 1.0
elif cmd == "PU":
pts = _hpgl_points(arg)
if pts:
current = pts[-1]
include(*current)
elif cmd == "PD":
pts = _hpgl_points(arg)
for target in pts:
if current is None:
current = target
include(*current)
continue
primitives.append(
("line", (current, target, pen_colors.get(pen, "#000000"), pen_width))
)
include(*current)
include(*target)
current = target
elif cmd == "CI" and current is not None:
radius = float(arg)
primitives.append(
("circle", (current, radius, pen_colors.get(pen, "#000000"), pen_width))
)
include(current[0] - radius, current[1] - radius)
include(current[0] + radius, current[1] + radius)
else:
raise RuntimeError(
f"unsupported HPGL {cmd} in vector symbol {symbol.get('name')}"
)
if not primitives or not math.isfinite(minx) or not math.isfinite(miny):
raise RuntimeError(f"vector symbol {symbol.get('name')} rendered no primitives")
pad = 1.5
width = max(1, int(math.ceil((maxx - minx) * unit + 2 * pad)))
height = max(1, int(math.ceil((maxy - miny) * unit + 2 * pad)))
def cv(point: tuple[float, float]) -> tuple[float, float]:
return (
(point[0] - minx) * unit + pad,
(point[1] - miny) * unit + pad,
)
parts: list[str] = []
for kind, data in primitives:
if kind == "line":
p1, p2, color, width_hpgl = data
x1, y1 = cv(p1)
x2, y2 = cv(p2)
stroke = max(1.0, float(width_hpgl))
parts.append(
f'<line x1="{x1:.3f}" y1="{y1:.3f}" x2="{x2:.3f}" y2="{y2:.3f}" '
f'stroke="{color}" stroke-width="{stroke:.3f}" stroke-linecap="round" />'
)
else:
center, radius, color, width_hpgl = data
cx, cy = cv(center)
r = max(0.5, float(radius) * unit)
stroke = max(1.0, float(width_hpgl))
parts.append(
f'<circle cx="{cx:.3f}" cy="{cy:.3f}" r="{r:.3f}" fill="none" '
f'stroke="{color}" stroke-width="{stroke:.3f}" />'
)
pivot = cv((float(vector["pivot_x"]), float(vector["pivot_y"])))
svg = (
f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" '
f'viewBox="0 0 {width} {height}">' + "".join(parts) + "</svg>"
)
return svg, {
"width": width,
"height": height,
"pivot_x": pivot[0],
"pivot_y": pivot[1],
}
def _sprite_entry(bitmap: dict[str, int]) -> dict[str, int]:
return {
"x": bitmap["x"],
@@ -270,7 +438,7 @@ def main() -> int:
direct_patterns = sorted({name for row in lookups for name in row["ap"]})
csp_names = sorted({name for row in lookups for name in row["cs"]})
sprite_symbols = (
sorted(name for name, symbol in symbols.items() if symbol.get("bitmap") is not None)
sorted(symbols)
if sprite_scope == "all"
else direct_symbols
)
@@ -283,6 +451,7 @@ def main() -> int:
pattern_metadata: dict[str, dict[str, Any]] = {}
missing_symbols: list[str] = []
vector_only_symbols: list[str] = []
vector_render_names: list[str] = []
out_of_bounds_symbols: list[str] = []
for name in sprite_symbols:
@@ -293,6 +462,8 @@ def main() -> int:
bitmap = symbol.get("bitmap")
if bitmap is None:
vector_only_symbols.append(name)
if symbol.get("vector") is not None:
vector_render_names.append(name)
continue
if (
bitmap["x"] < 0
@@ -323,13 +494,9 @@ def main() -> int:
continue
pattern_metadata[name] = dict(pattern)
collisions = sorted(set(metadata) & set(pattern_metadata))
if collisions:
raise SystemExit(f"sprite symbol/pattern name collision: {collisions}")
output = args.output
output.mkdir(parents=True, exist_ok=True)
shutil.copy2(atlas_path, output / "sprite.png")
sprite_json = {
name: _sprite_entry(metadata[name]["bitmap"])
for name in sorted(metadata)
@@ -338,6 +505,66 @@ def main() -> int:
name: _sprite_entry(pattern_metadata[name]["bitmap"])
for name in sorted(pattern_metadata)
})
if vector_render_names:
if shutil.which("convert") is None:
raise SystemExit("ImageMagick convert is required for vector-only S-52 symbols")
colors = color_tables[palette].get("colors") or {}
with tempfile.TemporaryDirectory(prefix="cm95-s52-symbol-") as td:
tmp = Path(td)
pngs: list[Path] = []
y = atlas_height
for index, name in enumerate(vector_render_names):
symbol = symbols[name]
svg, vector_meta = _vector_symbol_svg(symbol, colors)
svg_path = tmp / f"{index:03d}-{name}.svg"
png_path = tmp / f"{index:03d}-{name}.png"
svg_path.write_text(svg, encoding="utf-8")
subprocess.run(
["convert", "-background", "none", str(svg_path), str(png_path)],
check=True,
)
width, height = _png_size(png_path)
bitmap = {
"x": 0,
"y": y,
"width": width,
"height": height,
"pivot_x": int(round(vector_meta["pivot_x"])),
"pivot_y": int(round(vector_meta["pivot_y"])),
}
materialized = {
**symbol,
"bitmap": bitmap,
"maplibre_icon_offset": _icon_offset(bitmap),
"materialized_from_vector": True,
}
metadata[name] = materialized
sprite_json[name] = _sprite_entry(bitmap)
y += height
pngs.append(png_path)
subprocess.run(
[
"convert",
str(atlas_path),
*[str(p) for p in pngs],
"-background",
"none",
"-gravity",
"west",
"-append",
str(output / "sprite.png"),
],
check=True,
)
else:
shutil.copy2(atlas_path, output / "sprite.png")
collisions = sorted(set(metadata) & set(pattern_metadata))
if collisions:
raise SystemExit(f"sprite symbol/pattern name collision: {collisions}")
(output / "sprite.json").write_text(
json.dumps(sprite_json, ensure_ascii=False, indent=2), encoding="utf-8"
)

View File

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

View File

@@ -287,6 +287,110 @@ def _scan_pbf_usage(path: Path) -> dict[str, Any]:
}
def _merge_pbf_usages(usages: list[dict[str, Any]], label: str) -> dict[str, Any]:
global_values: dict[str, set[str]] = {field: set() for field in USAGE_VALUE_FIELDS}
layers: dict[str, dict[str, Any]] = {}
geometry_ignored: Counter[tuple[str, str, str]] = Counter()
render_kind_counts: dict[str, Counter[tuple[str, str, str]]] = {}
for usage in usages:
for field, values in (usage.get("global_values") or {}).items():
global_values.setdefault(field, set()).update(str(value) for value in values)
for layer_name, row in (usage.get("layers") or {}).items():
merged = layers.setdefault(
layer_name,
{
"feature_count": 0,
"fields_present": set(),
"field_values": {},
"field_priorities": {},
},
)
merged["feature_count"] += int(row.get("feature_count") or 0)
merged["fields_present"].update(row.get("fields_present") or [])
for field, values in (row.get("field_values") or {}).items():
merged["field_values"].setdefault(field, set()).update(
str(value) for value in values
)
for field, values in (row.get("field_priorities") or {}).items():
merged["field_priorities"].setdefault(field, set()).update(
str(value) for value in values
)
for row in usage.get("geometry_ignored") or []:
key = (
str(row.get("kind") or ""),
str(row.get("source_layer") or ""),
str(row.get("geometry_type") or ""),
)
geometry_ignored[key] += int(row.get("feature_count") or 0)
for kind, rows in (usage.get("render_kind_counts") or {}).items():
counter = render_kind_counts.setdefault(kind, Counter())
for row in rows:
key = (
str(row.get("source_layer") or ""),
str(row.get("priority") or ""),
str(row.get("category") or ""),
)
counter[key] += int(row.get("feature_count") or 0)
normalized_layers = {}
for layer_name, row in layers.items():
normalized_layers[layer_name] = {
"feature_count": int(row["feature_count"]),
"fields_present": sorted(row["fields_present"]),
"field_values": {
key: sorted(values)
for key, values in row["field_values"].items()
if values
},
"field_priorities": {
key: sorted(values)
for key, values in row["field_priorities"].items()
if values
},
}
return {
"pbf": label,
"pbf_count": len(usages),
"layers": normalized_layers,
"global_values": {
key: sorted(values) for key, values in global_values.items() if values
},
"geometry_ignored": [
{
"kind": kind,
"source_layer": layer,
"geometry_type": geometry,
"feature_count": count,
}
for (kind, layer, geometry), count in sorted(geometry_ignored.items())
],
"render_kind_counts": {
kind: [
{
"source_layer": layer,
"priority": priority,
"category": category,
"feature_count": count,
}
for (layer, priority, category), count in sorted(counter.items())
]
for kind, counter in render_kind_counts.items()
if counter
},
}
def _scan_pbf_directory(path: Path) -> dict[str, Any]:
files = sorted(path.rglob("*.pbf"))
if not files:
raise SystemExit(f"--usage-pbf-dir contains no PBF files: {path}")
usages = [_scan_pbf_usage(file) for file in files]
return _merge_pbf_usages(usages, str(path))
def _field_used(
usage: dict[str, Any] | None, layer_name: str, field: str, priority: str
) -> bool:
@@ -318,6 +422,7 @@ def _compile_style(
line_metadata: dict[str, dict[str, Any]],
sprite_index: dict[str, Any],
lookup_ir: list[dict[str, Any]],
resource_contract: dict[str, Any] | None = None,
usage: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], dict[str, Any]]:
source = dict(config["source"])
@@ -331,6 +436,10 @@ def _compile_style(
background_name = str(config.get("background_color", "DEPDW"))
background = _rgb_hex(colors.get(background_name, {"r": 212, "g": 234, "b": 238}))
usage_values = dict((usage or {}).get("global_values") or {})
upstream_undefined = dict((resource_contract or {}).get("upstream_undefined_references") or {})
undefined_symbols = set(str(x) for x in upstream_undefined.get("SY") or [])
undefined_patterns = set(str(x) for x in upstream_undefined.get("AP") or [])
undefined_lines = set(str(x) for x in upstream_undefined.get("LC") or [])
available_symbol_names = sorted(set(symbol_metadata) & set(sprite_index))
symbol_names = list(available_symbol_names)
@@ -367,6 +476,14 @@ def _compile_style(
used = set(usage_values.get("s52_line_complex_code") or [])
lc_mapping = {name: value for name, value in lc_mapping.items() if name in used}
fallback_symbol_name = "QUESMRK1" if "QUESMRK1" in available_symbol_names else None
fallback_symbol_offset = (
symbol_metadata.get("QUESMRK1", {}).get("maplibre_icon_offset", [0, 0])
if fallback_symbol_name
else [0, 0]
)
fallback_ap_key = available_ap_mapping.get("QUESMRK1")
text_offsets = _text_offset_pairs(lookup_ir)
lookup_opcode_priorities: dict[str, set[str]] = {
key: set() for key in (*STATIC_OPCODES, "TX/TE")
@@ -495,6 +612,43 @@ def _compile_style(
},
)
compiled_placements["AP"].add((layer_name, priority))
if (
"fill" in caps
and fallback_ap_key
and priority in opcode_priorities["AP"]
and _field_used(usage, layer_name, "s52_area_pattern_code", priority)
):
_add_layer(
layers,
compiled_counts,
"AP",
{
"id": f"{prefix}-ap-fallback-{layer_id}",
"type": "fill",
"source": source_id,
"source-layer": layer_name,
"filter": [
"all",
*_base_filter(
priority=priority,
categories=categories,
required_field="s52_area_pattern_code",
)[1:],
[
"!",
_supported_filter(
"s52_area_pattern_code",
list(available_ap_mapping),
),
],
],
"paint": {
"fill-pattern": fallback_ap_key,
"fill-opacity": float(config.get("area_pattern_opacity", 0.85)),
},
},
)
compiled_placements["AP"].add((layer_name, priority))
if (
"line" in caps
@@ -562,6 +716,49 @@ def _compile_style(
},
)
compiled_placements["LC"].add((layer_name, priority))
if (
"line" in caps
and fallback_symbol_name
and priority in opcode_priorities["LC"]
and _field_used(usage, layer_name, "s52_line_complex_code", priority)
):
_add_layer(
layers,
compiled_counts,
"LC",
{
"id": f"{prefix}-lc-fallback-{layer_id}",
"type": "symbol",
"source": source_id,
"source-layer": layer_name,
"filter": [
"all",
*_base_filter(
priority=priority,
categories=categories,
required_field="s52_line_complex_code",
)[1:],
[
"!",
_supported_filter(
"s52_line_complex_code",
list(available_lc_mapping),
),
],
],
"layout": {
"symbol-placement": "line",
"symbol-spacing": 80,
"icon-image": fallback_symbol_name,
"icon-anchor": "center",
"icon-offset": fallback_symbol_offset,
"icon-size": 1,
"icon-allow-overlap": True,
"icon-ignore-placement": True,
},
},
)
compiled_placements["LC"].add((layer_name, priority))
if (
"symbol" in caps
@@ -611,6 +808,60 @@ def _compile_style(
},
)
compiled_placements["SY"].add((layer_name, priority))
if (
"symbol" in caps
and fallback_symbol_name
and priority in opcode_priorities["SY"]
and _field_used(usage, layer_name, "s52_symbol_code", priority)
):
_add_layer(
layers,
compiled_counts,
"SY",
{
"id": f"{prefix}-sy-fallback-{layer_id}",
"type": "symbol",
"source": source_id,
"source-layer": layer_name,
"filter": [
"all",
*_base_filter(
priority=priority,
categories=categories,
required_field="s52_symbol_code",
forbidden_field="s52_sounding_symbol_sequence",
)[1:],
[
"!",
_supported_filter(
"s52_symbol_code",
available_symbol_names,
),
],
[
"!",
[
"all",
["has", "s52_light_symbol_code"],
[
"==",
["get", "s52_symbol_code"],
["get", "s52_light_symbol_code"],
],
],
],
],
"layout": {
"icon-image": fallback_symbol_name,
"icon-anchor": "center",
"icon-offset": fallback_symbol_offset,
"icon-size": 1,
"icon-allow-overlap": True,
"icon-ignore-placement": True,
},
},
)
compiled_placements["SY"].add((layer_name, priority))
if (
"symbol" in caps
@@ -649,6 +900,46 @@ def _compile_style(
},
)
compiled_placements["LIGHT_SY"].add((layer_name, priority))
if (
"symbol" in caps
and fallback_symbol_name
and _field_used(usage, layer_name, "s52_light_symbol_code", priority)
):
_add_layer(
layers,
compiled_counts,
"LIGHTS05-SY",
{
"id": f"{prefix}-lights05-sy-fallback-{layer_id}",
"type": "symbol",
"source": source_id,
"source-layer": layer_name,
"filter": [
"all",
*_base_filter(
priority=priority,
categories=categories,
required_field="s52_light_symbol_code",
)[1:],
[
"!",
_supported_filter(
"s52_light_symbol_code",
available_symbol_names,
),
],
],
"layout": {
"icon-image": fallback_symbol_name,
"icon-anchor": "center",
"icon-offset": fallback_symbol_offset,
"icon-size": 1,
"icon-allow-overlap": True,
"icon-ignore-placement": True,
},
},
)
compiled_placements["LIGHT_SY"].add((layer_name, priority))
if (
"text" in caps
@@ -772,6 +1063,21 @@ def _compile_style(
"layers": layers,
}
hard_gate_failures: list[dict[str, Any]] = []
if resource_contract is not None and resource_contract.get("status") != "PASS":
hard_gate_failures.append({
"code": "S52_RESOURCE_CONTRACT_NOT_PASS",
"values": list(resource_contract.get("failures") or []),
})
if (undefined_symbols or undefined_lines) and fallback_symbol_name is None:
hard_gate_failures.append({
"code": "S52_UNDEFINED_FALLBACK_SYMBOL_MISSING",
"values": ["QUESMRK1"],
})
if undefined_patterns and fallback_ap_key is None:
hard_gate_failures.append({
"code": "S52_UNDEFINED_FALLBACK_PATTERN_MISSING",
"values": ["QUESMRK1"],
})
if usage is not None:
contract_names = {str(row["name"]) for row in source_layers}
actual_names = set((usage.get("layers") or {}).keys())
@@ -783,7 +1089,9 @@ def _compile_style(
})
used_symbols = set(usage_values.get("s52_symbol_code") or [])
missing_used_symbols = sorted(used_symbols - set(available_symbol_names))
missing_used_symbols = sorted(
(used_symbols - set(available_symbol_names)) - undefined_symbols
)
if missing_used_symbols:
hard_gate_failures.append({
"code": "USED_SYMBOL_RESOURCE_GAP",
@@ -792,7 +1100,7 @@ def _compile_style(
used_light_symbols = set(usage_values.get("s52_light_symbol_code") or [])
missing_used_light_symbols = sorted(
used_light_symbols - set(available_symbol_names)
(used_light_symbols - set(available_symbol_names)) - undefined_symbols
)
if missing_used_light_symbols:
hard_gate_failures.append({
@@ -801,7 +1109,9 @@ def _compile_style(
})
used_patterns = set(usage_values.get("s52_area_pattern_code") or [])
missing_used_patterns = sorted(used_patterns - set(available_ap_mapping))
missing_used_patterns = sorted(
(used_patterns - set(available_ap_mapping)) - undefined_patterns
)
if missing_used_patterns:
hard_gate_failures.append({
"code": "USED_PATTERN_RESOURCE_GAP",
@@ -809,7 +1119,9 @@ def _compile_style(
})
used_complex_lines = set(usage_values.get("s52_line_complex_code") or [])
missing_used_complex_lines = sorted(used_complex_lines - set(available_lc_mapping))
missing_used_complex_lines = sorted(
(used_complex_lines - set(available_lc_mapping)) - undefined_lines
)
if missing_used_complex_lines:
hard_gate_failures.append({
"code": "USED_COMPLEX_LINE_RESOURCE_GAP",
@@ -891,6 +1203,12 @@ def _compile_style(
"schema": "navsea.cm95.s52-style-compiler-audit/v1",
"hard_gate_failures": hard_gate_failures,
"palette": config.get("palette", "DAY_BRIGHT"),
"resource_contract_status": (
resource_contract.get("status") if resource_contract is not None else None
),
"resource_contract_reference": (
resource_contract.get("reference") if resource_contract is not None else None
),
"source_id": source_id,
"source_layer_count": len(source_layers),
"display_categories": categories,
@@ -934,6 +1252,7 @@ def _compile_style(
if usage is None
else {
"pbf": usage.get("pbf"),
"pbf_count": int(usage.get("pbf_count") or 1),
"source_layer_count": len(usage.get("layers") or {}),
"feature_count": sum(
int(row.get("feature_count") or 0)
@@ -968,7 +1287,14 @@ def main() -> int:
type=Path,
help="Optionally scope generated layers/resources to fields actually present in one PBF tile.",
)
parser.add_argument(
"--usage-pbf-dir",
type=Path,
help="Scope generated layers/resources to the aggregate usage of all *.pbf files under a release/AOI directory.",
)
args = parser.parse_args()
if args.usage_pbf is not None and args.usage_pbf_dir is not None:
raise SystemExit("--usage-pbf and --usage-pbf-dir are mutually exclusive")
project_root = Path(__file__).resolve().parents[1]
config = _load_json(args.config)
@@ -981,6 +1307,7 @@ def main() -> int:
"line_metadata",
"sprite_index",
"lookup_ir",
"resource_contract",
)
missing = [name for name in required if not resources.get(name)]
if missing:
@@ -993,11 +1320,12 @@ def main() -> int:
colors = dict(color_tables[palette].get("colors") or {})
if not colors:
raise SystemExit(f"palette {palette} has no colors")
usage = (
_scan_pbf_usage(_resolve_path(project_root, args.usage_pbf))
if args.usage_pbf is not None
else None
)
if args.usage_pbf_dir is not None:
usage = _scan_pbf_directory(_resolve_path(project_root, args.usage_pbf_dir))
elif args.usage_pbf is not None:
usage = _scan_pbf_usage(_resolve_path(project_root, args.usage_pbf))
else:
usage = None
style, audit = _compile_style(
config=config,
@@ -1008,6 +1336,9 @@ def main() -> int:
line_metadata=_load_json(_resolve_path(project_root, resources["line_metadata"])),
sprite_index=_load_json(_resolve_path(project_root, resources["sprite_index"])),
lookup_ir=_load_json(_resolve_path(project_root, resources["lookup_ir"])),
resource_contract=_load_json(
_resolve_path(project_root, resources["resource_contract"])
),
usage=usage,
)

View File

@@ -155,7 +155,12 @@ def main() -> int:
ap.add_argument("--chartsymbols", required=True, type=Path)
ap.add_argument("--base-sprite-png", required=True, type=Path)
ap.add_argument("--base-sprite-json", required=True, type=Path)
ap.add_argument("--usage-audit", required=True, type=Path)
ap.add_argument("--usage-audit", type=Path)
ap.add_argument(
"--all-patterns",
action="store_true",
help="Compile every vector S-52 pattern in chartsymbols.xml; production resource bundles should use this.",
)
ap.add_argument(
"--usage-pbf",
type=Path,
@@ -174,7 +179,11 @@ def main() -> int:
if patterns is None:
raise SystemExit("missing patterns")
by_name = {_text(n, "name"): n for n in patterns.findall("pattern")}
usage = json.loads(args.usage_audit.read_text(encoding="utf-8"))
usage = (
json.loads(args.usage_audit.read_text(encoding="utf-8"))
if args.usage_audit is not None
else {}
)
lookup_used = {str(row[0]) for row in usage.get("pattern_resource_counts", [])}
pbf_used: set[str] = set()
if args.usage_pbf is not None:
@@ -191,13 +200,12 @@ def main() -> int:
if value not in (None, ""):
pbf_used.add(str(value))
used = sorted(lookup_used | pbf_used)
selected = sorted(by_name) if args.all_patterns else used
names = [
name for name in used
if name in by_name
and by_name[name].find("vector") is not None
and by_name[name].find("bitmap") is None
name for name in selected
if name in by_name and by_name[name].find("vector") is not None
]
missing = [name for name in used if name not in by_name]
missing = [name for name in selected if name not in by_name]
if missing:
raise SystemExit(f"missing patterns: {missing}")