adres-analyse/voorzieningen.py

#!/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 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)}"
        )
    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))