Spaces:
Sleeping
Sleeping
File size: 3,210 Bytes
74c43d2 |
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 |
"""Core logic for fetching geotags from addresses or coordinates."""
import re
from typing import List
import httpx
import h3
from .models import GeotagResult
# Regex for validating "latitude,longitude" input
COORD_REGEX = re.compile(r"^-?([1-8]?[0-9]|[1-9]0)\.{1}\d{1,15}\s*,\s*-?([1]?[0-7]?[0-9]|[1]?[1-8]?[0])\.{1}\d{1,15}$")
class GeotaggingError(Exception):
"""Base exception for geotagging errors."""
pass
class InvalidCoordinatesError(GeotaggingError):
"""Raised when coordinate input is invalid."""
pass
class GeocodingServiceError(GeotaggingError):
"""Raised when the external geocoding service fails."""
pass
class NoResultsFoundError(GeotaggingError):
"""Raised when no results are found for a query."""
pass
async def fetch_geotags(q: str, resolution: int) -> List[GeotagResult]:
"""
Fetches geotag information for a given query string (address or lat,lng).
Args:
q: The address string or 'latitude,longitude' pair.
resolution: The H3 resolution (0-15).
Returns:
A list of GeotagResult objects.
Raises:
InvalidCoordinatesError: If the input coordinates are malformed.
GeocodingServiceError: If there's an issue connecting to the geocoding service.
NoResultsFoundError: If the geocoding service returns no results.
"""
locations = []
if COORD_REGEX.match(q.replace(" ", "")):
try:
lat_str, lon_str = q.split(',')
lat, lon = float(lat_str), float(lon_str)
locations.append({
"lat": lat,
"lon": lon,
"display_name": f"Coordinates: {lat:.6f}, {lon:.6f}"
})
except (ValueError, IndexError):
raise InvalidCoordinatesError("Invalid coordinate format. Use 'latitude,longitude'.")
else:
headers = {'User-Agent': 'TrackEmDown/1.0 (nikhilsingh.io)'}
url = f"https://nominatim.openstreetmap.org/search?format=json&q={q}"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=10.0)
response.raise_for_status()
locations = response.json()
except httpx.RequestError as exc:
raise GeocodingServiceError(f"Error connecting to geocoding service: {exc}")
if not locations:
raise NoResultsFoundError("No results found for the given query.")
results = []
for loc in locations:
try:
lat, lon = float(loc['lat']), float(loc['lon'])
geotag = h3.latlng_to_cell(lat, lon, resolution)
results.append(
GeotagResult(
address=loc.get('display_name', 'N/A'),
latitude=lat,
longitude=lon,
geotag=geotag
)
)
except (ValueError, KeyError) as e:
# Skip malformed results from the external API
print(f"Skipping a location due to parsing error: {e}")
continue
if not results:
raise NoResultsFoundError("Geocoding service returned malformed data.")
return results
|