291 lines
10 KiB
Python
291 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""生成当前 NavSea PBF 的可取舍数据字典和分类统计。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
from collections import Counter, defaultdict
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import mapbox_vector_tile
|
|
|
|
|
|
TRACKED_FIELDS = (
|
|
"class_code",
|
|
"display_code",
|
|
"shape_class_code",
|
|
"canonical_object_type",
|
|
"object_family",
|
|
"detection_class",
|
|
"chart_symbol_code",
|
|
"chart_label_position_code",
|
|
"chart_label_subtext",
|
|
"chart_icon_image",
|
|
"chart_fill_pattern",
|
|
"chart_fill_style",
|
|
"chart_text_color",
|
|
"chart_text_style",
|
|
"chart_line_color",
|
|
"chart_line_width",
|
|
"chart_line_pattern",
|
|
"arc_id",
|
|
"arc_source_code",
|
|
"light_color_code",
|
|
"light_sector_mode",
|
|
"icon_id",
|
|
"source_layer",
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--pbf-root",
|
|
type=Path,
|
|
default=Path("/home/wwwroot/pbf-delivery-full-semantic-unified-20260808"),
|
|
)
|
|
parser.add_argument(
|
|
"--style",
|
|
type=Path,
|
|
default=Path("src/pbf/style.navsea-delivery-full-semantic-unified-20260808.json"),
|
|
)
|
|
parser.add_argument(
|
|
"--report-root",
|
|
type=Path,
|
|
default=Path("report/current_pbf_inventory_full_semantic_20260809"),
|
|
)
|
|
parser.add_argument("--workers", type=int, default=4)
|
|
return parser.parse_args()
|
|
|
|
|
|
def empty_result() -> dict[str, Any]:
|
|
return {
|
|
"tiles": Counter(),
|
|
"layers": Counter(),
|
|
"geometries": Counter(),
|
|
"property_keys": Counter(),
|
|
"property_key_layers": Counter(),
|
|
"values": Counter(),
|
|
"value_layers": Counter(),
|
|
}
|
|
|
|
|
|
def scan_chunk(paths: list[str]) -> dict[str, Any]:
|
|
result = empty_result()
|
|
for raw_path in paths:
|
|
path = Path(raw_path)
|
|
rel = path.relative_to(path.parents[2])
|
|
zoom = rel.parts[0]
|
|
result["tiles"][f"z{zoom}"] += 1
|
|
decoded = mapbox_vector_tile.decode(path.read_bytes())
|
|
for layer_name, layer in decoded.items():
|
|
features = layer.get("features", [])
|
|
result["layers"][layer_name] += len(features)
|
|
for feature in features:
|
|
geometry_type = (feature.get("geometry") or {}).get("type", "<none>")
|
|
result["geometries"][(layer_name, geometry_type)] += 1
|
|
properties = feature.get("properties") or {}
|
|
for key, value in properties.items():
|
|
result["property_keys"][key] += 1
|
|
result["property_key_layers"][(key, layer_name)] += 1
|
|
if key not in TRACKED_FIELDS or value is None:
|
|
continue
|
|
value_text = str(value)
|
|
result["values"][(key, value_text)] += 1
|
|
result["value_layers"][(key, value_text, layer_name)] += 1
|
|
return result
|
|
|
|
|
|
def merge(results: list[dict[str, Any]]) -> dict[str, Counter]:
|
|
merged = empty_result()
|
|
for result in results:
|
|
for key in merged:
|
|
merged[key].update(result[key])
|
|
return merged
|
|
|
|
|
|
def style_get_fields(style: dict[str, Any]) -> set[str]:
|
|
fields: set[str] = set()
|
|
|
|
def walk(value: Any) -> None:
|
|
if isinstance(value, dict):
|
|
for child in value.values():
|
|
walk(child)
|
|
elif isinstance(value, list):
|
|
if (
|
|
len(value) >= 2
|
|
and isinstance(value[0], str)
|
|
and value[0] in {"get", "has"}
|
|
and isinstance(value[1], str)
|
|
):
|
|
fields.add(value[1])
|
|
for child in value:
|
|
walk(child)
|
|
|
|
walk(style)
|
|
return fields
|
|
|
|
|
|
def write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, Any]]) -> None:
|
|
with path.open("w", encoding="utf-8-sig", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
paths = sorted(args.pbf_root.glob("*/*/*.pbf"))
|
|
if not paths:
|
|
raise SystemExit(f"no PBF tiles under {args.pbf_root}")
|
|
chunk_size = 128
|
|
chunks = [
|
|
[str(path) for path in paths[start : start + chunk_size]]
|
|
for start in range(0, len(paths), chunk_size)
|
|
]
|
|
with ProcessPoolExecutor(max_workers=args.workers) as executor:
|
|
results = list(executor.map(scan_chunk, chunks))
|
|
stats = merge(results)
|
|
|
|
args.report_root.mkdir(parents=True, exist_ok=True)
|
|
style = json.loads(args.style.read_text(encoding="utf-8"))
|
|
style_fields = style_get_fields(style)
|
|
|
|
layer_rows = []
|
|
for layer, feature_count in stats["layers"].most_common():
|
|
geometries = Counter(
|
|
{
|
|
geometry: count
|
|
for (layer_name, geometry), count in stats["geometries"].items()
|
|
if layer_name == layer
|
|
}
|
|
)
|
|
fields = sorted(
|
|
key
|
|
for (key, layer_name), count in stats["property_key_layers"].items()
|
|
if layer_name == layer and count
|
|
)
|
|
layer_rows.append(
|
|
{
|
|
"layer": layer,
|
|
"feature_count": feature_count,
|
|
"geometry_counts": json.dumps(geometries, ensure_ascii=False),
|
|
"property_count": len(fields),
|
|
"properties": ",".join(fields),
|
|
}
|
|
)
|
|
write_csv(
|
|
args.report_root / "图层分类.csv",
|
|
["layer", "feature_count", "geometry_counts", "property_count", "properties"],
|
|
layer_rows,
|
|
)
|
|
|
|
field_rows = []
|
|
for key, count in stats["property_keys"].most_common():
|
|
layer_counts = Counter(
|
|
{
|
|
layer: layer_count
|
|
for (field, layer), layer_count in stats["property_key_layers"].items()
|
|
if field == key
|
|
}
|
|
)
|
|
field_rows.append(
|
|
{
|
|
"field": key,
|
|
"feature_value_count": count,
|
|
"layer_count": len(layer_counts),
|
|
"layers": json.dumps(layer_counts, ensure_ascii=False),
|
|
"used_by_current_style": "yes" if key in style_fields else "no",
|
|
}
|
|
)
|
|
write_csv(
|
|
args.report_root / "字段分类.csv",
|
|
["field", "feature_value_count", "layer_count", "layers", "used_by_current_style"],
|
|
field_rows,
|
|
)
|
|
|
|
value_rows = []
|
|
for (field, value), count in stats["values"].most_common():
|
|
layer_counts = Counter(
|
|
{
|
|
layer: layer_count
|
|
for (tracked_field, tracked_value, layer), layer_count in stats["value_layers"].items()
|
|
if tracked_field == field and tracked_value == value
|
|
}
|
|
)
|
|
value_rows.append(
|
|
{
|
|
"field": field,
|
|
"value": value,
|
|
"feature_count": count,
|
|
"layers": json.dumps(layer_counts, ensure_ascii=False),
|
|
}
|
|
)
|
|
write_csv(
|
|
args.report_root / "字段值分类.csv",
|
|
["field", "value", "feature_count", "layers"],
|
|
value_rows,
|
|
)
|
|
|
|
total_features = sum(stats["layers"].values())
|
|
summary = {
|
|
"pbf_root": str(args.pbf_root),
|
|
"tile_count": len(paths),
|
|
"tiles_by_zoom": dict(sorted(stats["tiles"].items())),
|
|
"feature_count": total_features,
|
|
"layer_count": len(stats["layers"]),
|
|
"property_count": len(stats["property_keys"]),
|
|
"style_get_fields": sorted(style_fields),
|
|
}
|
|
(args.report_root / "summary.json").write_text(
|
|
json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
|
)
|
|
|
|
md: list[str] = [
|
|
"# 当前统一 Full PBF 数据分类报告",
|
|
"",
|
|
f"- PBF 根目录：`{args.pbf_root}`",
|
|
f"- 瓦片数：`{len(paths)}`",
|
|
f"- feature 总数：`{total_features}`",
|
|
f"- 图层数：`{len(stats['layers'])}`",
|
|
f"- 属性字段数：`{len(stats['property_keys'])}`",
|
|
"",
|
|
"## 按 zoom 的瓦片数量",
|
|
"",
|
|
"| zoom | 瓦片数 |",
|
|
"|---|---:|",
|
|
]
|
|
md.extend(f"| {zoom} | {count} |" for zoom, count in sorted(stats["tiles"].items()))
|
|
md.extend(["", "## 图层分类", "", "| 图层 | feature 数 | 几何类型 | 字段数 |", "|---|---:|---|---:|"])
|
|
for row in layer_rows:
|
|
md.append(
|
|
f"| `{row['layer']}` | {row['feature_count']} | `{row['geometry_counts']}` | {row['property_count']} |"
|
|
)
|
|
md.extend(
|
|
[
|
|
"",
|
|
"## 取舍说明",
|
|
"",
|
|
"- `geometry`、图层名、`canonical_object_type`、分类/显示码、`chart_*`、`arc_id`/`arc_source_code` 是渲染、对象审计或灯弧追溯的核心数据，不能在未做专项审计前删除。",
|
|
"- 只被点击详情或外部业务使用、但不被当前 style 读取的字段，属于可选保留项；应先确认 pickup、审计和后续重建是否依赖。",
|
|
"- 仅存在于 PBF、当前 style 不读取且不参与对象身份/分类/追溯的字段，才是候选删除项；具体字段和数量见 `字段分类.csv`。",
|
|
"- 所有细分值见 `字段值分类.csv`，不要仅依据当前出现次数为 0 的 style 分支删除上游分类能力。",
|
|
"",
|
|
"## 明细文件",
|
|
"",
|
|
"- `图层分类.csv`：每个 source-layer 的 feature、几何和字段分布。",
|
|
"- `字段分类.csv`：每个属性字段的出现量、所在图层和当前 style 是否读取。",
|
|
"- `字段值分类.csv`：分类码、显示码、对象类型、图片、纹理和灯弧等字段的逐值统计。",
|
|
]
|
|
)
|
|
(args.report_root / "当前PBF数据分类报告.md").write_text("\n".join(md) + "\n", encoding="utf-8")
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|