Spaces:
Runtime error
Runtime error
File size: 8,988 Bytes
4160511 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | 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] |