# pxy_routing/services/ors_provider.py from __future__ import annotations from typing import Any, Dict, Iterable, List, Tuple import math, os, time, random, requests # Optional graceful fallback try: from .crowfly_provider import CrowFlyRoutingProvider except Exception: CrowFlyRoutingProvider = None # fallback disabled if not available LatLon = Tuple[float, float] # (lat, lon) class ORSRoutingProvider: """ ORS isochrones with retries/backoff and optional crow-fly fallback. Env: ORS_BASE_URL e.g., https://api.openrouteservice.org ORS_API_KEY key (omit/blank for self-host) ORS_PROFILE driving-car | cycling-regular | foot-walking (default: driving-car) ORS_TIMEOUT_S request timeout seconds (default: 5) ORS_GENERALIZE generalization in meters for polygons (optional, e.g., 20) ORS_MAX_RANGE safety cap in minutes (optional; e.g., 45) # New hardening knobs: ORS_RETRY number of retries on 429/5xx (default: 2) ORS_BACKOFF_BASE_S base seconds for exponential backoff (default: 0.8) ORS_FALLBACK set to "crowfly" to degrade gracefully on errors """ def __init__( self, base_url: str | None = None, api_key: str | None = None, profile: str = "driving-car", timeout_s: int = 5, generalize: int | None = None, max_range_min: int | None = None, ): self.base_url = (base_url or os.getenv("ORS_BASE_URL") or "").rstrip("/") self.api_key = api_key if api_key is not None else os.getenv("ORS_API_KEY", "") self.profile = os.getenv("ORS_PROFILE", profile) self.timeout_s = int(os.getenv("ORS_TIMEOUT_S", str(timeout_s))) gen = os.getenv("ORS_GENERALIZE") self.generalize = int(gen) if (gen and gen.isdigit()) else generalize mr = os.getenv("ORS_MAX_RANGE") self.max_range_min = int(mr) if (mr and mr.isdigit()) else max_range_min # Hardening knobs self.retries = int(os.getenv("ORS_RETRY", "2")) self.backoff_base = float(os.getenv("ORS_BACKOFF_BASE_S", "0.8")) self.fallback_mode = (os.getenv("ORS_FALLBACK") or "").strip().lower() self._fallback = None if self.fallback_mode == "crowfly" and CrowFlyRoutingProvider: self._fallback = CrowFlyRoutingProvider() if not self.base_url: raise ValueError("ORS_BASE_URL is required for ORSRoutingProvider") self._iso_url = f"{self.base_url}/v2/isochrones/{self.profile}" self._headers = { "Content-Type": "application/json; charset=utf-8", "Accept": "application/json, application/geo+json", } if self.api_key: self._headers["Authorization"] = self.api_key # ---------- internals ---------- def _post(self, url: str, payload: Dict[str, Any]) -> requests.Response: attempts = 1 + max(0, self.retries) r = None for i in range(attempts): r = requests.post(url, json=payload, headers=self._headers, timeout=self.timeout_s) if r.status_code in (429, 502, 503, 504) and i < attempts - 1: delay = self.backoff_base * (2 ** i) * (0.75 + 0.5 * random.random()) time.sleep(delay) continue return r return r # type: ignore # ---------- public API ---------- def health(self) -> Dict[str, Any]: try: lat, lon = 19.4326, -99.1332 payload = {"locations": [[lon, lat]], "range": [60]} if self.generalize: payload["generalize"] = self.generalize r = self._post(self._iso_url, payload) ok = (r.status_code == 200) return {"provider": "ors", "ok": ok, "profile": self.profile, "base_url": self.base_url, "reason": None if ok else f"http {r.status_code}"} except Exception as e: return {"provider": "ors", "ok": False, "profile": self.profile, "base_url": self.base_url, "reason": f"{type(e).__name__}: {e}"} def isochrone(self, center: LatLon, minutes: int) -> Dict[str, Any]: if self.max_range_min and minutes > self.max_range_min: raise ValueError(f"minutes exceeds ORS_MAX_RANGE ({minutes} > {self.max_range_min})") lat, lon = center payload = {"locations": [[lon, lat]], "range": [int(minutes) * 60]} if self.generalize: payload["generalize"] = self.generalize r = self._post(self._iso_url, payload) if r.status_code != 200: if self._fallback is not None: feat = self._fallback.isochrone(center, minutes) feat["properties"]["provider"] = "ors_fallback_crowfly" return feat hint = { 400: "Bad request (profile/range/params).", 401: "Unauthorized (check ORS_API_KEY).", 403: "Forbidden (quota/key).", 404: "Profile/endpoint not found.", 413: "Payload too large.", 422: "Unprocessable (non-routable location or bad range).", 429: "Rate limited.", 500: "Server error.", 502: "Bad gateway.", 503: "Service unavailable.", 504: "Gateway timeout.", }.get(r.status_code, "Unexpected error.") raise RuntimeError(f"ORS isochrone error: HTTP {r.status_code}. {hint}") data = r.json() geom = _largest_polygon_geometry_from_ors(data) if not geom: if self._fallback is not None: feat = self._fallback.isochrone(center, minutes) feat["properties"]["provider"] = "ors_empty_fallback_crowfly" return feat raise RuntimeError("ORS returned no polygon geometry.") return { "type": "Feature", "properties": {"provider": "ors", "profile": self.profile, "minutes": minutes, "center": [lon, lat]}, "geometry": geom, } # Batch multiple ranges in one call (reduces rate-limit pressure) def isochrones(self, center: LatLon, minutes_list: List[int]) -> List[Dict[str, Any]]: lat, lon = center secs = [int(m) * 60 for m in minutes_list] payload = {"locations": [[lon, lat]], "range": secs} if self.generalize: payload["generalize"] = self.generalize r = self._post(self._iso_url, payload) if r.status_code != 200: # degrade by single-calls if fallback exists if self._fallback is not None: return [self.isochrone(center, m) for m in minutes_list] raise RuntimeError(f"ORS isochrones error: HTTP {r.status_code}") data = r.json() feats: List[Dict[str, Any]] = [] for feat in (data.get("features") or []): geom = _largest_polygon_geometry_from_ors({"features": [feat]}) if not geom: continue feats.append({ "type": "Feature", "properties": {"provider": "ors", "profile": self.profile}, "geometry": geom }) return feats # ---------- helpers ---------- def _largest_polygon_geometry_from_ors(fc: Dict[str, Any]) -> Dict[str, Any] | None: features = fc.get("features") or [] best_geom, best_area = None, -1.0 for feat in features: geom = feat.get("geometry") or {} gtype = geom.get("type"); coords = geom.get("coordinates") if not coords: continue if gtype == "Polygon": area = _polygon_area_m2(coords[0]) if area > best_area: best_area, best_geom = area, {"type":"Polygon","coordinates":coords} elif gtype == "MultiPolygon": for poly in coords: if not poly: continue ring = poly[0] area = _polygon_area_m2(ring) if area > best_area: best_area, best_geom = area, {"type":"Polygon","coordinates":[ring]} return best_geom def _polygon_area_m2(ring_lonlat: Iterable[Iterable[float]]) -> float: pts = list(ring_lonlat) if len(pts) < 3: return 0.0 if pts[0] != pts[-1]: pts = pts + [pts[0]] def merc(lon: float, lat: float) -> Tuple[float, float]: R = 6378137.0 x = math.radians(lon) * R lat = max(min(lat, 89.9999), -89.9999) y = math.log(math.tan(math.pi/4 + math.radians(lat)/2)) * R return x, y xs, ys = [], [] for lon, lat in pts: x, y = merc(float(lon), float(lat)); xs.append(x); ys.append(y) area = 0.0 for i in range(len(xs)-1): area += xs[i]*ys[i+1] - xs[i+1]*ys[i] return abs(area) * 0.5