tools/weather.py

import os
import json
import urllib.error
import urllib.parse
import urllib.request
from dotenv import load_dotenv
from smolagents.tools import Tool

load_dotenv()

class WeatherTool(Tool):
    name = "weather_data"
    description = "Fetches the current weather for a given location using Google Geocoding and Open-Meteo."
    inputs = {'location': {'type': 'string', 'description': 'The location to fetch the weather.'}}
    output_type = "string"

    #FORWARD
    # Cos'è: È il metodo che viene chiamato da smolagents quando il tool deve essere eseguito.
    # A cosa serve:
    # Contiene la logica principale del tool: riceve gli input, esegue le operazioni e restituisce il risultato.
    # Gli argomenti di forward devono corrispondere a quelli definiti in inputs.

    def forward(self, location: str) -> str:
        """
        Fetches the current weather for a given location.
        Args:
            location: The location to fetch the weather.
        Returns:
            A string describing the current weather.
        """

        base_url = "https://api.open-meteo.com/v1/forecast"

        key = os.getenv('GOOGLE_MAPS_API')
        maps_base_url = "https://maps.googleapis.com/maps/api/geocode/json"

        params = urllib.parse.urlencode(
            {
                "address" : f"{location}",
                "key" : f"{key}",
            }
        )

        url = f"{maps_base_url}?{params}"

        try:
            response = urllib.request.urlopen(url)
        except urllib.error.URLError:
            print(f"Error while fetching the request for latitude and longitude")
        else:
            result = json.load(response)
            try:
                latitude = result["results"][0]["geometry"]["location"]["lat"]
                longitude = result["results"][0]["geometry"]["location"]["lng"]
            except (KeyError, IndexError) as e:
                print(f"Error parsing the json to fetch lat/lon: {e}")

        param = urllib.parse.urlencode(
            {
                "latitude" : f"{latitude}",
                "longitude" : f"{longitude}",
                "current": "weather_code",
                "timezone": "Europe/Rome"
            }
        )

        url = f"{base_url}?{param}"

        try:
            weather = urllib.request.urlopen(url)
        except urllib.error.URLError:
            print(f"Error while fetching the request for the weather")
        else:
            result = json.load(weather)

            current_weather_code = result["current"]["weather_code"]

            weather_descriptions = {
                0: "Cielo sereno",
                1: "Cielo prevalentemente sereno",
                2: "Parzialmente nuvoloso",
                3: "Nuvoloso",
                45: "Nebbia",
                48: "Nebbia brinata",
                51: "Pioggerella leggera",
                53: "Pioggerella moderata",
                55: "Pioggerella intensa",
                56: "Pioggerella ghiacciata leggera",
                57: "Pioggerella ghiacciata intensa",
                61: "Pioggia leggera",
                63: "Pioggia moderata",
                65: "Pioggia forte",
                66: "Pioggia ghiacciata leggera",
                67: "Pioggia ghiacciata forte",
                71: "Nevicata leggera",
                73: "Nevicata moderata",
                75: "Nevicata forte",
                77: "Grandinata",
                80: "Rovesci leggeri",
                81: "Rovesci moderati",
                82: "Rovesci violenti",
                85: "Rovesci di neve leggeri",
                86: "Rovesci di neve forti",
                95: "Temporale",
                96: "Temporale con grandine leggera",
                99: "Temporale con grandine forte"
            }

            description = weather_descriptions.get(current_weather_code, "Condizione meteo sconosciuta")
            return description