audit full semantic extent and AOI visuals
This commit is contained in:
OpenAI Codex
368
navsea_geometry_native_audit.py
Normal file
368
navsea_geometry_native_audit.py
Normal file
@@ -0,0 +1,368 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from collections import Counter, defaultdict
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from mapbox_vector_tile.Mapbox import vector_tile_pb2
|
||||
|
||||
|
||||
DEFAULT_SOURCE_ROOT = Path(
|
||||
"/home/wwwroot/newpec/exported_auto/"
|
||||
"tile.mapple-on.jp__newpec-mvt-20260106__z___x___y_.pbf/tiles"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class LayerGeometryStats:
|
||||
point_count: int = 0
|
||||
min_x: int = 0
|
||||
max_x: int = 0
|
||||
min_y: int = 0
|
||||
max_y: int = 0
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="审计并按需修复 NavSea PBF 的几何 / extent / native 风险。")
|
||||
parser.add_argument("--target-root", type=Path, required=True)
|
||||
parser.add_argument("--label", required=True)
|
||||
parser.add_argument("--output-dir", type=Path, required=True)
|
||||
parser.add_argument("--source-root", type=Path, default=DEFAULT_SOURCE_ROOT)
|
||||
parser.add_argument("--apply-fix", action="store_true", help="若检测到 extent 与源瓦片不一致，则按源瓦片 extent 原地修复。")
|
||||
parser.add_argument("--sample-limit", type=int, default=30)
|
||||
parser.add_argument("--zoom", action="append", type=int, help="只审指定 zoom，可重复传入多次。")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def zigzag_decode(value: int) -> int:
|
||||
return (value >> 1) ^ (-(value & 1))
|
||||
|
||||
|
||||
def decode_feature_geometry(geometry: list[int]) -> LayerGeometryStats:
|
||||
stats = LayerGeometryStats()
|
||||
cursor_x = 0
|
||||
cursor_y = 0
|
||||
initialized = False
|
||||
idx = 0
|
||||
|
||||
while idx < len(geometry):
|
||||
command = geometry[idx]
|
||||
idx += 1
|
||||
command_id = command & 0x7
|
||||
command_count = command >> 3
|
||||
|
||||
if command_id in (1, 2): # MoveTo, LineTo
|
||||
for _ in range(command_count):
|
||||
if idx + 1 >= len(geometry):
|
||||
raise ValueError("geometry command truncated")
|
||||
cursor_x += zigzag_decode(geometry[idx])
|
||||
cursor_y += zigzag_decode(geometry[idx + 1])
|
||||
idx += 2
|
||||
|
||||
if not initialized:
|
||||
stats.min_x = stats.max_x = cursor_x
|
||||
stats.min_y = stats.max_y = cursor_y
|
||||
initialized = True
|
||||
else:
|
||||
stats.min_x = min(stats.min_x, cursor_x)
|
||||
stats.max_x = max(stats.max_x, cursor_x)
|
||||
stats.min_y = min(stats.min_y, cursor_y)
|
||||
stats.max_y = max(stats.max_y, cursor_y)
|
||||
stats.point_count += 1
|
||||
elif command_id == 7: # ClosePath
|
||||
continue
|
||||
else:
|
||||
raise ValueError(f"unsupported geometry command id={command_id}")
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
def merge_layer_stats(existing: LayerGeometryStats | None, incoming: LayerGeometryStats) -> LayerGeometryStats:
|
||||
if existing is None or existing.point_count == 0:
|
||||
return incoming
|
||||
if incoming.point_count == 0:
|
||||
return existing
|
||||
return LayerGeometryStats(
|
||||
point_count=existing.point_count + incoming.point_count,
|
||||
min_x=min(existing.min_x, incoming.min_x),
|
||||
max_x=max(existing.max_x, incoming.max_x),
|
||||
min_y=min(existing.min_y, incoming.min_y),
|
||||
max_y=max(existing.max_y, incoming.max_y),
|
||||
)
|
||||
|
||||
|
||||
def analyze_tile(tile_path: Path) -> dict[str, Any]:
|
||||
tile = vector_tile_pb2.tile()
|
||||
tile.ParseFromString(tile_path.read_bytes())
|
||||
|
||||
layer_summaries: list[dict[str, Any]] = []
|
||||
extent_values: set[int] = set()
|
||||
tile_outside_extent = False
|
||||
tile_max_overflow = 0
|
||||
|
||||
for layer in tile.layers:
|
||||
extent = int(layer.extent or 4096)
|
||||
extent_values.add(extent)
|
||||
aggregate: LayerGeometryStats | None = None
|
||||
for feature in layer.features:
|
||||
geom_stats = decode_feature_geometry(list(feature.geometry))
|
||||
aggregate = merge_layer_stats(aggregate, geom_stats)
|
||||
|
||||
if aggregate is None:
|
||||
aggregate = LayerGeometryStats()
|
||||
|
||||
outside_extent = False
|
||||
overflow = 0
|
||||
if aggregate.point_count:
|
||||
overflow = max(
|
||||
0,
|
||||
-aggregate.min_x,
|
||||
-aggregate.min_y,
|
||||
aggregate.max_x - extent,
|
||||
aggregate.max_y - extent,
|
||||
)
|
||||
outside_extent = overflow > 0
|
||||
|
||||
tile_outside_extent = tile_outside_extent or outside_extent
|
||||
tile_max_overflow = max(tile_max_overflow, overflow)
|
||||
layer_summaries.append(
|
||||
{
|
||||
"name": layer.name,
|
||||
"extent": extent,
|
||||
"point_count": aggregate.point_count,
|
||||
"min_x": aggregate.min_x,
|
||||
"max_x": aggregate.max_x,
|
||||
"min_y": aggregate.min_y,
|
||||
"max_y": aggregate.max_y,
|
||||
"outside_extent": outside_extent,
|
||||
"overflow": overflow,
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"extent_values": sorted(extent_values),
|
||||
"outside_extent": tile_outside_extent,
|
||||
"max_overflow": tile_max_overflow,
|
||||
"layers": layer_summaries,
|
||||
}
|
||||
|
||||
|
||||
def desired_extent_from_source(source_path: Path) -> int | None:
|
||||
if not source_path.exists():
|
||||
return None
|
||||
tile = vector_tile_pb2.tile()
|
||||
tile.ParseFromString(source_path.read_bytes())
|
||||
extents = sorted({int(layer.extent or 4096) for layer in tile.layers})
|
||||
if not extents:
|
||||
return None
|
||||
if len(extents) == 1:
|
||||
return extents[0]
|
||||
return max(extents)
|
||||
|
||||
|
||||
def repair_tile_extents(tile_path: Path, desired_extent: int) -> bool:
|
||||
tile = vector_tile_pb2.tile()
|
||||
raw = tile_path.read_bytes()
|
||||
tile.ParseFromString(raw)
|
||||
changed = False
|
||||
for layer in tile.layers:
|
||||
current = int(layer.extent or 4096)
|
||||
if current != desired_extent:
|
||||
layer.extent = desired_extent
|
||||
changed = True
|
||||
if changed:
|
||||
tile_path.write_bytes(tile.SerializeToString())
|
||||
return changed
|
||||
|
||||
|
||||
def hardlink_copy_tree(src_root: Path, dst_root: Path) -> None:
|
||||
if dst_root.exists():
|
||||
raise SystemExit(f"backup root already exists: {dst_root}")
|
||||
dst_root.mkdir(parents=True, exist_ok=True)
|
||||
os.system(f"cp -al {src_root}/. {dst_root}/")
|
||||
|
||||
|
||||
def write_report(
|
||||
output_dir: Path,
|
||||
label: str,
|
||||
target_root: Path,
|
||||
source_root: Path,
|
||||
summary: dict[str, Any],
|
||||
) -> tuple[Path, Path]:
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
report_json = output_dir / "geometry_native_audit.json"
|
||||
report_md = output_dir / "geometry_native_audit.md"
|
||||
report_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
lines = [
|
||||
f"# {label} 几何 / Extent / Native 风险审计",
|
||||
"",
|
||||
"## 范围",
|
||||
"",
|
||||
f"- target root: `{target_root}`",
|
||||
f"- source root: `{source_root}`",
|
||||
f"- tiles: `{summary['tile_count']}`",
|
||||
f"- repaired tiles: `{summary['repaired_tiles']}`",
|
||||
"",
|
||||
"## 总体结论",
|
||||
"",
|
||||
f"- 坐标超当前 extent 的瓦片数：`{summary['outside_extent_tiles']}`",
|
||||
f"- 与源瓦片 extent 不一致的瓦片数：`{summary['source_extent_mismatch_tiles']}`",
|
||||
f"- 可直接按源 extent 修复的瓦片数：`{summary['fixable_from_source_tiles']}`",
|
||||
"",
|
||||
"## Zoom 汇总",
|
||||
"",
|
||||
]
|
||||
|
||||
for zoom in sorted(summary["zoom_summary"], key=int):
|
||||
item = summary["zoom_summary"][zoom]
|
||||
lines.extend(
|
||||
[
|
||||
f"### z{zoom}",
|
||||
"",
|
||||
f"- tiles: `{item['tile_count']}`",
|
||||
f"- outside extent: `{item['outside_extent_tiles']}`",
|
||||
f"- source mismatch: `{item['source_extent_mismatch_tiles']}`",
|
||||
f"- extent counts: `{item['extent_counts']}`",
|
||||
"",
|
||||
]
|
||||
)
|
||||
|
||||
lines.extend(["## 代表问题样本", ""])
|
||||
for item in summary["sample_tiles"]:
|
||||
lines.extend(
|
||||
[
|
||||
f"### `{item['tile']}`",
|
||||
"",
|
||||
f"- extent values: `{item['extent_values']}`",
|
||||
f"- source extent: `{item['source_extent']}`",
|
||||
f"- outside extent: `{item['outside_extent']}`",
|
||||
f"- max overflow: `{item['max_overflow']}`",
|
||||
f"- fixable from source: `{item['fixable_from_source']}`",
|
||||
"",
|
||||
]
|
||||
)
|
||||
|
||||
report_md.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
return report_md, report_json
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
target_root = args.target_root
|
||||
source_root = args.source_root
|
||||
output_dir = args.output_dir
|
||||
|
||||
if not target_root.exists():
|
||||
raise SystemExit(f"target root not found: {target_root}")
|
||||
if not source_root.exists():
|
||||
raise SystemExit(f"source root not found: {source_root}")
|
||||
|
||||
zoom_summary: dict[str, dict[str, Any]] = defaultdict(
|
||||
lambda: {
|
||||
"tile_count": 0,
|
||||
"outside_extent_tiles": 0,
|
||||
"source_extent_mismatch_tiles": 0,
|
||||
"extent_counts": Counter(),
|
||||
}
|
||||
)
|
||||
sample_tiles: list[dict[str, Any]] = []
|
||||
repaired_tiles = 0
|
||||
outside_extent_tiles = 0
|
||||
source_extent_mismatch_tiles = 0
|
||||
fixable_from_source_tiles = 0
|
||||
|
||||
tile_paths = sorted(target_root.glob("*/*/*.pbf"))
|
||||
if args.zoom:
|
||||
zoom_filter = {str(item) for item in args.zoom}
|
||||
tile_paths = [path for path in tile_paths if path.parent.parent.name in zoom_filter]
|
||||
for tile_path in tile_paths:
|
||||
rel = tile_path.relative_to(target_root)
|
||||
zoom = rel.parts[0]
|
||||
analysis = analyze_tile(tile_path)
|
||||
source_extent = desired_extent_from_source(source_root / rel)
|
||||
current_extent = max(analysis["extent_values"]) if analysis["extent_values"] else None
|
||||
mismatch = source_extent is not None and current_extent is not None and source_extent != current_extent
|
||||
fixable = bool(
|
||||
mismatch
|
||||
and source_extent is not None
|
||||
and analysis["outside_extent"]
|
||||
and analysis["max_overflow"] <= max(0, source_extent - (current_extent or 0))
|
||||
)
|
||||
|
||||
zoom_item = zoom_summary[zoom]
|
||||
zoom_item["tile_count"] += 1
|
||||
for extent in analysis["extent_values"]:
|
||||
zoom_item["extent_counts"][extent] += 1
|
||||
if analysis["outside_extent"]:
|
||||
outside_extent_tiles += 1
|
||||
zoom_item["outside_extent_tiles"] += 1
|
||||
if mismatch:
|
||||
source_extent_mismatch_tiles += 1
|
||||
zoom_item["source_extent_mismatch_tiles"] += 1
|
||||
if fixable:
|
||||
fixable_from_source_tiles += 1
|
||||
|
||||
if args.apply_fix and fixable and source_extent is not None:
|
||||
if repair_tile_extents(tile_path, source_extent):
|
||||
repaired_tiles += 1
|
||||
analysis = analyze_tile(tile_path)
|
||||
|
||||
if len(sample_tiles) < args.sample_limit and (analysis["outside_extent"] or mismatch):
|
||||
sample_tiles.append(
|
||||
{
|
||||
"tile": str(rel),
|
||||
"extent_values": analysis["extent_values"],
|
||||
"source_extent": source_extent,
|
||||
"outside_extent": analysis["outside_extent"],
|
||||
"max_overflow": analysis["max_overflow"],
|
||||
"fixable_from_source": fixable,
|
||||
}
|
||||
)
|
||||
|
||||
normalized_zoom_summary = {
|
||||
zoom: {
|
||||
"tile_count": item["tile_count"],
|
||||
"outside_extent_tiles": item["outside_extent_tiles"],
|
||||
"source_extent_mismatch_tiles": item["source_extent_mismatch_tiles"],
|
||||
"extent_counts": dict(sorted(item["extent_counts"].items())),
|
||||
}
|
||||
for zoom, item in sorted(zoom_summary.items(), key=lambda kv: int(kv[0]))
|
||||
}
|
||||
|
||||
summary = {
|
||||
"label": args.label,
|
||||
"target_root": str(target_root),
|
||||
"source_root": str(source_root),
|
||||
"tile_count": len(tile_paths),
|
||||
"outside_extent_tiles": outside_extent_tiles,
|
||||
"source_extent_mismatch_tiles": source_extent_mismatch_tiles,
|
||||
"fixable_from_source_tiles": fixable_from_source_tiles,
|
||||
"repaired_tiles": repaired_tiles,
|
||||
"zoom_summary": normalized_zoom_summary,
|
||||
"sample_tiles": sample_tiles,
|
||||
}
|
||||
report_md, report_json = write_report(output_dir, args.label, target_root, source_root, summary)
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"report_md": str(report_md),
|
||||
"report_json": str(report_json),
|
||||
"outside_extent_tiles": outside_extent_tiles,
|
||||
"source_extent_mismatch_tiles": source_extent_mismatch_tiles,
|
||||
"fixable_from_source_tiles": fixable_from_source_tiles,
|
||||
"repaired_tiles": repaired_tiles,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user