#!/usr/bin/env python3 from __future__ import annotations import argparse import json from collections import Counter, defaultdict from dataclasses import dataclass from pathlib import Path from typing import Any import mapbox_vector_tile DEFAULT_RAW_ROOT = Path( "/home/wwwroot/newpec/exported_auto/" "tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles" ) DEFAULT_DELIVERY_ROOT = Path("/home/wwwroot/pbf-delivery-karatsu-20nm") DEFAULT_FINAL_ROOT = Path("/home/wwwroot/pbf-delivery-karatsu-20nm-final") DEFAULT_REPORT_ROOT = Path("/root/sourceserver/pbf/report/object_preservation_20nm") AUDIT_RULES = ( { "raw_layer": "p航行危険障害物", "delivery_layer": "navigation_hazard_point", "raw_match_field": "分類番号", "delivery_match_field": "class_code", "label": "navigation hazards", }, { "raw_layer": "p投錨注意障害物", "delivery_layer": "anchor_caution_hazard_point", "raw_match_field": "分類番号", "delivery_match_field": "class_code", "label": "anchor hazards", }, { "raw_layer": "p航路標識群", "delivery_layer": "navigation_marks", "raw_match_field": "表示用番号", "delivery_match_field": "display_code", "label": "navigation marks", }, { "raw_layer": "p施設・境界線等", "delivery_layer": "facility_boundary_point", "raw_match_field": "分類番号", "delivery_match_field": "class_code", "label": "facility points", }, { "raw_layer": "P航路ククリ", "delivery_layer": "route_outline", "raw_match_field": "分類番号", "delivery_match_field": "class_code", "label": "route outlines", }, ) MAX_EXAMPLES = 20 @dataclass(frozen=True) class TileCoord: z: int x: int y: int def canonical_json(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def parse_at_name(at_raw: Any) -> str | None: if not at_raw: return None try: parsed = json.loads(at_raw) except Exception: return None for item in parsed: if not isinstance(item, list) or len(item) != 2: continue if item[0] in {"名称", "船名"}: text = str(item[1]).strip() if text: return text return None def geometry_signature(feature: dict[str, Any]) -> str: geometry = feature.get("geometry") or {} return canonical_json(geometry) def build_tile_index(tile_path: Path, rules: tuple[dict[str, str], ...]) -> dict[str, list[dict[str, Any]]]: decoded = mapbox_vector_tile.decode(tile_path.read_bytes()) indexed: dict[str, list[dict[str, Any]]] = {} for rule in rules: indexed[rule["delivery_layer"]] = decoded.get(rule["delivery_layer"], {}).get("features", []) indexed[rule["raw_layer"]] = decoded.get(rule["raw_layer"], {}).get("features", []) return indexed def audit_one_target( *, target_name: str, raw_root: Path, target_root: Path, report_root: Path, ) -> tuple[Path, Path]: report_root.mkdir(parents=True, exist_ok=True) report_json = report_root / f"object_preservation_{target_name}.json" report_md = report_root / f"object_preservation_{target_name}.md" target_tiles = sorted(target_root.glob("*/*/*.pbf")) totals = Counter() rule_stats: dict[str, Counter[str]] = defaultdict(Counter) class_stats: dict[str, Counter[str]] = defaultdict(Counter) issue_examples: dict[str, list[dict[str, Any]]] = defaultdict(list) for target_tile in target_tiles: rel = target_tile.relative_to(target_root) raw_tile = raw_root / rel if not raw_tile.exists(): continue totals["tiles_compared"] += 1 z = int(rel.parts[0]) x = int(rel.parts[1]) y = int(Path(rel.parts[2]).stem) tile = TileCoord(z=z, x=x, y=y) raw_index = build_tile_index(raw_tile, AUDIT_RULES) target_index = build_tile_index(target_tile, AUDIT_RULES) all_target_geoms: dict[str, list[tuple[str, dict[str, Any]]]] = defaultdict(list) for rule in AUDIT_RULES: for feat in target_index[rule["delivery_layer"]]: all_target_geoms[geometry_signature(feat)].append((rule["delivery_layer"], feat)) for rule in AUDIT_RULES: raw_layer = rule["raw_layer"] delivery_layer = rule["delivery_layer"] raw_match_field = rule["raw_match_field"] delivery_match_field = rule["delivery_match_field"] key = f"{raw_layer}->{delivery_layer}" target_by_geom: dict[str, list[dict[str, Any]]] = defaultdict(list) for feat in target_index[delivery_layer]: target_by_geom[geometry_signature(feat)].append(feat) for raw_feat in raw_index[raw_layer]: totals["raw_objects"] += 1 rule_stats[key]["raw_objects"] += 1 raw_props = raw_feat.get("properties", {}) raw_match_value = raw_props.get(raw_match_field) class_key = f"{raw_layer}:{raw_match_value}" geom_sig = geometry_signature(raw_feat) name = raw_props.get("名称") or parse_at_name(raw_props.get("at")) candidates = target_by_geom.get(geom_sig, []) matched_class = False if not candidates: other_layer_hits = all_target_geoms.get(geom_sig, []) issue = "missing_object" if other_layer_hits: issue = "wrong_relayer" totals[issue] += 1 rule_stats[key][issue] += 1 class_stats[class_key][issue] += 1 if len(issue_examples[issue]) < MAX_EXAMPLES: issue_examples[issue].append( { "tile": [tile.z, tile.x, tile.y], "fid": raw_props.get("fid"), "name": name, "raw_layer": raw_layer, "delivery_layer": delivery_layer, "raw_match_field": raw_match_field, "raw_match_value": raw_match_value, "other_layer_hits": [ { "delivery_layer": layer_name, "delivery_match_field": delivery_match_field, "delivery_match_value": feat.get("properties", {}).get(delivery_match_field), "canonical_object_type": feat.get("properties", {}).get("canonical_object_type"), } for layer_name, feat in other_layer_hits[:5] ], } ) continue for candidate in candidates: target_props = candidate.get("properties", {}) if target_props.get(delivery_match_field) == raw_match_value: matched_class = True break if matched_class: totals["preserved"] += 1 rule_stats[key]["preserved"] += 1 class_stats[class_key]["preserved"] += 1 continue if all(candidate.get("properties", {}).get(delivery_match_field) is None for candidate in candidates): issue = "identifier_lost" else: issue = "identifier_mismatch" totals[issue] += 1 rule_stats[key][issue] += 1 class_stats[class_key][issue] += 1 if len(issue_examples[issue]) < MAX_EXAMPLES: issue_examples[issue].append( { "tile": [tile.z, tile.x, tile.y], "fid": raw_props.get("fid"), "name": name, "raw_layer": raw_layer, "delivery_layer": delivery_layer, "raw_match_field": raw_match_field, "raw_match_value": raw_match_value, "candidates": [ { "delivery_match_field": delivery_match_field, "delivery_match_value": candidate.get("properties", {}).get(delivery_match_field), "canonical_object_type": candidate.get("properties", {}).get("canonical_object_type"), "chart_symbol_code": candidate.get("properties", {}).get("chart_symbol_code"), "chart_icon_image": candidate.get("properties", {}).get("chart_icon_image"), } for candidate in candidates[:5] ], } ) top_class_issues = [] for class_key, stats in class_stats.items(): top_class_issues.append( { "class_key": class_key, "raw_objects": sum(stats.values()), "preserved": stats.get("preserved", 0), "missing_object": stats.get("missing_object", 0), "wrong_relayer": stats.get("wrong_relayer", 0), "identifier_lost": stats.get("identifier_lost", 0), "identifier_mismatch": stats.get("identifier_mismatch", 0), } ) top_class_issues.sort( key=lambda item: ( item["missing_object"] + item["wrong_relayer"] + item["identifier_lost"] + item["identifier_mismatch"], item["raw_objects"], ), reverse=True, ) payload = { "target_name": target_name, "raw_root": str(raw_root), "target_root": str(target_root), "totals": dict(totals), "rule_stats": {key: dict(stats) for key, stats in sorted(rule_stats.items())}, "top_class_issues": top_class_issues[:20], "issue_examples": issue_examples, } report_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") lines = [ f"# Object Preservation Audit: {target_name}", "", f"- raw root: `{raw_root}`", f"- target root: `{target_root}`", "", "## Totals", "", f"- tiles compared: `{totals.get('tiles_compared', 0)}`", f"- raw objects: `{totals.get('raw_objects', 0)}`", f"- preserved: `{totals.get('preserved', 0)}`", f"- missing object: `{totals.get('missing_object', 0)}`", f"- wrong relayer: `{totals.get('wrong_relayer', 0)}`", f"- identifier lost: `{totals.get('identifier_lost', 0)}`", f"- identifier mismatch: `{totals.get('identifier_mismatch', 0)}`", "", "## Per Rule", "", ] for key, stats in sorted(rule_stats.items()): lines.extend( [ f"### `{key}`", "", f"- raw objects: `{stats.get('raw_objects', 0)}`", f"- preserved: `{stats.get('preserved', 0)}`", f"- missing object: `{stats.get('missing_object', 0)}`", f"- wrong relayer: `{stats.get('wrong_relayer', 0)}`", f"- identifier lost: `{stats.get('identifier_lost', 0)}`", f"- identifier mismatch: `{stats.get('identifier_mismatch', 0)}`", "", ] ) lines.extend(["## Top Class Issues", ""]) for item in top_class_issues[:12]: lines.append( "- `{class_key}` raw=`{raw_objects}` preserved=`{preserved}` missing=`{missing_object}` " "wrong_relayer=`{wrong_relayer}` identifier_lost=`{identifier_lost}` identifier_mismatch=`{identifier_mismatch}`".format( **item ) ) lines.append("") for issue_name, examples in issue_examples.items(): if not examples: continue lines.extend([f"## Examples: `{issue_name}`", ""]) for example in examples: lines.append(f"- {canonical_json(example)}") lines.append("") report_md.write_text("\n".join(lines), encoding="utf-8") return report_md, report_json def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Audit raw-object preservation for hazards, navigation marks, facility points, and route outlines between original and delivery/final PBFs." ) parser.add_argument("--raw-root", type=Path, default=DEFAULT_RAW_ROOT) parser.add_argument("--delivery-root", type=Path, default=DEFAULT_DELIVERY_ROOT) parser.add_argument("--final-root", type=Path, default=DEFAULT_FINAL_ROOT) parser.add_argument("--report-root", type=Path, default=DEFAULT_REPORT_ROOT) return parser.parse_args() def main() -> None: args = parse_args() targets = ( ("delivery_20nm", args.delivery_root), ("final_20nm", args.final_root), ) for target_name, target_root in targets: report_md, report_json = audit_one_target( target_name=target_name, raw_root=args.raw_root, target_root=target_root, report_root=args.report_root, ) print(f"wrote {report_md}") print(f"wrote {report_json}") if __name__ == "__main__": main()