File size: 4,149 Bytes
c0ef6d4 28471a4 c0ef6d4 28471a4 c0ef6d4 28471a4 c0ef6d4 e0ec429 c0ef6d4 e0ec429 c0ef6d4 e0ec429 28471a4 e0ec429 28471a4 e0ec429 c0ef6d4 |
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 |
import requests
import os
from typing import Optional, Dict, Any
from utils.config import config
from functools import lru_cache
import time
class WeatherService:
"""Service for fetching weather information with caching"""
def __init__(self):
self.api_key = config.openweather_api_key
self.base_url = "http://api.openweathermap.org/data/2.5"
def _get_ttl_hash(self, seconds=300):
"""Helper function to invalidate cache periodically"""
return round(time.time() / seconds)
@lru_cache(maxsize=128)
def get_current_weather_cached(self, city: str, ttl_hash=None):
"""Cached version of weather API calls"""
return self.get_current_weather(city)
def get_current_weather(self, city: str) -> Optional[Dict[str, Any]]:
"""Get current weather for a city"""
if not self.api_key:
print("OpenWeather API key not configured")
return None
try:
params = {
'q': city,
'appid': self.api_key,
'units': 'metric' # Celsius
}
response = requests.get(
f"{self.base_url}/weather",
params=params,
timeout=10
)
if response.status_code == 200:
data = response.json()
return {
'city': data['name'],
'country': data['sys']['country'],
'temperature': data['main']['temp'],
'feels_like': data['main']['feels_like'],
'humidity': data['main']['humidity'],
'description': data['weather'][0]['description'],
'icon': data['weather'][0]['icon']
}
else:
print(f"Weather API error: {response.status_code} - {response.text}")
return None
except Exception as e:
print(f"Error fetching weather data: {e}")
return None
def get_forecast(self, city: str, days: int = 5) -> Optional[Dict[str, Any]]:
"""Get weather forecast for a city"""
if not self.api_key:
print("OpenWeather API key not configured")
return None
try:
params = {
'q': city,
'appid': self.api_key,
'units': 'metric',
'cnt': days
}
response = requests.get(
f"{self.base_url}/forecast",
params=params,
timeout=10
)
if response.status_code == 200:
data = response.json()
forecasts = []
for item in data['list']:
forecasts.append({
'datetime': item['dt_txt'],
'temperature': item['main']['temp'],
'description': item['weather'][0]['description'],
'icon': item['weather'][0]['icon']
})
return {
'city': data['city']['name'],
'country': data['city']['country'],
'forecasts': forecasts
}
else:
print(f"Forecast API error: {response.status_code} - {response.text}")
return None
except Exception as e:
print(f"Error fetching forecast data: {e}")
return None
def get_weather_summary(self, city="New York") -> str:
"""Get formatted weather summary with caching"""
try:
weather = self.get_current_weather_cached(
city,
ttl_hash=self._get_ttl_hash(300)
)
if weather:
return f"{weather.get('temperature', 'N/A')}°C, {weather.get('description', 'Clear skies')}"
else:
return "Clear skies"
except:
return "Clear skies"
# Global weather service instance
weather_service = WeatherService()
|