Edward Betts
feaefba03c
Web view was calling the OpenWeatherMap API directly on every request (home + all trip locations), causing >1000 calls/day. Now web_view.py uses cache_only=True everywhere so it never triggers live API calls — update.py (cron) is the sole API caller. Also warms home location cache in update_weather(), and increases TTL from 6h to 24h. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
"""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 = 24) -> 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,
|
|
cache_only: bool = False,
|
|
) -> list[dict]:
|
|
"""Return 8-day daily forecast for lat/lon, caching results for 24 hours.
|
|
|
|
If cache_only=True, return cached data if available (even if stale) and
|
|
never call the API. Returns [] if no cache exists.
|
|
"""
|
|
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]
|
|
|
|
if cache_only:
|
|
if os.path.exists(cache_file):
|
|
with open(cache_file) as f:
|
|
return json.load(f) # type: ignore[no-any-return]
|
|
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,
|
|
cache_only: bool = False,
|
|
) -> 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.
|
|
If cache_only=True, never call the API (returns stale or empty data).
|
|
"""
|
|
latlon = trip_latlon(trip)
|
|
if not latlon:
|
|
return {}
|
|
lat, lon = latlon
|
|
try:
|
|
forecasts = get_forecast(data_dir, api_key, lat, lon, cache_only=cache_only)
|
|
except Exception:
|
|
return {}
|
|
return {f["date"]: f for f in forecasts}
|