Add OpenWeatherMap weather forecasts. Closes #48
Show 8-day Bristol home weather on the index and weekends pages. Show destination weather per day on the trip list and trip page. Cache forecasts in ~/lib/data/weather/ and refresh via update.py. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
61e17d9c96
commit
b4f0a5bf5d
7 changed files with 217 additions and 0 deletions
83
agenda/weather.py
Normal file
83
agenda/weather.py
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
"""Weather forecast using OpenWeatherMap One Call API."""
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
|
||||
import pyowm
|
||||
|
||||
|
||||
def _cache_path(data_dir: str, lat: float, lon: float) -> str:
|
||||
"""Path for weather cache file."""
|
||||
weather_dir = os.path.join(data_dir, "weather")
|
||||
os.makedirs(weather_dir, exist_ok=True)
|
||||
return os.path.join(weather_dir, f"{lat:.2f}_{lon:.2f}.json")
|
||||
|
||||
|
||||
def _is_fresh(path: str, max_age_hours: int = 6) -> bool:
|
||||
"""Return True if the cache file exists and is recent enough."""
|
||||
if not os.path.exists(path):
|
||||
return False
|
||||
age = datetime.now().timestamp() - os.path.getmtime(path)
|
||||
return age < max_age_hours * 3600
|
||||
|
||||
|
||||
def get_forecast(data_dir: str, api_key: str, lat: float, lon: float) -> list[dict]:
|
||||
"""Return 8-day daily forecast for lat/lon, caching results for 6 hours."""
|
||||
cache_file = _cache_path(data_dir, lat, lon)
|
||||
|
||||
if _is_fresh(cache_file):
|
||||
with open(cache_file) as f:
|
||||
return json.load(f) # type: ignore[no-any-return]
|
||||
|
||||
owm = pyowm.OWM(api_key)
|
||||
mgr = owm.weather_manager()
|
||||
result = mgr.one_call(lat=lat, lon=lon)
|
||||
|
||||
forecasts = []
|
||||
for day in result.forecast_daily:
|
||||
dt = datetime.fromtimestamp(day.ref_time)
|
||||
temp = day.temperature("celsius")
|
||||
forecasts.append(
|
||||
{
|
||||
"date": dt.date().isoformat(),
|
||||
"status": day.status,
|
||||
"detailed_status": day.detailed_status,
|
||||
"temp_min": round(temp["min"]),
|
||||
"temp_max": round(temp["max"]),
|
||||
"precipitation_probability": day.precipitation_probability,
|
||||
"icon": day.weather_icon_name,
|
||||
}
|
||||
)
|
||||
|
||||
with open(cache_file, "w") as f:
|
||||
json.dump(forecasts, f)
|
||||
|
||||
return forecasts
|
||||
|
||||
|
||||
def trip_latlon(trip: object) -> tuple[float, float] | None:
|
||||
"""Return (lat, lon) for the primary destination of a trip, or None."""
|
||||
from agenda.types import Trip
|
||||
|
||||
assert isinstance(trip, Trip)
|
||||
for item in list(trip.accommodation) + list(trip.conferences):
|
||||
if "latitude" in item and "longitude" in item:
|
||||
return (float(item["latitude"]), float(item["longitude"]))
|
||||
return None
|
||||
|
||||
|
||||
def get_trip_weather(data_dir: str, api_key: str, trip: object) -> dict[str, dict]:
|
||||
"""Return forecast for a trip keyed by date ISO string.
|
||||
|
||||
Returns an empty dict if no location is known or the API call fails.
|
||||
"""
|
||||
latlon = trip_latlon(trip)
|
||||
if not latlon:
|
||||
return {}
|
||||
lat, lon = latlon
|
||||
try:
|
||||
forecasts = get_forecast(data_dir, api_key, lat, lon)
|
||||
except Exception:
|
||||
return {}
|
||||
return {f["date"]: f for f in forecasts}
|
||||
Loading…
Add table
Add a link
Reference in a new issue