""" Google Maps Provider Implementation Alternative map provider using Google Maps 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 GoogleMapsProvider(MapProvider): """ Google Maps API implementation. Documentation: https://developers.google.com/maps/documentation """ DEFAULT_API_URL = "https://maps.googleapis.com/maps/api" 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 "google_maps" async def geocode(self, address: str) -> Optional[GeocodingResult]: """Convert address to coordinates using Google Geocoding API""" try: url = f"{self.api_url}/geocode/json" params = { "address": address, "key": self.api_key } response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if data.get("status") != "OK" or not data.get("results"): return None result = data["results"][0] location = result["geometry"]["location"] components = {c["types"][0]: c["long_name"] for c in result.get("address_components", [])} return GeocodingResult( latitude=location["lat"], longitude=location["lng"], address=result.get("formatted_address", ""), place_id=result.get("place_id"), city=components.get("locality"), state=components.get("administrative_area_level_1"), country=components.get("country"), postal_code=components.get("postal_code"), formatted_address=result.get("formatted_address") ) except Exception as e: logger.error(f"Google geocoding error: {e}") return None async def reverse_geocode(self, latitude: float, longitude: float) -> Optional[GeocodingResult]: """Convert coordinates to address using Google Reverse Geocoding""" try: url = f"{self.api_url}/geocode/json" params = { "latlng": f"{latitude},{longitude}", "key": self.api_key } response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if data.get("status") != "OK" or not data.get("results"): return None result = data["results"][0] location = result["geometry"]["location"] components = {c["types"][0]: c["long_name"] for c in result.get("address_components", [])} return GeocodingResult( latitude=location["lat"], longitude=location["lng"], address=result.get("formatted_address", ""), place_id=result.get("place_id"), city=components.get("locality"), state=components.get("administrative_area_level_1"), country=components.get("country"), postal_code=components.get("postal_code"), formatted_address=result.get("formatted_address") ) except Exception as e: logger.error(f"Google 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 Google Distance Matrix API""" try: url = f"{self.api_url}/distancematrix/json" params = { "origins": f"{origin_lat},{origin_lng}", "destinations": f"{dest_lat},{dest_lng}", "mode": "driving", "key": self.api_key } response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if data.get("status") != "OK": return None element = data["rows"][0]["elements"][0] if element.get("status") != "OK": return None return DistanceResult( distance_km=element["distance"]["value"] / 1000, duration_minutes=int(element["duration"]["value"] / 60), origin=(origin_lat, origin_lng), destination=(dest_lat, dest_lng) ) except Exception as e: logger.error(f"Google 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 Google Directions API""" try: url = f"{self.api_url}/directions/json" params = { "origin": f"{origin_lat},{origin_lng}", "destination": f"{dest_lat},{dest_lng}", "mode": "driving", "key": self.api_key } if waypoints: wp_str = "|".join([f"{wp[0]},{wp[1]}" for wp in waypoints]) params["waypoints"] = wp_str response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if data.get("status") != "OK" or not data.get("routes"): return None route = data["routes"][0] leg = route["legs"][0] # Extract steps steps = [] for step in leg.get("steps", []): steps.append({ "instruction": step.get("html_instructions", ""), "distance_m": step.get("distance", {}).get("value", 0), "duration_s": step.get("duration", {}).get("value", 0), }) # Calculate total distance and duration total_distance = sum(leg["distance"]["value"] for leg in route["legs"]) total_duration = sum(leg["duration"]["value"] for leg in route["legs"]) return RouteResult( distance_km=total_distance / 1000, duration_minutes=int(total_duration / 60), polyline=route.get("overview_polyline", {}).get("points", ""), steps=steps, start_location=(origin_lat, origin_lng), end_location=(dest_lat, dest_lng) ) except Exception as e: logger.error(f"Google 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 Google Places API""" try: url = f"{self.api_url}/place/textsearch/json" params = { "query": query, "key": self.api_key } if latitude and longitude: params["location"] = f"{latitude},{longitude}" params["radius"] = 50000 # 50km radius response = await self._client.get(url, params=params) response.raise_for_status() data = response.json() if data.get("status") not in ["OK", "ZERO_RESULTS"]: return [] results = [] for place in data.get("results", [])[:limit]: location = place["geometry"]["location"] results.append(GeocodingResult( latitude=location["lat"], longitude=location["lng"], address=place.get("name", ""), place_id=place.get("place_id"), formatted_address=place.get("formatted_address") )) return results except Exception as e: logger.error(f"Google place search error: {e}") return [] async def close(self): """Close HTTP client""" await self._client.aclose()