v3.0: add family, neighbourhood and living environment analysis

voorzieningen.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Voorzieningenanalyse op basis van lokale PostGIS OSM-data + ORS Matrix."""
+"""Voorzieningenanalyse met lokale PostGIS-OSM-data en ORS Matrix-routes."""
 
 import json
 import math
@@ -45,113 +45,129 @@ def _matrix(lon, lat, pois, profiel):
         "metrics": ["distance", "duration"],
     }).encode()
     req = urllib.request.Request(
-        f"{ORS_BASE}/matrix/{profiel}",
-        data=payload,
-        headers={"Content-Type": "application/json"},
-        method="POST",
+        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))
+    return list(zip(data.get("distances", [[]])[0], data.get("durations", [[]])[0]))
 
 
 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]
+    """Verrijk POI's; een defect profiel maakt het hele rapport niet onbruikbaar."""
+    for profiel, afstand_attr, duur_attr in [
+        ("cycling-regular", "fiets_afstand_m", "fiets_duur_s"),
+        ("driving-car", "auto_afstand_m", "auto_duur_s"),
+    ]:
+        try:
+            routes = _matrix(lon, lat, pois, profiel)
+        except Exception:
+            routes = []
+        for poi, route in zip(pois, routes):
+            setattr(poi, afstand_attr, route[0])
+            setattr(poi, duur_attr, route[1])
     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}")
+def _haversine_m(lon1, lat1, lon2, lat2):
+    radius = 6_371_000
+    phi1, phi2 = math.radians(lat1), math.radians(lat2)
+    dphi, dlambda = math.radians(lat2 - lat1), math.radians(lon2 - lon1)
+    a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
+    return 2 * radius * math.asin(math.sqrt(a))
 
+
+def zoek_voorzieningen(lon, lat, categorie, limit=5, radius_m=10000):
+    views = {"supermarkt": "poi_supermarkten", "school": "poi_scholen"}
+    if categorie not in views:
+        raise ValueError(f"Onbekende lokale 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,
+    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)
+    FROM {views[categorie]}, 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))
+            cur.execute(f"SELECT * FROM ({sql}) q ORDER BY hemelsbreed_m ASC LIMIT %s", (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
-    ]
+    return [Voorziening(categorie, row["name"], float(row["lat"]), float(row["lon"]), float(row["hemelsbreed_m"])) for row in rows]
+
+
+def zoek_osm_voorzieningen(lon, lat, categorie, tags, limit=5, radius_m=10000):
+    """Vraag kleinschalige gezins-POI's live op via Overpass, met een vaste fallback-endpoint."""
+    south, west = lat - radius_m / 111_320, lon - radius_m / (111_320 * math.cos(math.radians(lat)))
+    north, east = lat + radius_m / 111_320, lon + radius_m / (111_320 * math.cos(math.radians(lat)))
+    tag_filter = "".join(f'["{key}"="{value}"]' for key, value in tags.items())
+    query = f"[out:json][timeout:45];(nwr{tag_filter}({south},{west},{north},{east}););out center;"
+    request = urllib.request.Request(
+        "https://overpass.kumi.systems/api/interpreter", data=query.encode(),
+        headers={"Content-Type": "application/x-www-form-urlencoded", "User-Agent": "adres-analyse/3.0"},
+    )
+    with urllib.request.urlopen(request, timeout=70) as response:
+        elements = json.loads(response.read()).get("elements", [])
+    found = []
+    for element in elements:
+        point = element.get("center", element)
+        if "lat" not in point or "lon" not in point:
+            continue
+        name = element.get("tags", {}).get("name") or categorie.title()
+        found.append(Voorziening(categorie, name, float(point["lat"]), float(point["lon"]), _haversine_m(lon, lat, point["lon"], point["lat"]), bron="OpenStreetMap/Overpass"))
+    # Een naam komt vaak als node én als vlak voor; behoud de dichtstbijzijnde.
+    unique = {}
+    for poi in sorted(found, key=lambda p: p.hemelsbreed_m):
+        unique.setdefault(poi.naam.lower(), poi)
+    return list(unique.values())[:limit]
 
 
 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}
+    result = {
+        "supermarkten": zoek_voorzieningen(lon, lat, "supermarkt", limit, 10000),
+        "scholen": zoek_voorzieningen(lon, lat, "school", limit, 15000),
+    }
+    for pois in result.values():
+        verrijk_met_routes(lon, lat, pois)
+    return result
+
+
+def analyse_gezinsvoorzieningen(lon, lat, limit=5):
+    specs = {
+        "kinderopvang": ({"amenity": "kindergarten"}, 10000),
+        "speelplekken": ({"leisure": "playground"}, 5000),
+        "huisartsen": ({"amenity": "doctors"}, 10000),
+        "apotheken": ({"amenity": "pharmacy"}, 10000),
+        "bibliotheken": ({"amenity": "library"}, 15000),
+        "sport": ({"leisure": "sports_centre"}, 10000),
+    }
+    result = {}
+    for key, (tags, radius) in specs.items():
+        try:
+            result[key] = zoek_osm_voorzieningen(lon, lat, key, tags, limit, radius)
+            verrijk_met_routes(lon, lat, result[key])
+        except Exception as exc:
+            result[key] = []
+            result[f"{key}_fout"] = str(exc)
+    return result
 
 
 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(".", ",")
+    return f"{m:.0f} m" if m < 1000 else 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"
+    return f"{max(1, round(s / 60))} min"
 
 
 def format_voorzieningen_markdown(titel, pois):
     lines = [titel]
     if not pois:
-        lines.append("Geen voorzieningen gevonden in zoekradius.")
-        return "\n".join(lines)
+        return "\n".join(lines + ["Geen voorzieningen gevonden of bron tijdelijk niet beschikbaar."])
     for p in pois:
-        lines.append(
-            f"• **{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)}"
-        )
+        lines.append(f"• **{p.naam}** — Hemelsbreed: {format_afstand(p.hemelsbreed_m)} | Fiets: {format_afstand(p.fiets_afstand_m)} / {format_duur(p.fiets_duur_s)} | 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))