Spaces:
Sleeping
Sleeping
| """ | |
| Location Services Tool for Address Resolution and Geocoding. | |
| This module provides comprehensive location services including: | |
| - Address to coordinates conversion (geocoding) | |
| - Support for various input formats (addresses, coordinates, place names) | |
| - Distance calculations between coordinates | |
| - Intelligent parsing of location strings | |
| The tool integrates with Nominatim (OpenStreetMap) for geocoding services | |
| and includes fallback mechanisms for reliable location resolution. | |
| Supported input formats: | |
| - "Málaga, Spain" (city, country) | |
| - "123 Main St, New York, NY" (full address) | |
| - "36.7156,-4.4044" (decimal coordinates) | |
| - "Tarifa Beach" (landmark/place name) | |
| Example: | |
| >>> tool = LocationTool() | |
| >>> result = tool.run(LocationInput(location_query="Lisbon, Portugal")) | |
| >>> print(f"Coordinates: {result.coordinates}") | |
| Author: Surf Spot Finder Team | |
| License: MIT | |
| """ | |
| from typing import Dict, Optional | |
| from pydantic import BaseModel, Field | |
| from geopy.geocoders import Nominatim | |
| from geopy.distance import geodesic | |
| import logging | |
| # Import from utils - this will work when run as a proper package | |
| try: | |
| from utils.location_parser import get_coordinates_from_location | |
| except ImportError: | |
| # Fallback for development/testing | |
| import sys | |
| import os | |
| sys.path.append(os.path.join(os.path.dirname(__file__), '..')) | |
| from utils.location_parser import get_coordinates_from_location | |
| logger = logging.getLogger(__name__) | |
| # ---------------------- Input / Output Schemas ---------------------- | |
| class LocationInput(BaseModel): | |
| """Input schema for location resolution. | |
| Attributes: | |
| location_query: Location string in any supported format. | |
| """ | |
| location_query: str = Field( | |
| description="Location name, address, or description (e.g., 'Malaga, Spain')" | |
| ) | |
| class LocationOutput(BaseModel): | |
| """Output schema for location resolution results. | |
| Attributes: | |
| success: Whether location was successfully resolved. | |
| coordinates: Dict with 'lat' and 'lon' keys if successful. | |
| formatted_address: Standardized address string from geocoder. | |
| error: Error message if resolution failed. | |
| """ | |
| success: bool | |
| coordinates: Optional[Dict[str, float]] = None | |
| formatted_address: str = "" | |
| error: str = "" | |
| class DistanceInput(BaseModel): | |
| """Input schema for distance calculation""" | |
| origin_lat: float = Field(description="Origin latitude") | |
| origin_lon: float = Field(description="Origin longitude") | |
| dest_lat: float = Field(description="Destination latitude") | |
| dest_lon: float = Field(description="Destination longitude") | |
| class DistanceOutput(BaseModel): | |
| """Output schema for distance calculation""" | |
| distance_km: float | |
| distance_miles: float | |
| # ---------------------- Tools ---------------------- | |
| class LocationTool: | |
| """MCP-compatible tool for location services""" | |
| name = "resolve_location" | |
| description = "Convert a location name or address into geographic coordinates" | |
| def __init__(self): | |
| self.geolocator = Nominatim(user_agent="surf-spot-finder") | |
| def run(self, input_data: LocationInput) -> LocationOutput: | |
| """Execute location resolution""" | |
| try: | |
| # Try Google Maps first | |
| try: | |
| coordinates = get_coordinates_from_location(input_data.location_query) | |
| if coordinates: | |
| if "lng" in coordinates: | |
| coordinates["lon"] = coordinates["lng"] | |
| return LocationOutput( | |
| success=True, | |
| coordinates=coordinates, | |
| formatted_address=input_data.location_query | |
| ) | |
| except Exception: | |
| logger.info("Google Maps unavailable, using Nominatim") | |
| # Fallback to Nominatim | |
| location = self.geolocator.geocode(input_data.location_query, timeout=10) | |
| if location: | |
| return LocationOutput( | |
| success=True, | |
| coordinates={ | |
| "lat": location.latitude, | |
| "lon": location.longitude, | |
| }, | |
| formatted_address=location.address, | |
| ) | |
| return LocationOutput( | |
| success=False, | |
| error=f"Could not find location: {input_data.location_query}", | |
| ) | |
| except Exception as e: | |
| logger.error(f"Location tool error: {e}") | |
| return LocationOutput(success=False, error=str(e)) | |
| class DistanceTool: | |
| """MCP-compatible tool for distance calculations""" | |
| name = "calculate_distance" | |
| description = "Calculate distance between two geographic points" | |
| def run(self, input_data: DistanceInput) -> DistanceOutput: | |
| origin = (input_data.origin_lat, input_data.origin_lon) | |
| dest = (input_data.dest_lat, input_data.dest_lon) | |
| distance = geodesic(origin, dest) | |
| return DistanceOutput( | |
| distance_km=round(distance.kilometers, 2), | |
| distance_miles=round(distance.miles, 2), | |
| ) | |
| # ---------------------- Registration ---------------------- | |
| def create_location_tool(): | |
| """Factory function to create the location tool""" | |
| tool = LocationTool() | |
| return { | |
| "name": tool.name, | |
| "description": tool.description, | |
| "input_schema": LocationInput.schema(), | |
| "function": tool.run, | |
| } | |
| def create_distance_tool(): | |
| """Factory function to create the distance tool""" | |
| tool = DistanceTool() | |
| return { | |
| "name": tool.name, | |
| "description": tool.description, | |
| "input_schema": DistanceInput.schema(), | |
| "function": tool.run, | |
| } | |