Files
CM95/tools/cm95_s52_style_compiler.py
2026-09-19 08:32:51 +08:00

1362 lines
55 KiB
Python
Executable File

#!/usr/bin/env python3
from __future__ import annotations
import argparse
from collections import Counter
import json
from pathlib import Path
import re
from typing import Any, Iterable
DISPLAY_PRIORITY_ORDER = (
"No data",
"Group 1",
"Area 1",
"Area 2",
"Point Symbol",
"Line Symbol",
"Area Symbol",
"Routing",
"Hazards",
"Mariners",
)
STATIC_OPCODES = ("SY", "AC", "AP", "LS", "LC", "TX", "TE")
def _load_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def _resolve_path(root: Path, value: str | Path) -> Path:
path = Path(value)
return path if path.is_absolute() else root / path
def _rgb_hex(value: dict[str, Any]) -> str:
return "#{:02x}{:02x}{:02x}".format(
int(value["r"]), int(value["g"]), int(value["b"])
)
def _safe_id(value: str) -> str:
return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") or "layer"
def _color_expression(
field: str, colors: dict[str, dict[str, Any]], fallback: str
) -> list[Any]:
expression: list[Any] = ["match", ["get", field]]
for name in sorted(colors):
expression.extend([name, _rgb_hex(colors[name])])
expression.append(fallback)
return expression
def _mapped_expression(field: str, mapping: dict[str, str], fallback: str) -> list[Any]:
expression: list[Any] = ["match", ["get", field]]
for name in sorted(mapping):
expression.extend([name, mapping[name]])
expression.append(fallback)
return expression
def _supported_filter(field: str, values: Iterable[str]) -> list[Any]:
choices = sorted(set(values))
return ["match", ["get", field], choices, True, False]
def _base_filter(
*,
priority: str,
categories: list[str],
required_field: str,
supported_values: list[str] | None = None,
forbidden_field: str | None = None,
) -> list[Any]:
clauses: list[Any] = [
["==", ["get", "s52_display_priority"], priority],
["match", ["get", "s52_display_category"], categories, True, False],
["has", required_field],
]
if supported_values is not None:
clauses.append(_supported_filter(required_field, supported_values))
if forbidden_field is not None:
clauses.append(["!", ["has", forbidden_field]])
return ["all", *clauses]
def _symbol_offset_expression(field: str, metadata: dict[str, dict[str, Any]]) -> list[Any]:
expression: list[Any] = ["match", ["get", field]]
for name in sorted(metadata):
offset = metadata[name].get("maplibre_icon_offset", [0, 0])
expression.extend([name, ["literal", offset]])
expression.append(["literal", [0, 0]])
return expression
def _text_offset_pairs(lookup_ir: list[dict[str, Any]]) -> list[tuple[str, str]]:
pairs: set[tuple[str, str]] = set()
for row in lookup_ir:
for rule in row.get("rules") or []:
args = rule.get("args") or []
if rule.get("opcode") == "TX" and len(args) >= 9:
layout = args[1:9]
elif rule.get("opcode") == "TE" and len(args) >= 10:
layout = args[2:10]
else:
continue
xoff, yoff = str(layout[4]).strip(), str(layout[5]).strip()
try:
float(xoff)
float(yoff)
except ValueError:
continue
pairs.add((xoff, yoff))
return sorted(pairs, key=lambda pair: (float(pair[0]), float(pair[1])))
def _text_offset_expression(slot: int, pairs: list[tuple[str, str]]) -> list[Any]:
key = [
"concat",
["to-string", ["get", f"s52_text{slot}_xoff"]],
",",
["to-string", ["get", f"s52_text{slot}_yoff"]],
]
expression: list[Any] = ["match", key]
for xoff, yoff in pairs:
expression.extend(
[f"{xoff},{yoff}", ["literal", [float(xoff), float(yoff)]]]
)
expression.append(["literal", [0.0, 0.0]])
return expression
def _text_anchor_expression(slot: int) -> list[Any]:
key = [
"concat",
["to-string", ["get", f"s52_text{slot}_hjust"]],
["to-string", ["get", f"s52_text{slot}_vjust"]],
]
return [
"match", key,
"11", "bottom",
"12", "center",
"13", "top",
"21", "bottom-right",
"22", "right",
"23", "top-right",
"31", "bottom-left",
"32", "left",
"33", "top-left",
"left",
]
def _layer_contract(config: dict[str, Any]) -> list[dict[str, Any]]:
rows = list(config.get("source_layers") or [])
if not rows:
raise SystemExit("config.source_layers must not be empty")
for row in rows:
if not row.get("name"):
raise SystemExit(f"source layer missing name: {row!r}")
return rows
USAGE_VALUE_FIELDS = (
"s52_symbol_code",
"s52_light_symbol_code",
"s52_area_color",
"s52_area_pattern_code",
"s52_line_color",
"s52_line_complex_code",
"s52_text1_color",
"s52_text2_color",
"s52_text3_color",
)
USAGE_PRESENCE_FIELDS = (
*USAGE_VALUE_FIELDS,
"s52_text_literal",
"s52_light_label_text",
"s52_light_rotation_deg",
"s52_text1_value",
"s52_text2_value",
"s52_text3_value",
)
def _scan_pbf_usage(path: Path) -> dict[str, Any]:
try:
import mapbox_vector_tile
except ImportError as exc:
raise SystemExit(
"--usage-pbf requires mapbox_vector_tile; run with the project .venv Python"
) from exc
decoded = mapbox_vector_tile.decode(path.read_bytes())
global_values: dict[str, set[str]] = {field: set() for field in USAGE_VALUE_FIELDS}
render_kind_counts: dict[str, Counter[tuple[str, str, str]]] = {
key: Counter()
for key in ("AC", "AP", "LS", "LC", "SY", "LIGHT_SY", "TXTE", "LIGHT_TX", "RUNTIME_SOUNDG", "RUNTIME_CA")
}
geometry_ignored: Counter[tuple[str, str, str]] = Counter()
layers: dict[str, Any] = {}
for layer_name, layer in decoded.items():
field_values: dict[str, set[str]] = {field: set() for field in USAGE_VALUE_FIELDS}
field_priorities: dict[str, set[str]] = {field: set() for field in USAGE_PRESENCE_FIELDS}
present: set[str] = set()
features = list(layer.get("features") or [])
for feature in features:
props = dict(feature.get("properties") or {})
priority = str(props.get("s52_display_priority") or "")
category = str(props.get("s52_display_category") or "")
placement = (layer_name, priority, category)
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"}
if "s52_area_color" in props:
if area_geometry:
render_kind_counts["AC"][placement] += 1
else:
geometry_ignored[("AC", layer_name, geometry_type)] += 1
if "s52_area_pattern_code" in props:
if area_geometry:
render_kind_counts["AP"][placement] += 1
else:
geometry_ignored[("AP", layer_name, geometry_type)] += 1
if "s52_line_color" in props:
if line_geometry:
render_kind_counts["LS"][placement] += 1
else:
geometry_ignored[("LS", layer_name, geometry_type)] += 1
if "s52_line_complex_code" in props:
if line_geometry:
render_kind_counts["LC"][placement] += 1
else:
geometry_ignored[("LC", layer_name, geometry_type)] += 1
if "s52_symbol_code" in props:
if props.get("s52_sounding_symbol_sequence"):
render_kind_counts["RUNTIME_SOUNDG"][placement] += 1
elif props.get("s52_light_symbol_code") == props.get("s52_symbol_code"):
# LIGHTS05 materialization may intentionally mirror the flare into
# s52_symbol_code. Count/render that pure LIGHTS symbol only once
# through LIGHT_SY; beacon/tower body symbols remain separate.
pass
else:
render_kind_counts["SY"][placement] += 1
if "s52_light_symbol_code" in props:
render_kind_counts["LIGHT_SY"][placement] += 1
if any(f"s52_text{slot}_value" in props for slot in (1, 2, 3)):
render_kind_counts["TXTE"][placement] += 1
if props.get("s52_csp_resolved") == "LIGHTS05" and "s52_text_literal" in props:
render_kind_counts["LIGHT_TX"][placement] += 1
if "s52_arc_color" in props:
render_kind_counts["RUNTIME_CA"][placement] += 1
for field in USAGE_PRESENCE_FIELDS:
if field not in props:
continue
present.add(field)
field_priorities[field].add(priority)
if field in field_values:
value = str(props[field])
field_values[field].add(value)
global_values[field].add(value)
layers[layer_name] = {
"feature_count": len(features),
"fields_present": sorted(present),
"field_values": {k: sorted(v) for k, v in field_values.items() if v},
"field_priorities": {k: sorted(v) for k, v in field_priorities.items() if v},
}
return {
"pbf": str(path),
"layers": layers,
"global_values": {k: sorted(v) for k, v in global_values.items() if v},
"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 _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:
if usage is None:
return True
layer = (usage.get("layers") or {}).get(layer_name)
if not layer or field not in set(layer.get("fields_present") or []):
return False
return priority in set((layer.get("field_priorities") or {}).get(field) or [])
def _add_layer(
layers: list[dict[str, Any]],
counts: Counter[str],
opcode: str,
layer: dict[str, Any],
) -> None:
layers.append(layer)
counts[opcode] += 1
def _compile_style(
*,
config: dict[str, Any],
colors: dict[str, dict[str, Any]],
symbol_metadata: dict[str, dict[str, Any]],
bitmap_pattern_metadata: dict[str, dict[str, Any]],
vector_pattern_metadata: dict[str, dict[str, Any]],
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"])
source_id = str(source.pop("id"))
sprite_url = str(config["sprite"])
categories = [str(x) for x in config.get("display_categories", [])]
if not categories:
categories = ["Displaybase", "Standard", "Other"]
fallback_name = str(config.get("fallback_color", "NODTA"))
fallback = _rgb_hex(colors.get(fallback_name, {"r": 163, "g": 180, "b": 183}))
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)
if usage is not None:
symbol_names = sorted(set(symbol_names) & set(usage_values.get("s52_symbol_code") or []))
symbol_offsets = {name: symbol_metadata[name] for name in symbol_names}
light_symbol_names = list(available_symbol_names)
if usage is not None:
light_symbol_names = sorted(
set(light_symbol_names) & set(usage_values.get("s52_light_symbol_code") or [])
)
light_symbol_offsets = {name: symbol_metadata[name] for name in light_symbol_names}
ap_mapping: dict[str, str] = {}
for name in bitmap_pattern_metadata:
if name in sprite_index:
ap_mapping[name] = name
for name, meta in vector_pattern_metadata.items():
sprite_key = str(meta.get("sprite_key") or "")
if sprite_key and sprite_key in sprite_index:
ap_mapping[name] = sprite_key
available_ap_mapping = dict(ap_mapping)
if usage is not None:
used = set(usage_values.get("s52_area_pattern_code") or [])
ap_mapping = {name: value for name, value in ap_mapping.items() if name in used}
lc_mapping: dict[str, str] = {}
for name, meta in line_metadata.items():
sprite_key = str(meta.get("sprite_key") or "")
if sprite_key and sprite_key in sprite_index:
lc_mapping[name] = sprite_key
available_lc_mapping = dict(lc_mapping)
if usage is not None:
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")
}
for row in lookup_ir:
priority = str(row.get("display_priority") or "")
for rule in row.get("rules") or []:
opcode = str(rule.get("opcode") or "")
if opcode in {"TX", "TE"}:
lookup_opcode_priorities["TX/TE"].add(priority)
elif opcode in lookup_opcode_priorities:
lookup_opcode_priorities[opcode].add(priority)
materialized_priority_fields = {
"AC": ("s52_area_color",),
"AP": ("s52_area_pattern_code",),
"LS": ("s52_line_color",),
"LC": ("s52_line_complex_code",),
"SY": ("s52_symbol_code",),
"TX/TE": ("s52_text1_value", "s52_text2_value", "s52_text3_value"),
}
materialized_opcode_priorities: dict[str, set[str]] = {
key: set() for key in lookup_opcode_priorities
}
if usage is not None:
for layer in (usage.get("layers") or {}).values():
field_priorities = layer.get("field_priorities") or {}
for opcode, fields in materialized_priority_fields.items():
for field in fields:
materialized_opcode_priorities[opcode].update(
str(value) for value in (field_priorities.get(field) or []) if value
)
opcode_priorities = {
key: set(lookup_opcode_priorities[key]) | set(materialized_opcode_priorities[key])
for key in lookup_opcode_priorities
}
source_layers = _layer_contract(config)
layers: list[dict[str, Any]] = [
{
"id": "s52-background",
"type": "background",
"paint": {"background-color": background},
}
]
compiled_counts: Counter[str] = Counter()
compiled_placements: dict[str, set[tuple[str, str]]] = {
key: set() for key in ("AC", "AP", "LS", "LC", "SY", "LIGHT_SY", "TXTE", "LIGHT_TX")
}
active_colors = colors
if usage is not None:
used_color_names = set()
for field in (
"s52_area_color",
"s52_line_color",
"s52_text1_color",
"s52_text2_color",
"s52_text3_color",
):
used_color_names.update(usage_values.get(field) or [])
active_colors = {name: colors[name] for name in used_color_names if name in colors}
if not active_colors:
active_colors = colors
color_area = _color_expression("s52_area_color", active_colors, fallback)
color_line = _color_expression("s52_line_color", active_colors, fallback)
for priority_index, priority in enumerate(DISPLAY_PRIORITY_ORDER):
prefix = f"s52-p{priority_index:02d}"
for contract in source_layers:
layer_name = str(contract["name"])
caps = set(contract.get("capabilities") or [])
layer_id = _safe_id(layer_name)
if (
"fill" in caps
and priority in opcode_priorities["AC"]
and _field_used(usage, layer_name, "s52_area_color", priority)
):
_add_layer(
layers,
compiled_counts,
"AC",
{
"id": f"{prefix}-ac-{layer_id}",
"type": "fill",
"source": source_id,
"source-layer": layer_name,
"filter": _base_filter(
priority=priority,
categories=categories,
required_field="s52_area_color",
),
"paint": {
"fill-color": color_area,
"fill-opacity": float(config.get("area_fill_opacity", 0.92)),
},
},
)
compiled_placements["AC"].add((layer_name, priority))
if (
"fill" in caps
and ap_mapping
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-{layer_id}",
"type": "fill",
"source": source_id,
"source-layer": layer_name,
"filter": _base_filter(
priority=priority,
categories=categories,
required_field="s52_area_pattern_code",
supported_values=list(ap_mapping),
),
"paint": {
"fill-pattern": _mapped_expression(
"s52_area_pattern_code", ap_mapping, next(iter(ap_mapping.values()))
),
"fill-opacity": float(config.get("area_pattern_opacity", 0.85)),
},
},
)
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
and priority in opcode_priorities["LS"]
and _field_used(usage, layer_name, "s52_line_color", priority)
):
_add_layer(
layers,
compiled_counts,
"LS",
{
"id": f"{prefix}-ls-{layer_id}",
"type": "line",
"source": source_id,
"source-layer": layer_name,
"filter": _base_filter(
priority=priority,
categories=categories,
required_field="s52_line_color",
),
"paint": {
"line-color": color_line,
"line-width": [
"to-number",
["get", "s52_line_width"],
float(config.get("line_width_fallback", 1.0)),
],
"line-dasharray": [
"case",
["==", ["get", "s52_line_style"], "DASH"],
["literal", [4, 2]],
["literal", [1, 0]],
],
},
},
)
compiled_placements["LS"].add((layer_name, priority))
if (
"line" in caps
and lc_mapping
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-{layer_id}",
"type": "line",
"source": source_id,
"source-layer": layer_name,
"filter": _base_filter(
priority=priority,
categories=categories,
required_field="s52_line_complex_code",
supported_values=list(lc_mapping),
),
"paint": {
"line-pattern": _mapped_expression(
"s52_line_complex_code", lc_mapping, next(iter(lc_mapping.values()))
),
"line-width": float(config.get("complex_line_width_px", 5.0)),
},
},
)
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
and symbol_names
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-{layer_id}",
"type": "symbol",
"source": source_id,
"source-layer": layer_name,
"filter": [
"all",
*_base_filter(
priority=priority,
categories=categories,
required_field="s52_symbol_code",
supported_values=symbol_names,
forbidden_field="s52_sounding_symbol_sequence",
)[1:],
[
"!",
[
"all",
["has", "s52_light_symbol_code"],
[
"==",
["get", "s52_symbol_code"],
["get", "s52_light_symbol_code"],
],
],
],
],
"layout": {
"icon-image": ["get", "s52_symbol_code"],
"icon-anchor": "center",
"icon-offset": _symbol_offset_expression("s52_symbol_code", symbol_offsets),
"icon-size": 1,
"icon-allow-overlap": True,
"icon-ignore-placement": True,
},
},
)
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
and light_symbol_names
and _field_used(usage, layer_name, "s52_light_symbol_code", priority)
):
_add_layer(
layers,
compiled_counts,
"LIGHTS05-SY",
{
"id": f"{prefix}-lights05-sy-{layer_id}",
"type": "symbol",
"source": source_id,
"source-layer": layer_name,
"filter": _base_filter(
priority=priority,
categories=categories,
required_field="s52_light_symbol_code",
supported_values=light_symbol_names,
),
"layout": {
"icon-image": ["get", "s52_light_symbol_code"],
"icon-anchor": "center",
"icon-offset": _symbol_offset_expression(
"s52_light_symbol_code", light_symbol_offsets
),
"icon-size": 1,
# OpenCPN's raster sprite already contains the final flare
# orientation. s52_light_rotation_deg is retained in the
# PBF contract for vector/runtime backends, but applying it
# again here rotates the raster flare a second time.
"icon-allow-overlap": True,
"icon-ignore-placement": True,
},
},
)
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
and _field_used(usage, layer_name, "s52_text_literal", priority)
):
_add_layer(
layers,
compiled_counts,
"LIGHTS05-TX",
{
"id": f"{prefix}-lights05-tx-{layer_id}",
"type": "symbol",
"source": source_id,
"source-layer": layer_name,
"filter": [
"all",
*_base_filter(
priority=priority,
categories=categories,
required_field="s52_text_literal",
)[1:],
["==", ["get", "s52_csp_resolved"], "LIGHTS05"],
],
"layout": {
"text-field": ["get", "s52_text_literal"],
"text-size": 10,
"text-offset": [1.2, 0.0],
"text-anchor": "left",
"text-allow-overlap": False,
"text-optional": True,
},
"paint": {
"text-color": _rgb_hex(
colors.get("CHBLK", {"r": 7, "g": 7, "b": 7})
),
"text-halo-color": "rgba(255,255,255,0.65)",
"text-halo-width": 0.5,
},
},
)
compiled_placements["LIGHT_TX"].add((layer_name, priority))
if "text" in caps and priority in opcode_priorities["TX/TE"]:
for slot in (1, 2, 3):
field = f"s52_text{slot}_value"
if not _field_used(usage, layer_name, field, priority):
continue
_add_layer(
layers,
compiled_counts,
"TX/TE",
{
"id": f"{prefix}-text{slot}-{layer_id}",
"type": "symbol",
"source": source_id,
"source-layer": layer_name,
"filter": _base_filter(
priority=priority,
categories=categories,
required_field=field,
),
"layout": {
"text-field": ["get", field],
"text-size": [
"to-number",
["get", f"s52_text{slot}_size"],
10,
],
"text-offset": _text_offset_expression(slot, text_offsets),
"text-anchor": _text_anchor_expression(slot),
"text-allow-overlap": False,
"text-optional": True,
},
"paint": {
"text-color": _color_expression(
f"s52_text{slot}_color", active_colors, _rgb_hex(colors.get("CHBLK", {"r": 7, "g": 7, "b": 7}))
),
"text-halo-color": "rgba(255,255,255,0.65)",
"text-halo-width": 0.5,
},
},
)
compiled_placements["TXTE"].add((layer_name, priority))
opcode_usage = Counter(
rule.get("opcode")
for row in lookup_ir
for rule in (row.get("rules") or [])
if rule.get("opcode")
)
referenced_sy = {
str(rule["args"][0]).strip("'\"")[:8]
for row in lookup_ir
for rule in (row.get("rules") or [])
if rule.get("opcode") == "SY" and rule.get("args")
}
referenced_ap = {
str(rule["args"][0]).strip("'\"")[:8]
for row in lookup_ir
for rule in (row.get("rules") or [])
if rule.get("opcode") == "AP" and rule.get("args")
}
referenced_lc = {
str(rule["args"][0]).strip("'\"")[:8]
for row in lookup_ir
for rule in (row.get("rules") or [])
if rule.get("opcode") == "LC" and rule.get("args")
}
style = {
"version": 8,
"name": str(config.get("name", "CM95 S-52 compiled style")),
"sprite": sprite_url,
"metadata": {
"cm95:s52-style-compiler": "v1",
"cm95:palette": config.get("palette", "DAY_BRIGHT"),
"cm95:static-opcodes": list(STATIC_OPCODES),
"cm95:runtime-opcodes": ["CA", "SOUNDG composition"],
},
"sources": {source_id: source},
"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())
unknown_layers = sorted(actual_names - contract_names)
if unknown_layers:
hard_gate_failures.append({
"code": "SOURCE_LAYER_CONTRACT_GAP",
"values": unknown_layers,
})
used_symbols = set(usage_values.get("s52_symbol_code") or [])
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",
"values": missing_used_symbols,
})
used_light_symbols = set(usage_values.get("s52_light_symbol_code") or [])
missing_used_light_symbols = sorted(
(used_light_symbols - set(available_symbol_names)) - undefined_symbols
)
if missing_used_light_symbols:
hard_gate_failures.append({
"code": "USED_LIGHT_SYMBOL_RESOURCE_GAP",
"values": missing_used_light_symbols,
})
used_patterns = set(usage_values.get("s52_area_pattern_code") or [])
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",
"values": missing_used_patterns,
})
used_complex_lines = set(usage_values.get("s52_line_complex_code") or [])
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",
"values": missing_used_complex_lines,
})
used_colors: set[str] = set()
for field in (
"s52_area_color",
"s52_line_color",
"s52_text1_color",
"s52_text2_color",
"s52_text3_color",
):
used_colors.update(usage_values.get(field) or [])
missing_used_colors = sorted(used_colors - set(colors))
if missing_used_colors:
hard_gate_failures.append({
"code": "USED_COLOR_RESOURCE_GAP",
"values": missing_used_colors,
})
light_text_count = sum(
int(row.get("feature_count") or 0)
for row in (usage.get("render_kind_counts") or {}).get("LIGHT_TX", [])
)
if light_text_count and "CHBLK" not in colors:
hard_gate_failures.append({
"code": "LIGHT_TEXT_COLOR_GAP",
"values": ["CHBLK"],
})
static_coverage: dict[str, Any] = {}
runtime_coverage: dict[str, Any] = {}
if usage is not None:
render_counts = usage.get("render_kind_counts") or {}
for kind in ("AC", "AP", "LS", "LC", "SY", "LIGHT_SY", "TXTE", "LIGHT_TX"):
rows = list(render_counts.get(kind) or [])
total = sum(int(row.get("feature_count") or 0) for row in rows)
covered = 0
gaps: list[dict[str, Any]] = []
for row in rows:
layer_name = str(row.get("source_layer") or "")
priority = str(row.get("priority") or "")
category = str(row.get("category") or "")
count = int(row.get("feature_count") or 0)
if (layer_name, priority) in compiled_placements[kind] and category in categories:
covered += count
else:
gaps.append({
"source_layer": layer_name,
"priority": priority,
"category": category,
"feature_count": count,
})
static_coverage[kind] = {
"feature_count": total,
"covered_feature_count": covered,
"gap_feature_count": total - covered,
"placement_count": len(rows),
"gaps": gaps,
}
if gaps:
hard_gate_failures.append({
"code": "MATERIALIZED_STATIC_COVERAGE_GAP",
"kind": kind,
"values": gaps,
})
for kind in ("RUNTIME_SOUNDG", "RUNTIME_CA"):
rows = list(render_counts.get(kind) or [])
runtime_coverage[kind] = {
"feature_count": sum(int(row.get("feature_count") or 0) for row in rows),
"placement_count": len(rows),
"placements": rows,
}
audit = {
"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,
"display_priority_order": list(DISPLAY_PRIORITY_ORDER),
"lookup_ir_count": len(lookup_ir),
"presentation_library_opcode_usage": dict(sorted(opcode_usage.items())),
"lookup_priority_usage": {
key: sorted(values)
for key, values in sorted(lookup_opcode_priorities.items())
if values
},
"materialized_priority_usage": {
key: sorted(values)
for key, values in sorted(materialized_opcode_priorities.items())
if values
},
"compiled_priority_usage": {
key: sorted(
values,
key=lambda value: (
DISPLAY_PRIORITY_ORDER.index(value)
if value in DISPLAY_PRIORITY_ORDER
else 999
),
)
for key, values in sorted(opcode_priorities.items())
if values
},
"compiled_layer_count": len(layers),
"compiled_layer_count_by_opcode": dict(sorted(compiled_counts.items())),
"sprite_symbol_count": len(symbol_names),
"light_symbol_count": len(light_symbol_names),
"area_pattern_count": len(ap_mapping),
"complex_line_count": len(lc_mapping),
"text_offset_pair_count": len(text_offsets),
"materialized_static_coverage": static_coverage,
"geometry_incompatible_materialized": (usage or {}).get("geometry_ignored") or [],
"runtime_coverage": runtime_coverage,
"usage_scope": (
None
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)
for row in (usage.get("layers") or {}).values()
),
"global_values": usage.get("global_values") or {},
}
),
"presentation_library_reference_gaps": {
"note": "Full 3057-row Presentation Library inventory; not a Dataset 1 usage hard gate.",
"symbols": sorted(referenced_sy - set(available_symbol_names)),
"patterns": sorted(referenced_ap - set(available_ap_mapping)),
"complex_lines": sorted(referenced_lc - set(available_lc_mapping)),
},
"runtime_boundary": {
"CA": "runtime overlay; screen-mm geometry is not compiled into static style",
"SOUNDG": "runtime glyph composition; static style does not split symbol sequences",
},
}
return style, audit
def main() -> int:
parser = argparse.ArgumentParser(
description="Compile CM95 materialized S-52 properties/resources into a MapLibre style."
)
parser.add_argument("--config", required=True, type=Path)
parser.add_argument("--output", required=True, type=Path)
parser.add_argument("--audit-output", type=Path)
parser.add_argument(
"--usage-pbf",
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)
resources = dict(config.get("resources") or {})
required = (
"colors",
"symbol_metadata",
"bitmap_pattern_metadata",
"vector_pattern_metadata",
"line_metadata",
"sprite_index",
"lookup_ir",
"resource_contract",
)
missing = [name for name in required if not resources.get(name)]
if missing:
raise SystemExit(f"config.resources missing: {missing}")
color_tables = _load_json(_resolve_path(project_root, resources["colors"]))
palette = str(config.get("palette", "DAY_BRIGHT"))
if palette not in color_tables:
raise SystemExit(f"unknown palette {palette}; choices={sorted(color_tables)}")
colors = dict(color_tables[palette].get("colors") or {})
if not colors:
raise SystemExit(f"palette {palette} has no colors")
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,
colors=colors,
symbol_metadata=_load_json(_resolve_path(project_root, resources["symbol_metadata"])),
bitmap_pattern_metadata=_load_json(_resolve_path(project_root, resources["bitmap_pattern_metadata"])),
vector_pattern_metadata=_load_json(_resolve_path(project_root, resources["vector_pattern_metadata"])),
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,
)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(style, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
audit_output = args.audit_output or args.output.with_suffix(".audit.json")
audit_output.parent.mkdir(parents=True, exist_ok=True)
audit_output.write_text(
json.dumps(audit, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
print(json.dumps(audit, ensure_ascii=False, indent=2))
if audit.get("hard_gate_failures"):
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())