57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
from pathlib import Path
|
|
|
|
from navsea_tile_reencode_guard import decode_tile_with_extents, encode_layers_preserving_extents
|
|
|
|
|
|
DELIVERY_ROOT = Path("/home/wwwroot/pbf-delivery-full-semantic-unified-20260808")
|
|
NAVMARK_ROOT = Path("/home/wwwroot/pbf-navmarks-semantic-unified-20260808")
|
|
|
|
|
|
def merge_one(nav_path_str: str) -> int:
|
|
nav_path = Path(nav_path_str)
|
|
rel = nav_path.relative_to(NAVMARK_ROOT)
|
|
target = DELIVERY_ROOT / rel
|
|
if not target.exists():
|
|
return 0
|
|
base, extents = decode_tile_with_extents(target.read_bytes())
|
|
nav, nav_extents = decode_tile_with_extents(nav_path.read_bytes())
|
|
nav_layer = nav.get("navigation_marks")
|
|
if nav_layer is None:
|
|
return 0
|
|
nav_layer = {"name": "navigation_marks", "features": nav_layer.get("features", [])}
|
|
merged = []
|
|
replaced = False
|
|
for layer_name, layer in base.items():
|
|
if layer_name == "navigation_marks":
|
|
merged.append(nav_layer)
|
|
replaced = True
|
|
else:
|
|
merged.append({"name": layer_name, "features": layer.get("features", [])})
|
|
if not replaced:
|
|
merged.append(nav_layer)
|
|
extents["navigation_marks"] = nav_extents.get("navigation_marks", 4096)
|
|
encoded = encode_layers_preserving_extents(merged, extents)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(dir=target.parent, prefix=target.name + ".", delete=False) as handle:
|
|
temp_path = Path(handle.name)
|
|
handle.write(encoded)
|
|
os.replace(temp_path, target)
|
|
return 1
|
|
|
|
|
|
def main() -> None:
|
|
paths = [str(path) for path in NAVMARK_ROOT.rglob("*.pbf")]
|
|
with ProcessPoolExecutor(max_workers=8) as executor:
|
|
merged = sum(executor.map(merge_one, paths, chunksize=64))
|
|
print(json.dumps({"navmark_tiles": len(paths), "merged_tiles": merged, "delivery_root": str(DELIVERY_ROOT)}, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|