164 lines
6.7 KiB
Python
164 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
||
"""审计九州语义图标测试版是否丢失对象、几何或非图标属性。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
from collections import Counter
|
||
from concurrent.futures import ProcessPoolExecutor
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import mapbox_vector_tile
|
||
|
||
|
||
EXPECTED_FIELDS = {"chart_icon_image", "icon_id", "arc_id"}
|
||
|
||
|
||
def stable(value: Any) -> str:
|
||
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||
|
||
|
||
def normalize_expected_properties(layer_name: str, properties: dict[str, Any]) -> None:
|
||
if (
|
||
layer_name == "anchor_caution_hazard_area"
|
||
and str(properties.get("class_code")) == "420"
|
||
and properties.get("chart_fill_pattern") == "fill-daytime-405"
|
||
):
|
||
properties["chart_fill_pattern"] = "fill-daytime-420"
|
||
if (
|
||
layer_name == "navigation_hazard_area"
|
||
and str(properties.get("class_code")) == "427"
|
||
and properties.get("chart_fill_pattern") == "fill-daytime-405"
|
||
):
|
||
properties["chart_fill_pattern"] = "fill-daytime-427"
|
||
|
||
|
||
def feature_signature(layer_name: str, feature: dict[str, Any]) -> str:
|
||
properties = dict(feature.get("properties") or {})
|
||
for field in EXPECTED_FIELDS:
|
||
properties.pop(field, None)
|
||
normalize_expected_properties(layer_name, properties)
|
||
return stable({
|
||
"id": feature.get("id"),
|
||
"geometry": feature.get("geometry"),
|
||
"properties": properties,
|
||
})
|
||
|
||
|
||
def audit_tile(args: tuple[str, str]) -> dict[str, Any]:
|
||
source_path, target_path = args
|
||
source = mapbox_vector_tile.decode(Path(source_path).read_bytes())
|
||
target = mapbox_vector_tile.decode(Path(target_path).read_bytes())
|
||
errors: list[dict[str, Any]] = []
|
||
source_layers = set(source)
|
||
target_layers = set(target)
|
||
if source_layers != target_layers:
|
||
errors.append({"kind": "layer_set", "source": sorted(source_layers), "target": sorted(target_layers)})
|
||
|
||
source_counts = Counter()
|
||
target_counts = Counter()
|
||
icon_counts = Counter()
|
||
old_prefix_counts = Counter()
|
||
for layer_name in sorted(source_layers | target_layers):
|
||
source_features = source.get(layer_name, {}).get("features", [])
|
||
target_features = target.get(layer_name, {}).get("features", [])
|
||
source_counts[layer_name] = len(source_features)
|
||
target_counts[layer_name] = len(target_features)
|
||
source_signatures = Counter(feature_signature(layer_name, feature) for feature in source_features)
|
||
target_signatures = Counter(feature_signature(layer_name, feature) for feature in target_features)
|
||
if source_signatures != target_signatures:
|
||
missing = source_signatures - target_signatures
|
||
extra = target_signatures - source_signatures
|
||
errors.append({
|
||
"kind": "feature_or_geometry_or_property",
|
||
"layer": layer_name,
|
||
"source_count": len(source_features),
|
||
"target_count": len(target_features),
|
||
"missing_count": sum(missing.values()),
|
||
"extra_count": sum(extra.values()),
|
||
})
|
||
for feature in target_features:
|
||
properties = feature.get("properties") or {}
|
||
for field in ("icon_id", "arc_id"):
|
||
value = properties.get(field)
|
||
if isinstance(value, str):
|
||
icon_counts[f"{field}:{value}"] += 1
|
||
for value in properties.values():
|
||
if isinstance(value, str) and (value.startswith("symbol-daytime-") or value.startswith("arc-daytime-")):
|
||
old_prefix_counts[value] += 1
|
||
return {
|
||
"tile": source_path,
|
||
"errors": errors,
|
||
"source_counts": dict(source_counts),
|
||
"target_counts": dict(target_counts),
|
||
"icon_counts": dict(icon_counts),
|
||
"old_prefix_counts": dict(old_prefix_counts),
|
||
}
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--source-root", type=Path, required=True)
|
||
parser.add_argument("--target-root", type=Path, required=True)
|
||
parser.add_argument("--report-dir", type=Path, required=True)
|
||
args = parser.parse_args()
|
||
pairs = []
|
||
for target in sorted(args.target_root.glob("*/*/*.pbf")):
|
||
source = args.source_root / target.relative_to(args.target_root)
|
||
if source.exists():
|
||
pairs.append((str(source), str(target)))
|
||
|
||
results: list[dict[str, Any]] = []
|
||
with ProcessPoolExecutor() as executor:
|
||
for result in executor.map(audit_tile, pairs, chunksize=16):
|
||
results.append(result)
|
||
|
||
error_rows = [error | {"tile": result["tile"]} for result in results for error in result["errors"]]
|
||
icon_counts = Counter()
|
||
old_prefix_counts = Counter()
|
||
source_features = Counter()
|
||
target_features = Counter()
|
||
for result in results:
|
||
icon_counts.update(result["icon_counts"])
|
||
old_prefix_counts.update(result["old_prefix_counts"])
|
||
source_features.update(result["source_counts"])
|
||
target_features.update(result["target_counts"])
|
||
|
||
payload = {
|
||
"source_root": str(args.source_root),
|
||
"target_root": str(args.target_root),
|
||
"tiles_target": len(list(args.target_root.glob("*/*/*.pbf"))),
|
||
"tiles_compared": len(results),
|
||
"tiles_with_errors": len({row["tile"] for row in error_rows}),
|
||
"errors": error_rows[:200],
|
||
"source_features": dict(source_features),
|
||
"target_features": dict(target_features),
|
||
"icon_counts": dict(icon_counts),
|
||
"old_prefix_counts": dict(old_prefix_counts),
|
||
"verdict": "PASS" if not error_rows and not old_prefix_counts else "FAIL",
|
||
}
|
||
args.report_dir.mkdir(parents=True, exist_ok=True)
|
||
(args.report_dir / "object_geometry_icon_audit.json").write_text(
|
||
json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
||
)
|
||
lines = [
|
||
"# 九州语义图标测试版对象 / 几何 / 图标审计",
|
||
"",
|
||
f"- 来源：`{args.source_root}`",
|
||
f"- 目标：`{args.target_root}`",
|
||
f"- 目标瓦片：`{payload['tiles_target']}`，实际比较：`{payload['tiles_compared']}`",
|
||
f"- 有错误瓦片：`{payload['tiles_with_errors']}`",
|
||
f"- 旧图标前缀残留：`{sum(old_prefix_counts.values())}`",
|
||
f"- 结论：`{payload['verdict']}`",
|
||
"",
|
||
"本审计忽略预期的 `chart_icon_image`、`icon_id`、`arc_id` 字段变化，逐图层比较对象数量、ID、几何和其余属性。",
|
||
]
|
||
(args.report_dir / "object_geometry_icon_audit.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|