feat(Q3): Open-Meteo weather fetcher with hourly caching
This commit is contained in:
57
vigilar/detection/weather.py
Normal file
57
vigilar/detection/weather.py
Normal file
@@ -0,0 +1,57 @@
|
||||
"""Open-Meteo weather fetcher with in-memory caching."""
|
||||
|
||||
import logging
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_WMO_CODES = {
|
||||
0: "Clear sky", 1: "Mainly clear", 2: "Partly cloudy", 3: "Overcast",
|
||||
45: "Fog", 48: "Depositing rime fog",
|
||||
51: "Light drizzle", 53: "Moderate drizzle", 55: "Dense drizzle",
|
||||
61: "Light rain", 63: "Moderate rain", 65: "Heavy rain",
|
||||
66: "Light freezing rain", 67: "Heavy freezing rain",
|
||||
71: "Light snow", 73: "Moderate snow", 75: "Heavy snow", 77: "Snow grains",
|
||||
80: "Light showers", 81: "Moderate showers", 82: "Violent showers",
|
||||
85: "Light snow showers", 86: "Heavy snow showers",
|
||||
95: "Thunderstorm", 96: "Thunderstorm with light hail", 99: "Thunderstorm with heavy hail",
|
||||
}
|
||||
|
||||
CACHE_TTL_S = 3600
|
||||
|
||||
|
||||
def _weather_code_to_text(code: int) -> str:
|
||||
return _WMO_CODES.get(code, "Unknown")
|
||||
|
||||
|
||||
class WeatherFetcher:
|
||||
def __init__(self):
|
||||
self._cache: dict[str, tuple[dict, float]] = {}
|
||||
|
||||
def get_conditions(self, lat: float, lon: float) -> dict | None:
|
||||
cache_key = f"{lat:.2f},{lon:.2f}"
|
||||
if cache_key in self._cache:
|
||||
data, ts = self._cache[cache_key]
|
||||
if time.time() - ts < CACHE_TTL_S:
|
||||
return data
|
||||
try:
|
||||
resp = requests.get(
|
||||
"https://api.open-meteo.com/v1/forecast",
|
||||
params={"latitude": lat, "longitude": lon, "current": "temperature_2m,weather_code"},
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return None
|
||||
j = resp.json()
|
||||
current = j.get("current", {})
|
||||
result = {
|
||||
"temperature_c": current.get("temperature_2m"),
|
||||
"conditions": _weather_code_to_text(current.get("weather_code", -1)),
|
||||
}
|
||||
self._cache[cache_key] = (result, time.time())
|
||||
return result
|
||||
except Exception:
|
||||
log.debug("Weather fetch failed", exc_info=True)
|
||||
return None
|
||||
Reference in New Issue
Block a user