tools/weather.py

import json
import requests
from typing import Any, Optional
from smolagents.tools import Tool


class WeatherForecast(Tool):
    name = "weather_forecast"
    description = "Performs a weather search via open-mateo with openstreetmaps to get latlon then returns the weather results."
    inputs = {'city': {'type': 'string', 'description': 'The name of the city required for weather gathering.'}}
    output_type = "string"

    def __init__(self, max_results=10, **kwargs):
        super().__init__()

    def forward(self, city) -> str:
        weather = self.get_weather_forecast(city)
        results = self.format_weather_for_agent(city, weather)
        print(results)
        if len(results) == 0:
            raise Exception("No results found! Try a less restrictive/shorter query.")
        
        return f"## Search Results\n\n{results}"

    def get_coordinates(self, city_name: str) -> [float, float]: 
        headers = {
            'User-Agent': 'MyGeocodingApp/1.0 (youremail@example.com)'  # Replace with your actual email
        }
        city_name = city_name.replace(" ", "+")
        base_url = "https://nominatim.openstreetmap.org/search"
        full_url = f"{base_url}?q={requests.utils.quote(city_name)}&format=json&limit=5"
    
        try:
            response = requests.get(full_url, headers=headers)
            response.raise_for_status()  # Raise an exception for bad status codes (4xx or 5xx)
            data = response.json()
    
            if data:  # Check if the response contains any results
                latitude = data[0]['lat']
                longitude = data[0]['lon']
                return latitude, longitude
            else:
                return None, None  # Return None if no results are found
    
        except requests.exceptions.RequestException as e:
            print(f"Error during geocoding: {e}")
            return None, None

    def get_weather_forecast(self, city: str) -> Optional[dict[str, Any]]:
        """
        Retrieves a 7-day weather forecast for a given city using the Open-Meteo API.

        Args:
            city: The name of the city.

        Returns:
            A dictionary containing the weather forecast data, or None if an error occurs
            or if the city's coordinates cannot be found.
        """
        latitude, longitude = self.get_coordinates(city)
        if latitude is None or longitude is None:
            print(f"Could not get coordinates for {city}.  Cannot retrieve weather forecast.")
            return None

        base_url = "https://api.open-meteo.com/v1/forecast"
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "hourly": "temperature_2m",
            "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
            "forecast_days": 7,
            "timezone": "auto"
        }

        try:
            response = requests.get(base_url, params=params)
            response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
            weather_data = response.json()
            return weather_data

        except requests.exceptions.RequestException as e:
            print(f"Error getting weather forecast: {e}")
            return None


    def format_weather_for_agent(self, city: str, weather_data: dict[str, Any]) -> str:
        """Formats the weather data into a string suitable for an agent's response."""

        if weather_data is None:
            return f"Could not retrieve weather forecast for {city}."

        daily = weather_data.get("daily", {})  # Use .get to handle missing data gracefully
        hourly = weather_data.get("hourly", {})

        # Example formatting (customize as needed)
        response_str = f"Weather forecast for {city}:\n"

        if daily:
            response_str += "Daily Forecast:\n"
            for i in range(len(daily.get("temperature_2m_max", []))): # Handle missing data
                max_temp = daily.get("temperature_2m_max", [None])[i]  # Use .get with default
                min_temp = daily.get("temperature_2m_min", [None])[i]
                precipitation = daily.get("precipitation_sum", [None])[i]
                response_str += (
                    f"Day {i+1}: Max: {max_temp}°C, Min: {min_temp}°C, Precipitation: {precipitation} mm\n"
                )

        if hourly:
            response_str += "\nHourly Forecast (Example):\n"
            # ... (Add hourly data formatting as needed) ...
            hourly_temps = hourly.get("temperature_2m", [])[:3] # Show first 3 hours
            for i, temp in enumerate(hourly_temps):
                response_str += f"Hour {i+1}: {temp}°C\n"
            # ...

        return response_str