Hermes Agent
174 lines
6.7 KiB
Python
174 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Voorzieningenanalyse met lokale PostGIS-OSM-data en ORS Matrix-routes."""
|
|
|
|
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())
|
|
return list(zip(data.get("distances", [[]])[0], data.get("durations", [[]])[0]))
|
|
|
|
|
|
def verrijk_met_routes(lon, lat, pois):
|
|
"""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 _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,
|
|
ST_Distance(geom::geography, origin.geog) AS hemelsbreed_m
|
|
FROM {views[categorie]}, origin
|
|
WHERE geom IS NOT NULL AND ST_DWithin(geom::geography, origin.geog, %s)
|
|
ORDER BY lower(name), hemelsbreed_m ASC
|
|
"""
|
|
with psycopg2.connect(DB_DSN) as conn:
|
|
with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
|
|
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, 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):
|
|
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."
|
|
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."
|
|
return f"{max(1, round(s / 60))} min"
|
|
|
|
|
|
def format_voorzieningen_markdown(titel, pois):
|
|
lines = [titel]
|
|
if not pois:
|
|
return "\n".join(lines + ["Geen voorzieningen gevonden of bron tijdelijk niet beschikbaar."])
|
|
for p in pois:
|
|
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)
|