From e5be7c198e7628c7e4aa9915dfb74f507583f165 Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Mon, 21 Sep 2026 06:09:39 +0800 Subject: [PATCH] feat: preserve ordered S-52 runtime rule chains --- src/cm95pbf/__init__.py | 4 +- src/cm95pbf/csp.py | 78 ++++++++++++++++++++++++++++--------- src/cm95pbf/mapping.py | 11 ++++++ src/cm95pbf/presentation.py | 31 +++++++++++++++ src/cm95pbf/s52.py | 31 +++++++++++++-- 5 files changed, 130 insertions(+), 25 deletions(-) diff --git a/src/cm95pbf/__init__.py b/src/cm95pbf/__init__.py index 28df7fe..fb76499 100644 --- a/src/cm95pbf/__init__.py +++ b/src/cm95pbf/__init__.py @@ -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" diff --git a/src/cm95pbf/csp.py b/src/cm95pbf/csp.py index be7cedd..73799a8 100644 --- a/src/cm95pbf/csp.py +++ b/src/cm95pbf/csp.py @@ -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 diff --git a/src/cm95pbf/mapping.py b/src/cm95pbf/mapping.py index ceb3bca..2b986fc 100644 --- a/src/cm95pbf/mapping.py +++ b/src/cm95pbf/mapping.py @@ -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", diff --git a/src/cm95pbf/presentation.py b/src/cm95pbf/presentation.py index 7b4e122..63eae4f 100644 --- a/src/cm95pbf/presentation.py +++ b/src/cm95pbf/presentation.py @@ -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]) diff --git a/src/cm95pbf/s52.py b/src/cm95pbf/s52.py index 3315965..ff24f8c 100644 --- a/src/cm95pbf/s52.py +++ b/src/cm95pbf/s52.py @@ -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: