diff --git a/src/cm95pbf/__init__.py b/src/cm95pbf/__init__.py index fb76499..cb00e1e 100644 --- a/src/cm95pbf/__init__.py +++ b/src/cm95pbf/__init__.py @@ -1,2 +1,2 @@ -__version__ = "0.4.13-s52-ordered-rule-ir" -BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.13-s52-ordered-rule-ir" +__version__ = "0.4.14-s52-opencpn-lookup-flow" +BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.14-s52-opencpn-lookup-flow" diff --git a/src/cm95pbf/s52.py b/src/cm95pbf/s52.py index 43e9aa8..e4523b4 100644 --- a/src/cm95pbf/s52.py +++ b/src/cm95pbf/s52.py @@ -1,5 +1,7 @@ from __future__ import annotations +import re +import struct from typing import Any, Iterable from .presentation import ( @@ -102,16 +104,36 @@ def normalize_cm93_object_class_for_s52(object_class: str) -> str: def normalize_cm93_object_for_s52( object_class: str, geometry_type: str, attrs: dict[str, Any] ) -> tuple[str, dict[str, Any]]: - """Normalize CM93 class/attributes into the S-57/S-52 presentation view.""" + """Normalize CM93 class/attributes into the fixed OpenCPN S-52 view.""" source_class = str(object_class or "") + lookup_class = normalize_cm93_object_class_for_s52(source_class) out = normalize_cm93_attributes_for_s52(attrs) + + if source_class == "$AREAS" and str(out.get("$SCODE") or "") == "II25": + out["$SCODE"] = "BACKGROUND" + if str(geometry_type).upper() == "POINT": catlmk = CM93_LANDMARK_CLASS_TO_CATLMK.get(source_class) if catlmk is not None: out.setdefault("CATLMK", catlmk) + + aton = ( + lookup_class.startswith("LIT") + or lookup_class.startswith("LIGHTS") + or lookup_class.startswith("BCN") + or lookup_class.startswith("_slgto") + or lookup_class.startswith("_boygn") + or lookup_class.startswith("_bcngn") + or lookup_class.startswith("_extgn") + or lookup_class.startswith("TOWERS") + or lookup_class.startswith("BOY") + ) + if aton and "OBJNAM" not in out and "INFORM" in out: + out["OBJNAM"] = out.pop("INFORM") + for key, value in CM93_CLASS_SYNTHETIC_ATTRIBUTES.get(source_class, {}).items(): out.setdefault(key, value) - return normalize_cm93_object_class_for_s52(source_class), out + return lookup_class, out CM93_COLMAR_TO_S57_COLOUR: dict[int, str] = { @@ -175,9 +197,15 @@ def _s57_list_string(value: Any) -> str | None: def normalize_cm93_attributes_for_s52(attrs: dict[str, Any]) -> dict[str, Any]: - """Return S-57/S-52-compatible attributes without mutating CM93 raw data.""" + """Return the fixed OpenCPN CM93-to-S57 typed attribute view.""" out = dict(attrs) + for key, value in list(out.items()): + if len(str(key)) == 6 and isinstance(value, (list, tuple)): + normalized = _s57_list_string(value) + if normalized is not None: + out[key] = normalized + if "COLOUR" in out: value = _s57_list_string(out.get("COLOUR")) if value is not None: @@ -191,6 +219,11 @@ def normalize_cm93_attributes_for_s52(attrs: dict[str, Any]) -> dict[str, Any]: if translated is not None: out["COLOUR"] = translated + for key in ("QUASOU", "CATLIT"): + value = out.get(key) + if isinstance(value, int): + out[key] = str(int(value)) + return out @@ -201,28 +234,41 @@ def split_lookup_condition(condition: str) -> tuple[str, str]: return text[:6].upper(), text[6:] +def _c_atoi(value: str) -> int: + """Minimal C atoi semantics used by fixed OpenCPN FindBestLUP.""" + match = re.match(r"^[\s]*([+-]?\d+)", str(value)) + return int(match.group(1)) if match else 0 + + +def _c_atof_float32(value: str) -> float: + """OpenCPN assigns atof() to float before exact REAL comparison.""" + try: + parsed = float(str(value).strip() or "0") + except ValueError: + parsed = 0.0 + return struct.unpack("f", struct.pack("f", parsed))[0] + + def lookup_condition_matches(condition: str, attrs: dict[str, Any]) -> bool: + """Match one condition as fixed OpenCPN FindBestLUP does.""" acronym, expected = split_lookup_condition(condition) actual_present = acronym in attrs and attrs.get(acronym) is not None if expected.startswith(" "): return actual_present if expected.startswith("?"): - return not actual_present + # Preserve the fixed-source quirk: this path never increments countATT. + return False if not actual_present: return False actual = attrs.get(acronym) - if isinstance(actual, (list, tuple)): - actual = _s57_list_string(actual) - elif isinstance(actual, bool): + if isinstance(actual, bool): actual = int(actual) - - if isinstance(actual, (int, float)): - try: - return abs(float(actual) - float(expected)) < 1e-6 - except (TypeError, ValueError): - return False + if isinstance(actual, int): + return actual == _c_atoi(expected) + if isinstance(actual, float): + return actual == _c_atof_float32(expected) return str(actual) == expected @@ -234,7 +280,7 @@ def _lookup_sequence(row: dict[str, Any]) -> int: def sort_lookup_candidates(rows: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: - """Match OpenCPN CompareLUPObjects: most conditions first, then sequence.""" + """Match fixed OpenCPN CompareLUPObjects.""" return sorted( rows, key=lambda row: ( @@ -244,6 +290,10 @@ def sort_lookup_candidates(rows: Iterable[dict[str, Any]]) -> list[dict[str, Any ) +def _has_lookup_visible_attributes(attrs: dict[str, Any]) -> bool: + return any(len(str(key)) == 6 and value is not None for key, value in attrs.items()) + + def find_best_lookup( rows: Iterable[dict[str, Any]], attrs: dict[str, Any], @@ -251,6 +301,7 @@ def find_best_lookup( table_name: str | None = None, geometry_type: str | None = None, ) -> dict[str, Any] | None: + """Reproduce non-strict fixed OpenCPN FindBestLUP selection.""" normalized = normalize_cm93_attributes_for_s52(attrs) candidates = [] for row in rows: @@ -260,16 +311,29 @@ def find_best_lookup( continue candidates.append(row) - fallback: dict[str, Any] | None = None - for row in sort_lookup_candidates(candidates): + ordered = sort_lookup_candidates(candidates) + if not ordered: + return None + + first = ordered[0] + if not _has_lookup_visible_attributes(normalized): + return first + + for row in ordered: conditions = row.get("attribute_conditions_raw") or [] if not conditions: - if fallback is None: - fallback = row continue - if all(lookup_condition_matches(c, normalized) for c in conditions): + count_att = sum( + 1 for condition in conditions + if lookup_condition_matches(condition, normalized) + ) + if count_att == len(conditions): return row - return fallback + + for row in ordered: + if not (row.get("attribute_conditions_raw") or []): + return row + return first def primary_symbol_code(row: dict[str, Any] | None) -> str | None: