#!/usr/bin/env python3 """Fetch all Austrian public-transport stops from OpenStreetMap (Overpass API) and write a deduplicated snapshot to data/stops_at.csv.gz. Data © OpenStreetMap contributors, licensed under the ODbL. Run this only when you want to refresh the snapshot; the import into the database reads the checked-in CSV and needs no network access. python scripts/fetch_stops.py """ from __future__ import annotations import csv import gzip import json import sys import unicodedata import urllib.parse import urllib.request from pathlib import Path OUT_PATH = Path(__file__).resolve().parent.parent / "data" / "stops_at.csv.gz" OVERPASS_ENDPOINTS = [ "https://overpass-api.de/api/interpreter", "https://overpass.kumi.systems/api/interpreter", "https://overpass.private.coffee/api/interpreter", ] # Named nodes in Austria that represent a boardable stop ("Haltestelle"). OVERPASS_QUERY = """ [out:json][timeout:600]; area["ISO3166-1"="AT"][admin_level=2]->.at; ( node(area.at)["highway"="bus_stop"]["name"]; node(area.at)["railway"="tram_stop"]["name"]; node(area.at)["railway"="station"]["name"]; node(area.at)["railway"="halt"]["name"]; node(area.at)["public_transport"="station"]["name"]; node(area.at)["amenity"="bus_station"]["name"]; ); out body; """ CSV_FIELDS = [ "osm_type", "osm_id", "name", "stop_type", "municipality", "latitude", "longitude", ] def normalize_name(value: str) -> str: """Lowercase, fold accents and ß so search is diacritic-insensitive. The same function lives in stops_import.py; keep them in sync. """ value = value.strip().lower().replace("ß", "ss") decomposed = unicodedata.normalize("NFKD", value) return "".join(ch for ch in decomposed if not unicodedata.combining(ch)) def classify(tags: dict) -> str: if tags.get("station") == "subway" or tags.get("subway") == "yes": return "subway" if tags.get("railway") in {"station", "halt"}: return "train" if tags.get("railway") == "tram_stop" or tags.get("tram") == "yes": return "tram" if tags.get("amenity") == "bus_station": return "bus_station" if tags.get("highway") == "bus_stop" or tags.get("bus") == "yes": return "bus" if tags.get("public_transport") == "station": return "other" return "other" def municipality_of(tags: dict) -> str: for key in ("addr:city", "is_in:municipality", "is_in:city", "is_in"): value = tags.get(key) if value: return value.split(",")[0].strip() return "" def fetch_raw() -> dict: body = urllib.parse.urlencode({"data": OVERPASS_QUERY}).encode() last_error: Exception | None = None for endpoint in OVERPASS_ENDPOINTS: try: print(f"Querying {endpoint} ...", file=sys.stderr) req = urllib.request.Request( endpoint, data=body, headers={"User-Agent": "pbt-stop-import/1.0"} ) with urllib.request.urlopen(req, timeout=600) as resp: return json.load(resp) except Exception as exc: # noqa: BLE001 - try the next mirror last_error = exc print(f" failed: {exc}", file=sys.stderr) raise SystemExit(f"All Overpass endpoints failed; last error: {last_error}") TYPE_RANK = {"subway": 5, "train": 4, "bus_station": 3, "tram": 2, "bus": 1, "other": 0} # Nodes with the same name closer than this are treated as one stop (OSM keeps a # separate node per direction for most bus stops, ~20-60 m apart). MERGE_METERS = 250.0 def _dist_m(a: dict, b: dict) -> float: import math lat1, lon1, lat2, lon2 = map( math.radians, (a["latitude"], a["longitude"], b["latitude"], b["longitude"]), ) dlat, dlon = lat2 - lat1, lon2 - lon1 h = math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2 return 2 * 6_371_000 * math.asin(min(1.0, math.sqrt(h))) def build_rows(payload: dict) -> list[dict]: # Collect candidate nodes grouped by normalized name. by_name: dict[str, list[dict]] = {} for element in payload.get("elements", []): if element.get("type") != "node": continue tags = element.get("tags", {}) name = (tags.get("name") or "").strip() lat = element.get("lat") lon = element.get("lon") if not name or lat is None or lon is None: continue node = { "osm_type": "node", "osm_id": int(element["id"]), "name": name, "stop_type": classify(tags), "municipality": municipality_of(tags), "latitude": round(float(lat), 6), "longitude": round(float(lon), 6), } by_name.setdefault(normalize_name(name), []).append(node) rows: list[dict] = [] for nodes in by_name.values(): # Greedy single-link clustering within the name group. clusters: list[list[dict]] = [] for node in sorted(nodes, key=lambda n: n["osm_id"]): for cluster in clusters: if any(_dist_m(node, member) <= MERGE_METERS for member in cluster): cluster.append(node) break else: clusters.append([node]) for cluster in clusters: lead = max( cluster, key=lambda n: (TYPE_RANK[n["stop_type"]], n["municipality"] != "", -n["osm_id"]), ) rows.append( { "osm_type": "node", # lowest id in the cluster -> stable identity across refreshes "osm_id": min(n["osm_id"] for n in cluster), "name": lead["name"], "stop_type": lead["stop_type"], "municipality": lead["municipality"] or next((n["municipality"] for n in cluster if n["municipality"]), ""), "latitude": round(sum(n["latitude"] for n in cluster) / len(cluster), 6), "longitude": round(sum(n["longitude"] for n in cluster) / len(cluster), 6), } ) rows.sort(key=lambda r: (normalize_name(r["name"]), r["osm_id"])) return rows def main() -> None: payload = fetch_raw() rows = build_rows(payload) if len(rows) < 10_000: raise SystemExit( f"Only {len(rows)} stops parsed - refusing to overwrite the snapshot. " "Overpass probably returned a partial result; try again later." ) OUT_PATH.parent.mkdir(parents=True, exist_ok=True) with gzip.open(OUT_PATH, "wt", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=CSV_FIELDS) writer.writeheader() writer.writerows(rows) by_type: dict[str, int] = {} for row in rows: by_type[row["stop_type"]] = by_type.get(row["stop_type"], 0) + 1 print(f"Wrote {len(rows)} stops to {OUT_PATH}", file=sys.stderr) print(f" by type: {by_type}", file=sys.stderr) if __name__ == "__main__": main()