Initial Vibe coded project

flight_map/maps.py

@@ -0,0 +1,127 @@
+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 LineString
+
+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__)
+
+
+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});'
+        ");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 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
+        geometry = element.get("geometry")
+        if not isinstance(geometry, list) or len(geometry) < 2:
+            raise ValueError("Map way has no usable geometry; export using 'out geom'.")
+        try:
+            positions = [(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 positions
+        ):
+            raise ValueError("Map way coordinates must be finite and within geographic limits.")
+        points = [projection.point(lon, lat) for lon, lat in positions]
+        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