|
|
"""
|
|
|
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
|
|
|
|
