1353 lines
51 KiB
Python
1353 lines
51 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
NavSea render audit for original vs engineering tiles/styles.
|
|
|
|
This script compares one original style/tile set against one engineering style/tile set.
|
|
|
|
Audit goal:
|
|
- use legacy fid as the primary object identity
|
|
- fall back to geometry + stable legacy properties when fid is missing
|
|
- evaluate each style layer against decoded tile features
|
|
- extract per-object render observations (icon/text/line/fill)
|
|
- compare original and engineering render observations at tile-instance scope
|
|
- emit Markdown and JSON audit reports for review
|
|
- persist the latest audit result into MySQL for fid-level trace-back
|
|
|
|
Important scope note:
|
|
- render comparison is performed per tile feature instance
|
|
- identity is "fid first", but zoom/tile instance is preserved because rendering is zoom-sensitive
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import mapbox_vector_tile
|
|
import pymysql
|
|
|
|
from navsea_tile_builder import NavSeaFidCodec
|
|
|
|
|
|
DEFAULT_ORIGINAL_STYLE_PATH = Path("/mnt/sda1/www/newpec/style.patched.local.json")
|
|
DEFAULT_ENGINEERING_STYLE_PATH = Path("/mnt/sda1/www/newpec/navsea-engineering.json")
|
|
DEFAULT_ORIGINAL_TILE_ROOT = Path(
|
|
"/home/wwwroot/newpec/exported_auto/"
|
|
"tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles"
|
|
)
|
|
DEFAULT_ENGINEERING_TILE_ROOT = Path("/home/wwwroot/pbf-engineering-karatsu-10nm")
|
|
DEFAULT_REPORT_MD_PATH = Path("/root/weather/NavSea_Original_vs_Engineering_Render_Audit_Karatsu_10nm.md")
|
|
DEFAULT_REPORT_JSON_PATH = Path("/root/weather/NavSea_Original_vs_Engineering_Render_Audit_Karatsu_10nm.json")
|
|
DEFAULT_AUDIT_NAME = "navsea_original_vs_engineering_karatsu_10nm"
|
|
DB_BATCH_SIZE = 1000
|
|
MAX_EXAMPLES = 30
|
|
|
|
ORIGINAL_STYLE_PATH = DEFAULT_ORIGINAL_STYLE_PATH
|
|
ENGINEERING_STYLE_PATH = DEFAULT_ENGINEERING_STYLE_PATH
|
|
ORIGINAL_TILE_ROOT = DEFAULT_ORIGINAL_TILE_ROOT
|
|
ENGINEERING_TILE_ROOT = DEFAULT_ENGINEERING_TILE_ROOT
|
|
REPORT_MD_PATH = DEFAULT_REPORT_MD_PATH
|
|
REPORT_JSON_PATH = DEFAULT_REPORT_JSON_PATH
|
|
AUDIT_NAME = DEFAULT_AUDIT_NAME
|
|
SKIP_DB_PERSIST = False
|
|
COMPARISON_LABEL = "工程版"
|
|
FID_CODEC: NavSeaFidCodec | None = None
|
|
MATCH_ON_FID_ONLY = False
|
|
|
|
DERIVED_PROPERTY_PREFIXES = (
|
|
"canonical_",
|
|
"semantic_",
|
|
"detection_",
|
|
"render_",
|
|
"chart_",
|
|
"light_",
|
|
"hazard_",
|
|
"area_",
|
|
"source_layer_",
|
|
)
|
|
DERIVED_PROPERTY_KEYS = {
|
|
"fid_algo_id",
|
|
"fid_key_id",
|
|
"fid_navsea_int",
|
|
"fid_legacy_raw",
|
|
"normalization_bundle_id",
|
|
"source_layer_rule_id",
|
|
"trace_status",
|
|
"feature_id",
|
|
"depth_value_m",
|
|
"clearance_height_m",
|
|
"least_depth_m",
|
|
"bearing_deg",
|
|
}
|
|
STABLE_FALLBACK_KEYS = (
|
|
"分類番号",
|
|
"形状分類番号",
|
|
"表示用番号",
|
|
"灯色",
|
|
"灯略記",
|
|
"明弧/分孤",
|
|
"表示位置",
|
|
"名称",
|
|
"名称補助",
|
|
"日本語地名",
|
|
"英文字地名",
|
|
"水深値(m)",
|
|
"高さ(m)",
|
|
"高さ/深度(m)",
|
|
"角度",
|
|
)
|
|
EXPRESSION_OPS = {
|
|
"get",
|
|
"match",
|
|
"coalesce",
|
|
"concat",
|
|
"number",
|
|
"literal",
|
|
"rgba",
|
|
"case",
|
|
"has",
|
|
"any",
|
|
"all",
|
|
"==",
|
|
"!=",
|
|
">=",
|
|
"<=",
|
|
">",
|
|
"<",
|
|
"/",
|
|
"*",
|
|
"interpolate",
|
|
"step",
|
|
"zoom",
|
|
}
|
|
|
|
DEPTH_NUMERIC_SOURCE_LAYERS = {
|
|
"L海底地形",
|
|
"L概略等深線",
|
|
"L等深線",
|
|
}
|
|
CLEARANCE_NUMERIC_SOURCE_LAYERS = {
|
|
"p高さ制限",
|
|
}
|
|
SAFETY_ICON_SOURCE_LAYERS = {
|
|
"p航路標識群",
|
|
"p航路標識",
|
|
"P航路標識",
|
|
"p灯台",
|
|
"p灯浮標",
|
|
"p灯立標",
|
|
"p灯標",
|
|
}
|
|
SUPPORT_FILL_SOURCE_LAYERS = {
|
|
"P穴",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DbConfig:
|
|
host: str = "localhost"
|
|
port: int = 3306
|
|
user: str = "root"
|
|
password: str = "2chi9ks2"
|
|
database: str = "pbf_analysis"
|
|
unix_socket: str | None = "/tmp/mysql.sock"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuditFeature:
|
|
dataset: str
|
|
tile_z: int
|
|
tile_x: int
|
|
tile_y: int
|
|
layer_name: str
|
|
geom_type: str
|
|
object_id: str
|
|
fid_legacy: str | None
|
|
native_feature_id: int | None
|
|
properties: dict[str, Any]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RenderObservation:
|
|
component_type: str
|
|
style_layer_id: str
|
|
signature: str
|
|
payload: dict[str, Any]
|
|
|
|
|
|
def canonical_json(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def hash_text(text: str) -> str:
|
|
return hashlib.sha1(text.encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def text_or_none(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|
|
|
|
|
|
def numeric_or_text(value: Any) -> float | str | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
text = text_or_none(value)
|
|
if text is None:
|
|
return None
|
|
try:
|
|
return float(text)
|
|
except ValueError:
|
|
return text
|
|
|
|
|
|
def normalize_color(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, str):
|
|
return value
|
|
if isinstance(value, list):
|
|
return canonical_json(value)
|
|
return str(value)
|
|
|
|
|
|
def compare_numbers(lhs: Any, rhs: Any, op: str) -> bool:
|
|
lhs_num = numeric_or_text(lhs)
|
|
rhs_num = numeric_or_text(rhs)
|
|
if isinstance(lhs_num, float) and isinstance(rhs_num, float):
|
|
if op == ">=":
|
|
return lhs_num >= rhs_num
|
|
if op == "<=":
|
|
return lhs_num <= rhs_num
|
|
if op == ">":
|
|
return lhs_num > rhs_num
|
|
if op == "<":
|
|
return lhs_num < rhs_num
|
|
lhs_text = "" if lhs is None else str(lhs)
|
|
rhs_text = "" if rhs is None else str(rhs)
|
|
if op == ">=":
|
|
return lhs_text >= rhs_text
|
|
if op == "<=":
|
|
return lhs_text <= rhs_text
|
|
if op == ">":
|
|
return lhs_text > rhs_text
|
|
return lhs_text < rhs_text
|
|
|
|
|
|
def evaluate_expression(expr: Any, feature: AuditFeature, zoom: int) -> Any:
|
|
if not isinstance(expr, list):
|
|
return expr
|
|
if not expr:
|
|
return expr
|
|
head = expr[0]
|
|
if not isinstance(head, str) or head not in EXPRESSION_OPS:
|
|
return [evaluate_expression(item, feature, zoom) for item in expr]
|
|
|
|
if head == "get":
|
|
key = expr[1]
|
|
return feature.properties.get(key)
|
|
if head == "literal":
|
|
return expr[1]
|
|
if head == "coalesce":
|
|
for item in expr[1:]:
|
|
value = evaluate_expression(item, feature, zoom)
|
|
if value not in (None, ""):
|
|
return value
|
|
return None
|
|
if head == "concat":
|
|
return "".join("" if (value := evaluate_expression(item, feature, zoom)) is None else str(value) for item in expr[1:])
|
|
if head == "number":
|
|
value = evaluate_expression(expr[1], feature, zoom)
|
|
parsed = numeric_or_text(value)
|
|
if isinstance(parsed, float):
|
|
return parsed
|
|
fallback = evaluate_expression(expr[2], feature, zoom) if len(expr) > 2 else None
|
|
return fallback
|
|
if head == "rgba":
|
|
values = [evaluate_expression(item, feature, zoom) for item in expr[1:5]]
|
|
return f"rgba({values[0]},{values[1]},{values[2]},{values[3]})"
|
|
if head == "has":
|
|
key = expr[1]
|
|
return key in feature.properties and feature.properties.get(key) not in (None, "")
|
|
if head == "zoom":
|
|
return zoom
|
|
if head in {"==", "!="}:
|
|
lhs = evaluate_expression(expr[1], feature, zoom)
|
|
rhs = evaluate_expression(expr[2], feature, zoom)
|
|
result = lhs == rhs
|
|
return result if head == "==" else not result
|
|
if head in {">=", "<=", ">", "<"}:
|
|
lhs = evaluate_expression(expr[1], feature, zoom)
|
|
rhs = evaluate_expression(expr[2], feature, zoom)
|
|
return compare_numbers(lhs, rhs, head)
|
|
if head == "any":
|
|
return any(bool(evaluate_expression(item, feature, zoom)) for item in expr[1:])
|
|
if head == "all":
|
|
return all(bool(evaluate_expression(item, feature, zoom)) for item in expr[1:])
|
|
if head == "case":
|
|
clauses = expr[1:]
|
|
for idx in range(0, len(clauses) - 1, 2):
|
|
if bool(evaluate_expression(clauses[idx], feature, zoom)):
|
|
return evaluate_expression(clauses[idx + 1], feature, zoom)
|
|
return evaluate_expression(clauses[-1], feature, zoom) if clauses else None
|
|
if head == "match":
|
|
value = evaluate_expression(expr[1], feature, zoom)
|
|
arms = expr[2:]
|
|
fallback = arms[-1] if arms else None
|
|
for idx in range(0, len(arms) - 1, 2):
|
|
label = arms[idx]
|
|
result = arms[idx + 1]
|
|
if isinstance(label, list):
|
|
if value in [evaluate_expression(item, feature, zoom) for item in label]:
|
|
return evaluate_expression(result, feature, zoom)
|
|
else:
|
|
if value == evaluate_expression(label, feature, zoom):
|
|
return evaluate_expression(result, feature, zoom)
|
|
return evaluate_expression(fallback, feature, zoom)
|
|
if head == "/":
|
|
lhs = evaluate_expression(expr[1], feature, zoom)
|
|
rhs = evaluate_expression(expr[2], feature, zoom)
|
|
lhs_num = numeric_or_text(lhs)
|
|
rhs_num = numeric_or_text(rhs)
|
|
if isinstance(lhs_num, float) and isinstance(rhs_num, float) and rhs_num != 0:
|
|
return lhs_num / rhs_num
|
|
return None
|
|
if head == "*":
|
|
lhs = evaluate_expression(expr[1], feature, zoom)
|
|
rhs = evaluate_expression(expr[2], feature, zoom)
|
|
lhs_num = numeric_or_text(lhs)
|
|
rhs_num = numeric_or_text(rhs)
|
|
if isinstance(lhs_num, float) and isinstance(rhs_num, float):
|
|
return lhs_num * rhs_num
|
|
return None
|
|
if head == "step":
|
|
input_value = evaluate_expression(expr[1], feature, zoom)
|
|
input_num = numeric_or_text(input_value)
|
|
if not isinstance(input_num, float):
|
|
return evaluate_expression(expr[2], feature, zoom)
|
|
result = evaluate_expression(expr[2], feature, zoom)
|
|
stops = expr[3:]
|
|
for idx in range(0, len(stops), 2):
|
|
stop = evaluate_expression(stops[idx], feature, zoom)
|
|
stop_num = numeric_or_text(stop)
|
|
if not isinstance(stop_num, float):
|
|
continue
|
|
if idx + 1 >= len(stops):
|
|
break
|
|
if input_num >= stop_num:
|
|
result = evaluate_expression(stops[idx + 1], feature, zoom)
|
|
else:
|
|
break
|
|
return result
|
|
if head == "interpolate":
|
|
input_value = evaluate_expression(expr[2], feature, zoom)
|
|
input_num = numeric_or_text(input_value)
|
|
if not isinstance(input_num, float):
|
|
return None
|
|
stops = expr[3:]
|
|
prev_stop = None
|
|
prev_value = None
|
|
for idx in range(0, len(stops), 2):
|
|
stop_num = numeric_or_text(evaluate_expression(stops[idx], feature, zoom))
|
|
stop_value = evaluate_expression(stops[idx + 1], feature, zoom) if idx + 1 < len(stops) else None
|
|
if not isinstance(stop_num, float):
|
|
continue
|
|
if input_num == stop_num:
|
|
return stop_value
|
|
if input_num < stop_num:
|
|
return prev_value if prev_value is not None else stop_value
|
|
prev_stop = stop_num
|
|
prev_value = stop_value
|
|
return prev_value
|
|
return None
|
|
|
|
|
|
def layer_visible(layer: dict[str, Any], zoom: int) -> bool:
|
|
if zoom < int(layer.get("minzoom", 0)):
|
|
return False
|
|
maxzoom = layer.get("maxzoom")
|
|
if maxzoom is not None and zoom >= int(maxzoom):
|
|
return False
|
|
layout = layer.get("layout") or {}
|
|
if layout.get("visibility") == "none":
|
|
return False
|
|
return True
|
|
|
|
|
|
def layer_matches_feature(layer: dict[str, Any], feature: AuditFeature) -> bool:
|
|
if layer.get("source-layer") != feature.layer_name:
|
|
return False
|
|
if not layer_visible(layer, feature.tile_z):
|
|
return False
|
|
filter_expr = layer.get("filter")
|
|
if filter_expr is None:
|
|
return True
|
|
return bool(evaluate_expression(filter_expr, feature, feature.tile_z))
|
|
|
|
|
|
def evaluate_style_value(layer: dict[str, Any], section: str, key: str, feature: AuditFeature) -> Any:
|
|
payload = layer.get(section) or {}
|
|
if key not in payload:
|
|
return None
|
|
return evaluate_expression(payload[key], feature, feature.tile_z)
|
|
|
|
|
|
def build_signature(component_type: str, payload: dict[str, Any]) -> str:
|
|
normalized = {
|
|
k: payload[k]
|
|
for k in sorted(payload)
|
|
if k != "layer_id" and payload[k] not in (None, "", [], {})
|
|
}
|
|
return f"{component_type}:{canonical_json(normalized)}"
|
|
|
|
|
|
def collect_layer_observations(layer: dict[str, Any], feature: AuditFeature) -> list[RenderObservation]:
|
|
observations: list[RenderObservation] = []
|
|
layer_id = str(layer["id"])
|
|
layer_type = str(layer.get("type", ""))
|
|
|
|
if layer_type == "symbol":
|
|
icon_image = evaluate_style_value(layer, "layout", "icon-image", feature)
|
|
text_value = evaluate_style_value(layer, "layout", "text-field", feature)
|
|
text_color = evaluate_style_value(layer, "paint", "text-color", feature)
|
|
text_anchor = evaluate_style_value(layer, "layout", "text-anchor", feature)
|
|
text_size = evaluate_style_value(layer, "layout", "text-size", feature)
|
|
icon_size = evaluate_style_value(layer, "layout", "icon-size", feature)
|
|
if icon_image not in (None, ""):
|
|
payload = {
|
|
"layer_id": layer_id,
|
|
"icon_image": icon_image,
|
|
"icon_size": icon_size,
|
|
}
|
|
observations.append(
|
|
RenderObservation(
|
|
component_type="icon",
|
|
style_layer_id=layer_id,
|
|
signature=build_signature("icon", payload),
|
|
payload=payload,
|
|
)
|
|
)
|
|
if text_value not in (None, ""):
|
|
payload = {
|
|
"layer_id": layer_id,
|
|
"text_value": str(text_value),
|
|
"text_color": normalize_color(text_color),
|
|
"text_anchor": text_anchor,
|
|
"text_size": text_size,
|
|
}
|
|
observations.append(
|
|
RenderObservation(
|
|
component_type="text",
|
|
style_layer_id=layer_id,
|
|
signature=build_signature("text", payload),
|
|
payload=payload,
|
|
)
|
|
)
|
|
return observations
|
|
|
|
if layer_type == "line":
|
|
payload = {
|
|
"layer_id": layer_id,
|
|
"line_color": normalize_color(evaluate_style_value(layer, "paint", "line-color", feature)),
|
|
"line_width": evaluate_style_value(layer, "paint", "line-width", feature),
|
|
"line_dasharray": evaluate_style_value(layer, "paint", "line-dasharray", feature),
|
|
"line_pattern": evaluate_style_value(layer, "paint", "line-pattern", feature),
|
|
}
|
|
observations.append(
|
|
RenderObservation(
|
|
component_type="line",
|
|
style_layer_id=layer_id,
|
|
signature=build_signature("line", payload),
|
|
payload=payload,
|
|
)
|
|
)
|
|
return observations
|
|
|
|
if layer_type == "fill":
|
|
payload = {
|
|
"layer_id": layer_id,
|
|
"fill_color": normalize_color(evaluate_style_value(layer, "paint", "fill-color", feature)),
|
|
"fill_pattern": evaluate_style_value(layer, "paint", "fill-pattern", feature),
|
|
"fill_outline_color": normalize_color(evaluate_style_value(layer, "paint", "fill-outline-color", feature)),
|
|
"fill_opacity": evaluate_style_value(layer, "paint", "fill-opacity", feature),
|
|
}
|
|
observations.append(
|
|
RenderObservation(
|
|
component_type="fill",
|
|
style_layer_id=layer_id,
|
|
signature=build_signature("fill", payload),
|
|
payload=payload,
|
|
)
|
|
)
|
|
return observations
|
|
|
|
return observations
|
|
|
|
|
|
def collect_render_observations(style: dict[str, Any], feature: AuditFeature) -> list[RenderObservation]:
|
|
observations: list[RenderObservation] = []
|
|
for layer in style.get("layers", []):
|
|
if "source-layer" not in layer:
|
|
continue
|
|
if not layer_matches_feature(layer, feature):
|
|
continue
|
|
observations.extend(collect_layer_observations(layer, feature))
|
|
return observations
|
|
|
|
|
|
def load_style(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def iter_tile_paths(root: Path) -> list[Path]:
|
|
return sorted(root.glob("*/*/*.pbf"))
|
|
|
|
|
|
def stable_fallback_properties(properties: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
key: properties[key]
|
|
for key in STABLE_FALLBACK_KEYS
|
|
if key in properties and properties[key] not in (None, "")
|
|
}
|
|
|
|
|
|
def geometry_hash(geometry: dict[str, Any]) -> str:
|
|
return hash_text(canonical_json(geometry))
|
|
|
|
|
|
def extract_legacy_fid(properties: dict[str, Any], native_feature_id: int | None = None) -> str | None:
|
|
legacy = properties.get("fid_legacy_raw")
|
|
if legacy not in (None, ""):
|
|
try:
|
|
return str(int(str(legacy)))
|
|
except ValueError:
|
|
return str(legacy)
|
|
fid = properties.get("fid")
|
|
if fid not in (None, ""):
|
|
fid_text = str(fid).strip()
|
|
# Original tiles store legacy fid as decimal integers. Only treat
|
|
# an 8-char property fid as encrypted hex when it is not plain digits.
|
|
if fid_text.isdigit():
|
|
return str(int(fid_text))
|
|
if FID_CODEC is not None and len(fid_text) == 8:
|
|
try:
|
|
decoded = FID_CODEC.decrypt_number(int(fid_text, 16))
|
|
return str(decoded)
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
return str(int(fid_text))
|
|
except ValueError:
|
|
return fid_text
|
|
if native_feature_id is not None:
|
|
if FID_CODEC is not None:
|
|
try:
|
|
decoded = FID_CODEC.decrypt_number(int(native_feature_id))
|
|
return str(decoded)
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
feature_id_int = int(native_feature_id)
|
|
if feature_id_int > 2147483647:
|
|
feature_id_int -= 4294967296
|
|
return str(feature_id_int)
|
|
except ValueError:
|
|
return str(native_feature_id)
|
|
return None
|
|
|
|
|
|
def object_identity(
|
|
layer_name: str,
|
|
geometry: dict[str, Any],
|
|
properties: dict[str, Any],
|
|
native_feature_id: int | None = None,
|
|
) -> tuple[str, str | None]:
|
|
geom_type = str(geometry.get("type", ""))
|
|
legacy_source_layer = text_or_none(properties.get("source_layer_jp")) or layer_name
|
|
legacy_fid = extract_legacy_fid(properties, native_feature_id)
|
|
if legacy_fid:
|
|
if MATCH_ON_FID_ONLY:
|
|
return f"fid:{legacy_fid}|geom:{geom_type}", legacy_fid
|
|
return f"fid:{legacy_fid}|src:{legacy_source_layer}|geom:{geom_type}", legacy_fid
|
|
fallback_payload = {
|
|
"source_layer": legacy_source_layer,
|
|
"geom_type": geom_type,
|
|
"geometry_hash": geometry_hash(geometry),
|
|
"stable_props": stable_fallback_properties(properties),
|
|
}
|
|
return f"fallback:{hash_text(canonical_json(fallback_payload))}", None
|
|
|
|
|
|
def feature_instance_id(feature: AuditFeature) -> str:
|
|
if MATCH_ON_FID_ONLY:
|
|
return f"{feature.object_id}|z:{feature.tile_z}|x:{feature.tile_x}|y:{feature.tile_y}"
|
|
return (
|
|
f"{feature.object_id}|z:{feature.tile_z}|x:{feature.tile_x}|"
|
|
f"y:{feature.tile_y}|layer:{feature.layer_name}"
|
|
)
|
|
|
|
|
|
def decode_tile_instances(dataset: str, tile_root: Path, style: dict[str, Any], tile_path: Path) -> dict[str, dict[str, Any]]:
|
|
instances: dict[str, dict[str, Any]] = {}
|
|
rel = tile_path.relative_to(tile_root)
|
|
z = int(rel.parts[0])
|
|
x = int(rel.parts[1])
|
|
y = int(tile_path.stem)
|
|
decoded = mapbox_vector_tile.decode(tile_path.read_bytes())
|
|
for layer_name, payload in decoded.items():
|
|
for feature in payload.get("features", []):
|
|
geometry = feature.get("geometry") or {}
|
|
properties = dict(feature.get("properties") or {})
|
|
native_feature_id = feature.get("id")
|
|
native_feature_id = int(native_feature_id) if isinstance(native_feature_id, int) else None
|
|
object_id, legacy_fid = object_identity(layer_name, geometry, properties, native_feature_id)
|
|
audit_feature = AuditFeature(
|
|
dataset=dataset,
|
|
tile_z=z,
|
|
tile_x=x,
|
|
tile_y=y,
|
|
layer_name=layer_name,
|
|
geom_type=str(geometry.get("type", "")),
|
|
object_id=object_id,
|
|
fid_legacy=legacy_fid,
|
|
native_feature_id=native_feature_id,
|
|
properties=properties,
|
|
)
|
|
instance_id = feature_instance_id(audit_feature)
|
|
instances[instance_id] = {
|
|
"feature": audit_feature,
|
|
"observations": collect_render_observations(style, audit_feature),
|
|
}
|
|
return instances
|
|
|
|
|
|
def summarize_observations(observations: list[RenderObservation]) -> dict[str, list[str]]:
|
|
grouped: dict[str, set[str]] = defaultdict(set)
|
|
for observation in observations:
|
|
grouped[observation.component_type].add(observation.signature)
|
|
return {key: sorted(values) for key, values in grouped.items()}
|
|
|
|
|
|
def summarize_observation_payloads(observations: list[RenderObservation]) -> dict[str, list[dict[str, Any]]]:
|
|
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for observation in observations:
|
|
grouped[observation.component_type].append(dict(observation.payload))
|
|
return {
|
|
key: sorted(values, key=canonical_json)
|
|
for key, values in grouped.items()
|
|
}
|
|
|
|
|
|
def normalize_component_map(component_map: dict[str, list[str]]) -> dict[str, tuple[str, ...]]:
|
|
return {key: tuple(values) for key, values in sorted(component_map.items())}
|
|
|
|
|
|
def normalize_payload_map(payload_map: dict[str, list[dict[str, Any]]]) -> dict[str, tuple[dict[str, Any], ...]]:
|
|
return {
|
|
key: tuple(sorted(values, key=canonical_json))
|
|
for key, values in sorted(payload_map.items())
|
|
}
|
|
|
|
|
|
def compare_instance(
|
|
original: dict[str, Any] | None,
|
|
engineering: dict[str, Any] | None,
|
|
) -> dict[str, Any]:
|
|
if original is None:
|
|
eng_feature = engineering["feature"]
|
|
engineering_components = normalize_component_map(
|
|
summarize_observations(engineering["observations"])
|
|
)
|
|
engineering_payloads = normalize_payload_map(
|
|
summarize_observation_payloads(engineering["observations"])
|
|
)
|
|
return {
|
|
"status": "extra_in_engineering",
|
|
"object_instance_id": feature_instance_id(eng_feature),
|
|
"object_id": eng_feature.object_id,
|
|
"fid_legacy": eng_feature.fid_legacy,
|
|
"tile": f"{eng_feature.tile_z}/{eng_feature.tile_x}/{eng_feature.tile_y}",
|
|
"tile_z": eng_feature.tile_z,
|
|
"tile_x": eng_feature.tile_x,
|
|
"tile_y": eng_feature.tile_y,
|
|
"source_layer": eng_feature.layer_name,
|
|
"canonical_object_type": text_or_none(eng_feature.properties.get("canonical_object_type")),
|
|
"original_components": {},
|
|
"engineering_components": engineering_components,
|
|
"original_payloads": {},
|
|
"engineering_payloads": engineering_payloads,
|
|
}
|
|
if engineering is None:
|
|
orig_feature = original["feature"]
|
|
original_components = normalize_component_map(
|
|
summarize_observations(original["observations"])
|
|
)
|
|
original_payloads = normalize_payload_map(
|
|
summarize_observation_payloads(original["observations"])
|
|
)
|
|
return {
|
|
"status": "missing_in_engineering",
|
|
"object_instance_id": feature_instance_id(orig_feature),
|
|
"object_id": orig_feature.object_id,
|
|
"fid_legacy": orig_feature.fid_legacy,
|
|
"tile": f"{orig_feature.tile_z}/{orig_feature.tile_x}/{orig_feature.tile_y}",
|
|
"tile_z": orig_feature.tile_z,
|
|
"tile_x": orig_feature.tile_x,
|
|
"tile_y": orig_feature.tile_y,
|
|
"source_layer": orig_feature.layer_name,
|
|
"canonical_object_type": None,
|
|
"original_components": original_components,
|
|
"engineering_components": {},
|
|
"original_payloads": original_payloads,
|
|
"engineering_payloads": {},
|
|
}
|
|
|
|
orig_feature = original["feature"]
|
|
eng_feature = engineering["feature"]
|
|
original_components = normalize_component_map(summarize_observations(original["observations"]))
|
|
engineering_components = normalize_component_map(summarize_observations(engineering["observations"]))
|
|
original_payloads = normalize_payload_map(summarize_observation_payloads(original["observations"]))
|
|
engineering_payloads = normalize_payload_map(summarize_observation_payloads(engineering["observations"]))
|
|
|
|
if original_components == engineering_components:
|
|
status = "exact_match"
|
|
elif not original_components and engineering_components:
|
|
status = "extra_in_engineering"
|
|
elif original_components and not engineering_components:
|
|
status = "missing_in_engineering"
|
|
else:
|
|
status = "mismatch"
|
|
|
|
return {
|
|
"status": status,
|
|
"object_instance_id": feature_instance_id(orig_feature),
|
|
"object_id": orig_feature.object_id,
|
|
"fid_legacy": orig_feature.fid_legacy or eng_feature.fid_legacy,
|
|
"tile": f"{orig_feature.tile_z}/{orig_feature.tile_x}/{orig_feature.tile_y}",
|
|
"tile_z": orig_feature.tile_z,
|
|
"tile_x": orig_feature.tile_x,
|
|
"tile_y": orig_feature.tile_y,
|
|
"source_layer": text_or_none(eng_feature.properties.get("source_layer_jp")) or orig_feature.layer_name,
|
|
"canonical_object_type": text_or_none(eng_feature.properties.get("canonical_object_type")),
|
|
"original_components": original_components,
|
|
"engineering_components": engineering_components,
|
|
"original_payloads": original_payloads,
|
|
"engineering_payloads": engineering_payloads,
|
|
}
|
|
|
|
|
|
def format_component_map(component_map: dict[str, tuple[str, ...]] | dict[str, list[str]]) -> str:
|
|
if not component_map:
|
|
return "none"
|
|
parts = []
|
|
for key in sorted(component_map):
|
|
values = component_map[key]
|
|
parts.append(f"{key}={list(values)}")
|
|
return "; ".join(parts)
|
|
|
|
|
|
def has_component(component_map: dict[str, tuple[str, ...]] | dict[str, list[str]], component_type: str) -> bool:
|
|
values = component_map.get(component_type)
|
|
return bool(values)
|
|
|
|
|
|
def get_component_payloads(
|
|
payload_map: dict[str, tuple[dict[str, Any], ...]] | dict[str, list[dict[str, Any]]],
|
|
component_type: str,
|
|
) -> list[dict[str, Any]]:
|
|
values = payload_map.get(component_type) or ()
|
|
return [dict(value) for value in values]
|
|
|
|
|
|
def payload_has_nontransparent_fill(payloads: list[dict[str, Any]]) -> bool:
|
|
for payload in payloads:
|
|
color = text_or_none(payload.get("fill_color"))
|
|
opacity = payload.get("fill_opacity")
|
|
if opacity == 0:
|
|
continue
|
|
if color in (None, "rgba(0,0,0,0)", "rgba(0, 0, 0, 0)"):
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
|
|
def payload_has_transparent_fill(payloads: list[dict[str, Any]]) -> bool:
|
|
for payload in payloads:
|
|
color = text_or_none(payload.get("fill_color"))
|
|
opacity = payload.get("fill_opacity")
|
|
if opacity == 0:
|
|
return True
|
|
if color in ("rgba(0,0,0,0)", "rgba(0, 0, 0, 0)"):
|
|
return True
|
|
return False
|
|
|
|
|
|
def text_payloads_with_numeric_value(payloads: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
numeric_payloads: list[dict[str, Any]] = []
|
|
for payload in payloads:
|
|
text_value = text_or_none(payload.get("text_value"))
|
|
if text_value is None:
|
|
continue
|
|
if any(char.isdigit() for char in text_value):
|
|
numeric_payloads.append(payload)
|
|
return numeric_payloads
|
|
|
|
|
|
def classify_annotation_issue(item: dict[str, Any]) -> str | None:
|
|
original_components = item["original_components"]
|
|
engineering_components = item["engineering_components"]
|
|
original_has_text = has_component(original_components, "text")
|
|
engineering_has_text = has_component(engineering_components, "text")
|
|
original_has_icon = has_component(original_components, "icon")
|
|
engineering_has_icon = has_component(engineering_components, "icon")
|
|
|
|
if original_has_text and not engineering_has_text:
|
|
return "text_missing_in_engineering"
|
|
if not original_has_text and engineering_has_text:
|
|
return "extra_text_in_engineering"
|
|
if original_has_icon and not engineering_has_icon:
|
|
return "icon_missing_in_engineering"
|
|
if not original_has_icon and engineering_has_icon:
|
|
return "extra_icon_in_engineering"
|
|
return None
|
|
|
|
|
|
def classify_style_semantic_issue(item: dict[str, Any]) -> str | None:
|
|
source_layer = item["source_layer"] or "unknown"
|
|
original_payloads = item.get("original_payloads", {})
|
|
engineering_payloads = item.get("engineering_payloads", {})
|
|
|
|
original_fill_payloads = get_component_payloads(original_payloads, "fill")
|
|
engineering_fill_payloads = get_component_payloads(engineering_payloads, "fill")
|
|
original_text_payloads = get_component_payloads(original_payloads, "text")
|
|
engineering_text_payloads = get_component_payloads(engineering_payloads, "text")
|
|
original_icon_payloads = get_component_payloads(original_payloads, "icon")
|
|
engineering_icon_payloads = get_component_payloads(engineering_payloads, "icon")
|
|
|
|
if source_layer in SUPPORT_FILL_SOURCE_LAYERS:
|
|
if payload_has_transparent_fill(original_fill_payloads) and payload_has_nontransparent_fill(engineering_fill_payloads):
|
|
return "support_layer_strengthened_fill"
|
|
|
|
original_numeric_texts = text_payloads_with_numeric_value(original_text_payloads)
|
|
engineering_numeric_texts = text_payloads_with_numeric_value(engineering_text_payloads)
|
|
|
|
if source_layer in DEPTH_NUMERIC_SOURCE_LAYERS and original_numeric_texts and not engineering_numeric_texts:
|
|
return "depth_numeric_missing"
|
|
if source_layer in CLEARANCE_NUMERIC_SOURCE_LAYERS and original_numeric_texts and not engineering_numeric_texts:
|
|
return "clearance_numeric_missing"
|
|
if source_layer in SAFETY_ICON_SOURCE_LAYERS and original_icon_payloads and not engineering_icon_payloads:
|
|
return "safety_icon_missing"
|
|
|
|
return None
|
|
|
|
|
|
def write_reports(
|
|
*,
|
|
original_count: int,
|
|
engineering_count: int,
|
|
result_count: int,
|
|
status_counter: Counter[str],
|
|
source_layer_counter: Counter[tuple[str, str]],
|
|
mismatch_examples: list[dict[str, Any]],
|
|
annotation_issue_counter: Counter[tuple[str, str]],
|
|
annotation_issue_examples: list[dict[str, Any]],
|
|
style_semantic_issue_counter: Counter[tuple[str, str]],
|
|
style_semantic_issue_examples: list[dict[str, Any]],
|
|
) -> None:
|
|
payload = {
|
|
"original_feature_instances": original_count,
|
|
"engineering_feature_instances": engineering_count,
|
|
"result_count": result_count,
|
|
"status_counts": dict(status_counter),
|
|
"source_layer_issue_counts": [
|
|
{"status": status, "source_layer": source_layer, "count": count}
|
|
for (status, source_layer), count in source_layer_counter.most_common(30)
|
|
],
|
|
"annotation_issue_counts": [
|
|
{"issue": issue, "source_layer": source_layer, "count": count}
|
|
for (issue, source_layer), count in annotation_issue_counter.most_common(30)
|
|
],
|
|
"style_semantic_issue_counts": [
|
|
{"issue": issue, "source_layer": source_layer, "count": count}
|
|
for (issue, source_layer), count in style_semantic_issue_counter.most_common(30)
|
|
],
|
|
"mismatch_examples": mismatch_examples,
|
|
"annotation_issue_examples": annotation_issue_examples,
|
|
"style_semantic_issue_examples": style_semantic_issue_examples,
|
|
}
|
|
REPORT_JSON_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
lines = [
|
|
f"# NavSea 原始版 vs {COMPARISON_LABEL}渲染审计报告",
|
|
"",
|
|
"## 范围",
|
|
"",
|
|
f"- 原始样式: `{ORIGINAL_STYLE_PATH}`",
|
|
f"- {COMPARISON_LABEL}样式: `{ENGINEERING_STYLE_PATH}`",
|
|
f"- 原始瓦片根目录: `{ORIGINAL_TILE_ROOT}`",
|
|
f"- {COMPARISON_LABEL}瓦片根目录: `{ENGINEERING_TILE_ROOT}`",
|
|
f"- 原始 feature 实例数: `{original_count}`",
|
|
f"- {COMPARISON_LABEL} feature 实例数: `{engineering_count}`",
|
|
f"- 审计结果数: `{result_count}`",
|
|
"",
|
|
"## 匹配口径",
|
|
"",
|
|
"- 主键优先使用 legacy `fid`。",
|
|
"- 没有 `fid` 的对象，回退到 `geometry + 稳定旧属性`。",
|
|
"- 审计粒度保留 tile 实例，因为渲染具有 zoom 敏感性。",
|
|
"",
|
|
"## 结果统计",
|
|
"",
|
|
]
|
|
for status, count in status_counter.most_common():
|
|
lines.append(f"- `{status}`: `{count}`")
|
|
|
|
lines.extend([
|
|
"",
|
|
"## 主要问题层",
|
|
"",
|
|
])
|
|
if not source_layer_counter:
|
|
lines.append("- 没有发现差异。")
|
|
else:
|
|
for (status, source_layer), count in source_layer_counter.most_common(20):
|
|
lines.append(f"- `{status}` | `{source_layer}` | `{count}`")
|
|
|
|
lines.extend([
|
|
"",
|
|
"## 标注专项问题",
|
|
"",
|
|
])
|
|
if not annotation_issue_counter:
|
|
lines.append("- 没有发现标注专项问题。")
|
|
else:
|
|
for (issue, source_layer), count in annotation_issue_counter.most_common(20):
|
|
lines.append(f"- `{issue}` | `{source_layer}` | `{count}`")
|
|
|
|
lines.extend([
|
|
"",
|
|
"## Style 语义专项问题",
|
|
"",
|
|
])
|
|
if not style_semantic_issue_counter:
|
|
lines.append("- 没有发现 style 语义专项问题。")
|
|
else:
|
|
for (issue, source_layer), count in style_semantic_issue_counter.most_common(20):
|
|
lines.append(f"- `{issue}` | `{source_layer}` | `{count}`")
|
|
|
|
lines.extend([
|
|
"",
|
|
"## 差异样例",
|
|
"",
|
|
])
|
|
if not mismatch_examples:
|
|
lines.append("- 没有差异样例。")
|
|
else:
|
|
for item in mismatch_examples:
|
|
lines.append(
|
|
f"- `{item['status']}` | tile=`{item['tile']}` | source_layer=`{item['source_layer']}` | "
|
|
f"fid=`{item['fid_legacy']}` | object=`{item['canonical_object_type'] or 'n/a'}`"
|
|
)
|
|
lines.append(f" original: {format_component_map(item['original_components'])}")
|
|
lines.append(f" engineering: {format_component_map(item['engineering_components'])}")
|
|
|
|
lines.extend([
|
|
"",
|
|
"## 标注问题样例",
|
|
"",
|
|
])
|
|
if not annotation_issue_examples:
|
|
lines.append("- 没有标注问题样例。")
|
|
else:
|
|
for item in annotation_issue_examples:
|
|
lines.append(
|
|
f"- `{item['annotation_issue']}` | tile=`{item['tile']}` | source_layer=`{item['source_layer']}` | "
|
|
f"fid=`{item['fid_legacy']}` | object=`{item['canonical_object_type'] or 'n/a'}`"
|
|
)
|
|
lines.append(f" original: {format_component_map(item['original_components'])}")
|
|
lines.append(f" engineering: {format_component_map(item['engineering_components'])}")
|
|
|
|
lines.extend([
|
|
"",
|
|
"## Style 语义问题样例",
|
|
"",
|
|
])
|
|
if not style_semantic_issue_examples:
|
|
lines.append("- 没有 style 语义问题样例。")
|
|
else:
|
|
for item in style_semantic_issue_examples:
|
|
lines.append(
|
|
f"- `{item['style_semantic_issue']}` | tile=`{item['tile']}` | source_layer=`{item['source_layer']}` | "
|
|
f"fid=`{item['fid_legacy']}` | object=`{item['canonical_object_type'] or 'n/a'}`"
|
|
)
|
|
lines.append(f" original: {format_component_map(item['original_components'])}")
|
|
lines.append(f" engineering: {format_component_map(item['engineering_components'])}")
|
|
|
|
REPORT_MD_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def db_connect() -> pymysql.Connection:
|
|
cfg = DbConfig()
|
|
kwargs: dict[str, Any] = {
|
|
"host": cfg.host,
|
|
"port": cfg.port,
|
|
"user": cfg.user,
|
|
"password": cfg.password,
|
|
"database": cfg.database,
|
|
"charset": "utf8mb4",
|
|
"autocommit": False,
|
|
}
|
|
if cfg.unix_socket:
|
|
kwargs["unix_socket"] = cfg.unix_socket
|
|
return pymysql.connect(**kwargs)
|
|
|
|
|
|
def ensure_audit_tables(cur: pymysql.cursors.Cursor) -> None:
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS navsea_render_audit_run (
|
|
audit_name VARCHAR(128) NOT NULL,
|
|
original_style_path VARCHAR(512) NOT NULL,
|
|
engineering_style_path VARCHAR(512) NOT NULL,
|
|
original_tile_root VARCHAR(512) NOT NULL,
|
|
engineering_tile_root VARCHAR(512) NOT NULL,
|
|
original_feature_instances INT NOT NULL,
|
|
engineering_feature_instances INT NOT NULL,
|
|
result_count INT NOT NULL,
|
|
status_counts_json LONGTEXT NOT NULL,
|
|
source_layer_issue_counts_json LONGTEXT NOT NULL,
|
|
report_md_path VARCHAR(512) NOT NULL,
|
|
report_json_path VARCHAR(512) NOT NULL,
|
|
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (audit_name)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
|
"""
|
|
)
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS navsea_render_audit_result (
|
|
audit_name VARCHAR(64) NOT NULL,
|
|
object_instance_key VARCHAR(64) NOT NULL,
|
|
object_instance_id TEXT NOT NULL,
|
|
object_id VARCHAR(255) NOT NULL,
|
|
fid_legacy VARCHAR(64) DEFAULT NULL,
|
|
tile_z INT NOT NULL,
|
|
tile_x INT NOT NULL,
|
|
tile_y INT NOT NULL,
|
|
source_layer VARCHAR(100) DEFAULT NULL,
|
|
canonical_object_type VARCHAR(191) DEFAULT NULL,
|
|
status VARCHAR(32) NOT NULL,
|
|
original_components_json LONGTEXT NOT NULL,
|
|
engineering_components_json LONGTEXT NOT NULL,
|
|
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (audit_name, object_instance_key),
|
|
KEY idx_render_audit_fid (fid_legacy),
|
|
KEY idx_render_audit_status (status),
|
|
KEY idx_render_audit_layer (source_layer)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
|
"""
|
|
)
|
|
|
|
|
|
def insert_result_batch(cur: pymysql.cursors.Cursor, rows: list[tuple[Any, ...]]) -> None:
|
|
if not rows:
|
|
return
|
|
cur.executemany(
|
|
"""
|
|
INSERT INTO navsea_render_audit_result (
|
|
audit_name,
|
|
object_instance_key,
|
|
object_instance_id,
|
|
object_id,
|
|
fid_legacy,
|
|
tile_z,
|
|
tile_x,
|
|
tile_y,
|
|
source_layer,
|
|
canonical_object_type,
|
|
status,
|
|
original_components_json,
|
|
engineering_components_json
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
""",
|
|
rows,
|
|
)
|
|
|
|
|
|
def persist_audit_run(
|
|
cur: pymysql.cursors.Cursor,
|
|
*,
|
|
original_count: int,
|
|
engineering_count: int,
|
|
result_count: int,
|
|
status_counter: Counter[str],
|
|
source_layer_counter: Counter[tuple[str, str]],
|
|
) -> None:
|
|
source_layer_issue_counts = [
|
|
{"status": status, "source_layer": source_layer, "count": count}
|
|
for (status, source_layer), count in source_layer_counter.most_common(100)
|
|
]
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO navsea_render_audit_run (
|
|
audit_name,
|
|
original_style_path,
|
|
engineering_style_path,
|
|
original_tile_root,
|
|
engineering_tile_root,
|
|
original_feature_instances,
|
|
engineering_feature_instances,
|
|
result_count,
|
|
status_counts_json,
|
|
source_layer_issue_counts_json,
|
|
report_md_path,
|
|
report_json_path
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
""",
|
|
(
|
|
AUDIT_NAME,
|
|
str(ORIGINAL_STYLE_PATH),
|
|
str(ENGINEERING_STYLE_PATH),
|
|
str(ORIGINAL_TILE_ROOT),
|
|
str(ENGINEERING_TILE_ROOT),
|
|
original_count,
|
|
engineering_count,
|
|
result_count,
|
|
json.dumps(dict(status_counter), ensure_ascii=False),
|
|
json.dumps(source_layer_issue_counts, ensure_ascii=False),
|
|
str(REPORT_MD_PATH),
|
|
str(REPORT_JSON_PATH),
|
|
),
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Audit original vs engineering render outputs.")
|
|
parser.add_argument("--audit-name", default=DEFAULT_AUDIT_NAME)
|
|
parser.add_argument("--comparison-label", default="工程版")
|
|
parser.add_argument("--fid-key")
|
|
parser.add_argument("--match-on-fid-only", action="store_true")
|
|
parser.add_argument("--original-style", type=Path, default=DEFAULT_ORIGINAL_STYLE_PATH)
|
|
parser.add_argument("--engineering-style", type=Path, default=DEFAULT_ENGINEERING_STYLE_PATH)
|
|
parser.add_argument("--original-tile-root", type=Path, default=DEFAULT_ORIGINAL_TILE_ROOT)
|
|
parser.add_argument("--engineering-tile-root", type=Path, default=DEFAULT_ENGINEERING_TILE_ROOT)
|
|
parser.add_argument("--report-md", type=Path, default=DEFAULT_REPORT_MD_PATH)
|
|
parser.add_argument("--report-json", type=Path, default=DEFAULT_REPORT_JSON_PATH)
|
|
parser.add_argument("--skip-db-persist", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
global AUDIT_NAME
|
|
global ORIGINAL_STYLE_PATH
|
|
global ENGINEERING_STYLE_PATH
|
|
global ORIGINAL_TILE_ROOT
|
|
global ENGINEERING_TILE_ROOT
|
|
global REPORT_MD_PATH
|
|
global REPORT_JSON_PATH
|
|
global SKIP_DB_PERSIST
|
|
global COMPARISON_LABEL
|
|
global FID_CODEC
|
|
global MATCH_ON_FID_ONLY
|
|
|
|
args = parse_args()
|
|
AUDIT_NAME = args.audit_name
|
|
COMPARISON_LABEL = args.comparison_label
|
|
FID_CODEC = NavSeaFidCodec(args.fid_key.encode("utf-8")) if args.fid_key else None
|
|
MATCH_ON_FID_ONLY = args.match_on_fid_only
|
|
ORIGINAL_STYLE_PATH = args.original_style
|
|
ENGINEERING_STYLE_PATH = args.engineering_style
|
|
ORIGINAL_TILE_ROOT = args.original_tile_root
|
|
ENGINEERING_TILE_ROOT = args.engineering_tile_root
|
|
REPORT_MD_PATH = args.report_md
|
|
REPORT_JSON_PATH = args.report_json
|
|
SKIP_DB_PERSIST = args.skip_db_persist
|
|
|
|
original_style = load_style(ORIGINAL_STYLE_PATH)
|
|
engineering_style = load_style(ENGINEERING_STYLE_PATH)
|
|
engineering_tiles = iter_tile_paths(ENGINEERING_TILE_ROOT)
|
|
|
|
original_count = 0
|
|
engineering_count = 0
|
|
result_count = 0
|
|
status_counter: Counter[str] = Counter()
|
|
source_layer_counter: Counter[tuple[str, str]] = Counter()
|
|
mismatch_examples: list[dict[str, Any]] = []
|
|
annotation_issue_counter: Counter[tuple[str, str]] = Counter()
|
|
annotation_issue_examples: list[dict[str, Any]] = []
|
|
style_semantic_issue_counter: Counter[tuple[str, str]] = Counter()
|
|
style_semantic_issue_examples: list[dict[str, Any]] = []
|
|
|
|
if SKIP_DB_PERSIST:
|
|
for engineering_tile in engineering_tiles:
|
|
rel = engineering_tile.relative_to(ENGINEERING_TILE_ROOT)
|
|
original_tile = ORIGINAL_TILE_ROOT / rel
|
|
if not original_tile.exists():
|
|
continue
|
|
|
|
original_instances = decode_tile_instances("original", ORIGINAL_TILE_ROOT, original_style, original_tile)
|
|
engineering_instances = decode_tile_instances(
|
|
"engineering", ENGINEERING_TILE_ROOT, engineering_style, engineering_tile
|
|
)
|
|
original_count += len(original_instances)
|
|
engineering_count += len(engineering_instances)
|
|
|
|
all_instance_ids = sorted(set(original_instances) | set(engineering_instances))
|
|
for instance_id in all_instance_ids:
|
|
item = compare_instance(original_instances.get(instance_id), engineering_instances.get(instance_id))
|
|
result_count += 1
|
|
status_counter[item["status"]] += 1
|
|
if item["status"] != "exact_match":
|
|
source_layer_counter[(item["status"], item["source_layer"] or "unknown")] += 1
|
|
if len(mismatch_examples) < MAX_EXAMPLES:
|
|
mismatch_examples.append(item)
|
|
annotation_issue = classify_annotation_issue(item)
|
|
if annotation_issue:
|
|
source_layer = item["source_layer"] or "unknown"
|
|
annotation_issue_counter[(annotation_issue, source_layer)] += 1
|
|
if len(annotation_issue_examples) < MAX_EXAMPLES:
|
|
annotation_issue_examples.append(
|
|
{
|
|
**item,
|
|
"annotation_issue": annotation_issue,
|
|
}
|
|
)
|
|
style_semantic_issue = classify_style_semantic_issue(item)
|
|
if style_semantic_issue:
|
|
source_layer = item["source_layer"] or "unknown"
|
|
style_semantic_issue_counter[(style_semantic_issue, source_layer)] += 1
|
|
if len(style_semantic_issue_examples) < MAX_EXAMPLES:
|
|
style_semantic_issue_examples.append(
|
|
{
|
|
**item,
|
|
"style_semantic_issue": style_semantic_issue,
|
|
}
|
|
)
|
|
else:
|
|
with db_connect() as conn:
|
|
with conn.cursor() as cur:
|
|
ensure_audit_tables(cur)
|
|
cur.execute("TRUNCATE TABLE navsea_render_audit_result")
|
|
cur.execute("TRUNCATE TABLE navsea_render_audit_run")
|
|
conn.commit()
|
|
|
|
rows: list[tuple[Any, ...]] = []
|
|
for engineering_tile in engineering_tiles:
|
|
rel = engineering_tile.relative_to(ENGINEERING_TILE_ROOT)
|
|
original_tile = ORIGINAL_TILE_ROOT / rel
|
|
if not original_tile.exists():
|
|
continue
|
|
|
|
original_instances = decode_tile_instances("original", ORIGINAL_TILE_ROOT, original_style, original_tile)
|
|
engineering_instances = decode_tile_instances(
|
|
"engineering", ENGINEERING_TILE_ROOT, engineering_style, engineering_tile
|
|
)
|
|
original_count += len(original_instances)
|
|
engineering_count += len(engineering_instances)
|
|
|
|
all_instance_ids = sorted(set(original_instances) | set(engineering_instances))
|
|
for instance_id in all_instance_ids:
|
|
item = compare_instance(original_instances.get(instance_id), engineering_instances.get(instance_id))
|
|
result_count += 1
|
|
status_counter[item["status"]] += 1
|
|
if item["status"] != "exact_match":
|
|
source_layer_counter[(item["status"], item["source_layer"] or "unknown")] += 1
|
|
if len(mismatch_examples) < MAX_EXAMPLES:
|
|
mismatch_examples.append(item)
|
|
annotation_issue = classify_annotation_issue(item)
|
|
if annotation_issue:
|
|
source_layer = item["source_layer"] or "unknown"
|
|
annotation_issue_counter[(annotation_issue, source_layer)] += 1
|
|
if len(annotation_issue_examples) < MAX_EXAMPLES:
|
|
annotation_issue_examples.append(
|
|
{
|
|
**item,
|
|
"annotation_issue": annotation_issue,
|
|
}
|
|
)
|
|
style_semantic_issue = classify_style_semantic_issue(item)
|
|
if style_semantic_issue:
|
|
source_layer = item["source_layer"] or "unknown"
|
|
style_semantic_issue_counter[(style_semantic_issue, source_layer)] += 1
|
|
if len(style_semantic_issue_examples) < MAX_EXAMPLES:
|
|
style_semantic_issue_examples.append(
|
|
{
|
|
**item,
|
|
"style_semantic_issue": style_semantic_issue,
|
|
}
|
|
)
|
|
|
|
rows.append(
|
|
(
|
|
AUDIT_NAME,
|
|
hash_text(item["object_instance_id"]),
|
|
item["object_instance_id"],
|
|
item["object_id"],
|
|
item["fid_legacy"],
|
|
item["tile_z"],
|
|
item["tile_x"],
|
|
item["tile_y"],
|
|
item["source_layer"],
|
|
item["canonical_object_type"],
|
|
item["status"],
|
|
json.dumps(item["original_components"], ensure_ascii=False),
|
|
json.dumps(item["engineering_components"], ensure_ascii=False),
|
|
)
|
|
)
|
|
if len(rows) >= DB_BATCH_SIZE:
|
|
insert_result_batch(cur, rows)
|
|
conn.commit()
|
|
rows.clear()
|
|
|
|
insert_result_batch(cur, rows)
|
|
persist_audit_run(
|
|
cur,
|
|
original_count=original_count,
|
|
engineering_count=engineering_count,
|
|
result_count=result_count,
|
|
status_counter=status_counter,
|
|
source_layer_counter=source_layer_counter,
|
|
)
|
|
conn.commit()
|
|
|
|
write_reports(
|
|
original_count=original_count,
|
|
engineering_count=engineering_count,
|
|
result_count=result_count,
|
|
status_counter=status_counter,
|
|
source_layer_counter=source_layer_counter,
|
|
mismatch_examples=mismatch_examples,
|
|
annotation_issue_counter=annotation_issue_counter,
|
|
annotation_issue_examples=annotation_issue_examples,
|
|
style_semantic_issue_counter=style_semantic_issue_counter,
|
|
style_semantic_issue_examples=style_semantic_issue_examples,
|
|
)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"original_feature_instances": original_count,
|
|
"engineering_feature_instances": engineering_count,
|
|
"results": result_count,
|
|
"report_md": str(REPORT_MD_PATH),
|
|
"report_json": str(REPORT_JSON_PATH),
|
|
"audit_name": AUDIT_NAME,
|
|
"db_run_table": "navsea_render_audit_run",
|
|
"db_result_table": "navsea_render_audit_result",
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|