fix: align S-52 lookup and mariner semantics

This commit is contained in:
OpenAI Codex
2026-09-21 06:13:00 +08:00
parent e5be7c198e
commit 043b022413
3 changed files with 133 additions and 13 deletions

View File

@@ -21,6 +21,8 @@ class S52MarinerSettings:
deep_contour_m: float = 6.0 deep_contour_m: float = 6.0
two_shades: bool = False two_shades: bool = False
symbolized_boundaries: 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 # 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 # profile into tile properties. The fixed S52MarinerSettings path remains
# available for OpenCPN/reference rendering. # available for OpenCPN/reference rendering.
MARINER_DEPENDENT_CSPS = frozenset( 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, settings: S52MarinerSettings,
*, *,
next_safe_contour_m: float | None = None, next_safe_contour_m: float | None = None,
safe_dashed_instruction: str | None = None,
safe_solid_instruction: str | None = None,
) -> tuple[str, bool]: ) -> tuple[str, bool]:
safe = False safe = False
if object_class == "DEPARE": if object_class == "DEPARE":
@@ -117,8 +121,16 @@ def resolve_depcnt02_instruction(
except (TypeError, ValueError): except (TypeError, ValueError):
quapos = 0 quapos = 0
if 2 <= quapos < 10: if 2 <= quapos < 10:
return ("LS(DASH,2,DEPSC)" if safe else "LS(DASH,1,DEPCN)"), safe return (
return ("LS(SOLD,2,DEPSC)" if safe else "LS(SOLD,1,DEPCN)"), safe (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: try:
litchr = int(attrs.get("LITCHR")) litchr = int(attrs.get("LITCHR"))
except (TypeError, ValueError): 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" text += f" {sigper:.1f}s" if abs(round(sigper) - sigper) > 0.01 else f" {sigper:.0f}s"
height = _optional_float(attrs.get("HEIGHT")) height = _optional_float(attrs.get("HEIGHT"))
if height is not None: 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")) valnmr = _optional_float(attrs.get("VALNMR"))
if valnmr is not None and not has_sector: if valnmr is not None and not has_sector:
text += f" {valnmr:.0f}Nm" text += f" {valnmr:.0f}Nm"
@@ -238,7 +253,9 @@ def _light_description(attrs: dict[str, Any]) -> str:
return text.strip() 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")) catlit = _int_list(attrs.get("CATLIT"))
valnmr = _optional_float(attrs.get("VALNMR")) valnmr = _optional_float(attrs.get("VALNMR"))
if valnmr is None: 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), "s52_sector_sweep_deg": float(sweep),
}) })
description = _light_description(attrs) description = _light_description(attrs, height_unit=settings.height_unit)
if description: if description:
safe = description.replace("'", "’") safe = description.replace("'", "’")
instruction += f";TX('{safe}',3,2,3,'15110',2,0,CHBLK,23)" instruction += f";TX('{safe}',3,2,3,'15110',2,0,CHBLK,23)"
@@ -319,9 +336,23 @@ def resolve_soundg03_instruction(
depth_value = 99999.0 depth_value = 99999.0
if depth_value < -1000.0: if depth_value < -1000.0:
depth_value = 0.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 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] = [] codes: list[str] = []
swept = 6 in _int_list(attrs.get("TECSOU")) swept = 6 in _int_list(attrs.get("TECSOU"))
low_quality = bool( low_quality = bool(
@@ -385,11 +416,13 @@ def resolve_soundg03_instruction(
extra = { extra = {
"s52_soundg_point_csp": "SOUNDG03", "s52_soundg_point_csp": "SOUNDG03",
"s52_sounding_depth_m": float(depth_value_in), "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_prefix": prefix,
"s52_sounding_symbol_sequence": ",".join(codes), "s52_sounding_symbol_sequence": ",".join(codes),
"s52_sounding_swept": int(swept), "s52_sounding_swept": int(swept),
"s52_sounding_low_quality": int(low_quality), "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 return instruction, extra
@@ -491,10 +524,15 @@ def _runtime_hazard_depth(
return 0.0 return 0.0
if catwrk in {4, 5}: if catwrk in {4, 5}:
return -15.0 return -15.0
if "CATWRK" in attrs and attrs.get("CATWRK") not in {None, ""}:
return None
if watlev == 3: if watlev == 3:
return 0.01 return 0.01
if watlev == 5: if watlev == 5:
return 0.0 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 return -15.0
expsou = _int_attr(attrs, "EXPSOU", 0) expsou = _int_attr(attrs, "EXPSOU", 0)
@@ -528,6 +566,35 @@ def materialize_mariner_runtime_facts(
"s52_runtime_csp": name, "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"}: if name in {"DEPARE01", "DEPARE02"}:
depth_min = _optional_float(attrs.get("DRVAL1")) depth_min = _optional_float(attrs.get("DRVAL1"))
depth_max = _optional_float(attrs.get("DRVAL2")) 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(): for source_key, target_key in scalar_codes.items():
if attrs.get(source_key) not in {None, ""}: if attrs.get(source_key) not in {None, ""}:
out[target_key] = _int_attr(attrs, source_key, 0) 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 = { list_codes = {
"QUASOU": "s52_quasou_codes", "QUASOU": "s52_quasou_codes",
@@ -955,6 +1025,8 @@ def resolve_csp_hints(
next_safe_contour_m: float | None = None, next_safe_contour_m: float | None = None,
sounding_depth_m: float | None = None, sounding_depth_m: float | None = None,
associated_depth_ranges: list[dict[str, Any]] | 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]: ) -> dict[str, Any]:
name = str(csp_name or "")[:8] name = str(csp_name or "")[:8]
instruction: str | None = None instruction: str | None = None
@@ -968,10 +1040,15 @@ def resolve_csp_hints(
attrs, attrs,
settings, settings,
next_safe_contour_m=next_safe_contour_m, 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) extra["s52_is_safety_contour"] = int(safe)
if safe:
extra["s52_display_base_promoted"] = 1
extra["s52_scamin_override"] = 100000001
elif name == "LIGHTS05": elif name == "LIGHTS05":
instruction, light_extra = resolve_lights05_instruction(attrs) instruction, light_extra = resolve_lights05_instruction(attrs, settings)
extra.update(light_extra) extra.update(light_extra)
elif name == "SOUNDG02": elif name == "SOUNDG02":
if sounding_depth_m is None: if sounding_depth_m is None:

View File

@@ -67,6 +67,11 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset(
"s52_expsou_code", "s52_expsou_code",
"s52_catobs_code", "s52_catobs_code",
"s52_catwrk_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_color",
"s52_line_complex_code", "s52_line_complex_code",
"s52_line_count", "s52_line_count",
@@ -85,6 +90,12 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset(
"s52_source_object_class", "s52_source_object_class",
"s52_source_geometry", "s52_source_geometry",
"s52_scamin_override", "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_color",
"s52_light_arc_radius_mm", "s52_light_arc_radius_mm",
"s52_light_mode", "s52_light_mode",
@@ -111,6 +122,8 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset(
"s52_symbol_rule_sequence", "s52_symbol_rule_sequence",
"s52_symbol_sequence", "s52_symbol_sequence",
"s52_sounding_depth_m", "s52_sounding_depth_m",
"s52_sounding_display_depth",
"s52_sounding_depth_unit",
"s52_sounding_is_shallow", "s52_sounding_is_shallow",
"s52_sounding_low_quality", "s52_sounding_low_quality",
"s52_sounding_prefix", "s52_sounding_prefix",

View File

@@ -448,6 +448,18 @@ class S52LookupRegistry:
geometry_type=lookup_geometry, 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( def properties(
self, self,
object_class: str, object_class: str,
@@ -482,12 +494,13 @@ class S52LookupRegistry:
for token in str(out.get("s52_csp") or "").split(",") for token in str(out.get("s52_csp") or "").split(",")
if token.strip() if token.strip()
] ]
safe_dashed_instruction = self._plain_area_instruction("SAFECD")
safe_solid_instruction = self._plain_area_instruction("SAFECN")
unresolved: list[str] = [] unresolved: list[str] = []
resolved_by_procedure: dict[str, dict[str, Any]] = {} resolved_by_procedure: dict[str, dict[str, Any]] = {}
for procedure in procedures: for procedure in procedures:
if self.mariner_mode == "runtime" and is_mariner_dependent_csp(procedure): if self.mariner_mode == "runtime" and is_mariner_dependent_csp(procedure):
out.update( runtime_facts = materialize_mariner_runtime_facts(
materialize_mariner_runtime_facts(
procedure, procedure,
lookup_class, lookup_class,
str(geometry_type), str(geometry_type),
@@ -495,7 +508,22 @@ class S52LookupRegistry:
sounding_depth_m=sounding_depth_m, sounding_depth_m=sounding_depth_m,
associated_depth_ranges=associated_depth_ranges, 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) unresolved.append(procedure)
continue continue
resolved = resolve_csp_hints( resolved = resolve_csp_hints(
@@ -507,6 +535,8 @@ class S52LookupRegistry:
next_safe_contour_m=next_safe_contour_m, next_safe_contour_m=next_safe_contour_m,
sounding_depth_m=sounding_depth_m, sounding_depth_m=sounding_depth_m,
associated_depth_ranges=associated_depth_ranges, associated_depth_ranges=associated_depth_ranges,
safe_dashed_instruction=safe_dashed_instruction,
safe_solid_instruction=safe_solid_instruction,
) )
if resolved: if resolved:
out.update(resolved) out.update(resolved)