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

feat: preserve ordered S-52 runtime rule chains

Browse Source
This commit is contained in:
OpenAI Codex
2026-09-21 06:09:39 +08:00
parent 8a2ceb0f5f
commit e5be7c198e
5 changed files with 130 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/cm95pbf/__init__.py
View File

@@ -1,2 +1,2 @@
__version__ = "0.4.12-opencpn-cm93-coverage"
BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.12-opencpn-cm93-coverage"
__version__ = "0.4.13-s52-ordered-rule-ir"
BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.13-s52-ordered-rule-ir"

78
src/cm95pbf/csp.py
View File

@@ -5,7 +5,9 @@ import math
from typing import Any
from .presentation import (
materialize_rule_ir,
materialize_symbol_hints,
materialize_text_hints,
parse_instruction_string,
static_rule_hints,
)
@@ -26,7 +28,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"}
{"DEPARE01", "DEPARE02", "DEPCNT02", "SOUNDG02", "OBSTRN04", "WRECKS02", "RESARE02"}
)
@@ -597,6 +599,16 @@ def resolve_obstrn04_instruction(
*,
associated_depth_ranges: list[dict[str, Any]] | None = None,
) -> tuple[str, dict[str, Any]]:
quality = resolve_quapnt01_instruction(attrs)
name_text = (
"TX(OBJNAM,1,2,3,'15118',-1,-1,CHBLK,26)"
if attrs.get("OBJNAM") not in {None, ""}
else ""
)
def finish(*parts: str) -> str:
return ";".join(part.strip(";") for part in parts if part)
valsou = _optional_float(attrs.get("VALSOU"))
depth_value = valsou
if depth_value is None and str(geometry_type).upper() == "AREA":
@@ -625,7 +637,7 @@ def resolve_obstrn04_instruction(
geom = str(geometry_type).upper()
if geom == "POINT":
if udw:
return udw, extra
return finish(udw, quality, name_text), extra
sounding = False
watlev = _int_attr(attrs, "WATLEV", -9)
if valsou is not None:
@@ -671,32 +683,45 @@ def resolve_obstrn04_instruction(
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
instruction += ";" + snd if snd else ""
extra.update(snd_extra)
return instruction, extra
return finish(instruction, quality, name_text), extra
if geom == "LINE":
quapos = _int_attr(attrs, "QUAPOS", 0)
if 2 <= quapos < 10:
return ("LC(LOWACC41)" if udw else "LC(LOWACC31)"), extra
return finish(
"LC(LOWACC41)" if udw else "LC(LOWACC31)", name_text
), extra
if udw:
return "LS(DOTT,2,CHBLK)", extra
return finish("LS(DOTT,2,CHBLK)", name_text), extra
if valsou is None or valsou <= 20.0:
return "LS(DOTT,2,CHBLK)", extra
return "LS(DASH,2,CHBLK)", extra
line = "LS(DOTT,2,CHBLK)"
if valsou is not None:
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
extra.update(snd_extra)
line = finish(line, snd)
return finish(line, name_text), extra
return finish("LS(DASH,2,CHBLK)", name_text), extra
# AREA
if udw:
return "AC(DEPVS);AP(FOULAR01);LS(DOTT,2,CHBLK);" + udw, extra
return finish(
"AC(DEPVS);AP(FOULAR01);LS(DOTT,2,CHBLK)", udw, quality, name_text
), 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
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
extra.update(snd_extra)
return finish(line, snd, quality, name_text), extra
if watlev in {1, 2}:
return "AC(CHBRN);LS(SOLD,2,CSTLN)", extra
return finish("AC(CHBRN);LS(SOLD,2,CSTLN)", quality, name_text), extra
if watlev == 4:
return "AC(DEPIT);LS(DASH,2,CSTLN)", extra
return finish("AC(DEPIT);LS(DASH,2,CSTLN)", quality, name_text), 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
return finish(
"AC(DEPVS);AP(FOULAR01);LS(DOTT,2,CHBLK)", quality, name_text
), extra
return finish("AC(DEPVS);LS(DOTT,2,CHBLK)", quality, name_text), extra
def resolve_wrecks02_instruction(
@@ -737,9 +762,14 @@ def resolve_wrecks02_instruction(
extra = dict(udw_extra)
extra.update({"s52_hazard_procedure": "WRECKS02", "s52_hazard_depth_m": float(depth_value)})
geom = str(geometry_type).upper()
quality = resolve_quapnt01_instruction(attrs)
def finish(*parts: str) -> str:
return ";".join(part.strip(";") for part in parts if part)
if geom == "POINT":
if udw:
return udw, extra
return finish(udw, quality), 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)"
@@ -749,14 +779,16 @@ def resolve_wrecks02_instruction(
if snd:
instruction += ";" + snd
extra.update(snd_extra)
return instruction, extra
return finish(instruction, quality), extra
if catwrk == 1 and watlev == 3:
return "SY(WRECKS04)", extra
return finish("SY(WRECKS04)", quality), extra
if catwrk == 2 and watlev == 3:
return "SY(WRECKS05)", extra
return finish("SY(WRECKS05)", quality), extra
if catwrk in {4, 5} or watlev in {1, 2, 4, 5}:
return "SY(WRECKS01)", extra
return "SY(WRECKS05)", extra
return finish("SY(WRECKS01)", quality), extra
if "CATWRK" not in attrs or "WATLEV" not in attrs:
return quality, extra
return finish("SY(WRECKS05)", quality), extra
quapos = _int_attr(attrs, "QUAPOS", 0)
if 2 <= quapos < 10:
@@ -774,6 +806,7 @@ def resolve_wrecks02_instruction(
instruction = line
if valsou is not None and valsou <= 20.0:
instruction += ";" + udw if udw else ""
instruction = finish(instruction, quality)
snd, snd_extra = resolve_soundg03_instruction(attrs, valsou, settings)
if snd:
instruction += ";" + snd
@@ -783,6 +816,9 @@ def resolve_wrecks02_instruction(
instruction += f";AC({color})"
if udw:
instruction += ";" + udw
instruction = finish(instruction, quality)
else:
instruction = finish(instruction, udw, quality)
return instruction, extra
@@ -969,6 +1005,7 @@ def resolve_csp_hints(
hints = static_rule_hints(instruction)
hints.update(materialize_symbol_hints(instruction, attrs))
hints.update(materialize_text_hints(instruction, attrs))
if name == "LIGHTS05" and hints.get("s52_symbol_code"):
extra["s52_light_symbol_code"] = hints["s52_symbol_code"]
hints.update(extra)
@@ -976,4 +1013,7 @@ def resolve_csp_hints(
hints["s52_csp_rule_opcodes"] = ",".join(
rule.opcode for rule in parse_instruction_string(instruction)
)
hints["s52_rule_chain"] = materialize_rule_ir(
instruction, attrs, source="csp", procedure=name
)
return hints

11
src/cm95pbf/mapping.py
View File

@@ -73,7 +73,18 @@ DELIVERY_PROPERTY_ALLOWLIST = frozenset(
"s52_line_sequence",
"s52_line_style",
"s52_line_width",
"s52_line1_style",
"s52_line1_width",
"s52_line1_color",
"s52_line2_style",
"s52_line2_width",
"s52_line2_color",
"s52_rule_opcodes",
"s52_rule_chain",
"s52_rule_chain_version",
"s52_source_object_class",
"s52_source_geometry",
"s52_scamin_override",
"s52_light_arc_color",
"s52_light_arc_radius_mm",
"s52_light_mode",

31
src/cm95pbf/presentation.py
View File

@@ -167,6 +167,32 @@ def materialize_symbol_hints(instruction: str, attrs: dict[str, Any]) -> dict[st
return out
def materialize_rule_ir(
instruction: str,
attrs: dict[str, Any],
*,
source: str,
procedure: str | None = None,
) -> list[dict[str, Any]]:
"""Return an ordered, executable representation of an S-52 rule string."""
out: list[dict[str, Any]] = []
for rule in parse_instruction_string(instruction):
item = rule.to_dict()
item.update({"source": source, "procedure": procedure})
if rule.opcode in {"TX", "TE"}:
text = materialize_text_hints(rule.raw, attrs)
if text.get("s52_text1_value") not in (None, ""):
item["text"] = {
key.removeprefix("s52_text1_"): value
for key, value in text.items()
if key.startswith("s52_text1_")
}
if rule.opcode == "OP":
item["side_effect"] = "upstream_parser_noop"
out.append(item)
return out
class PresentationParseError(ValueError):
pass
@@ -359,6 +385,11 @@ def static_rule_hints(instruction: str) -> dict[str, Any]:
line_rules = [
rule for rule in rules if rule.opcode == "LS" and len(rule.args) >= 3
]
for index, rule in enumerate(line_rules, 1):
prefix = f"s52_line{index}_"
out[prefix + "style"] = rule.args[0].strip().strip("'\"")
out[prefix + "width"] = rule.args[1].strip().strip("'\"")
out[prefix + "color"] = rule.args[2].strip().strip("'\"")
if len(line_rules) > 1:
out["s52_line_sequence"] = "|".join(
",".join(arg.strip().strip("'\"") for arg in rule.args[:3])

31
src/cm95pbf/s52.py
View File

@@ -3,8 +3,10 @@ from __future__ import annotations
from typing import Any, Iterable
from .presentation import (
materialize_rule_ir,
materialize_symbol_hints,
materialize_text_hints,
parse_instruction_string,
static_rule_hints,
)
from .csp import (
@@ -30,12 +32,8 @@ S52_DISPLAY_PRIORITY_BY_CODE = {
def _apply_presentation_side_effects(out: dict[str, Any]) -> None:
override = str(out.get("s52_priority_override") or "")
if override and override[0] in S52_DISPLAY_PRIORITY_BY_CODE:
out["s52_display_priority"] = S52_DISPLAY_PRIORITY_BY_CODE[override[0]]
if int(out.get("s52_display_base_promoted") or 0) == 1:
out["s52_display_category"] = "Displaybase"
out["s52_display_priority"] = "Hazards"
OPENCPN_CM93_CLASS_SUBSTITUTIONS: dict[str, str] = {
@@ -485,6 +483,7 @@ class S52LookupRegistry:
if token.strip()
]
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(
@@ -511,8 +510,32 @@ class S52LookupRegistry:
)
if resolved:
out.update(resolved)
resolved_by_procedure[procedure] = resolved
else:
unresolved.append(procedure)
ordered_chain: list[dict[str, Any]] = []
for rule in parse_instruction_string(instruction):
if rule.opcode != "CS" or not rule.args:
ordered_chain.extend(
materialize_rule_ir(rule.raw, normalized_attrs, source="lookup")
)
continue
procedure = rule.args[0].strip().strip("'\"")[:8]
if procedure in unresolved:
dynamic = materialize_rule_ir(
rule.raw, normalized_attrs, source="lookup"
)[0]
dynamic["dynamic"] = True
dynamic["procedure"] = procedure
ordered_chain.append(dynamic)
else:
ordered_chain.extend(
list(resolved_by_procedure.get(procedure, {}).get("s52_rule_chain") or [])
)
out["s52_rule_chain_version"] = 1
out["s52_rule_chain"] = ordered_chain
out["s52_source_object_class"] = lookup_class
out["s52_source_geometry"] = str(geometry_type).upper()
if unresolved:
out["s52_csp"] = ",".join(unresolved)
else: