""" Nominatim Geocoding Service Provides address search and geocoding using OpenStreetMap Nominatim API. For development: uses public API (rate-limited) For production: should point to self-hosted Nominatim instance """ import httpx from typing import List, Dict, Any, Optional import structlog from asyncio import sleep logger = structlog.get_logger() class NominatimService: """ Nominatim geocoding and address search service. Uses OpenStreetMap Nominatim API for address autocomplete and geocoding. Respects rate limits and usage policy. """ # For development: public API (rate-limited to 1 req/sec) # For production: should be overridden with self-hosted instance DEFAULT_BASE_URL = "https://nominatim.openstreetmap.org" def __init__(self, base_url: Optional[str] = None, user_agent: str = "BakeryIA-Forecasting/1.0"): """ Initialize Nominatim service. Args: base_url: Nominatim server URL (defaults to public API) user_agent: User agent for API requests (required by Nominatim policy) """ self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") self.user_agent = user_agent self.headers = { "User-Agent": self.user_agent } # Rate limiting for public API (1 request per second) self.is_public_api = self.base_url == self.DEFAULT_BASE_URL self.min_request_interval = 1.0 if self.is_public_api else 0.0 logger.info( "Nominatim service initialized", base_url=self.base_url, is_public_api=self.is_public_api, rate_limit=f"{self.min_request_interval}s" if self.is_public_api else "none" ) async def search_address( self, query: str, country_code: str = "es", limit: int = 10 ) -> List[Dict[str, Any]]: """ Search for addresses matching query (autocomplete). Args: query: Address search query country_code: ISO country code to restrict search (default: Spain) limit: Maximum number of results Returns: List of address suggestions with display_name, lat, lon, osm_id, etc. """ if not query or len(query.strip()) < 3: logger.warning("Search query too short", query=query) return [] try: # Rate limiting for public API if self.is_public_api: await sleep(self.min_request_interval) async with httpx.AsyncClient(timeout=10.0) as client: response = await client.get( f"{self.base_url}/search", params={ "q": query, "format": "json", "addressdetails": 1, "countrycodes": country_code, "limit": limit, "accept-language": "es" }, headers=self.headers ) response.raise_for_status() results = response.json() # Parse and enrich results addresses = [] for result in results: addresses.append({ "display_name": result.get("display_name"), "lat": float(result.get("lat")), "lon": float(result.get("lon")), "osm_type": result.get("osm_type"), "osm_id": result.get("osm_id"), "place_id": result.get("place_id"), "type": result.get("type"), "class": result.get("class"), "address": result.get("address", {}), "boundingbox": result.get("boundingbox", []) }) logger.info( "Address search completed", query=query, result_count=len(addresses) ) return addresses except httpx.HTTPError as e: logger.error( "Nominatim API request failed", query=query, error=str(e) ) return [] except Exception as e: logger.error( "Unexpected error in address search", query=query, error=str(e), exc_info=True ) return [] async def geocode_address( self, address: str, country_code: str = "es" ) -> Optional[Dict[str, Any]]: """ Geocode an address to get coordinates. Args: address: Full address string country_code: ISO country code Returns: Dictionary with lat, lon, display_name, address components or None """ results = await self.search_address(address, country_code, limit=1) if not results: logger.warning("No geocoding results found", address=address) return None result = results[0] logger.info( "Address geocoded successfully", address=address, lat=result["lat"], lon=result["lon"] ) return result async def reverse_geocode( self, latitude: float, longitude: float ) -> Optional[Dict[str, Any]]: """ Reverse geocode coordinates to get address. Args: latitude: Latitude coordinate longitude: Longitude coordinate Returns: Dictionary with address information or None """ try: # Rate limiting for public API if self.is_public_api: await sleep(self.min_request_interval) async with httpx.AsyncClient(timeout=10.0) as client: response = await client.get( f"{self.base_url}/reverse", params={ "lat": latitude, "lon": longitude, "format": "json", "addressdetails": 1, "accept-language": "es" }, headers=self.headers ) response.raise_for_status() result = response.json() address_info = { "display_name": result.get("display_name"), "lat": float(result.get("lat")), "lon": float(result.get("lon")), "osm_type": result.get("osm_type"), "osm_id": result.get("osm_id"), "place_id": result.get("place_id"), "address": result.get("address", {}), "boundingbox": result.get("boundingbox", []) } logger.info( "Reverse geocoding completed", lat=latitude, lon=longitude, address=address_info["display_name"] ) return address_info except httpx.HTTPError as e: logger.error( "Nominatim reverse geocoding failed", lat=latitude, lon=longitude, error=str(e) ) return None except Exception as e: logger.error( "Unexpected error in reverse geocoding", lat=latitude, lon=longitude, error=str(e), exc_info=True ) return None async def validate_coordinates( self, latitude: float, longitude: float ) -> bool: """ Validate that coordinates point to a real location. Args: latitude: Latitude to validate longitude: Longitude to validate Returns: True if coordinates are valid, False otherwise """ if not (-90 <= latitude <= 90 and -180 <= longitude <= 180): return False result = await self.reverse_geocode(latitude, longitude) return result is not None async def health_check(self) -> bool: """ Check if Nominatim service is accessible. Returns: True if service is healthy, False otherwise """ try: async with httpx.AsyncClient(timeout=5.0) as client: response = await client.get( f"{self.base_url}/status", params={"format": "json"}, headers=self.headers ) return response.status_code == 200 except Exception as e: logger.error( "Nominatim health check failed", error=str(e) ) return False