""" Mapbox Map Provider Implementation Default map provider using Mapbox APIs """ import httpx from typing import List, Optional, Tuple import logging from app.services.map_provider.base_provider import ( MapProvider, GeocodingResult, RouteResult, DistanceResult ) logger = logging.getLogger(__name__) class MapboxProvider(MapProvider): """ Mapbox API implementation. Documentation: https://docs.mapbox.com/api/ """ DEFAULT_API_URL = "https://api.mapbox.com" def __init__(self, api_key: str, api_url: Optional[str] = None, settings: Optional[dict] = None): super().__init__(api_key, api_url or self.DEFAULT_API_URL, settings) self._client = httpx.AsyncClient(timeout=30.0) @property def provider_name(self) -> str: return "mapbox" async def geocode(self, address: str) -> Optional[GeocodingResult]: """Convert address to coordinates using Mapbox Geocoding API""" try: url = f"{self.api_url}/geocoding/v5/mapbox.places/{address}.json" params = { "access_token": self.api_key, "limit": 1, "types": "address,place,locality,neighborhood" } response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if not data.get("features"): return None feature = data["features"][0] coords = feature["geometry"]["coordinates"] context = {c.get("id", "").split(".")[0]: c.get("text") for c in feature.get("context", [])} return GeocodingResult( latitude=coords[1], longitude=coords[0], address=feature.get("text", ""), place_id=feature.get("id"), city=context.get("place"), state=context.get("region"), country=context.get("country"), postal_code=context.get("postcode"), formatted_address=feature.get("place_name") ) except Exception as e: logger.error(f"Mapbox geocoding error: {e}") return None async def reverse_geocode(self, latitude: float, longitude: float) -> Optional[GeocodingResult]: """Convert coordinates to address using Mapbox Reverse Geocoding""" try: url = f"{self.api_url}/geocoding/v5/mapbox.places/{longitude},{latitude}.json" params = { "access_token": self.api_key, "types": "address,place" } response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if not data.get("features"): return None feature = data["features"][0] coords = feature["geometry"]["coordinates"] context = {c.get("id", "").split(".")[0]: c.get("text") for c in feature.get("context", [])} return GeocodingResult( latitude=coords[1], longitude=coords[0], address=feature.get("text", ""), place_id=feature.get("id"), city=context.get("place"), state=context.get("region"), country=context.get("country"), postal_code=context.get("postcode"), formatted_address=feature.get("place_name") ) except Exception as e: logger.error(f"Mapbox reverse geocoding error: {e}") return None async def calculate_distance( self, origin_lat: float, origin_lng: float, dest_lat: float, dest_lng: float ) -> Optional[DistanceResult]: """Calculate distance using Mapbox Directions API""" try: url = f"{self.api_url}/directions/v5/mapbox/driving/{origin_lng},{origin_lat};{dest_lng},{dest_lat}" params = { "access_token": self.api_key, "overview": "false" } response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if not data.get("routes"): return None route = data["routes"][0] return DistanceResult( distance_km=route["distance"] / 1000, # Convert meters to km duration_minutes=int(route["duration"] / 60), # Convert seconds to minutes origin=(origin_lat, origin_lng), destination=(dest_lat, dest_lng) ) except Exception as e: logger.error(f"Mapbox distance calculation error: {e}") return None async def get_route( self, origin_lat: float, origin_lng: float, dest_lat: float, dest_lng: float, waypoints: Optional[List[Tuple[float, float]]] = None ) -> Optional[RouteResult]: """Get driving route using Mapbox Directions API""" try: # Build coordinates string coords = f"{origin_lng},{origin_lat}" if waypoints: for wp in waypoints: coords += f";{wp[1]},{wp[0]}" coords += f";{dest_lng},{dest_lat}" url = f"{self.api_url}/directions/v5/mapbox/driving/{coords}" params = { "access_token": self.api_key, "overview": "full", "geometries": "polyline", "steps": "true" } response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if not data.get("routes"): return None route = data["routes"][0] # Extract steps steps = [] for leg in route.get("legs", []): for step in leg.get("steps", []): steps.append({ "instruction": step.get("maneuver", {}).get("instruction", ""), "distance_m": step.get("distance", 0), "duration_s": step.get("duration", 0), }) return RouteResult( distance_km=route["distance"] / 1000, duration_minutes=int(route["duration"] / 60), polyline=route.get("geometry", ""), steps=steps, start_location=(origin_lat, origin_lng), end_location=(dest_lat, dest_lng) ) except Exception as e: logger.error(f"Mapbox route calculation error: {e}") return None async def search_places( self, query: str, latitude: Optional[float] = None, longitude: Optional[float] = None, limit: int = 10 ) -> List[GeocodingResult]: """Search for places using Mapbox Geocoding API""" try: url = f"{self.api_url}/geocoding/v5/mapbox.places/{query}.json" params = { "access_token": self.api_key, "limit": min(limit, 10), "types": "address,poi,place,locality,neighborhood" } # Add proximity bias if coordinates provided if latitude and longitude: params["proximity"] = f"{longitude},{latitude}" response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() results = [] for feature in data.get("features", []): coords = feature["geometry"]["coordinates"] context = {c.get("id", "").split(".")[0]: c.get("text") for c in feature.get("context", [])} results.append(GeocodingResult( latitude=coords[1], longitude=coords[0], address=feature.get("text", ""), place_id=feature.get("id"), city=context.get("place"), state=context.get("region"), country=context.get("country"), postal_code=context.get("postcode"), formatted_address=feature.get("place_name") )) return results except Exception as e: logger.error(f"Mapbox place search error: {e}") return [] async def close(self): """Close HTTP client""" await self._client.aclose()