| """
|
| Geocoding Service - Uses Nominatim (OpenStreetMap) for free geocoding
|
| """
|
| import httpx
|
| from typing import Optional, Tuple
|
| import asyncio
|
|
|
|
|
| NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
|
| USER_AGENT = "NUMIDIUM/1.0 (Intelligence System)"
|
|
|
|
|
| async def geocode(location_name: str) -> Optional[Tuple[float, float]]:
|
| """
|
| Convert a location name to coordinates using Nominatim.
|
| Returns (latitude, longitude) or None if not found.
|
|
|
| Note: Nominatim has rate limits (1 request/second), so be careful with batch operations.
|
| """
|
| try:
|
| async with httpx.AsyncClient(timeout=10.0) as client:
|
| response = await client.get(
|
| NOMINATIM_URL,
|
| params={
|
| "q": location_name,
|
| "format": "json",
|
| "limit": 1,
|
| "addressdetails": 0
|
| },
|
| headers={
|
| "User-Agent": USER_AGENT
|
| }
|
| )
|
|
|
| if response.status_code == 200:
|
| data = response.json()
|
| if data and len(data) > 0:
|
| lat = float(data[0]["lat"])
|
| lon = float(data[0]["lon"])
|
| return (lat, lon)
|
|
|
| return None
|
|
|
| except Exception as e:
|
| print(f"Geocoding error for '{location_name}': {e}")
|
| return None
|
|
|
|
|
| async def geocode_batch(location_names: list[str], delay: float = 1.0) -> dict[str, Tuple[float, float]]:
|
| """
|
| Geocode multiple locations with proper rate limiting.
|
| Returns a dict mapping location names to (lat, lon) tuples.
|
| """
|
| results = {}
|
|
|
| for name in location_names:
|
| coords = await geocode(name)
|
| if coords:
|
| results[name] = coords
|
|
|
| await asyncio.sleep(delay)
|
|
|
| return results
|
|
|
