Spaces:
Sleeping
Sleeping
File size: 3,631 Bytes
f59a149 | 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 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | import time
import logging
import requests
from typing import Optional, Tuple
from ..config import settings
logger = logging.getLogger("pharmashield.geocoder")
class Geocoder:
"""
Geocoding service using Google Maps API (if key available)
or falling back to OpenStreetMap's Nominatim.
"""
def __init__(self):
self.google_key = settings.GOOGLE_MAPS_API_KEY
self.nominatim_url = "https://nominatim.openstreetmap.org/search"
self.google_url = "https://maps.googleapis.com/maps/api/geocode/json"
self.headers = {
"User-Agent": "PharmaShield-MapAnalytics/1.0"
}
def get_coordinates(self, location_name: str) -> Optional[Tuple[float, float]]:
"""
Fetches latitude and longitude for a given location name.
"""
if self.google_key:
return self._google_geocode(location_name)
return self._nominatim_geocode(location_name)
def _google_geocode(self, location_name: str) -> Optional[Tuple[float, float]]:
"""Geocoding via Google Maps API."""
try:
params = {
"address": location_name,
"key": self.google_key
}
logger.info(f"Google Geocoding: {location_name}")
response = requests.get(self.google_url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if data.get("status") == "OK" and data.get("results"):
loc = data["results"][0]["geometry"]["location"]
lat, lng = loc["lat"], loc["lng"]
logger.info(f"Google found coords for {location_name}: {lat}, {lng}")
return lat, lng
else:
logger.warning(f"Google Geocoding failed for {location_name}: {data.get('status')}")
return None
except Exception as e:
logger.error(f"Google Geocoding exception for {location_name}: {e}")
return None
def _nominatim_geocode(self, location_name: str) -> Optional[Tuple[float, float]]:
"""Geocoding via Nominatim (with rate limiting)."""
try:
params = {
"q": location_name,
"format": "json",
"limit": 1
}
logger.info(f"Nominatim Geocoding: {location_name}")
response = requests.get(self.nominatim_url, params=params, headers=self.headers, timeout=10)
response.raise_for_status()
data = response.json()
if data and len(data) > 0:
lat = float(data[0]["lat"])
lon = float(data[0]["lon"])
logger.info(f"Nominatim found coords for {location_name}: {lat}, {lon}")
return lat, lon
else:
logger.warning(f"No coordinates found via Nominatim for {location_name}")
return None
except Exception as e:
logger.error(f"Nominatim Geocoding failed for {location_name}: {e}")
return None
finally:
# Respect Nominatim's 1 request per second usage policy
time.sleep(1.1)
def geocode_batch(self, locations: list[str]) -> dict[str, Tuple[float, float]]:
"""
Geocodes a list of locations and returns a dictionary mapping name to coordinates.
"""
results = {}
for loc in locations:
coords = self.get_coordinates(loc)
if coords:
results[loc] = coords
return results
# Singleton instance
geocoder = Geocoder()
|