perfect-postcode/pipeline/transform/property_border_tiles.py

138 lines
4.7 KiB
Python

"""Build PMTiles polygon tiles for the INSPIRE property-border overlay.
Reads the HM Land Registry INSPIRE Index Polygons (per-local-authority GML ZIPs
in EPSG:27700), reprojects each parcel to WGS84, and tiles the outlines with
tippecanoe. The dashboard serves the resulting archive through
``/api/overlays/property-borders`` and renders it as thin outlines only at the
postcode zoom level.
The same ZIPs are already downloaded for postcode-boundary generation; this
target re-uses :func:`parse_inspire_zip` to stay self-contained and is wired to
the ``$(INSPIRE_STAMP)`` make dependency rather than the boundary cache.
Data: HM Land Registry INSPIRE Index Polygons, Open Government Licence v3.0.
Boundaries are indicative "general boundaries", not the legal extent of title.
"""
from __future__ import annotations
import argparse
import shutil
import subprocess
import tempfile
from pathlib import Path
import numpy as np
import shapely
from pyproj import Transformer
from shapely.geometry import Polygon
from tqdm import tqdm
from pipeline.local_temp import local_tmp_dir
from pipeline.transform.postcode_boundaries.inspire import parse_inspire_zip
def _require_tippecanoe() -> str:
executable = shutil.which("tippecanoe")
if executable is None:
raise RuntimeError(
"tippecanoe is required to build property border PMTiles. "
"Install tippecanoe and rerun this target."
)
return executable
def _write_property_geojsonseq(inspire_dir: Path, output_path: Path) -> int:
"""Stream INSPIRE parcels to a WGS84 GeoJSONSeq file, one feature per line.
Features carry no properties — the overlay only draws outlines, so dropping
attributes keeps the tiles as small as possible. Reprojection and GeoJSON
encoding are vectorised per ZIP (one local authority) to bound memory while
staying in shapely's C path.
"""
to_wgs84 = Transformer.from_crs("EPSG:27700", "EPSG:4326", always_xy=True)
zip_files = sorted(inspire_dir.glob("*.zip"))
if not zip_files:
raise RuntimeError(f"No INSPIRE ZIP files found in {inspire_dir}")
feature_count = 0
with output_path.open("w") as file:
for zip_path in tqdm(zip_files, desc="INSPIRE ZIPs", unit="file"):
rings = parse_inspire_zip(zip_path) # list of Nx2 (easting, northing)
if not rings:
continue
geoms = np.array([Polygon(coords) for coords in rings], dtype=object)
# interleaved=False → transform(x, y) called once with full arrays.
geoms = shapely.transform(geoms, to_wgs84.transform, interleaved=False)
for geometry_json in shapely.to_geojson(geoms):
file.write('{"type":"Feature","properties":{},"geometry":')
file.write(geometry_json)
file.write("}\n")
feature_count += 1
return feature_count
def build_property_border_tiles(
inspire_dir: Path,
output_path: Path,
min_zoom: int,
max_zoom: int,
) -> None:
tippecanoe = _require_tippecanoe()
output_path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(dir=local_tmp_dir()) as tmp:
ndjson_path = Path(tmp) / "property_borders.geojsonseq"
feature_count = _write_property_geojsonseq(inspire_dir, ndjson_path)
print(f"Writing {feature_count:,} INSPIRE parcel polygons")
subprocess.run(
[
tippecanoe,
"--force",
"--output",
str(output_path),
"--layer",
"property_borders",
"--minimum-zoom",
str(min_zoom),
"--maximum-zoom",
str(max_zoom),
# Borders are only meaningful at street level; thin the densest
# tiles at low zoom but keep full geometry at max zoom.
"--drop-smallest-as-needed",
"--simplify-only-low-zooms",
"--extend-zooms-if-still-dropping",
"--temporary-directory",
tmp,
str(ndjson_path),
],
check=True,
)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--inspire", type=Path, required=True, help="INSPIRE ZIP directory"
)
parser.add_argument(
"--output", type=Path, required=True, help="Output .pmtiles path"
)
parser.add_argument("--min-zoom", type=int, default=12)
parser.add_argument("--max-zoom", type=int, default=16)
args = parser.parse_args()
build_property_border_tiles(
inspire_dir=args.inspire,
output_path=args.output,
min_zoom=args.min_zoom,
max_zoom=args.max_zoom,
)
if __name__ == "__main__":
main()