373 lines
14 KiB
Python
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
|