tei/CM95
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: resolve S-52 topmarks and light fields

Browse Source
This commit is contained in:
OpenAI Codex
2026-09-17 07:09:30 +08:00
parent 8e51e388c1
commit ef36ec3599
5 changed files with 231 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
config/s52/cm95_s52_poc_v1.json
View File

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

62
src/cm95pbf/builder.py
View File

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

14
src/cm95pbf/mapping.py
View File

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

143
src/cm95pbf/s52.py
View File

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

14
tools/cm95_s52_maplibre_poc.py
View File

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