Files
pbf/navsea_mapping_registry.py
2026-04-08 19:32:25 +08:00

373 lines
14 KiB
Python

from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import pymysql
@dataclass(frozen=True)
class FieldValueRule:
field_name_jp: str
legacy_value: str
standardized_field: str
standardized_value: str
geom_scope: tuple[str, ...]
source_layer_scope: tuple[str, ...]
canonical_object_scope: tuple[str, ...]
def matches(self, context: dict[str, str]) -> bool:
if self.geom_scope and context.get("geom_type", "") not in self.geom_scope:
return False
if self.source_layer_scope and context.get("source_layer", "") not in self.source_layer_scope:
return False
if self.canonical_object_scope and context.get("canonical_object_type", "") not in self.canonical_object_scope:
return False
return True
@dataclass(frozen=True)
class FieldNameRule:
field_name_jp: str
field_name_std: str
field_group_name: str
keep_in_delivery: bool
keep_in_engineering: bool
normalization_class: str
@dataclass(frozen=True)
class RenderRule:
rule_id: str
priority: int
match_expr: dict[str, Any]
output: dict[str, Any]
def matches(self, context: dict[str, Any]) -> bool:
return evaluate_match_expr(self.match_expr, context)
def normalize_csv_scope(value: str | None) -> tuple[str, ...]:
if not value:
return ()
return tuple(part.strip() for part in str(value).split(",") if part.strip())
def evaluate_match_expr(match_expr: dict[str, Any], context: dict[str, Any]) -> bool:
clauses = match_expr.get("all") or []
for clause in clauses:
field = clause.get("field")
op = clause.get("op")
expected = clause.get("value")
actual = context.get(str(field))
actual_text = "" if actual is None else str(actual)
if op == "eq":
if actual_text != str(expected):
return False
elif op == "neq":
if actual_text == str(expected):
return False
elif op == "in":
if actual_text not in {str(item) for item in (expected or [])}:
return False
elif op == "contains":
if str(expected) not in actual_text:
return False
elif op == "contains_any":
values = [str(item) for item in (expected or [])]
if not any(item in actual_text for item in values):
return False
elif op == "is_null":
if actual not in (None, ""):
return False
elif op == "not_null":
if actual in (None, ""):
return False
else:
return False
return True
def load_local_field_name_rules() -> dict[str, FieldNameRule]:
mapping_path = Path(__file__).resolve().parent / "tasks" / "pbf" / "mappings" / "navsea_field_name_rules_v1.yaml"
if not mapping_path.exists():
return {}
rules: dict[str, FieldNameRule] = {}
current: dict[str, str] = {}
for raw_line in mapping_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or line in {"ruleset: navsea_field_name_rules_v1", "rules:"}:
continue
if line.startswith("- field_name_jp:"):
if current.get("field_name_jp") and current.get("field_name_std"):
rules[current["field_name_jp"]] = FieldNameRule(
field_name_jp=current["field_name_jp"],
field_name_std=current["field_name_std"],
field_group_name=current.get("field_group", ""),
keep_in_delivery=current.get("keep_in_delivery", "true").lower() == "true",
keep_in_engineering=current.get("keep_in_engineering", "true").lower() == "true",
normalization_class=current.get("normalization_class", ""),
)
current = {"field_name_jp": line.split(":", 1)[1].strip()}
continue
if not current or ":" not in line:
continue
key, value = line.split(":", 1)
current[key.strip()] = value.strip()
if current.get("field_name_jp") and current.get("field_name_std"):
rules[current["field_name_jp"]] = FieldNameRule(
field_name_jp=current["field_name_jp"],
field_name_std=current["field_name_std"],
field_group_name=current.get("field_group", ""),
keep_in_delivery=current.get("keep_in_delivery", "true").lower() == "true",
keep_in_engineering=current.get("keep_in_engineering", "true").lower() == "true",
normalization_class=current.get("normalization_class", ""),
)
return rules
def load_local_source_layer_rules() -> dict[str, dict[str, str]]:
mapping_path = Path(__file__).resolve().parent / "tasks" / "pbf" / "mappings" / "navsea_source_layer_rules_v1.yaml"
if not mapping_path.exists():
return {}
rules: dict[str, dict[str, str]] = {}
current_jp: str | None = None
current_std: str | None = None
for raw_line in mapping_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or line in {"ruleset: navsea_source_layer_rules_v1", "rules:"}:
continue
if line.startswith("- source_layer_jp:"):
if current_jp and current_std:
rules[current_jp] = {
"source_layer_std": current_std,
"source_layer_rule_id": f"SLR_LOCAL:{current_jp}",
}
current_jp = line.split(":", 1)[1].strip()
current_std = None
continue
if line.startswith("source_layer_std:") and current_jp:
current_std = line.split(":", 1)[1].strip()
if current_jp and current_std:
rules[current_jp] = {
"source_layer_std": current_std,
"source_layer_rule_id": f"SLR_LOCAL:{current_jp}",
}
manual_aliases = {
"P投錨注意障害物透明": "anchor_caution_hazard_area",
"P投錨注意障害物": "anchor_caution_hazard_area",
"P航行危険障害物": "navigation_hazard_area",
"P錨泊地等": "anchorage_area",
"P橋りょう等構造物": "bridge_structure",
"L701": "depth_zone_700",
"L738": "depth_zone_739",
}
for source_layer_jp, source_layer_std in manual_aliases.items():
rules.setdefault(
source_layer_jp,
{
"source_layer_std": source_layer_std,
"source_layer_rule_id": f"SLR_LOCAL:{source_layer_jp}",
},
)
return rules
class NavSeaMappingRegistry:
def __init__(
self,
bundle_id: str,
bundle_version: str,
source_layer_rules: dict[str, dict[str, str]],
field_name_rules: dict[str, FieldNameRule],
field_value_rules: dict[tuple[str, str], list[FieldValueRule]],
render_rules: list[RenderRule],
) -> None:
self.bundle_id = bundle_id
self.bundle_version = bundle_version
self.source_layer_rules = source_layer_rules
self.field_name_rules = field_name_rules
self.field_value_rules = field_value_rules
self.render_rules = render_rules
@classmethod
def load(
cls,
conn: pymysql.Connection,
bundle_id: str | None = None,
) -> "NavSeaMappingRegistry":
with conn.cursor() as cur:
if bundle_id:
cur.execute(
"""
SELECT bundle_id, bundle_version
FROM navsea_rule_bundle
WHERE bundle_id = %s
ORDER BY created_at DESC
LIMIT 1
""",
(bundle_id,),
)
else:
cur.execute(
"""
SELECT bundle_id, bundle_version
FROM navsea_rule_bundle
ORDER BY created_at DESC
LIMIT 1
"""
)
bundle = cur.fetchone()
if not bundle:
raise RuntimeError("navsea mapping registry is empty")
resolved_bundle_id = str(bundle["bundle_id"])
resolved_bundle_version = str(bundle["bundle_version"])
cur.execute(
"""
SELECT source_layer_jp, source_layer_std
FROM navsea_source_layer_rules
WHERE bundle_id = %s AND bundle_version = %s
""",
(resolved_bundle_id, resolved_bundle_version),
)
source_layer_rules = {
str(row["source_layer_jp"]): {
"source_layer_std": str(row["source_layer_std"]),
"source_layer_rule_id": f"SLR:{row['source_layer_jp']}",
}
for row in cur.fetchall()
}
source_layer_rules.update(load_local_source_layer_rules())
cur.execute(
"""
SELECT
field_name_jp,
field_name_std,
field_group_name,
keep_in_delivery,
keep_in_engineering,
normalization_class
FROM navsea_field_name_rules
WHERE bundle_id = %s AND bundle_version = %s
""",
(resolved_bundle_id, resolved_bundle_version),
)
field_name_rules = {
str(row["field_name_jp"]): FieldNameRule(
field_name_jp=str(row["field_name_jp"]),
field_name_std=str(row["field_name_std"]),
field_group_name=str(row["field_group_name"]),
keep_in_delivery=bool(row["keep_in_delivery"]),
keep_in_engineering=bool(row["keep_in_engineering"]),
normalization_class=str(row["normalization_class"]),
)
for row in cur.fetchall()
}
field_name_rules.update(load_local_field_name_rules())
cur.execute(
"""
SELECT
field_name_jp,
legacy_value,
standardized_field,
standardized_value,
geom_scope,
source_layer_scope,
canonical_object_scope
FROM navsea_field_value_rules
WHERE bundle_id = %s AND bundle_version = %s
ORDER BY rule_priority ASC, rule_id ASC, rule_revision ASC
""",
(resolved_bundle_id, resolved_bundle_version),
)
field_value_rules: dict[tuple[str, str], list[FieldValueRule]] = {}
for row in cur.fetchall():
key = (str(row["field_name_jp"]), str(row["legacy_value"]))
field_value_rules.setdefault(key, []).append(
FieldValueRule(
field_name_jp=str(row["field_name_jp"]),
legacy_value=str(row["legacy_value"]),
standardized_field=str(row["standardized_field"]),
standardized_value=str(row["standardized_value"]),
geom_scope=normalize_csv_scope(row.get("geom_scope")),
source_layer_scope=normalize_csv_scope(row.get("source_layer_scope")),
canonical_object_scope=normalize_csv_scope(row.get("canonical_object_scope")),
)
)
cur.execute(
"""
SELECT rule_id, priority, match_expr_json, output_json
FROM navsea_render_rules
WHERE bundle_id = %s AND bundle_version = %s AND enabled = 1
ORDER BY priority ASC, rule_id ASC, rule_revision ASC
""",
(resolved_bundle_id, resolved_bundle_version),
)
render_rules = []
for row in cur.fetchall():
render_rules.append(
RenderRule(
rule_id=str(row["rule_id"]),
priority=int(row["priority"]),
match_expr=json.loads(row["match_expr_json"]),
output=json.loads(row["output_json"]),
)
)
return cls(
bundle_id=resolved_bundle_id,
bundle_version=resolved_bundle_version,
source_layer_rules=source_layer_rules,
field_name_rules=field_name_rules,
field_value_rules=field_value_rules,
render_rules=render_rules,
)
def resolve_source_layer(self, source_layer_jp: str) -> tuple[str, str]:
rule = self.source_layer_rules.get(source_layer_jp)
if not rule:
return source_layer_jp, f"SLR:{source_layer_jp}:identity"
return rule["source_layer_std"], rule["source_layer_rule_id"]
def get_field_name_rule(self, field_name: str) -> FieldNameRule | None:
return self.field_name_rules.get(field_name)
def standardize_field_value(
self,
field_name_jp: str,
legacy_value: object,
standardized_field: str,
*,
context: dict[str, str] | None = None,
) -> str | None:
if legacy_value in (None, ""):
return None
key = (field_name_jp, str(legacy_value))
rules = self.field_value_rules.get(key, [])
match_context = context or {}
for rule in rules:
if rule.standardized_field != standardized_field:
continue
if rule.matches(match_context):
return rule.standardized_value
return None
def resolve_render_rule(self, context: dict[str, Any]) -> tuple[dict[str, Any], str | None]:
for rule in self.render_rules:
if rule.matches(context):
return dict(rule.output), rule.rule_id
return {}, None