980 lines
34 KiB
Python
980 lines
34 KiB
Python
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
import math
|
||
from typing import Any
|
||
|
||
from .presentation import (
|
||
materialize_symbol_hints,
|
||
parse_instruction_string,
|
||
static_rule_hints,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class S52MarinerSettings:
|
||
shallow_contour_m: float = 2.0
|
||
safety_contour_m: float = 3.0
|
||
safety_depth_m: float = 3.0
|
||
deep_contour_m: float = 6.0
|
||
two_shades: bool = False
|
||
symbolized_boundaries: bool = False
|
||
|
||
|
||
# These CSPs depend on mariner-selected depth thresholds. NavSea delivery PBF
|
||
# must preserve their chart facts instead of freezing the current comparison
|
||
# profile into tile properties. The fixed S52MarinerSettings path remains
|
||
# available for OpenCPN/reference rendering.
|
||
MARINER_DEPENDENT_CSPS = frozenset(
|
||
{"DEPARE01", "DEPARE02", "DEPCNT02", "SOUNDG02", "OBSTRN04", "WRECKS02"}
|
||
)
|
||
|
||
|
||
def is_mariner_dependent_csp(csp_name: str) -> bool:
|
||
return str(csp_name or "")[:8] in MARINER_DEPENDENT_CSPS
|
||
|
||
|
||
def _float(attrs: dict[str, Any], key: str, default: float) -> float:
|
||
try:
|
||
return float(attrs.get(key, default))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def resolve_depare01_instruction(
|
||
object_class: str,
|
||
attrs: dict[str, Any],
|
||
settings: S52MarinerSettings,
|
||
) -> str:
|
||
drval1_found = attrs.get("DRVAL1") not in {None, ""}
|
||
drval1 = _float(attrs, "DRVAL1", -1.0)
|
||
drval2 = _float(attrs, "DRVAL2", drval1 + 0.01)
|
||
if drval2 <= drval1:
|
||
drval2 = drval1 + 0.01
|
||
|
||
color = "DEPIT"
|
||
if drval1 >= 0.0 and drval2 > 0.0:
|
||
color = "DEPVS"
|
||
|
||
if settings.two_shades:
|
||
if (
|
||
drval1 >= settings.safety_contour_m
|
||
and drval2 > settings.safety_contour_m
|
||
):
|
||
color = "DEPDW"
|
||
else:
|
||
if (
|
||
drval1 >= settings.shallow_contour_m
|
||
and drval2 > settings.shallow_contour_m
|
||
):
|
||
color = "DEPMS"
|
||
if (
|
||
drval1 >= settings.safety_contour_m
|
||
and drval2 > settings.safety_contour_m
|
||
):
|
||
color = "DEPMD"
|
||
if (
|
||
drval1 >= settings.deep_contour_m
|
||
and drval2 > settings.deep_contour_m
|
||
):
|
||
color = "DEPDW"
|
||
|
||
if object_class == "DRGARE" and not drval1_found:
|
||
color = "DEPMD"
|
||
|
||
instruction = f"AC({color})"
|
||
if object_class == "DRGARE":
|
||
instruction += ";AP(DRGARE01);LS(DASH,1,CHGRF)"
|
||
return instruction
|
||
|
||
|
||
def resolve_depcnt02_instruction(
|
||
object_class: str,
|
||
attrs: dict[str, Any],
|
||
settings: S52MarinerSettings,
|
||
*,
|
||
next_safe_contour_m: float | None = None,
|
||
) -> tuple[str, bool]:
|
||
safe = False
|
||
if object_class == "DEPARE":
|
||
drval1 = _float(attrs, "DRVAL1", 0.0)
|
||
drval2 = _float(attrs, "DRVAL2", drval1)
|
||
if drval1 <= settings.safety_contour_m <= drval2:
|
||
safe = True
|
||
elif next_safe_contour_m is not None:
|
||
safe = abs(drval1 - next_safe_contour_m) < 1e-4
|
||
else:
|
||
valdco = _float(attrs, "VALDCO", 0.0)
|
||
if abs(valdco - settings.safety_contour_m) < 1e-4:
|
||
safe = True
|
||
elif next_safe_contour_m is not None:
|
||
safe = abs(valdco - next_safe_contour_m) < 1e-4
|
||
|
||
try:
|
||
quapos = int(attrs.get("QUAPOS") or 0)
|
||
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
|
||
|
||
|
||
|
||
def _int_list(value: Any) -> list[int]:
|
||
if value is None or value == "":
|
||
return []
|
||
if isinstance(value, (list, tuple)):
|
||
raw = list(value)
|
||
elif isinstance(value, str):
|
||
raw = [part.strip() for part in value.strip().strip("[]").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 _optional_float(value: Any) -> float | None:
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _light_color_ref(colours: list[int], *, sector: bool = False) -> str:
|
||
if len(colours) == 1:
|
||
if 3 in colours:
|
||
return "LITRD"
|
||
if 4 in colours:
|
||
return "LITGN"
|
||
# OpenCPN uses orange (11) for sectors, amber (9) for flares/circles.
|
||
if any(code in colours for code in (1, 6, 11 if sector else 9)):
|
||
return "LITYW"
|
||
return "CHMGD"
|
||
if len(colours) == 2:
|
||
if 1 in colours and 3 in colours:
|
||
return "LITRD"
|
||
if 1 in colours and 4 in colours:
|
||
return "LITGN"
|
||
return "CHMGD"
|
||
|
||
|
||
def _light_flare_symbol(colours: list[int]) -> str:
|
||
ref = _light_color_ref(colours)
|
||
return {
|
||
"LITRD": "LIGHTS11",
|
||
"LITGN": "LIGHTS12",
|
||
"LITYW": "LIGHTS13",
|
||
}.get(ref, "LITDEF11")
|
||
|
||
|
||
def _light_radius_mm(valnmr: float) -> float:
|
||
if valnmr > 0:
|
||
if valnmr < 7.0:
|
||
return 3.0
|
||
if valnmr < 15.0:
|
||
return 10.0
|
||
if valnmr < 30.0:
|
||
return 15.0
|
||
return 20.0
|
||
return 3.0
|
||
|
||
|
||
_LIGHT_CHARACTER = {
|
||
1: "F", 2: "Fl", 3: "LFl", 4: "Q", 5: "VQ", 6: "UQ",
|
||
7: "Iso", 8: "Occ", 9: "IQ", 10: "IVQ", 11: "IUQ", 12: "Mo",
|
||
13: "F + Fl", 14: "Fl + LFl", 15: "Occ + Fl", 16: "F + LFl",
|
||
17: "Al Occ", 18: "Al LFl", 19: "Al Fl", 20: "Al Grp",
|
||
21: "F", 22: "F", 23: "F", 24: "F", 25: "Q + LFl",
|
||
26: "VQ + LFl", 27: "UQ + LFl", 28: "Alt", 29: "F + Alt",
|
||
}
|
||
|
||
|
||
def _light_description(attrs: dict[str, Any]) -> str:
|
||
try:
|
||
litchr = int(attrs.get("LITCHR"))
|
||
except (TypeError, ValueError):
|
||
litchr = -9
|
||
text = _LIGHT_CHARACTER.get(litchr, "")
|
||
two_groups = litchr in {13, 14, 15, 16, 25, 26, 27, 29}
|
||
siggrp = str(attrs.get("SIGGRP") or "").strip()
|
||
if two_groups:
|
||
groups = [part for part in siggrp.replace("(", ")").split(")") if part]
|
||
first = groups[0][0] if groups else "?"
|
||
left, right = text.split(" + ", 1)
|
||
text = f"{left}({first}) + {right}"
|
||
if len(groups) > 1 and groups[1] != "1":
|
||
text += f"({groups[1]})"
|
||
elif siggrp and siggrp != "(1)":
|
||
text += siggrp
|
||
|
||
# Description suppression depends on SECTR1 presence, not sector validity.
|
||
has_sector = "SECTR1" in attrs
|
||
if not has_sector:
|
||
colour_codes = {1: "W", 3: "R", 4: "G", 6: "Y"}
|
||
colours = "".join(colour_codes.get(code, "") for code in _int_list(attrs.get("COLOUR")))
|
||
if colours:
|
||
text += f" {colours}"
|
||
|
||
sigper = _optional_float(attrs.get("SIGPER"))
|
||
if sigper is not None:
|
||
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"
|
||
valnmr = _optional_float(attrs.get("VALNMR"))
|
||
if valnmr is not None and not has_sector:
|
||
text += f" {valnmr:.0f}Nm"
|
||
if litchr in {21, 23}:
|
||
text += " (vert)"
|
||
elif litchr in {22, 24}:
|
||
text += " (horz)"
|
||
return text.strip()
|
||
|
||
|
||
def resolve_lights05_instruction(attrs: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
||
catlit = _int_list(attrs.get("CATLIT"))
|
||
valnmr = _optional_float(attrs.get("VALNMR"))
|
||
if valnmr is None:
|
||
valnmr = 9.0
|
||
|
||
sectr1 = _optional_float(attrs.get("SECTR1"))
|
||
sectr2 = _optional_float(attrs.get("SECTR2"))
|
||
has_sector = sectr1 not in (None, -9.0) and sectr2 not in (None, -9.0)
|
||
colours = _int_list(attrs.get("COLOUR")) or [12]
|
||
extra: dict[str, Any] = {
|
||
"s52_light_nominal_range_nm": float(valnmr),
|
||
}
|
||
|
||
if 8 in catlit or 11 in catlit:
|
||
instruction = "SY(LIGHTS82)"
|
||
extra["s52_light_mode"] = "special"
|
||
elif 9 in catlit:
|
||
instruction = "SY(LIGHTS81)"
|
||
extra["s52_light_mode"] = "special"
|
||
elif not has_sector:
|
||
if any(code in catlit for code in (1, 14)):
|
||
instruction = "SY(QUESMRK1)"
|
||
extra["s52_light_mode"] = "directional_unknown"
|
||
elif valnmr < 10.0:
|
||
instruction = f"SY({_light_flare_symbol(colours)},135)"
|
||
extra["s52_light_mode"] = "flare"
|
||
extra["s52_light_rotation_deg"] = 135.0
|
||
else:
|
||
radius = _light_radius_mm(valnmr)
|
||
color = _light_color_ref(colours)
|
||
radius_adjust = {"LITRD": 1.0, "LITGN": 0.0, "LITYW": 2.0}.get(color, 3.0 if colours == [12] else 5.0)
|
||
instruction = f"CA(OUTLW,4,{color},2,0,360,{radius + radius_adjust:.1f},0)"
|
||
extra["s52_light_mode"] = "all_round"
|
||
else:
|
||
assert sectr1 is not None and sectr2 is not None
|
||
sweep = (sectr2 - sectr1) % 360.0
|
||
if sweep < 1.0:
|
||
radius = _light_radius_mm(valnmr)
|
||
color = _light_color_ref(colours)
|
||
radius_adjust = {"LITRD": 1.0, "LITGN": 0.0, "LITYW": 2.0}.get(color, 3.0 if colours == [12] else 5.0)
|
||
instruction = f"CA(OUTLW,4,{color},2,0,360,{radius + radius_adjust:.1f},0)"
|
||
extra["s52_light_mode"] = "all_round"
|
||
else:
|
||
arc_radius = 10.0 if 0 < valnmr < 15.0 else 15.0 if 15.0 <= valnmr < 30.0 else 20.0
|
||
color = _light_color_ref(colours, sector=True)
|
||
outline_color, outline_width, arc_width = "OUTLW", 4, 2
|
||
litvis = _int_list(attrs.get("LITVIS"))
|
||
if any(code in litvis for code in (3, 7, 8)):
|
||
outline_color, outline_width, color, arc_width = "CHBLK", 4, "CHBRN", 1
|
||
extra["s52_light_obscured"] = 1
|
||
start = (sectr1 + 180.0) % 360.0
|
||
end = (sectr2 + 180.0) % 360.0
|
||
if end <= start:
|
||
end += 360.0
|
||
instruction = f"CA({outline_color},{outline_width},{color},{arc_width},{start:.1f},{end:.1f},{arc_radius:.1f},25.0)"
|
||
extra.update({
|
||
"s52_light_mode": "sector",
|
||
"s52_sector_start_deg": float(sectr1),
|
||
"s52_sector_end_deg": float(sectr2),
|
||
"s52_sector_sweep_deg": float(sweep),
|
||
})
|
||
|
||
description = _light_description(attrs)
|
||
if description:
|
||
safe = description.replace("'", "’")
|
||
instruction += f";TX('{safe}',3,2,3,'15110',2,0,CHBLK,23)"
|
||
extra["s52_light_label_text"] = description
|
||
return instruction, extra
|
||
|
||
|
||
def resolve_soundg03_instruction(
|
||
attrs: dict[str, Any],
|
||
depth_value_in: float,
|
||
settings: S52MarinerSettings,
|
||
) -> tuple[str, dict[str, Any]]:
|
||
depth_value = float(depth_value_in)
|
||
if depth_value > 40000.0:
|
||
depth_value = 99999.0
|
||
if depth_value < -1000.0:
|
||
depth_value = 0.0
|
||
depth_value += 0.01 if depth_value > 0.0 else -0.01
|
||
|
||
prefix = "SOUNDS" if depth_value <= settings.safety_depth_m else "SOUNDG"
|
||
codes: list[str] = []
|
||
swept = 6 in _int_list(attrs.get("TECSOU"))
|
||
low_quality = bool(
|
||
set(_int_list(attrs.get("QUASOU"))) & {3, 4, 5, 8, 9}
|
||
or 18 in _int_list(attrs.get("STATUS"))
|
||
or any(2 <= q < 10 for q in _int_list(attrs.get("QUAPOS")))
|
||
)
|
||
if swept:
|
||
codes.append(f"{prefix}B1")
|
||
if low_quality:
|
||
codes.append(f"{prefix}C2")
|
||
|
||
leading = int(abs(depth_value))
|
||
absolute = abs(depth_value)
|
||
|
||
if absolute < 10.0:
|
||
fraction = int(abs((absolute - leading) * 10))
|
||
codes.append(f"{prefix}1{leading}")
|
||
if fraction > 0:
|
||
codes.append(f"{prefix}5{fraction}")
|
||
if depth_value < 0.0:
|
||
codes.append(f"{prefix}A1")
|
||
elif absolute < 31.0 and (absolute - math.floor(leading)) != 0.0:
|
||
fraction = int(abs(absolute - math.floor(leading)) * 10)
|
||
if leading >= 10:
|
||
codes.append(f"{prefix}2{leading // 10}")
|
||
first_digit = math.floor(leading / 10)
|
||
second_digit = int(math.floor(leading - first_digit * 10))
|
||
codes.append(f"{prefix}1{second_digit}")
|
||
if fraction > 0:
|
||
codes.append(f"{prefix}5{fraction}")
|
||
if depth_value < 0.0:
|
||
codes.append(f"{prefix}A1")
|
||
elif absolute < 100.0:
|
||
first_digit = int(math.floor(leading / 10))
|
||
second_digit = int(math.floor(leading - first_digit * 10))
|
||
if depth_value < 0.0:
|
||
codes.extend([f"{prefix}2{first_digit}", f"{prefix}1{second_digit}", f"{prefix}A1"])
|
||
else:
|
||
codes.extend([f"{prefix}1{first_digit}", f"{prefix}0{second_digit}"])
|
||
elif depth_value < 1000.0:
|
||
first = int(math.floor(leading / 100))
|
||
second = int(math.floor((leading - first * 100) / 10))
|
||
third = int(math.floor(leading - first * 100 - second * 10))
|
||
codes.extend([f"{prefix}2{first}", f"{prefix}1{second}", f"{prefix}0{third}"])
|
||
elif depth_value < 10000.0:
|
||
first = int(math.floor(leading / 1000))
|
||
second = int(math.floor((leading - first * 1000) / 100))
|
||
third = int(math.floor((leading - first * 1000 - second * 100) / 10))
|
||
last = int(math.floor(leading - first * 1000 - second * 100 - third * 10))
|
||
codes.extend([f"{prefix}2{first}", f"{prefix}1{second}", f"{prefix}0{third}", f"{prefix}4{last}"])
|
||
else:
|
||
first = int(math.floor(leading / 10000))
|
||
second = int(math.floor((leading - first * 10000) / 1000))
|
||
third = int(math.floor((leading - first * 10000 - second * 1000) / 100))
|
||
fourth = int(math.floor((leading - first * 10000 - second * 1000 - third * 100) / 10))
|
||
last = int(math.floor(leading - first * 10000 - second * 1000 - third * 100 - fourth * 10))
|
||
codes.extend([f"{prefix}3{first}", f"{prefix}2{second}", f"{prefix}1{third}", f"{prefix}0{fourth}", f"{prefix}4{last}"])
|
||
|
||
instruction = ";".join(f"SY({code})" for code in codes)
|
||
extra = {
|
||
"s52_soundg_point_csp": "SOUNDG03",
|
||
"s52_sounding_depth_m": float(depth_value_in),
|
||
"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),
|
||
}
|
||
return instruction, extra
|
||
|
||
|
||
def _int_attr(attrs: dict[str, Any], key: str, default: int = -9) -> int:
|
||
try:
|
||
return int(attrs.get(key))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _hazard_depth_context(
|
||
associated_depth_ranges: list[dict[str, Any]] | None,
|
||
) -> list[dict[str, Any]]:
|
||
return list(associated_depth_ranges or [])
|
||
|
||
|
||
def resolve_udwhaz03_instruction(
|
||
attrs: dict[str, Any],
|
||
depth_value: float | None,
|
||
settings: S52MarinerSettings,
|
||
*,
|
||
associated_depth_ranges: list[dict[str, Any]] | None = None,
|
||
) -> tuple[str, dict[str, Any]]:
|
||
danger = False
|
||
expsou = _int_attr(attrs, "EXPSOU", 0)
|
||
if depth_value is None and expsou != 1:
|
||
danger = True
|
||
if not danger and (
|
||
expsou == 1
|
||
or (depth_value is not None and depth_value <= settings.safety_contour_m)
|
||
):
|
||
for item in _hazard_depth_context(associated_depth_ranges):
|
||
primitive = str(item.get("geometry_type") or "AREA").upper()
|
||
try:
|
||
drval1 = float(item.get("drval1") or 0.0)
|
||
except (TypeError, ValueError):
|
||
drval1 = 0.0
|
||
try:
|
||
drval2 = float(item.get("drval2") or 0.0)
|
||
except (TypeError, ValueError):
|
||
drval2 = 0.0
|
||
if primitive == "LINE":
|
||
if drval2 < settings.safety_contour_m:
|
||
danger = True
|
||
break
|
||
elif drval1 >= settings.safety_contour_m and expsou != 1:
|
||
danger = True
|
||
break
|
||
|
||
extra = {
|
||
"s52_isolated_danger": int(danger),
|
||
"s52_display_base_promoted": int(danger),
|
||
}
|
||
if depth_value is not None:
|
||
extra["s52_hazard_depth_m"] = float(depth_value)
|
||
if not danger:
|
||
return "", extra
|
||
watlev = _int_attr(attrs, "WATLEV", 0)
|
||
if watlev in {1, 2}:
|
||
return "", extra
|
||
return "SY(ISODGR51)", extra
|
||
|
||
|
||
def _area_least_depth(
|
||
associated_depth_ranges: list[dict[str, Any]] | None,
|
||
) -> float | None:
|
||
values: list[float] = []
|
||
for item in _hazard_depth_context(associated_depth_ranges):
|
||
try:
|
||
values.append(float(item.get("drval1")))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return min(values) if values else None
|
||
|
||
|
||
def _runtime_hazard_depth(
|
||
csp_name: str,
|
||
object_class: str,
|
||
geometry_type: str,
|
||
attrs: dict[str, Any],
|
||
associated_depth_ranges: list[dict[str, Any]] | None,
|
||
) -> float | None:
|
||
valsou = _optional_float(attrs.get("VALSOU"))
|
||
if valsou is not None:
|
||
return valsou
|
||
if str(geometry_type).upper() == "AREA":
|
||
area_depth = _area_least_depth(associated_depth_ranges)
|
||
if area_depth is not None:
|
||
return area_depth
|
||
|
||
name = str(csp_name or "")[:8]
|
||
watlev = _int_attr(attrs, "WATLEV", -9)
|
||
if name == "WRECKS02":
|
||
catwrk = _int_attr(attrs, "CATWRK", -9)
|
||
if catwrk == 1:
|
||
return 20.0
|
||
if catwrk == 2:
|
||
return 0.0
|
||
if catwrk in {4, 5}:
|
||
return -15.0
|
||
if watlev == 3:
|
||
return 0.01
|
||
if watlev == 5:
|
||
return 0.0
|
||
return -15.0
|
||
|
||
expsou = _int_attr(attrs, "EXPSOU", 0)
|
||
if expsou == 1:
|
||
return None
|
||
if _int_attr(attrs, "CATOBS", 0) == 6:
|
||
return 0.01
|
||
if watlev == 5:
|
||
return 0.0
|
||
if watlev == 3:
|
||
return 0.01
|
||
return -15.0
|
||
|
||
|
||
def materialize_mariner_runtime_facts(
|
||
csp_name: str,
|
||
object_class: str,
|
||
geometry_type: str,
|
||
attrs: dict[str, Any],
|
||
*,
|
||
sounding_depth_m: float | None = None,
|
||
associated_depth_ranges: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Emit mariner-neutral inputs needed by NavSea runtime presentation."""
|
||
name = str(csp_name or "")[:8]
|
||
if name not in MARINER_DEPENDENT_CSPS:
|
||
return {}
|
||
|
||
out: dict[str, Any] = {
|
||
"s52_mariner_dependent": 1,
|
||
"s52_runtime_csp": name,
|
||
}
|
||
|
||
if name in {"DEPARE01", "DEPARE02"}:
|
||
depth_min = _optional_float(attrs.get("DRVAL1"))
|
||
depth_max = _optional_float(attrs.get("DRVAL2"))
|
||
if depth_min is not None:
|
||
out["depth_min_m"] = depth_min
|
||
if depth_max is not None:
|
||
out["depth_max_m"] = depth_max
|
||
|
||
if name == "DEPCNT02":
|
||
contour = _optional_float(attrs.get("VALDCO"))
|
||
if contour is not None:
|
||
out["contour_depth_m"] = contour
|
||
|
||
if name == "SOUNDG02" and sounding_depth_m is not None:
|
||
out["s52_sounding_depth_m"] = float(sounding_depth_m)
|
||
out["s52_sounding_swept"] = int(6 in _int_list(attrs.get("TECSOU")))
|
||
out["s52_sounding_low_quality"] = int(
|
||
bool(
|
||
set(_int_list(attrs.get("QUASOU"))) & {3, 4, 5, 8, 9}
|
||
or 18 in _int_list(attrs.get("STATUS"))
|
||
or any(2 <= q < 10 for q in _int_list(attrs.get("QUAPOS")))
|
||
)
|
||
)
|
||
|
||
if name in {"OBSTRN04", "WRECKS02"}:
|
||
hazard_depth = _runtime_hazard_depth(
|
||
name, object_class, geometry_type, attrs, associated_depth_ranges
|
||
)
|
||
if hazard_depth is not None:
|
||
out["s52_hazard_depth_m"] = float(hazard_depth)
|
||
ranges = _hazard_depth_context(associated_depth_ranges)
|
||
lows = [_optional_float(item.get("drval1")) for item in ranges]
|
||
highs = [_optional_float(item.get("drval2")) for item in ranges]
|
||
lows = [value for value in lows if value is not None]
|
||
highs = [value for value in highs if value is not None]
|
||
if lows:
|
||
out["s52_surrounding_max_drval1_m"] = max(lows)
|
||
if highs:
|
||
out["s52_surrounding_min_drval2_m"] = min(highs)
|
||
|
||
scalar_codes = {
|
||
"QUAPOS": "s52_quapos_code",
|
||
"WATLEV": "s52_watlev_code",
|
||
"EXPSOU": "s52_expsou_code",
|
||
"CATOBS": "s52_catobs_code",
|
||
"CATWRK": "s52_catwrk_code",
|
||
}
|
||
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)
|
||
|
||
list_codes = {
|
||
"QUASOU": "s52_quasou_codes",
|
||
"TECSOU": "s52_tecsou_codes",
|
||
"STATUS": "s52_status_codes",
|
||
}
|
||
for source_key, target_key in list_codes.items():
|
||
values = _int_list(attrs.get(source_key))
|
||
if values:
|
||
out[target_key] = values
|
||
return out
|
||
|
||
|
||
def resolve_obstrn04_instruction(
|
||
object_class: str,
|
||
geometry_type: str,
|
||
attrs: dict[str, Any],
|
||
settings: S52MarinerSettings,
|
||
*,
|
||
associated_depth_ranges: list[dict[str, Any]] | None = None,
|
||
) -> tuple[str, dict[str, Any]]:
|
||
valsou = _optional_float(attrs.get("VALSOU"))
|
||
depth_value = valsou
|
||
if depth_value is None and str(geometry_type).upper() == "AREA":
|
||
depth_value = _area_least_depth(associated_depth_ranges)
|
||
if depth_value is None:
|
||
catobs = _int_attr(attrs, "CATOBS", 0)
|
||
watlev = _int_attr(attrs, "WATLEV", 0)
|
||
expsou = _int_attr(attrs, "EXPSOU", 0)
|
||
if expsou != 1:
|
||
if catobs == 6:
|
||
depth_value = 0.01
|
||
elif watlev == 5:
|
||
depth_value = 0.0
|
||
elif watlev == 3:
|
||
depth_value = 0.01
|
||
else:
|
||
depth_value = -15.0
|
||
|
||
udw, extra = resolve_udwhaz03_instruction(
|
||
attrs,
|
||
depth_value,
|
||
settings,
|
||
associated_depth_ranges=associated_depth_ranges,
|
||
)
|
||
extra["s52_hazard_procedure"] = "OBSTRN04"
|
||
geom = str(geometry_type).upper()
|
||
if geom == "POINT":
|
||
if udw:
|
||
return udw, extra
|
||
sounding = False
|
||
watlev = _int_attr(attrs, "WATLEV", -9)
|
||
if valsou is not None:
|
||
if valsou <= 20.0:
|
||
if object_class == "UWTROC":
|
||
if watlev in {4, 5}:
|
||
instruction = "SY(UWTROC04)"
|
||
else:
|
||
instruction = "SY(DANGER51)"
|
||
sounding = True
|
||
else:
|
||
if watlev == -9:
|
||
instruction = "SY(DANGER01)"
|
||
sounding = True
|
||
elif watlev in {1, 2}:
|
||
instruction = "SY(LNDARE01)"
|
||
elif watlev == 3:
|
||
instruction = "SY(DANGER52)"
|
||
sounding = True
|
||
elif watlev in {4, 5}:
|
||
instruction = "SY(DANGER53)"
|
||
sounding = True
|
||
else:
|
||
instruction = "SY(DANGER51)"
|
||
sounding = True
|
||
else:
|
||
instruction = "SY(DANGER52)"
|
||
sounding = True
|
||
else:
|
||
if object_class == "UWTROC":
|
||
instruction = (
|
||
"SY(LNDARE01)" if watlev == 2 else
|
||
"SY(UWTROC03)" if watlev == 3 else
|
||
"SY(UWTROC04)"
|
||
)
|
||
else:
|
||
instruction = (
|
||
"SY(OBSTRN11)" if watlev in {1, 2} else
|
||
"SY(OBSTRN03)" if watlev in {4, 5} else
|
||
"SY(OBSTRN01)"
|
||
)
|
||
if sounding and valsou is not None:
|
||
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
|
||
instruction += ";" + snd if snd else ""
|
||
extra.update(snd_extra)
|
||
return instruction, extra
|
||
|
||
if geom == "LINE":
|
||
quapos = _int_attr(attrs, "QUAPOS", 0)
|
||
if 2 <= quapos < 10:
|
||
return ("LC(LOWACC41)" if udw else "LC(LOWACC31)"), extra
|
||
if udw:
|
||
return "LS(DOTT,2,CHBLK)", extra
|
||
if valsou is None or valsou <= 20.0:
|
||
return "LS(DOTT,2,CHBLK)", extra
|
||
return "LS(DASH,2,CHBLK)", extra
|
||
|
||
# AREA
|
||
if udw:
|
||
return "AC(DEPVS);AP(FOULAR01);LS(DOTT,2,CHBLK);" + udw, extra
|
||
watlev = _int_attr(attrs, "WATLEV", -9)
|
||
if valsou is not None:
|
||
line = "LS(DOTT,2,CHBLK)" if valsou <= 20.0 else "LS(DASH,2,CHBLK)"
|
||
return line, extra
|
||
if watlev in {1, 2}:
|
||
return "AC(CHBRN);LS(SOLD,2,CSTLN)", extra
|
||
if watlev == 4:
|
||
return "AC(DEPIT);LS(DASH,2,CSTLN)", extra
|
||
if _int_attr(attrs, "CATOBS", -9) == 6 and watlev in {3, 5}:
|
||
return "AC(DEPVS);AP(FOULAR01);LS(DOTT,2,CHBLK)", extra
|
||
return "AC(DEPVS);LS(DOTT,2,CHBLK)", extra
|
||
|
||
|
||
def resolve_wrecks02_instruction(
|
||
geometry_type: str,
|
||
attrs: dict[str, Any],
|
||
settings: S52MarinerSettings,
|
||
*,
|
||
associated_depth_ranges: list[dict[str, Any]] | None = None,
|
||
) -> tuple[str, dict[str, Any]]:
|
||
valsou = _optional_float(attrs.get("VALSOU"))
|
||
watlev = _int_attr(attrs, "WATLEV", -9)
|
||
catwrk = _int_attr(attrs, "CATWRK", -9)
|
||
depth_value = valsou
|
||
if depth_value is None and str(geometry_type).upper() == "AREA":
|
||
depth_value = _area_least_depth(associated_depth_ranges)
|
||
if depth_value is None:
|
||
if catwrk == 1:
|
||
depth_value = 20.0
|
||
elif catwrk == 2:
|
||
depth_value = 0.0
|
||
elif catwrk in {4, 5}:
|
||
depth_value = -15.0
|
||
elif watlev == 3:
|
||
depth_value = 0.01
|
||
elif watlev == 5:
|
||
depth_value = 0.0
|
||
else:
|
||
depth_value = -15.0
|
||
|
||
quasou7 = 7 in _int_list(attrs.get("QUASOU"))
|
||
if quasou7:
|
||
udw, udw_extra = "", {"s52_isolated_danger": 0, "s52_display_base_promoted": 0}
|
||
else:
|
||
udw, udw_extra = resolve_udwhaz03_instruction(
|
||
attrs, depth_value, settings,
|
||
associated_depth_ranges=associated_depth_ranges,
|
||
)
|
||
extra = dict(udw_extra)
|
||
extra.update({"s52_hazard_procedure": "WRECKS02", "s52_hazard_depth_m": float(depth_value)})
|
||
geom = str(geometry_type).upper()
|
||
if geom == "POINT":
|
||
if udw:
|
||
return udw, extra
|
||
if valsou is not None:
|
||
symbol = "DANGER51" if valsou < settings.safety_contour_m else "DANGER52"
|
||
instruction = f"SY({symbol});TX('Wk',3,1,2,'15110',2,0,CHBLK,21)"
|
||
if quasou7:
|
||
instruction += ";SY(WRECKS07)"
|
||
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
|
||
if snd:
|
||
instruction += ";" + snd
|
||
extra.update(snd_extra)
|
||
return instruction, extra
|
||
if catwrk == 1 and watlev == 3:
|
||
return "SY(WRECKS04)", extra
|
||
if catwrk == 2 and watlev == 3:
|
||
return "SY(WRECKS05)", extra
|
||
if catwrk in {4, 5} or watlev in {1, 2, 4, 5}:
|
||
return "SY(WRECKS01)", extra
|
||
return "SY(WRECKS05)", extra
|
||
|
||
quapos = _int_attr(attrs, "QUAPOS", 0)
|
||
if 2 <= quapos < 10:
|
||
line = "LC(LOWACC41)"
|
||
elif udw or (valsou is not None and valsou <= 20.0):
|
||
line = "LS(DOTT,2,CHBLK)"
|
||
elif valsou is not None:
|
||
line = "LS(DASH,2,CHBLK)"
|
||
elif watlev in {1, 2}:
|
||
line = "LS(SOLD,2,CSTLN)"
|
||
elif watlev == 4:
|
||
line = "LS(DASH,2,CSTLN)"
|
||
else:
|
||
line = "LS(DOTT,2,CSTLN)"
|
||
instruction = line
|
||
if valsou is not None and valsou <= 20.0:
|
||
instruction += ";" + udw if udw else ""
|
||
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
|
||
if snd:
|
||
instruction += ";" + snd
|
||
extra.update(snd_extra)
|
||
elif valsou is None:
|
||
color = "CHBRN" if watlev in {1, 2} else "DEPIT" if watlev == 4 else "DEPVS"
|
||
instruction += f";AC({color})"
|
||
if udw:
|
||
instruction += ";" + udw
|
||
return instruction, extra
|
||
|
||
|
||
def resolve_quapnt01_instruction(attrs: dict[str, Any]) -> str:
|
||
quapos = _int_attr(attrs, "QUAPOS", 10)
|
||
if not (2 <= quapos < 10):
|
||
return ""
|
||
if quapos == 4:
|
||
return "SY(QUAPOS01)"
|
||
if quapos == 5:
|
||
return "SY(QUAPOS02)"
|
||
if quapos in {7, 8}:
|
||
return "SY(QUAPOS03)"
|
||
return "SY(LOWACC03)"
|
||
|
||
|
||
def resolve_qualin01_instruction(
|
||
object_class: str,
|
||
attrs: dict[str, Any],
|
||
) -> str:
|
||
quapos_present = attrs.get("QUAPOS") is not None
|
||
quapos = _int_attr(attrs, "QUAPOS", 0)
|
||
if quapos_present:
|
||
if 2 <= quapos < 10:
|
||
return "LC(LOWACC21)"
|
||
return ""
|
||
if object_class == "COALNE":
|
||
conrad = _int_attr(attrs, "CONRAD", 0)
|
||
if conrad == 1:
|
||
return "LS(SOLD,3,CHMGF);LS(SOLD,1,CSTLN)"
|
||
return "LS(SOLD,1,CSTLN)"
|
||
|
||
|
||
def resolve_quapos01_instruction(
|
||
object_class: str,
|
||
geometry_type: str,
|
||
attrs: dict[str, Any],
|
||
) -> str:
|
||
if str(geometry_type).upper() == "LINE":
|
||
return resolve_qualin01_instruction(object_class, attrs)
|
||
return resolve_quapnt01_instruction(attrs)
|
||
|
||
|
||
def resolve_slcons03_instruction(
|
||
geometry_type: str,
|
||
attrs: dict[str, Any],
|
||
) -> str:
|
||
geom = str(geometry_type).upper()
|
||
quapos_present = attrs.get("QUAPOS") is not None
|
||
quapos = _int_attr(attrs, "QUAPOS", 0)
|
||
|
||
if geom == "POINT":
|
||
if quapos_present and 2 <= quapos < 10:
|
||
return "SY(LOWACC01)"
|
||
return ""
|
||
|
||
instructions: list[str] = []
|
||
if geom == "AREA":
|
||
instructions.append("AP(CROSSX01)")
|
||
|
||
if quapos_present:
|
||
if 2 <= quapos < 10:
|
||
instructions.append("LC(LOWACC01)")
|
||
return ";".join(instructions)
|
||
|
||
condtn = _int_attr(attrs, "CONDTN", 0)
|
||
catslc = _int_attr(attrs, "CATSLC", 0)
|
||
watlev = _int_attr(attrs, "WATLEV", 0)
|
||
if condtn in {1, 2}:
|
||
instructions.append("LS(DASH,1,CSTLN)")
|
||
elif catslc in {6, 15, 16}:
|
||
instructions.append("LS(SOLD,4,CSTLN)")
|
||
elif watlev == 2:
|
||
instructions.append("LS(SOLD,2,CSTLN)")
|
||
elif watlev in {3, 4}:
|
||
instructions.append("LS(DASH,2,CSTLN)")
|
||
else:
|
||
instructions.append("LS(SOLD,2,CSTLN)")
|
||
return ";".join(instructions)
|
||
|
||
def resolve_restrn01_instruction(attrs: dict[str, Any]) -> str:
|
||
restrn = set(_int_list(attrs.get("RESTRN")))
|
||
if not restrn:
|
||
return ""
|
||
if restrn & {7, 8, 14}:
|
||
symbol = "ENTRES61" if restrn & {1, 2, 3, 4, 5, 6} else "ENTRES71" if restrn & {9, 10, 11, 12, 13} else "ENTRES51"
|
||
elif restrn & {1, 2}:
|
||
symbol = "ACHRES61" if restrn & {3, 4, 5, 6} else "ACHRES71" if restrn & {9, 10, 11, 12, 13} else "ACHRES51"
|
||
elif restrn & {3, 4, 5, 6}:
|
||
symbol = "FSHRES71" if restrn & {9, 10, 11, 12, 13} else "FSHRES51"
|
||
else:
|
||
symbol = "INFARE51" if restrn & {9, 10, 11, 12, 13} else "RSRDEF51"
|
||
return f"SY({symbol})"
|
||
|
||
|
||
def resolve_resare02_instruction(
|
||
attrs: dict[str, Any], settings: S52MarinerSettings
|
||
) -> str:
|
||
restrn = set(_int_list(attrs.get("RESTRN")))
|
||
catrea = set(_int_list(attrs.get("CATREA")))
|
||
group_a = {1, 8, 9, 12, 14, 19, 21, 25}
|
||
group_b = {4, 5, 6, 7, 10, 18, 20, 22, 23, 24}
|
||
priority = False
|
||
if "RESTRN" in attrs:
|
||
if restrn & {7, 8, 14}:
|
||
symbol = "ENTRES61" if (restrn & {1,2,3,4,5,6} or catrea & group_a) else "ENTRES71" if (restrn & {9,10,11,12,13} or catrea & group_b) else "ENTRES51"
|
||
line_code = "RESARE51"; priority = True
|
||
elif restrn & {1, 2}:
|
||
symbol = "ACHRES61" if (restrn & {3,4,5,6} or catrea & group_a) else "ACHRES71" if (restrn & {9,10,11,12,13} or catrea & group_b) else "RESTRN51"
|
||
line_code = "RESARE51"; priority = True
|
||
elif restrn & {3,4,5,6}:
|
||
symbol = "FSHRES51" if catrea & group_a else "FSHRES71" if (restrn & {9,10,11,12,13} or catrea & group_b) else "FSHRES51"
|
||
line_code = "FSHRES51"; priority = True
|
||
else:
|
||
symbol = "INFARE51" if restrn & {9,10,11,12,13} else "RSRDEF51"
|
||
line_code = "CTYARE51"
|
||
else:
|
||
symbol = "CTYARE71" if (catrea & group_a and catrea & group_b) else "CTYARE51" if catrea & group_a else "INFARE51" if catrea & group_b else "RSRDEF51"
|
||
line_code = "CTYARE51"
|
||
line = f"LC({line_code})" if settings.symbolized_boundaries else "LS(DASH,2,CHMGD)"
|
||
parts = [line, f"SY({symbol})"]
|
||
if priority:
|
||
parts.insert(0, "OP(6---)")
|
||
return ";".join(parts)
|
||
|
||
|
||
def resolve_csp_hints(
|
||
csp_name: str,
|
||
object_class: str,
|
||
geometry_type: str,
|
||
attrs: dict[str, Any],
|
||
settings: S52MarinerSettings,
|
||
*,
|
||
next_safe_contour_m: float | None = None,
|
||
sounding_depth_m: float | None = None,
|
||
associated_depth_ranges: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
name = str(csp_name or "")[:8]
|
||
instruction: str | None = None
|
||
extra: dict[str, Any] = {}
|
||
|
||
if name in {"DEPARE01", "DEPARE02"}:
|
||
instruction = resolve_depare01_instruction(object_class, attrs, settings)
|
||
elif name == "DEPCNT02":
|
||
instruction, safe = resolve_depcnt02_instruction(
|
||
object_class,
|
||
attrs,
|
||
settings,
|
||
next_safe_contour_m=next_safe_contour_m,
|
||
)
|
||
extra["s52_is_safety_contour"] = int(safe)
|
||
elif name == "LIGHTS05":
|
||
instruction, light_extra = resolve_lights05_instruction(attrs)
|
||
extra.update(light_extra)
|
||
elif name == "SOUNDG02":
|
||
if sounding_depth_m is None:
|
||
return {}
|
||
instruction, sounding_extra = resolve_soundg03_instruction(
|
||
attrs, sounding_depth_m, settings
|
||
)
|
||
extra.update(sounding_extra)
|
||
elif name == "OBSTRN04":
|
||
instruction, hazard_extra = resolve_obstrn04_instruction(
|
||
object_class, geometry_type=geometry_type,
|
||
attrs=attrs, settings=settings,
|
||
associated_depth_ranges=associated_depth_ranges,
|
||
)
|
||
extra.update(hazard_extra)
|
||
elif name == "WRECKS02":
|
||
instruction, hazard_extra = resolve_wrecks02_instruction(
|
||
geometry_type, attrs, settings, associated_depth_ranges=associated_depth_ranges
|
||
)
|
||
extra.update(hazard_extra)
|
||
elif name == "QUAPOS01":
|
||
instruction = resolve_quapos01_instruction(object_class, geometry_type, attrs)
|
||
elif name == "SLCONS03":
|
||
instruction = resolve_slcons03_instruction(geometry_type, attrs)
|
||
elif name == "RESTRN01":
|
||
instruction = resolve_restrn01_instruction(attrs)
|
||
elif name == "RESARE02":
|
||
instruction = resolve_resare02_instruction(attrs, settings)
|
||
else:
|
||
return {}
|
||
|
||
hints = static_rule_hints(instruction)
|
||
hints.update(materialize_symbol_hints(instruction, attrs))
|
||
if name == "LIGHTS05" and hints.get("s52_symbol_code"):
|
||
extra["s52_light_symbol_code"] = hints["s52_symbol_code"]
|
||
hints.update(extra)
|
||
hints["s52_csp_resolved"] = name
|
||
hints["s52_csp_rule_opcodes"] = ",".join(
|
||
rule.opcode for rule in parse_instruction_string(instruction)
|
||
)
|
||
return hints
|