import requests from datetime import datetime from pydantic import BaseModel, Field class Tools: class Valves(BaseModel): api_key: str = Field( default="", description="OpenWeather API Key is required", ) units: str = Field( default="imperial", # Fixed typo: 'imperial' instead of 'imperical' description="imperial or metric units of measure", ) city: str = Field( default="Indianapolis,IN", description="City,ST", ) def __init__(self): self.valves = self.Valves() def get_current_weather(self) -> str: """ Get the current weather. :return: The current weather information or an error message. """ if not self.valves.api_key: return "Valve for API key is not set." base_url = "http://api.openweathermap.org/data/2.5/weather" params = { "q": self.valves.city, "appid": self.valves.api_key, "units": self.valves.units, } try: response = requests.get(base_url, params=params) response.raise_for_status() # Raise HTTPError for bad responses (4xx and 5xx) data = response.json() if data.get("cod") != 200: return f"Error fetching weather data: {data.get('message')}" weather_description = data["weather"][0]["description"] temperature = data["main"]["temp"] feels_like = data["main"]["feels_like"] humidity = data["main"]["humidity"] wind_speed = data["wind"]["speed"] unit_symbol = "°F" if self.valves.units == "imperial" else "°C" speed_unit = ( "Mile Per Hour" if self.valves.units == "imperial" else "Meters Per Second" ) return ( f"Weather in {self.valves.city}: {temperature}{unit_symbol}, and feels like {feels_like}{unit_symbol}, {weather_description}, " f"Humidity: {humidity}%, Wind Speed: {wind_speed} {speed_unit}" ) except requests.RequestException as e: return f"Error fetching weather data: {str(e)}" def get_extended_forecast(self) -> str: """ Get an extended forecast (5-day forecast). The information returned is for future dates, so when responding to the user, use future-tense statements. :return: The extended forecast information or an error message. """ if not self.valves.api_key: return "Valve for API key is not set." # Using the 'forecast' endpoint for extended forecast data. base_url = "http://api.openweathermap.org/data/2.5/forecast" params = { "q": self.valves.city, "appid": self.valves.api_key, "units": self.valves.units, } try: response = requests.get(base_url, params=params) response.raise_for_status() data = response.json() # OpenWeather returns a string code "200" for success on the forecast endpoint. if data.get("cod") != "200": return f"Error fetching forecast data: {data.get('message')}" forecast_items = data.get("list", []) if not forecast_items: return "No forecast data available." # Build a string that contains a forecast summary for each time slot. forecast_lines = [] unit_symbol = "°F" if self.valves.units == "imperial" else "°C" for item in forecast_items: # Convert the Unix timestamp to a human-readable date/time. dt = datetime.fromtimestamp(item["dt"]) dt_str = dt.strftime("%Y-%m-%d %H:%M") temp = item["main"]["temp"] description = item["weather"][0]["description"] forecast_lines.append(f"{dt_str}: {temp}{unit_symbol}, {description}") return "\n".join(forecast_lines) except requests.RequestException as e: return f"Error fetching forecast data: {str(e)}"