109 lines
5.1 KiB
Python
109 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""只读汇总全国实际 PBF 使用情况，复用原 S-52 compiler 编译轻量对比样式。"""
|
||
from __future__ import annotations
|
||
import importlib.util
|
||
import argparse
|
||
import math
|
||
import os
|
||
import json
|
||
from pathlib import Path
|
||
import sys
|
||
import time
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
RELEASE = "cm95-d1-62a85bd40d3a-ae61b8866e80"
|
||
PBF_DIR = ROOT / "build" / "releases-s52-v2" / RELEASE / "pbf" / "delivery"
|
||
OUT = ROOT / "build" / "s52_style_compiler_japan_compare_v1"
|
||
SPEC = importlib.util.spec_from_file_location("cm95_s52_style_compiler", ROOT / "tools" / "cm95_s52_style_compiler.py")
|
||
assert SPEC and SPEC.loader
|
||
compiler = importlib.util.module_from_spec(SPEC)
|
||
SPEC.loader.exec_module(compiler)
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="全国 Style 按真实 PBF 的代表性样本编译")
|
||
parser.add_argument("--all", action="store_true", help="扫描全部 PBF（资源密集，默认采用分层样本）")
|
||
args = parser.parse_args()
|
||
# 用 scandir 逐级枚举，避免为 177383 个文件创建和排序 Path 对象。
|
||
strides = {9: 4, 10: 20, 11: 100, 12: 400}
|
||
files = []
|
||
by_zoom = {}
|
||
all_count = 0
|
||
with os.scandir(PBF_DIR) as zoom_items:
|
||
zoom_dirs = sorted((d for d in zoom_items if d.is_dir() and d.name.isdigit()), key=lambda d: int(d.name))
|
||
for zoom_dir in zoom_dirs:
|
||
z = int(zoom_dir.name)
|
||
with os.scandir(zoom_dir.path) as x_items:
|
||
x_dirs = sorted((d for d in x_items if d.is_dir() and d.name.isdigit()), key=lambda d: int(d.name))
|
||
for x_dir in x_dirs:
|
||
with os.scandir(x_dir.path) as y_items:
|
||
entries = sorted((d for d in y_items if d.is_file() and d.name.endswith('.pbf')), key=lambda d: int(d.name[:-4]))
|
||
for entry in entries:
|
||
i = by_zoom.get(z, 0)
|
||
by_zoom[z] = i + 1
|
||
all_count += 1
|
||
if args.all or z <= 8 or i % strides.get(z, 1) == 0:
|
||
files.append(Path(entry.path))
|
||
if not all_count:
|
||
raise SystemExit('全国 PBF 目录为空')
|
||
ports = [(140.7288,41.7688),(140.8719,38.2682),(139.77,35.58),
|
||
(136.75,34.73),(135.32,34.63),(132.44,34.30),(130.38,33.59),
|
||
(129.97,33.44),(129.87,32.74),(130.56,31.56),(127.68,26.21),
|
||
(124.16,24.34)]
|
||
seen = set(files)
|
||
for lon,lat in ports:
|
||
for z in (10,11,12):
|
||
n = 1 << z
|
||
x = int((lon+180)/360*n)
|
||
y = int((1-math.asinh(math.tan(math.radians(lat)))/math.pi)/2*n)
|
||
path = PBF_DIR / str(z) / str(x) / (str(y)+".pbf")
|
||
if path.is_file() and path not in seen:
|
||
files.append(path)
|
||
seen.add(path)
|
||
files.sort()
|
||
print("SAMPLE", len(files), "of", all_count, "PBF; all production tiles remain served", flush=True)
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
total = 0
|
||
agg = None
|
||
chunk = []
|
||
begun = time.monotonic()
|
||
for path in files:
|
||
chunk.append(compiler._scan_pbf_usage(path))
|
||
total += 1
|
||
if len(chunk) == 200 or total == len(files):
|
||
one = compiler._merge_pbf_usages(chunk, "national chunk")
|
||
agg = one if agg is None else compiler._merge_pbf_usages([agg, one], "national aggregate")
|
||
chunk.clear()
|
||
if total % 10000 == 0 or total == len(files):
|
||
print(f"USAGE {total}/{len(files)} elapsed={int(time.monotonic()-begun)}s", flush=True)
|
||
assert agg is not None
|
||
agg["pbf_count"] = total
|
||
agg["pbf"] = str(PBF_DIR)
|
||
(OUT / "usage_scope.json").write_text(json.dumps({
|
||
"release_id": RELEASE, "mode": "all" if args.all else "deterministic_stratified_preview",
|
||
"all_pbf_count": all_count, "scanned_pbf_count": total, "strides": strides,
|
||
"ports": ports, "note": "仅控制浏览器 Style 的显示规则集合，不改变或删减全国 PBF 数据"
|
||
}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
usage_path = OUT / "national_pbf_usage.json"
|
||
usage_path.write_text(json.dumps(agg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
# 不修改原编译器；将完整全国聚合 usage 交给原有编译入口。
|
||
compiler._scan_pbf_directory = lambda path: agg
|
||
sys.argv = [
|
||
str(ROOT / "tools" / "cm95_s52_style_compiler.py"),
|
||
"--config", str(ROOT / "config" / "s52" / "cm95_s52_style_compiler_japan_compare_v1.json"),
|
||
"--usage-pbf-dir", str(PBF_DIR),
|
||
"--output", str(OUT / "style.s52.json"),
|
||
"--audit-output", str(OUT / "style.s52.audit.json"),
|
||
]
|
||
result = int(compiler.main())
|
||
style = json.loads((OUT / "style.s52.json").read_text(encoding="utf-8"))
|
||
audit = json.loads((OUT / "style.s52.audit.json").read_text(encoding="utf-8"))
|
||
print("COMPLETE", json.dumps({"pbf_count": total, "layer_count": len(style["layers"]),
|
||
"style_bytes": (OUT / "style.s52.json").stat().st_size,
|
||
"hard_gate_failures": audit.get("hard_gate_failures"),
|
||
"resource_contract_status": audit.get("resource_contract_status")}, ensure_ascii=False), flush=True)
|
||
return result
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|