From 043b02241335904c848f6d500c4bd02697958372 Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Mon, 21 Sep 2026 06:13:00 +0800 Subject: [PATCH] fix: align S-52 lookup and mariner semantics --- src/cm95pbf/csp.py | 97 +++++++++++++++++++++++++++++++++++++----- src/cm95pbf/mapping.py | 13 ++++++ src/cm95pbf/s52.py | 36 ++++++++++++++-- 3 files changed, 133 insertions(+), 13 deletions(-) diff --git a/src/cm95pbf/csp.py b/src/cm95pbf/csp.py index 73799a8..e73093b 100644 --- a/src/cm95pbf/csp.py +++ b/src/cm95pbf/csp.py @@ -21,6 +21,8 @@ class S52MarinerSettings: deep_contour_m: float = 6.0 two_shades: bool = False symbolized_boundaries: bool = False + depth_unit: str = "metres" + height_unit: str = "metres" # These CSPs depend on mariner-selected depth thresholds. NavSea delivery PBF @@ -28,7 +30,7 @@ class S52MarinerSettings: # profile into tile properties. The fixed S52MarinerSettings path remains # available for OpenCPN/reference rendering. MARINER_DEPENDENT_CSPS = frozenset( - {"DEPARE01", "DEPARE02", "DEPCNT02", "SOUNDG02", "OBSTRN04", "WRECKS02", "RESARE02"} + {"DEPARE01", "DEPARE02", "DEPCNT02", "SOUNDG02", "OBSTRN04", "WRECKS02", "RESARE02", "LIGHTS05"} ) @@ -96,6 +98,8 @@ def resolve_depcnt02_instruction( settings: S52MarinerSettings, *, next_safe_contour_m: float | None = None, + safe_dashed_instruction: str | None = None, + safe_solid_instruction: str | None = None, ) -> tuple[str, bool]: safe = False if object_class == "DEPARE": @@ -117,8 +121,16 @@ def resolve_depcnt02_instruction( except (TypeError, ValueError): quapos = 0 if 2 <= quapos < 10: - return ("LS(DASH,2,DEPSC)" if safe else "LS(DASH,1,DEPCN)"), safe - return ("LS(SOLD,2,DEPSC)" if safe else "LS(SOLD,1,DEPCN)"), safe + return ( + (safe_dashed_instruction or "LS(DASH,2,DEPSC)") + if safe else "LS(DASH,1,DEPCN)" + ), safe + if quapos != 0: + return "", safe + return ( + (safe_solid_instruction or "LS(SOLD,2,DEPSC)") + if safe else "LS(SOLD,1,DEPCN)" + ), safe @@ -196,7 +208,7 @@ _LIGHT_CHARACTER = { } -def _light_description(attrs: dict[str, Any]) -> str: +def _light_description(attrs: dict[str, Any], *, height_unit: str = "metres") -> str: try: litchr = int(attrs.get("LITCHR")) except (TypeError, ValueError): @@ -227,7 +239,10 @@ def _light_description(attrs: dict[str, Any]) -> str: text += f" {sigper:.1f}s" if abs(round(sigper) - sigper) > 0.01 else f" {sigper:.0f}s" height = _optional_float(attrs.get("HEIGHT")) if height is not None: - text += f" {height:.0f}m" + if height_unit == "feet": + text += f" {math.floor(height / 0.3048 + 0.5):.0f}ft" + else: + text += f" {math.floor(height + 0.5):.0f}m" valnmr = _optional_float(attrs.get("VALNMR")) if valnmr is not None and not has_sector: text += f" {valnmr:.0f}Nm" @@ -238,7 +253,9 @@ def _light_description(attrs: dict[str, Any]) -> str: return text.strip() -def resolve_lights05_instruction(attrs: dict[str, Any]) -> tuple[str, dict[str, Any]]: +def resolve_lights05_instruction( + attrs: dict[str, Any], settings: S52MarinerSettings = S52MarinerSettings() +) -> tuple[str, dict[str, Any]]: catlit = _int_list(attrs.get("CATLIT")) valnmr = _optional_float(attrs.get("VALNMR")) if valnmr is None: @@ -301,7 +318,7 @@ def resolve_lights05_instruction(attrs: dict[str, Any]) -> tuple[str, dict[str, "s52_sector_sweep_deg": float(sweep), }) - description = _light_description(attrs) + description = _light_description(attrs, height_unit=settings.height_unit) if description: safe = description.replace("'", "’") instruction += f";TX('{safe}',3,2,3,'15110',2,0,CHBLK,23)" @@ -319,9 +336,23 @@ def resolve_soundg03_instruction( depth_value = 99999.0 if depth_value < -1000.0: depth_value = 0.0 + safety_depth = settings.safety_depth_m + if settings.depth_unit == "feet": + depth_value /= 0.3048 + safety_depth /= 0.3048 + elif settings.depth_unit == "fathoms": + depth_value /= 0.3048 * 6.0 + safety_depth /= 0.3048 * 6.0 + elif settings.depth_unit != "metres": + raise ValueError(f"unsupported depth unit: {settings.depth_unit}") depth_value += 0.01 if depth_value > 0.0 else -0.01 - prefix = "SOUNDS" if depth_value <= settings.safety_depth_m else "SOUNDG" + # OpenCPN suppresses fractional feet below 31ft by rounding to one or two + # displayed digits before selecting glyphs. + if settings.depth_unit == "feet" and 0.0 < abs(depth_value) < 31.0: + depth_value = math.copysign(math.floor(abs(depth_value) + 0.5), depth_value) + + prefix = "SOUNDS" if depth_value <= safety_depth else "SOUNDG" codes: list[str] = [] swept = 6 in _int_list(attrs.get("TECSOU")) low_quality = bool( @@ -385,11 +416,13 @@ def resolve_soundg03_instruction( extra = { "s52_soundg_point_csp": "SOUNDG03", "s52_sounding_depth_m": float(depth_value_in), + "s52_sounding_display_depth": float(depth_value), + "s52_sounding_depth_unit": settings.depth_unit, "s52_sounding_prefix": prefix, "s52_sounding_symbol_sequence": ",".join(codes), "s52_sounding_swept": int(swept), "s52_sounding_low_quality": int(low_quality), - "s52_sounding_is_shallow": int(depth_value <= settings.safety_depth_m), + "s52_sounding_is_shallow": int(depth_value <= safety_depth), } return instruction, extra @@ -491,10 +524,15 @@ def _runtime_hazard_depth( return 0.0 if catwrk in {4, 5}: return -15.0 + if "CATWRK" in attrs and attrs.get("CATWRK") not in {None, ""}: + return None if watlev == 3: return 0.01 if watlev == 5: return 0.0 + if "WATLEV" in attrs and attrs.get("WATLEV") not in {None, ""}: + if watlev not in {1, 2, 4, 6}: + return None return -15.0 expsou = _int_attr(attrs, "EXPSOU", 0) @@ -528,6 +566,35 @@ def materialize_mariner_runtime_facts( "s52_runtime_csp": name, } + if name == "RESARE02": + plain = resolve_resare02_instruction( + attrs, S52MarinerSettings(symbolized_boundaries=False) + ) + symbolized = resolve_resare02_instruction( + attrs, S52MarinerSettings(symbolized_boundaries=True) + ) + out["s52_resare_plain_rule_chain"] = materialize_rule_ir( + plain, attrs, source="csp", procedure=name + ) + out["s52_resare_symbolized_rule_chain"] = materialize_rule_ir( + symbolized, attrs, source="csp", procedure=name + ) + + if name == "LIGHTS05": + metres, metres_extra = resolve_lights05_instruction( + attrs, S52MarinerSettings(height_unit="metres") + ) + feet, _ = resolve_lights05_instruction( + attrs, S52MarinerSettings(height_unit="feet") + ) + out.update(metres_extra) + out["s52_lights_metres_rule_chain"] = materialize_rule_ir( + metres, attrs, source="csp", procedure=name + ) + out["s52_lights_feet_rule_chain"] = materialize_rule_ir( + feet, attrs, source="csp", procedure=name + ) + if name in {"DEPARE01", "DEPARE02"}: depth_min = _optional_float(attrs.get("DRVAL1")) depth_max = _optional_float(attrs.get("DRVAL2")) @@ -578,6 +645,9 @@ def materialize_mariner_runtime_facts( for source_key, target_key in scalar_codes.items(): if attrs.get(source_key) not in {None, ""}: out[target_key] = _int_attr(attrs, source_key, 0) + out[f"{target_key}_present"] = int( + source_key in attrs and attrs.get(source_key) not in {None, ""} + ) list_codes = { "QUASOU": "s52_quasou_codes", @@ -955,6 +1025,8 @@ def resolve_csp_hints( next_safe_contour_m: float | None = None, sounding_depth_m: float | None = None, associated_depth_ranges: list[dict[str, Any]] | None = None, + safe_dashed_instruction: str | None = None, + safe_solid_instruction: str | None = None, ) -> dict[str, Any]: name = str(csp_name or "")[:8] instruction: str | None = None @@ -968,10 +1040,15 @@ def resolve_csp_hints( attrs, settings, next_safe_contour_m=next_safe_contour_m, + safe_dashed_instruction=safe_dashed_instruction, + safe_solid_instruction=safe_solid_instruction, ) extra["s52_is_safety_contour"] = int(safe) + if safe: + extra["s52_display_base_promoted"] = 1 + extra["s52_scamin_override"] = 100000001 elif name == "LIGHTS05": - instruction, light_extra = resolve_lights05_instruction(attrs) + instruction, light_extra = resolve_lights05_instruction(attrs, settings) extra.update(light_extra) elif name == "SOUNDG02": if sounding_depth_m is None: diff --git a/src/cm95pbf/mapping.py b/src/cm95pbf/mapping.py index 2b986fc..2fd682a 100644 --- a/src/cm95pbf/mapping.py +++ b/src/cm95pbf/mapping.py @@ -67,6 +67,11 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset( "s52_expsou_code", "s52_catobs_code", "s52_catwrk_code", + "s52_quapos_code_present", + "s52_watlev_code_present", + "s52_expsou_code_present", + "s52_catobs_code_present", + "s52_catwrk_code_present", "s52_line_color", "s52_line_complex_code", "s52_line_count", @@ -85,6 +90,12 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset( "s52_source_object_class", "s52_source_geometry", "s52_scamin_override", + "s52_safe_dashed_rule_chain", + "s52_safe_solid_rule_chain", + "s52_resare_plain_rule_chain", + "s52_resare_symbolized_rule_chain", + "s52_lights_metres_rule_chain", + "s52_lights_feet_rule_chain", "s52_light_arc_color", "s52_light_arc_radius_mm", "s52_light_mode", @@ -111,6 +122,8 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset( "s52_symbol_rule_sequence", "s52_symbol_sequence", "s52_sounding_depth_m", + "s52_sounding_display_depth", + "s52_sounding_depth_unit", "s52_sounding_is_shallow", "s52_sounding_low_quality", "s52_sounding_prefix", diff --git a/src/cm95pbf/s52.py b/src/cm95pbf/s52.py index ff24f8c..43e9aa8 100644 --- a/src/cm95pbf/s52.py +++ b/src/cm95pbf/s52.py @@ -448,6 +448,18 @@ class S52LookupRegistry: geometry_type=lookup_geometry, ) + def _plain_area_instruction(self, object_class: str) -> str | None: + row = find_best_lookup( + self.by_class.get(object_class, []), + {}, + table_name="Plain", + geometry_type="Area", + ) + if row is None: + return None + instruction = str(row.get("instruction") or "").strip() + return instruction or None + def properties( self, object_class: str, @@ -482,12 +494,13 @@ class S52LookupRegistry: for token in str(out.get("s52_csp") or "").split(",") if token.strip() ] + safe_dashed_instruction = self._plain_area_instruction("SAFECD") + safe_solid_instruction = self._plain_area_instruction("SAFECN") unresolved: list[str] = [] resolved_by_procedure: dict[str, dict[str, Any]] = {} for procedure in procedures: if self.mariner_mode == "runtime" and is_mariner_dependent_csp(procedure): - out.update( - materialize_mariner_runtime_facts( + runtime_facts = materialize_mariner_runtime_facts( procedure, lookup_class, str(geometry_type), @@ -495,7 +508,22 @@ class S52LookupRegistry: sounding_depth_m=sounding_depth_m, associated_depth_ranges=associated_depth_ranges, ) - ) + if procedure == "DEPCNT02": + if safe_dashed_instruction: + runtime_facts["s52_safe_dashed_rule_chain"] = materialize_rule_ir( + safe_dashed_instruction, + normalized_attrs, + source="lookup", + procedure="SAFECD", + ) + if safe_solid_instruction: + runtime_facts["s52_safe_solid_rule_chain"] = materialize_rule_ir( + safe_solid_instruction, + normalized_attrs, + source="lookup", + procedure="SAFECN", + ) + out.update(runtime_facts) unresolved.append(procedure) continue resolved = resolve_csp_hints( @@ -507,6 +535,8 @@ class S52LookupRegistry: next_safe_contour_m=next_safe_contour_m, sounding_depth_m=sounding_depth_m, associated_depth_ranges=associated_depth_ranges, + safe_dashed_instruction=safe_dashed_instruction, + safe_solid_instruction=safe_solid_instruction, ) if resolved: out.update(resolved)