from typing import Any, Optional from smolagents.tools import Tool import requests from datetime import datetime, timedelta import json import os from dotenv import load_dotenv class WeatherTool(Tool): name = "weather_forecast" description = "Obtient les prévisions météorologiques pour un pays/ville spécifique à une date donnée. Utilise l'API gratuite OpenWeatherMap 2.5." inputs = { 'location': {'type': 'string', 'description': 'Le nom de la ville ou du pays pour lequel obtenir la météo (ex: "Paris", "London", "Tokyo")'}, 'date': {'type': 'string', 'description': 'La date pour laquelle obtenir la météo au format YYYY-MM-DD (optionnel, par défaut aujourd\'hui)', 'nullable': True}, 'api_key': {'type': 'string', 'description': 'Clé API OpenWeatherMap (optionnel si définie dans les variables d\'environnement)', 'nullable': True} } output_type = "string" def __init__(self, api_key: Optional[str] = None): super().__init__() # Charger les variables d'environnement depuis le fichier .env load_dotenv() # Utiliser la clé API fournie, sinon celle du .env self.api_key = api_key or os.getenv('OPENWEATHER_API_KEY') self.base_url = "http://api.openweathermap.org/data/2.5" def forward(self, location: str, date: Optional[str] = None, api_key: Optional[str] = None) -> str: try: # Utiliser la clé API fournie ou celle par défaut used_api_key = api_key or self.api_key if not used_api_key: return "Erreur: Clé API OpenWeatherMap requise. Ajoutez OPENWEATHER_API_KEY dans votre fichier .env ou obtenez une clé gratuite sur https://openweathermap.org/api" # Parser la date si fournie target_date = None if date: try: target_date = datetime.strptime(date, "%Y-%m-%d") except ValueError: return f"Erreur: Format de date invalide. Utilisez YYYY-MM-DD (ex: 2024-01-15)" # Obtenir les coordonnées de la localisation geo_url = f"http://api.openweathermap.org/geo/1.0/direct" geo_params = { 'q': location, 'limit': 1, 'appid': used_api_key } geo_response = requests.get(geo_url, params=geo_params, timeout=10) geo_response.raise_for_status() geo_data = geo_response.json() if not geo_data: return f"Erreur: Localisation '{location}' non trouvée. Essayez avec le nom d'une ville ou d'un pays plus précis." lat = geo_data[0]['lat'] lon = geo_data[0]['lon'] country = geo_data[0].get('country', '') city_name = geo_data[0]['name'] # Utiliser l'API gratuite return self._get_weather(lat, lon, city_name, country, target_date, used_api_key) except requests.exceptions.Timeout: return "Erreur: Délai d'attente dépassé. Veuillez réessayer." except requests.exceptions.HTTPError as e: if e.response.status_code == 401: return "Erreur 401: Clé API invalide ou non activée. Vérifiez votre clé API OpenWeatherMap et assurez-vous qu'elle est activée (peut prendre quelques heures après création)." elif e.response.status_code == 429: return "Erreur 429: Limite de requêtes dépassée. Attendez avant de refaire une requête." else: return f"Erreur HTTP {e.response.status_code}: {str(e)}" except requests.exceptions.RequestException as e: return f"Erreur de requête: {str(e)}" except Exception as e: return f"Erreur inattendue: {str(e)}" def _get_weather(self, lat: float, lon: float, city_name: str, country: str, target_date: Optional[datetime], api_key: str) -> str: """Utilise l'API gratuite 2.5""" if not target_date or target_date.date() == datetime.now().date(): # Météo actuelle weather_url = f"{self.base_url}/weather" params = { 'lat': lat, 'lon': lon, 'appid': api_key, 'units': 'metric', 'lang': 'fr' } response = requests.get(weather_url, params=params, timeout=10) response.raise_for_status() data = response.json() return self._format_current_weather(data, city_name, country) elif target_date and target_date <= datetime.now() + timedelta(days=5): # Prévisions sur 5 jours forecast_url = f"{self.base_url}/forecast" params = { 'lat': lat, 'lon': lon, 'appid': api_key, 'units': 'metric', 'lang': 'fr' } response = requests.get(forecast_url, params=params, timeout=10) response.raise_for_status() data = response.json() return self._format_forecast_weather(data, city_name, country, target_date) else: return "Erreur: Les prévisions ne sont disponibles que pour les 5 prochains jours maximum." def _format_current_weather(self, data: dict, city_name: str, country: str) -> str: """Formate les données météo actuelles""" try: weather = data['weather'][0] main = data['main'] wind = data.get('wind', {}) result = f"🌤️ **Météo actuelle pour {city_name}, {country}**\n\n" result += f"**Conditions:** {weather['description'].title()}\n" result += f"**Température:** {main['temp']:.1f}°C (ressenti: {main['feels_like']:.1f}°C)\n" result += f"**Humidité:** {main['humidity']}%\n" result += f"**Pression:** {main['pressure']} hPa\n" if 'speed' in wind: result += f"**Vent:** {wind['speed']} m/s" if 'deg' in wind: result += f" ({self._wind_direction(wind['deg'])})" result += "\n" if 'visibility' in data: result += f"**Visibilité:** {data['visibility']/1000:.1f} km\n" return result except KeyError as e: return f"Erreur lors du formatage des données météo: {str(e)}" def _format_forecast_weather(self, data: dict, city_name: str, country: str, target_date: datetime) -> str: """Formate les prévisions météo pour une date spécifique""" try: target_date_str = target_date.strftime("%Y-%m-%d") # Trouver les prévisions pour la date cible forecasts_for_date = [] for forecast in data['list']: forecast_date = datetime.fromtimestamp(forecast['dt']).strftime("%Y-%m-%d") if forecast_date == target_date_str: forecasts_for_date.append(forecast) if not forecasts_for_date: return f"Aucune prévision disponible pour le {target_date_str}" result = f"🌤️ **Prévisions météo pour {city_name}, {country} - {target_date.strftime('%d/%m/%Y')}**\n\n" for i, forecast in enumerate(forecasts_for_date): time = datetime.fromtimestamp(forecast['dt']).strftime("%H:%M") weather = forecast['weather'][0] main = forecast['main'] wind = forecast.get('wind', {}) result += f"**{time}:**\n" result += f" • Conditions: {weather['description'].title()}\n" result += f" • Température: {main['temp']:.1f}°C (ressenti: {main['feels_like']:.1f}°C)\n" result += f" • Humidité: {main['humidity']}%\n" if 'speed' in wind: result += f" • Vent: {wind['speed']} m/s" if 'deg' in wind: result += f" ({self._wind_direction(wind['deg'])})" result += "\n" if i < len(forecasts_for_date) - 1: result += "\n" return result except KeyError as e: return f"Erreur lors du formatage des prévisions: {str(e)}" def _wind_direction(self, degrees: float) -> str: """Convertit les degrés en direction du vent""" directions = ["N", "NNE", "NE", "ENE", "E", "ESE", "SE", "SSE", "S", "SSO", "SO", "OSO", "O", "ONO", "NO", "NNO"] index = round(degrees / 22.5) % 16 return directions[index]