From ef36ec3599ccaf0bd8cba3e14716ebc527401b09 Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Thu, 17 Sep 2026 07:09:30 +0800 Subject: [PATCH] feat: resolve S-52 topmarks and light fields --- config/s52/cm95_s52_poc_v1.json | 3 +- src/cm95pbf/builder.py | 62 +++++++++++++- src/cm95pbf/mapping.py | 14 ++++ src/cm95pbf/s52.py | 143 ++++++++++++++++++++++++++++++++ tools/cm95_s52_maplibre_poc.py | 14 +++- 5 files changed, 231 insertions(+), 5 deletions(-) diff --git a/config/s52/cm95_s52_poc_v1.json b/config/s52/cm95_s52_poc_v1.json index b6a34d8..bee69e3 100644 --- a/config/s52/cm95_s52_poc_v1.json +++ b/config/s52/cm95_s52_poc_v1.json @@ -17,5 +17,6 @@ "conditional_symbology": "explicit_builder_derived_or_runtime_renderer", "production_release": "do_not_touch_in_poc" }, - "point_table": "Paper" + "point_table": "Paper", + "sprite_scope": "all" } diff --git a/src/cm95pbf/builder.py b/src/cm95pbf/builder.py index b5d75c4..5d89d73 100644 --- a/src/cm95pbf/builder.py +++ b/src/cm95pbf/builder.py @@ -56,7 +56,7 @@ from .policy import ( mercator_to_lonlat, xyz_tile_bounds, ) -from .s52 import S52LookupRegistry +from .s52 import S52LookupRegistry, resolve_lights05, resolve_topmar01_symbol @dataclass(frozen=True) @@ -431,6 +431,7 @@ class Cm95TileBuilder: selected.append(row) consumed: set[int] = set() + s52_context = self._build_s52_context(selected) nav_outputs = self._compose_navigation_aids( selected, manifest, @@ -468,6 +469,7 @@ class Cm95TileBuilder: manifest, row, trace_rows, + s52_context, ): status_counts["S52_FALLBACK"] += 1 continue @@ -1464,6 +1466,15 @@ class Cm95TileBuilder: props.setdefault("canonical_object_type", "minor_light") props.setdefault("chart_symbol_code", "minor_light") props.setdefault("chart_icon_image", "light_minor") + if self.s52 is not None: + light_s52 = self.s52.properties( + light.object_class, light.geometry_type, light.attrs + ) + for key, value in light_s52.items(): + if key != "s52_csp": + props.setdefault(key, value) + if light_s52.get("s52_csp") == "LIGHTS05": + props.update(resolve_lights05(light.attrs)) direct_name = first_direct_name(light.attrs) if direct_name: props.setdefault("chart_label_text", direct_name) @@ -1666,7 +1677,12 @@ class Cm95TileBuilder: light.geometry_type, lattrs, ) - if light_s52.get("s52_csp"): + for key, value in light_s52.items(): + if key != "s52_csp": + props.setdefault(key, value) + if light_s52.get("s52_csp") == "LIGHTS05": + props.update(resolve_lights05(lattrs)) + elif light_s52.get("s52_csp"): props["s52_csp"] = light_s52["s52_csp"] clean_props = { @@ -1680,6 +1696,35 @@ class Cm95TileBuilder: "conflict": name_conflict, } + def _build_s52_context(self, rows: list[SourceFeature]) -> dict[str, set[Any]]: + floating_relations: set[Any] = set() + floating_points: set[Any] = set() + for row in rows: + if row.geometry_type != "POINT": + continue + if not (row.object_class.startswith("BOY") or row.object_class in {"LITFLT", "LITVES"}): + continue + relation = row.attrs.get("_cm93_relation20_raw") + if relation is not None: + floating_relations.add((row.cell_id, str(relation))) + geom = row.geom_wgs84 + if getattr(geom, "geom_type", None) == "Point": + floating_points.add((float(geom.x), float(geom.y))) + return { + "floating_relations": floating_relations, + "floating_points": floating_points, + } + + @staticmethod + def _topmar_is_floating(row: SourceFeature, context: dict[str, set[Any]]) -> bool: + relation = row.attrs.get("_cm93_relation20_raw") + if relation is not None and (row.cell_id, str(relation)) in context["floating_relations"]: + return True + geom = row.geom_wgs84 + if getattr(geom, "geom_type", None) == "Point": + return (float(geom.x), float(geom.y)) in context["floating_points"] + return False + def _s52_properties(self, row: SourceFeature) -> dict[str, Any]: if self.s52 is None: return {} @@ -1695,6 +1740,7 @@ class Cm95TileBuilder: manifest, row: SourceFeature, trace_rows, + s52_context: dict[str, set[Any]], ) -> bool: if self.s52 is None or row.geometry_type != "POINT": return False @@ -1702,6 +1748,15 @@ class Cm95TileBuilder: return False s52_props = self._s52_properties(row) + resolved_csp = None + if row.object_class == "TOPMAR" and s52_props.get("s52_csp") == "TOPMAR01": + floating = self._topmar_is_floating(row, s52_context) + s52_props["s52_symbol_code"] = resolve_topmar01_symbol( + row.attrs, floating=floating + ) + s52_props["s52_csp_resolved"] = "TOPMAR01" + s52_props.pop("s52_csp", None) + resolved_csp = "TOPMAR01" if not s52_props.get("s52_symbol_code") and not s52_props.get("s52_csp"): return False @@ -1733,6 +1788,7 @@ class Cm95TileBuilder: "native_scale": row.native_scale, "mapping_rule_id": f"S52-FALLBACK-{row.object_class}", "mapping_status": "S52_FALLBACK", + **({"s52_resolved_csp": resolved_csp} if resolved_csp else {}), } ) return True @@ -2001,6 +2057,8 @@ class Cm95TileBuilder: ) -> dict[str, Any]: props = dict(props) for key, value in self._s52_properties(row).items(): + if key == "s52_csp" and props.get("s52_csp_resolved"): + continue props.setdefault(key, value) props = _sanitize_properties(props) diff --git a/src/cm95pbf/mapping.py b/src/cm95pbf/mapping.py index 1a416ed..5b051e3 100644 --- a/src/cm95pbf/mapping.py +++ b/src/cm95pbf/mapping.py @@ -33,7 +33,21 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset( "name_ja", "place_name_en", "s52_csp", + "s52_csp_resolved", "s52_display_category", + "s52_light_arc_color", + "s52_light_arc_radius_mm", + "s52_light_mode", + "s52_light_nominal_range_nm", + "s52_light_obscured", + "s52_light_rotation_deg", + "s52_light_sector_radius_mm", + "s52_light_symbol_code", + "s52_sector_display_end_deg", + "s52_sector_display_start_deg", + "s52_sector_end_deg", + "s52_sector_start_deg", + "s52_sector_sweep_deg", "s52_display_priority", "s52_lookup_rcid", "s52_symbol_code", diff --git a/src/cm95pbf/s52.py b/src/cm95pbf/s52.py index 267c141..c75dc7d 100644 --- a/src/cm95pbf/s52.py +++ b/src/cm95pbf/s52.py @@ -25,6 +25,30 @@ CM93_COLMAR_TO_S57_COLOUR: dict[int, str] = { } +TOPMAR01_FLOATING_SYMBOLS: dict[int, str] = { + 1: "TOPMAR02", 2: "TOPMAR04", 3: "TOPMAR10", 4: "TOPMAR12", + 5: "TOPMAR13", 6: "TOPMAR14", 7: "TOPMAR65", 8: "TOPMAR17", + 9: "TOPMAR16", 10: "TOPMAR08", 11: "TOPMAR07", 12: "TOPMAR14", + 13: "TOPMAR05", 14: "TOPMAR06", 17: "TMARDEF2", 18: "TOPMAR10", + 19: "TOPMAR13", 20: "TOPMAR14", 21: "TOPMAR13", 22: "TOPMAR14", + 23: "TOPMAR14", 24: "TOPMAR02", 25: "TOPMAR04", 26: "TOPMAR10", + 27: "TOPMAR17", 28: "TOPMAR18", 29: "TOPMAR02", 30: "TOPMAR17", + 31: "TOPMAR14", 32: "TOPMAR10", 33: "TMARDEF2", +} + +TOPMAR01_RIGID_SYMBOLS: dict[int, str] = { + 1: "TOPMAR22", 2: "TOPMAR24", 3: "TOPMAR30", 4: "TOPMAR32", + 5: "TOPMAR33", 6: "TOPMAR34", 7: "TOPMAR85", 8: "TOPMAR86", + 9: "TOPMAR36", 10: "TOPMAR28", 11: "TOPMAR27", 12: "TOPMAR14", + 13: "TOPMAR25", 14: "TOPMAR26", 15: "TOPMAR88", 16: "TOPMAR87", + 17: "TMARDEF1", 18: "TOPMAR30", 19: "TOPMAR33", 20: "TOPMAR34", + 21: "TOPMAR33", 22: "TOPMAR34", 23: "TOPMAR34", 24: "TOPMAR22", + 25: "TOPMAR24", 26: "TOPMAR30", 27: "TOPMAR86", 28: "TOPMAR89", + 29: "TOPMAR22", 30: "TOPMAR86", 31: "TOPMAR14", 32: "TOPMAR30", + 33: "TMARDEF1", +} + + def _s57_list_string(value: Any) -> str | None: if value is None: return None @@ -152,6 +176,111 @@ def conditional_procedures(row: dict[str, Any] | None) -> tuple[str, ...]: return tuple(str(x) for x in (row.get("cs") or [])) +def _s52_int_list(value: Any) -> list[int]: + if value is None or value == "": + return [] + if isinstance(value, (list, tuple)): + raw = value + elif isinstance(value, str): + text = value.strip().strip("[]") + raw = [part.strip() for part in text.split(",") if part.strip()] + else: + raw = [value] + out: list[int] = [] + for item in raw: + try: + out.append(int(item)) + except (TypeError, ValueError): + continue + return out + + +def _s52_float(value: Any, default: float | None = None) -> float | None: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _lights05_color(colours: list[int]) -> str: + if len(colours) == 1: + if 3 in colours: + return "red" + if 4 in colours: + return "green" + if any(code in colours for code in (1, 6, 11)): + return "yellow_white" + return "magenta" + if len(colours) == 2: + if 1 in colours and 3 in colours: + return "red" + if 1 in colours and 4 in colours: + return "green" + return "magenta" + + +def _lights05_flare_symbol(colours: list[int]) -> str: + color = _lights05_color(colours) + if color == "red": + return "LIGHTS11" + if color == "green": + return "LIGHTS12" + if color == "yellow_white": + return "LIGHTS13" + return "LITDEF11" + + +def resolve_lights05(attrs: dict[str, Any]) -> dict[str, Any]: + """Resolve OpenCPN PLIB LIGHTS05/LIGHTS06 into runtime-friendly fields.""" + catlit = _s52_int_list(attrs.get("CATLIT")) + if 8 in catlit or 11 in catlit: + return {"s52_csp_resolved": "LIGHTS05", "s52_light_mode": "special", "s52_light_symbol_code": "LIGHTS82"} + if 9 in catlit: + return {"s52_csp_resolved": "LIGHTS05", "s52_light_mode": "special", "s52_light_symbol_code": "LIGHTS81"} + + colours = _s52_int_list(attrs.get("COLOUR")) or [12] + valnmr = _s52_float(attrs.get("VALNMR"), 9.0) or 9.0 + sectr1 = _s52_float(attrs.get("SECTR1")) + sectr2 = _s52_float(attrs.get("SECTR2")) + arc_color = _lights05_color(colours) + litvis = _s52_int_list(attrs.get("LITVIS")) + obscured = any(code in litvis for code in (3, 7, 8)) + if obscured: + arc_color = "obscured" + + base: dict[str, Any] = { + "s52_csp_resolved": "LIGHTS05", + "s52_light_arc_color": arc_color, + "s52_light_nominal_range_nm": float(valnmr), + } + if sectr1 is None or sectr2 is None: + if valnmr < 10.0: + base.update({"s52_light_mode": "flare", "s52_light_symbol_code": _lights05_flare_symbol(colours), "s52_light_rotation_deg": 135.0}) + return base + radius = 3.0 if valnmr < 7.0 else 10.0 if valnmr < 15.0 else 15.0 if valnmr < 30.0 else 20.0 + base.update({"s52_light_mode": "all_round", "s52_light_arc_radius_mm": radius}) + return base + + sweep = (sectr2 - sectr1) % 360.0 + if sweep < 1.0: + base.update({"s52_light_mode": "all_round", "s52_light_arc_radius_mm": 10.0 if valnmr < 15.0 else 15.0 if valnmr < 30.0 else 20.0}) + return base + + arc_radius = 10.0 if 0 < valnmr < 15.0 else 15.0 if valnmr < 30.0 else 20.0 + base.update({ + "s52_light_mode": "sector", + "s52_sector_start_deg": float(sectr1), + "s52_sector_end_deg": float(sectr2), + "s52_sector_sweep_deg": float(sweep), + "s52_sector_display_start_deg": float((sectr1 + 180.0) % 360.0), + "s52_sector_display_end_deg": float((sectr2 + 180.0) % 360.0), + "s52_light_arc_radius_mm": float(arc_radius), + "s52_light_sector_radius_mm": 25.0, + "s52_light_obscured": bool(obscured), + }) + return base + + class S52LookupRegistry: """Read extracted S-52 lookups once and expose PBF-ready properties.""" @@ -206,3 +335,17 @@ class S52LookupRegistry: if procedures: out["s52_csp"] = ",".join(procedures) return {key: value for key, value in out.items() if value != ""} + + +def resolve_topmar01_symbol(attrs: dict[str, Any], *, floating: bool) -> str: + """Resolve S-52 TOPMAR01 exactly as OpenCPN PLIB for known TOPSHP values.""" + value = attrs.get("TOPSHP") + if value is None or value == "": + return "QUESMRK1" + try: + topshp = int(value) + except (TypeError, ValueError): + return "QUESMRK1" + if floating: + return TOPMAR01_FLOATING_SYMBOLS.get(topshp, "TMARDEF2") + return TOPMAR01_RIGID_SYMBOLS.get(topshp, "TMARDEF1") diff --git a/tools/cm95_s52_maplibre_poc.py b/tools/cm95_s52_maplibre_poc.py index 652a95c..19798aa 100644 --- a/tools/cm95_s52_maplibre_poc.py +++ b/tools/cm95_s52_maplibre_poc.py @@ -218,6 +218,9 @@ def main() -> int: targets_text = args.targets or ",".join(config.get("targets", DEFAULT_TARGETS)) targets = {x.strip().upper() for x in targets_text.split(",") if x.strip()} palette = args.palette or config.get("palette", "DAY_BRIGHT") + sprite_scope = str(config.get("sprite_scope", "all")) + if sprite_scope not in {"all", "target"}: + raise SystemExit(f"invalid sprite_scope: {sprite_scope}") source_id = config.get("source_id", "navsea_delivery") source_layer = config.get("source_layer", "navigation_marks") sprite_url = config.get("sprite_url", "./sprite") @@ -243,12 +246,17 @@ def main() -> int: atlas_width, atlas_height = _png_size(atlas_path) direct_symbols = sorted({name for row in lookups for name in row["sy"]}) 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) + if sprite_scope == "all" + else direct_symbols + ) metadata: dict[str, dict[str, Any]] = {} missing_symbols: list[str] = [] vector_only_symbols: list[str] = [] out_of_bounds_symbols: list[str] = [] - for name in direct_symbols: + for name in sprite_symbols: symbol = symbols.get(name) if symbol is None: missing_symbols.append(name) @@ -303,6 +311,7 @@ def main() -> int: json.dumps(style, ensure_ascii=False, indent=2), encoding="utf-8" ) by_class = Counter(row["object_class"] for row in lookups) + direct_missing = sorted(name for name in direct_symbols if name not in symbols) report = { "schema": "navsea.cm95.s52-maplibre-poc/v1", "targets": sorted(targets), @@ -314,8 +323,9 @@ def main() -> int: "conditional_lookup_count": len(conditional_rows), "conditional_procedures": csp_names, "direct_symbol_reference_count": len(direct_symbols), + "sprite_scope": sprite_scope, "sprite_entry_count": len(sprite_json), - "missing_symbol_definitions": missing_symbols, + "missing_symbol_definitions": direct_missing, "vector_only_symbol_definitions": vector_only_symbols, "atlas_out_of_bounds": out_of_bounds_symbols, "production_boundary": {