FlightPlot/flight_map/maps.py

import hashlib
import json
import logging
import math
import os
import tempfile
from pathlib import Path
from typing import Any
from urllib.parse import urlencode

from shapely.geometry import GeometryCollection, LineString, Polygon
from shapely.geometry.base import BaseGeometry
from shapely.ops import polygonize_full, unary_union

from flight_map.geo import Bounds, Projection
from flight_map.network import request_json

OVERPASS_URL = "https://overpass-api.de/api/interpreter"
LOGGER = logging.getLogger(__name__)


class WaterMaskError(ValueError):
    """Water geometry cannot reliably distinguish land from water."""


def validate_map_data(data: Any) -> dict[str, Any]:
    if not isinstance(data, dict) or not isinstance(data.get("elements"), list):
        raise ValueError("Map data must be an Overpass JSON object containing elements.")
    if data.get("remark"):
        raise ValueError(f"Overpass could not complete the map query: {data['remark']}")
    return data


def map_data(
    bounds: Bounds, cache: Path, local_file: Path | None = None, *, refresh: bool = False
) -> dict[str, Any]:
    if local_file and refresh:
        raise ValueError("--refresh-map cannot be combined with --map-data.")
    bbox = f"{bounds.south},{bounds.west},{bounds.north},{bounds.east}"
    query = (
        "[out:json][timeout:60];("
        f'way["highway"]({bbox});way["waterway"]({bbox});'
        f'way["building"]({bbox});way["natural"="water"]({bbox});'
        f'way["waterway"="riverbank"]({bbox});'
        f'relation["natural"="water"]({bbox});relation["waterway"="riverbank"]({bbox});'
        ");out geom;"
    )
    key = hashlib.sha256((OVERPASS_URL + "\n" + query).encode()).hexdigest()
    path = local_file or cache / f"osm-{key}.json"
    if path.exists() and (local_file or not refresh):
        LOGGER.info("Reading %s map data: %s", "local" if local_file else "cached", path)
        with path.open() as stream:
            try:
                data = json.load(stream)
            except (ValueError, UnicodeError) as exc:
                raise ValueError(
                    f"Invalid map JSON in {path}; correct the file or use --refresh-map "
                    "to replace a damaged cache entry."
                ) from exc
    elif local_file:
        raise ValueError(f"Local map data file does not exist: {local_file}")
    else:
        LOGGER.info("Downloading vector map data from Overpass for %s", bounds.values)
        data = request_json(
            OVERPASS_URL,
            data=urlencode({"data": query}).encode(),
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        validate_map_data(data)
        cache.mkdir(parents=True, exist_ok=True)
        temporary: Path | None = None
        try:
            with tempfile.NamedTemporaryFile(
                mode="w", encoding="utf-8", dir=cache, suffix=".tmp", delete=False
            ) as stream:
                temporary = Path(stream.name)
                json.dump(data, stream)
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(temporary, path)
        finally:
            if temporary and temporary.exists():
                temporary.unlink()
        LOGGER.info("Cached vector map data: %s", path)
    return validate_map_data(data)


def positions(geometry: Any) -> list[tuple[float, float]]:
    if not isinstance(geometry, list) or len(geometry) < 2:
        raise ValueError("Map way has no usable geometry; export using 'out geom'.")
    try:
        result = [(float(p["lon"]), float(p["lat"])) for p in geometry]
    except (KeyError, ValueError, TypeError) as exc:
        raise ValueError("Map way contains invalid geographic coordinates.") from exc
    if any(
        not math.isfinite(lon)
        or not math.isfinite(lat)
        or not -180 <= lon <= 180
        or not -90 <= lat <= 90
        for lon, lat in result
    ):
        raise ValueError("Map way coordinates must be finite and within geographic limits.")
    return result


def water_areas(data: dict[str, Any]) -> BaseGeometry:
    """Assemble closed OSM water areas, retaining islands in multipolygon relations."""
    areas: list[BaseGeometry] = []
    member_ids = set()
    for element in data["elements"]:
        if not isinstance(element, dict):
            raise ValueError("Map data contains a malformed feature.")
        tags = element.get("tags", {})
        if not isinstance(tags, dict):
            raise ValueError("Map feature tags must be an object.")
        is_water = tags.get("natural") == "water" or tags.get("waterway") == "riverbank"
        if element.get("type") != "relation" or not is_water:
            continue
        rings: dict[str, list[LineString]] = {"outer": [], "inner": []}
        for member in element.get("members", []):
            if member.get("type") != "way":
                continue
            role = member.get("role", "") or "outer"
            if role in rings:
                rings[role].append(LineString(positions(member.get("geometry"))))
                member_ids.add(member["ref"])
        combined = {}
        for role, segments in rings.items():
            polygons, cuts, dangles, invalid = polygonize_full(segments)
            if not cuts.is_empty or not dangles.is_empty or not invalid.is_empty:
                raise WaterMaskError("Water multipolygon has incomplete rings; cannot mask land.")
            combined[role] = unary_union(list(polygons.geoms))
        if combined["outer"].is_empty:
            raise WaterMaskError("Water multipolygon has no outer ring.")
        areas.append(combined["outer"].difference(combined["inner"]))
    for element in data["elements"]:
        tags = element.get("tags", {})
        if (
            element.get("type") != "way"
            or element.get("id") in member_ids
            or not (tags.get("natural") == "water" or tags.get("waterway") == "riverbank")
        ):
            continue
        points = positions(element.get("geometry"))
        if len(points) < 4 or points[0] != points[-1]:
            raise WaterMaskError(
                f"Water way {element.get('id', 'unknown')} has an unclosed outline; "
                "cannot mask land."
            )
        polygon = Polygon(points)
        if not polygon.is_valid:
            raise WaterMaskError("Water area has invalid geometry; cannot mask land.")
        areas.append(polygon)
    return unary_union(areas) if areas else GeometryCollection()


def map_lines(
    data: dict[str, Any], projection: Projection, *, min_building_size: float = 0
) -> list[LineString]:
    result = []
    for element in data["elements"]:
        if not isinstance(element, dict):
            raise ValueError("Map data contains a malformed feature.")
        if element.get("type") != "way":
            continue
        tags = element.get("tags", {})
        if not isinstance(tags, dict):
            raise ValueError("Map feature tags must be an object.")
        if tags.get("highway") in {
            "footway",
            "path",
            "steps",
            "cycleway",
            "bridleway",
            "corridor",
            "construction",
            "proposed",
        }:
            continue
        points = [projection.point(lon, lat) for lon, lat in positions(element.get("geometry"))]
        line = LineString(points)
        if "building" in tags:
            west, south, east, north = line.bounds
            if min(east - west, north - south) < min_building_size:
                continue
        result.append(line)
    if not result:
        raise ValueError("No drawable map features were found for this area.")
    return result