planet/scripts/build_earth_boundary_pmtiles.py

#!/usr/bin/env python3
"""Build the production Earth boundary PMTiles artifact from collected sources."""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
import shutil
import subprocess
import tempfile
from pathlib import Path
from typing import Any


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--admin0-source", type=Path, required=True)
    parser.add_argument("--coastline-source", type=Path, required=True)
    parser.add_argument("--claims-source", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--manifest", type=Path, required=True)
    parser.add_argument("--build-input-hash", required=True)
    parser.add_argument("--pov-policy", type=Path, default=None)
    parser.add_argument("--min-zoom", type=int, default=0)
    parser.add_argument("--max-zoom", type=int, default=10)
    return parser.parse_args()


def require_tool(name: str) -> str:
    path = shutil.which(name)
    if not path:
        raise RuntimeError(f"required external tool not found in PATH: {name}")
    return path


def load_geojson(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        payload = json.load(f)
    if payload.get("type") != "FeatureCollection":
        raise ValueError(f"{path} must be a GeoJSON FeatureCollection")
    return payload


def iter_layer_features(payload: dict[str, Any], layer: str) -> list[dict[str, Any]]:
    out = []
    for feature in payload.get("features") or []:
        if not feature.get("geometry"):
            continue
        props = dict(feature.get("properties") or {})
        props["tippecanoe"] = {"layer": layer}
        out.append(
            {
                "type": "Feature",
                "geometry": feature["geometry"],
                "properties": props,
            }
        )
    return out


def write_ndjson(path: Path, features: list[dict[str, Any]]) -> None:
    with path.open("w", encoding="utf-8") as f:
        for feature in features:
            f.write(json.dumps(feature, ensure_ascii=False, separators=(",", ":")))
            f.write("\n")


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def main() -> None:
    args = parse_args()
    tippecanoe = require_tool("tippecanoe")
    pmtiles = require_tool("pmtiles")

    admin0 = load_geojson(args.admin0_source)
    coastline = load_geojson(args.coastline_source)
    claims = load_geojson(args.claims_source)
    features = [
        *iter_layer_features(admin0, "boundary_admin0"),
        *iter_layer_features(coastline, "coastline"),
        *iter_layer_features(claims, "claim_line"),
    ]
    if not features:
        raise RuntimeError("no features available for PMTiles build")

    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.manifest.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix="planet-earth-boundaries-") as tmp:
        tmp_dir = Path(tmp)
        ndjson_path = tmp_dir / "earth-boundaries.ndjson"
        mbtiles_path = tmp_dir / "earth-boundaries.mbtiles"
        write_ndjson(ndjson_path, features)

        subprocess.run(
            [
                tippecanoe,
                "--force",
                "--read-parallel",
                "--minimum-zoom",
                str(args.min_zoom),
                "--maximum-zoom",
                str(args.max_zoom),
                "--no-tile-compression",
                "--output",
                str(mbtiles_path),
                str(ndjson_path),
            ],
            check=True,
        )
        subprocess.run([pmtiles, "convert", str(mbtiles_path), str(args.output)], check=True)

    manifest = {
        "version": "china-pov-v1",
        "builtAt": dt.datetime.now(dt.UTC).replace(microsecond=0).isoformat(),
        "tileProvider": "pmtiles-mvt",
        "format": "pmtiles+mvt",
        "buildInputHash": args.build_input_hash,
        "pmtiles": {
            "path": os.path.relpath(args.output, args.manifest.parent),
            "sha256": sha256_file(args.output),
        },
        "tiles": {
            "minZoom": args.min_zoom,
            "maxZoom": args.max_zoom,
        },
        "layers": ["boundary_admin0", "boundary_disputed_internal", "coastline", "claim_line"],
        "povPolicy": {
            "path": str(args.pov_policy) if args.pov_policy else None,
            "applied": False,
            "note": "This builder packages collected sources. Full China POV geometry overlay is still a source-prep responsibility.",
        },
        "base": None,
        "hoverIndex": None,
    }
    with args.manifest.open("w", encoding="utf-8") as f:
        json.dump(manifest, f, ensure_ascii=False, indent=2)
        f.write("\n")

    print(
        json.dumps(
            {
                "status": "success",
                "output": str(args.output),
                "features": len(features),
                "format": "pmtiles+mvt",
            },
            ensure_ascii=False,
            sort_keys=True,
        )
    )


if __name__ == "__main__":
    main()