fix: align CM93 lookup flow with OpenCPN
This commit is contained in:
@@ -1,2 +1,2 @@
|
|||||||
__version__ = "0.4.13-s52-ordered-rule-ir"
|
__version__ = "0.4.14-s52-opencpn-lookup-flow"
|
||||||
BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.13-s52-ordered-rule-ir"
|
BUILDER_CONTRACT_VERSION = "cm95pbf-builder/0.4.14-s52-opencpn-lookup-flow"
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import re
|
||||||
|
import struct
|
||||||
from typing import Any, Iterable
|
from typing import Any, Iterable
|
||||||
|
|
||||||
from .presentation import (
|
from .presentation import (
|
||||||
@@ -102,16 +104,36 @@ def normalize_cm93_object_class_for_s52(object_class: str) -> str:
|
|||||||
def normalize_cm93_object_for_s52(
|
def normalize_cm93_object_for_s52(
|
||||||
object_class: str, geometry_type: str, attrs: dict[str, Any]
|
object_class: str, geometry_type: str, attrs: dict[str, Any]
|
||||||
) -> tuple[str, 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 "")
|
source_class = str(object_class or "")
|
||||||
|
lookup_class = normalize_cm93_object_class_for_s52(source_class)
|
||||||
out = normalize_cm93_attributes_for_s52(attrs)
|
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":
|
if str(geometry_type).upper() == "POINT":
|
||||||
catlmk = CM93_LANDMARK_CLASS_TO_CATLMK.get(source_class)
|
catlmk = CM93_LANDMARK_CLASS_TO_CATLMK.get(source_class)
|
||||||
if catlmk is not None:
|
if catlmk is not None:
|
||||||
out.setdefault("CATLMK", catlmk)
|
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():
|
for key, value in CM93_CLASS_SYNTHETIC_ATTRIBUTES.get(source_class, {}).items():
|
||||||
out.setdefault(key, value)
|
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] = {
|
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]:
|
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)
|
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:
|
if "COLOUR" in out:
|
||||||
value = _s57_list_string(out.get("COLOUR"))
|
value = _s57_list_string(out.get("COLOUR"))
|
||||||
if value is not None:
|
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:
|
if translated is not None:
|
||||||
out["COLOUR"] = translated
|
out["COLOUR"] = translated
|
||||||
|
|
||||||
|
for key in ("QUASOU", "CATLIT"):
|
||||||
|
value = out.get(key)
|
||||||
|
if isinstance(value, int):
|
||||||
|
out[key] = str(int(value))
|
||||||
|
|
||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
@@ -201,28 +234,41 @@ def split_lookup_condition(condition: str) -> tuple[str, str]:
|
|||||||
return text[:6].upper(), text[6:]
|
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:
|
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)
|
acronym, expected = split_lookup_condition(condition)
|
||||||
actual_present = acronym in attrs and attrs.get(acronym) is not None
|
actual_present = acronym in attrs and attrs.get(acronym) is not None
|
||||||
|
|
||||||
if expected.startswith(" "):
|
if expected.startswith(" "):
|
||||||
return actual_present
|
return actual_present
|
||||||
if expected.startswith("?"):
|
if expected.startswith("?"):
|
||||||
return not actual_present
|
# Preserve the fixed-source quirk: this path never increments countATT.
|
||||||
|
return False
|
||||||
if not actual_present:
|
if not actual_present:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
actual = attrs.get(acronym)
|
actual = attrs.get(acronym)
|
||||||
if isinstance(actual, (list, tuple)):
|
if isinstance(actual, bool):
|
||||||
actual = _s57_list_string(actual)
|
|
||||||
elif isinstance(actual, bool):
|
|
||||||
actual = int(actual)
|
actual = int(actual)
|
||||||
|
if isinstance(actual, int):
|
||||||
if isinstance(actual, (int, float)):
|
return actual == _c_atoi(expected)
|
||||||
try:
|
if isinstance(actual, float):
|
||||||
return abs(float(actual) - float(expected)) < 1e-6
|
return actual == _c_atof_float32(expected)
|
||||||
except (TypeError, ValueError):
|
|
||||||
return False
|
|
||||||
return str(actual) == 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]]:
|
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(
|
return sorted(
|
||||||
rows,
|
rows,
|
||||||
key=lambda row: (
|
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(
|
def find_best_lookup(
|
||||||
rows: Iterable[dict[str, Any]],
|
rows: Iterable[dict[str, Any]],
|
||||||
attrs: dict[str, Any],
|
attrs: dict[str, Any],
|
||||||
@@ -251,6 +301,7 @@ def find_best_lookup(
|
|||||||
table_name: str | None = None,
|
table_name: str | None = None,
|
||||||
geometry_type: str | None = None,
|
geometry_type: str | None = None,
|
||||||
) -> dict[str, Any] | None:
|
) -> dict[str, Any] | None:
|
||||||
|
"""Reproduce non-strict fixed OpenCPN FindBestLUP selection."""
|
||||||
normalized = normalize_cm93_attributes_for_s52(attrs)
|
normalized = normalize_cm93_attributes_for_s52(attrs)
|
||||||
candidates = []
|
candidates = []
|
||||||
for row in rows:
|
for row in rows:
|
||||||
@@ -260,16 +311,29 @@ def find_best_lookup(
|
|||||||
continue
|
continue
|
||||||
candidates.append(row)
|
candidates.append(row)
|
||||||
|
|
||||||
fallback: dict[str, Any] | None = None
|
ordered = sort_lookup_candidates(candidates)
|
||||||
for row in 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 []
|
conditions = row.get("attribute_conditions_raw") or []
|
||||||
if not conditions:
|
if not conditions:
|
||||||
if fallback is None:
|
|
||||||
fallback = row
|
|
||||||
continue
|
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 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:
|
def primary_symbol_code(row: dict[str, Any] | None) -> str | None:
|
||||||
|
|||||||
Reference in New Issue
Block a user