Add 20nm object preservation audit and final hazard icon fix
This commit is contained in:
OpenAI Codex
329
navsea_object_preservation_audit.py
Normal file
329
navsea_object_preservation_audit.py
Normal file
@@ -0,0 +1,329 @@
|
||||
#!/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")
|
||||
|
||||
HAZARD_LAYER_RULES = (
|
||||
{
|
||||
"raw_layer": "p航行危険障害物",
|
||||
"delivery_layer": "navigation_hazard_point",
|
||||
"raw_class_field": "分類番号",
|
||||
"delivery_class_field": "class_code",
|
||||
"label": "navigation hazards",
|
||||
},
|
||||
{
|
||||
"raw_layer": "p投錨注意障害物",
|
||||
"delivery_layer": "anchor_caution_hazard_point",
|
||||
"raw_class_field": "分類番号",
|
||||
"delivery_class_field": "class_code",
|
||||
"label": "anchor hazards",
|
||||
},
|
||||
)
|
||||
|
||||
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, HAZARD_LAYER_RULES)
|
||||
target_index = build_tile_index(target_tile, HAZARD_LAYER_RULES)
|
||||
|
||||
all_target_geoms: dict[str, list[tuple[str, dict[str, Any]]]] = defaultdict(list)
|
||||
for rule in HAZARD_LAYER_RULES:
|
||||
for feat in target_index[rule["delivery_layer"]]:
|
||||
all_target_geoms[geometry_signature(feat)].append((rule["delivery_layer"], feat))
|
||||
|
||||
for rule in HAZARD_LAYER_RULES:
|
||||
raw_layer = rule["raw_layer"]
|
||||
delivery_layer = rule["delivery_layer"]
|
||||
raw_class_field = rule["raw_class_field"]
|
||||
delivery_class_field = rule["delivery_class_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_class = raw_props.get(raw_class_field)
|
||||
class_key = f"{raw_layer}:{raw_class}"
|
||||
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_class": raw_class,
|
||||
"other_layer_hits": [
|
||||
{
|
||||
"delivery_layer": layer_name,
|
||||
"delivery_class": feat.get("properties", {}).get(delivery_class_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_class_field) == raw_class:
|
||||
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_class_field) is None for candidate in candidates):
|
||||
issue = "raw_class_lost"
|
||||
else:
|
||||
issue = "class_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_class": raw_class,
|
||||
"candidates": [
|
||||
{
|
||||
"delivery_class": candidate.get("properties", {}).get(delivery_class_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),
|
||||
"raw_class_lost": stats.get("raw_class_lost", 0),
|
||||
"class_mismatch": stats.get("class_mismatch", 0),
|
||||
}
|
||||
)
|
||||
top_class_issues.sort(
|
||||
key=lambda item: (
|
||||
item["missing_object"] + item["wrong_relayer"] + item["raw_class_lost"] + item["class_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"- raw class lost: `{totals.get('raw_class_lost', 0)}`",
|
||||
f"- class mismatch: `{totals.get('class_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"- raw class lost: `{stats.get('raw_class_lost', 0)}`",
|
||||
f"- class mismatch: `{stats.get('class_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}` class_lost=`{raw_class_lost}` class_mismatch=`{class_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 hazard layers 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()
|
||||
Reference in New Issue
Block a user