#!/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()