#!/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_line1_color", "s52_line2_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_presentations", "s52_light_rotation_deg", "s52_text1_value", "s52_text2_value", "s52_text3_value", "s52_line1_style", "s52_line1_width", "s52_line2_style", "s52_line2_width", ) def _parse_light_presentations(props: dict[str, Any]) -> list[dict[str, Any]]: raw = props.get("s52_light_presentations") if raw in (None, ""): return [] if isinstance(raw, str): try: raw = json.loads(raw) except Exception: return [] if not isinstance(raw, list): return [] return [dict(item) for item in raw if isinstance(item, dict)] 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") } 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(): 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) 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"} 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 line_slots = [ slot for slot in (1, 2) if f"s52_line{slot}_color" in props ] if line_slots or "s52_line_color" in props: if line_geometry: render_kind_counts["LS"][placement] += max(1, len(line_slots)) 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 light_presentations = _parse_light_presentations(props) if light_presentations: for presentation in light_presentations: nested_placement = ( layer_name, str(presentation.get("s52_display_priority") or priority), str(presentation.get("s52_display_category") or category), ) if presentation.get("s52_arc_color"): render_kind_counts["RUNTIME_CA"][nested_placement] += 1 elif "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}, "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()) ], "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]]] = {} runtime_csp_counts: Counter[str] = Counter() 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 ) 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 ""), 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 }, "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()) ], "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", "s52_line1_color", "s52_line2_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_line1_color", "s52_line2_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"]: for slot in (1, 2): color_field = f"s52_line{slot}_color" if not _field_used(usage, layer_name, color_field, priority): continue style_field = f"s52_line{slot}_style" width_field = f"s52_line{slot}_width" _add_layer( layers, compiled_counts, "LS", { "id": f"{prefix}-ls{slot}-{layer_id}", "type": "line", "source": source_id, "source-layer": layer_name, "filter": _base_filter( priority=priority, categories=categories, required_field=color_field, ), "layout": {"line-cap": "round"}, "paint": { "line-color": _color_expression( color_field, active_colors, fallback ), "line-width": [ "to-number", ["get", width_field], float(config.get("line_width_fallback", 1.0)), ], "line-dasharray": [ "match", ["get", style_field], "DASH", ["literal", [4, 2]], "DOTT", ["literal", [1, 2]], ["literal", [1, 0]], ], }, }, ) compiled_placements["LS"].add((layer_name, priority)) if _field_used(usage, layer_name, "s52_line_color", priority): _add_layer( layers, compiled_counts, "LS", { "id": f"{prefix}-ls-legacy-{layer_id}", "type": "line", "source": source_id, "source-layer": layer_name, "filter": _base_filter( priority=priority, categories=categories, required_field="s52_line_color", forbidden_field="s52_line1_color", ), "layout": {"line-cap": "round"}, "paint": { "line-color": color_line, "line-width": ["to-number", ["get", "s52_line_width"], 1], "line-dasharray": [ "match", ["get", "s52_line_style"], "DASH", ["literal", [4, 2]], "DOTT", ["literal", [1, 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_line1_color", "s52_line2_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, } 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, "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())