File size: 1,982 Bytes
3c11834 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
"""
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
# Respect Nominatim rate limits
await asyncio.sleep(delay)
return results
|