Release v2.0 - add Netherlands POI analysis

voorzieningen.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""Voorzieningenanalyse op basis van lokale PostGIS OSM-data + ORS Matrix."""
+
+import json
+import math
+import os
+import urllib.request
+from dataclasses import dataclass, asdict
+
+import psycopg2
+import psycopg2.extras
+
+DB_DSN = os.environ.get(
+    "ADRES_ANALYSE_DB_DSN",
+    "host=127.0.0.1 port=5433 dbname=adresanalyse user=adres password=adres",
+)
+ORS_BASE = os.environ.get("ORS_BASE", "http://192.168.1.71:9080/ors/v2")
+
+
+@dataclass
+class Voorziening:
+    categorie: str
+    naam: str
+    lat: float
+    lon: float
+    hemelsbreed_m: float
+    fiets_afstand_m: float | None = None
+    fiets_duur_s: float | None = None
+    auto_afstand_m: float | None = None
+    auto_duur_s: float | None = None
+    bron: str = "OSM/PostGIS"
+
+    def to_dict(self):
+        return asdict(self)
+
+
+def _matrix(lon, lat, pois, profiel):
+    if not pois:
+        return []
+    locations = [[lon, lat]] + [[p.lon, p.lat] for p in pois]
+    payload = json.dumps({
+        "locations": locations,
+        "sources": [0],
+        "destinations": list(range(1, len(locations))),
+        "metrics": ["distance", "duration"],
+    }).encode()
+    req = urllib.request.Request(
+        f"{ORS_BASE}/matrix/{profiel}",
+        data=payload,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=60) as resp:
+        data = json.loads(resp.read())
+    distances = data.get("distances", [[]])[0]
+    durations = data.get("durations", [[]])[0]
+    return list(zip(distances, durations))
+
+
+def verrijk_met_routes(lon, lat, pois):
+    fiets = _matrix(lon, lat, pois, "cycling-regular")
+    auto = _matrix(lon, lat, pois, "driving-car")
+    for i, poi in enumerate(pois):
+        if i < len(fiets):
+            poi.fiets_afstand_m, poi.fiets_duur_s = fiets[i]
+        if i < len(auto):
+            poi.auto_afstand_m, poi.auto_duur_s = auto[i]
+    return pois
+
+
+def zoek_voorzieningen(lon, lat, categorie, limit=5, radius_m=10000):
+    if categorie == "supermarkt":
+        view = "poi_supermarkten"
+    elif categorie == "school":
+        view = "poi_scholen"
+    else:
+        raise ValueError(f"Onbekende categorie: {categorie}")
+
+    sql = f"""
+    WITH origin AS (
+      SELECT ST_SetSRID(ST_MakePoint(%s, %s), 4326)::geography AS geog
+    )
+    SELECT DISTINCT ON (lower(name))
+      name,
+      ST_Y(geom) AS lat,
+      ST_X(geom) AS lon,
+      ST_Distance(geom::geography, origin.geog) AS hemelsbreed_m
+    FROM {view}, origin
+    WHERE geom IS NOT NULL
+      AND ST_DWithin(geom::geography, origin.geog, %s)
+    ORDER BY lower(name), hemelsbreed_m ASC
+    """
+    # DISTINCT ON kiest dichtstbijzijnde per naam; buitenquery sorteert weer op afstand.
+    wrapped = f"SELECT * FROM ({sql}) q ORDER BY hemelsbreed_m ASC LIMIT %s"
+    with psycopg2.connect(DB_DSN) as conn:
+        with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+            cur.execute(wrapped, (lon, lat, radius_m, limit))
+            rows = cur.fetchall()
+    return [
+        Voorziening(
+            categorie=categorie,
+            naam=row["name"],
+            lat=float(row["lat"]),
+            lon=float(row["lon"]),
+            hemelsbreed_m=float(row["hemelsbreed_m"]),
+        )
+        for row in rows
+    ]
+
+
+def analyse_voorzieningen(lon, lat, limit=5):
+    supermarkten = zoek_voorzieningen(lon, lat, "supermarkt", limit=limit, radius_m=10000)
+    scholen = zoek_voorzieningen(lon, lat, "school", limit=limit, radius_m=15000)
+    verrijk_met_routes(lon, lat, supermarkten)
+    verrijk_met_routes(lon, lat, scholen)
+    return {"supermarkten": supermarkten, "scholen": scholen}
+
+
+def format_afstand(m):
+    if m is None:
+        return "n.b."
+    if m < 1000:
+        return f"{m:.0f} m"
+    return f"{m/1000:.1f} km".replace(".", ",")
+
+
+def format_duur(s):
+    if s is None:
+        return "n.b."
+    minutes = max(1, round(s / 60))
+    return f"{minutes} min"
+
+
+def format_voorzieningen_markdown(titel, pois):
+    lines = [titel]
+    if not pois:
+        lines.append("Geen voorzieningen gevonden in zoekradius.")
+        return "\n".join(lines)
+    for i, p in enumerate(pois, 1):
+        lines.extend([
+            f"{i}. **{p.naam}**",
+            f"   Hemelsbreed: {format_afstand(p.hemelsbreed_m)}",
+            f"   Fiets: {format_afstand(p.fiets_afstand_m)} / {format_duur(p.fiets_duur_s)}",
+            f"   Auto: {format_afstand(p.auto_afstand_m)} / {format_duur(p.auto_duur_s)}",
+        ])
+    return "\n".join(lines)
+
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser()
+    parser.add_argument("lon", type=float)
+    parser.add_argument("lat", type=float)
+    parser.add_argument("--limit", type=int, default=5)
+    args = parser.parse_args()
+    result = analyse_voorzieningen(args.lon, args.lat, args.limit)
+    print(json.dumps({k: [p.to_dict() for p in v] for k, v in result.items()}, indent=2, ensure_ascii=False))