355 lines
14 KiB
Python
355 lines
14 KiB
Python
#!/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()
|