144 lines
5.0 KiB
Python
144 lines
5.0 KiB
Python
import requests
|
|
from pprint import pprint
|
|
from models import POI
|
|
|
|
OVERPASS_URL = "https://overpass-api.de/api/interpreter"
|
|
|
|
|
|
# REMARK:
|
|
# zwei Strategien:
|
|
# Fail-fast: Ein Fehler bricht alles ab → sinnvoll, wenn jedes Element kritisch ist
|
|
# Best-effort: Fehlerhafte Elemente überspringen, Rest verarbeiten → sinnvoll bei OSM-Daten, wo einzelne Einträge unvollständig sein können
|
|
|
|
|
|
class OverpassApiError(Exception):
|
|
pass
|
|
|
|
|
|
def load_pois(overpass_query: str, bbox: tuple) -> list[POI]:
|
|
"""Führt Fetch und Parse zusammen aus."""
|
|
raw = _fetch_overpass(overpass_query=overpass_query, bbox=bbox)
|
|
return _parse_pois(raw)
|
|
|
|
|
|
def _fetch_overpass(overpass_query: str, bbox: tuple) -> dict:
|
|
"""
|
|
Fragt die Overpass API nach Bergbahnen in der angegebenen Bounding Box ab.
|
|
Sendet einen HTTP-POST-Request an die Overpass API und gibt die geparste
|
|
JSON-Antwort zurück.
|
|
|
|
Args:
|
|
overpass_query (str): Overpass-QL-Query mit dem Platzhalter {bbox}.
|
|
Beispiel:
|
|
'[out:json][timeout:5];
|
|
(node["aerialway"="station"]({bbox}););
|
|
out center body;'
|
|
bbox (tuple): Bounding Box als 4-Tuple in Dezimalgrad:
|
|
(south, west, north, east)
|
|
Beispiel Davos: (46.72, 9.70, 46.92, 10.00)
|
|
Beispiel Schweiz: (45.8, 5.9, 47.8, 10.5)
|
|
|
|
Returns:
|
|
dict: Geparste JSON-Antwort der Overpass API. Die Antwort enthält
|
|
unter dem Schlüssel "elements" eine Liste von OSM-Objekten
|
|
(nodes und ways) mit ihren Tags und Koordinaten.
|
|
Beispiel:
|
|
{
|
|
"elements": [
|
|
{
|
|
"type": "node",
|
|
"id": 123456,
|
|
"lat": 46.8, "lon": 9.8,
|
|
"tags": {"aerialway": "station", "name": "Jakobshorn"}
|
|
},
|
|
...
|
|
]
|
|
}
|
|
|
|
Raises:
|
|
OverpassApiError: Wenn die API nicht innerhalb des gesetzten Timeouts
|
|
antwortet (clientseitig, unabhängig vom serverseitigen
|
|
Timeout im Query).
|
|
OverpassApiError: Wenn der Request aus einem anderen Grund fehlschlägt
|
|
(z.B. 429 Too Many Requests, 504 Gateway Timeout,
|
|
Netzwerkfehler).
|
|
"""
|
|
|
|
bbox_str = ",".join(map(str, bbox))
|
|
query = overpass_query.format(bbox=bbox_str)
|
|
|
|
try:
|
|
response = requests.post(
|
|
OVERPASS_URL,
|
|
data={"data": query},
|
|
timeout=5,
|
|
headers={"User-Agent": "CDS Exercise"},
|
|
)
|
|
response.raise_for_status() # prüft den HTTP-Statuscode der Antwort und wirft eine Exception, wenn es ein Fehler war (requests.HTTPError)
|
|
except requests.Timeout as exc:
|
|
raise OverpassApiError("Overpass-API Timeout") from exc
|
|
except requests.RequestException as exc:
|
|
raise OverpassApiError("Overpass-API Request fehlgeschlagen") from exc
|
|
|
|
data = response.json()
|
|
|
|
# zusätzliche Fehlermöglichkeit -> Status ist zwar 200, aber Liste mit Ergebnissen ist leer...
|
|
if "remark" in data:
|
|
raise OverpassApiError(f"Overpass Query-Fehler: {data['remark']}")
|
|
|
|
return data
|
|
|
|
|
|
def _parse_poi(data: dict) -> POI:
|
|
""" Wandelt ein einzelnes Overpass-Element in ein POI-Objekt um.
|
|
|
|
:param data: dictionary mit Daten für ein POI-Objekt
|
|
:return: POI-Objekt
|
|
"""
|
|
try:
|
|
return POI(
|
|
id=data['id'],
|
|
type=data.get('type', ''),
|
|
lat=float(data.get("lat") or data["center"]["lat"]),
|
|
lon=float(data.get("lon") or data["center"]["lon"]),
|
|
tags=data.get('tags', {}),
|
|
)
|
|
except KeyError as exc:
|
|
raise OverpassApiError("Feld in API - Antwort fehlt") from exc
|
|
except (TypeError, ValueError) as exc:
|
|
raise OverpassApiError("API - Antwort hat falsches Format ") from exc
|
|
|
|
|
|
|
|
def _parse_pois(raw: dict) -> list[POI]:
|
|
"""Extrahiert alle Elemente aus der API-Antwort und parst sie.
|
|
Fehlerhafte Elemente werden übersprungen und geloggt.
|
|
"""
|
|
pois = []
|
|
for element in raw.get("elements", []):
|
|
try:
|
|
pois.append(_parse_poi(element))
|
|
except OverpassApiError as exc:
|
|
print(f"POI übersprungen (id={element.get('id', '?')}): {exc}")
|
|
return pois
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
BERGBAHN_QUERY = """
|
|
[out:json][timeout:2][maxsize:500000];
|
|
(
|
|
node["aerialway"="station"]({bbox});
|
|
way["aerialway"="station"]({bbox});
|
|
node["railway"="funicular"]({bbox});
|
|
way["railway"="funicular"]({bbox});
|
|
node["railway"="station"]["funicular"="yes"]({bbox});
|
|
);
|
|
out center body;
|
|
"""
|
|
|
|
bbox = (46.72, 9.70, 46.92, 10.00)
|
|
# bbox = (45.8, 5.9, 47.8, 10.5)
|
|
result = load_pois(overpass_query=BERGBAHN_QUERY, bbox=bbox)
|
|
pprint(result) |