Add weather sync job, API endpoints and settings UI
Fetches daily ERA5 weather from Open-Meteo (no API key). Configurable location, timezone and sync time via Settings. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
fc15bc5d9a
commit
5fbf155f62
4 changed files with 671 additions and 1 deletions
213
backend/jobs/weather_sync.py
Normal file
213
backend/jobs/weather_sync.py
Normal file
|
|
@ -0,0 +1,213 @@
|
|||
"""
|
||||
Weather data sync - fetches daily weather from Open-Meteo archive API (ERA5).
|
||||
Location: Stow on the Wold, GL54 1JX (lat=51.9253, lon=-1.7272, elev=230m).
|
||||
No API key required. Stores one row per day, upserts on re-fetch.
|
||||
"""
|
||||
import logging
|
||||
from datetime import date, timedelta
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
from dateutil.relativedelta import relativedelta
|
||||
from sqlalchemy import text
|
||||
|
||||
from database import SyncSessionLocal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
OPEN_METEO_URL = "https://archive-api.open-meteo.com/v1/archive"
|
||||
|
||||
# Fallback defaults — overridden by system_config at runtime
|
||||
_DEFAULT_LATITUDE = 51.9253
|
||||
_DEFAULT_LONGITUDE = -1.7272
|
||||
_DEFAULT_TIMEZONE = "Europe/London"
|
||||
|
||||
DAILY_VARIABLES = [
|
||||
"temperature_2m_max",
|
||||
"temperature_2m_min",
|
||||
"temperature_2m_mean",
|
||||
"precipitation_sum",
|
||||
"rain_sum",
|
||||
"snowfall_sum",
|
||||
"snow_depth_max",
|
||||
"windspeed_10m_max",
|
||||
"sunshine_duration",
|
||||
"weathercode",
|
||||
]
|
||||
|
||||
|
||||
async def sync_weather_data(
|
||||
from_date: date,
|
||||
to_date: date,
|
||||
triggered_by: str = "scheduler"
|
||||
) -> dict:
|
||||
"""Fetch daily weather from Open-Meteo and upsert into weather_data."""
|
||||
logger.info(f"Starting weather sync {from_date} to {to_date}")
|
||||
|
||||
db = SyncSessionLocal()
|
||||
try:
|
||||
# Read location config from system_config (falls back to defaults if not set)
|
||||
cfg_result = db.execute(
|
||||
text("""
|
||||
SELECT config_key, config_value FROM system_config
|
||||
WHERE config_key IN ('weather_latitude', 'weather_longitude', 'weather_timezone')
|
||||
""")
|
||||
)
|
||||
cfg = {row.config_key: row.config_value for row in cfg_result.fetchall()}
|
||||
latitude = float(cfg.get("weather_latitude") or _DEFAULT_LATITUDE)
|
||||
longitude = float(cfg.get("weather_longitude") or _DEFAULT_LONGITUDE)
|
||||
timezone = cfg.get("weather_timezone") or _DEFAULT_TIMEZONE
|
||||
|
||||
db.execute(
|
||||
text("""
|
||||
INSERT INTO sync_log (sync_type, source, started_at, status, date_from, date_to, triggered_by)
|
||||
VALUES ('weather', 'open_meteo', NOW(), 'running', :from_date, :to_date, :triggered_by)
|
||||
"""),
|
||||
{"from_date": from_date, "to_date": to_date, "triggered_by": triggered_by}
|
||||
)
|
||||
db.commit()
|
||||
|
||||
params = {
|
||||
"latitude": latitude,
|
||||
"longitude": longitude,
|
||||
"start_date": from_date.isoformat(),
|
||||
"end_date": to_date.isoformat(),
|
||||
"daily": ",".join(DAILY_VARIABLES),
|
||||
"timezone": timezone,
|
||||
}
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
response = await client.get(OPEN_METEO_URL, params=params)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
daily = data.get("daily", {})
|
||||
dates = daily.get("time", [])
|
||||
|
||||
if not dates:
|
||||
raise Exception("No data returned from Open-Meteo API")
|
||||
|
||||
records_upserted = 0
|
||||
|
||||
for i, date_str in enumerate(dates):
|
||||
def val(key, idx=i):
|
||||
v = daily.get(key, [])
|
||||
return v[idx] if idx < len(v) else None
|
||||
|
||||
sunshine_s = val("sunshine_duration")
|
||||
sunshine_h = round(sunshine_s / 3600, 2) if sunshine_s is not None else None
|
||||
|
||||
db.execute(
|
||||
text("""
|
||||
INSERT INTO weather_data (
|
||||
date,
|
||||
temperature_max, temperature_min, temperature_mean,
|
||||
precipitation_sum, rain_sum, snowfall_sum, snow_depth_max,
|
||||
windspeed_max, sunshine_hours, weathercode,
|
||||
fetched_at
|
||||
) VALUES (
|
||||
:date,
|
||||
:temp_max, :temp_min, :temp_mean,
|
||||
:precip, :rain, :snowfall, :snow_depth,
|
||||
:wind, :sunshine, :weathercode,
|
||||
NOW()
|
||||
)
|
||||
ON CONFLICT (date) DO UPDATE SET
|
||||
temperature_max = :temp_max,
|
||||
temperature_min = :temp_min,
|
||||
temperature_mean = :temp_mean,
|
||||
precipitation_sum = :precip,
|
||||
rain_sum = :rain,
|
||||
snowfall_sum = :snowfall,
|
||||
snow_depth_max = :snow_depth,
|
||||
windspeed_max = :wind,
|
||||
sunshine_hours = :sunshine,
|
||||
weathercode = :weathercode,
|
||||
fetched_at = NOW()
|
||||
"""),
|
||||
{
|
||||
"date": date_str,
|
||||
"temp_max": val("temperature_2m_max"),
|
||||
"temp_min": val("temperature_2m_min"),
|
||||
"temp_mean": val("temperature_2m_mean"),
|
||||
"precip": val("precipitation_sum"),
|
||||
"rain": val("rain_sum"),
|
||||
"snowfall": val("snowfall_sum"),
|
||||
"snow_depth": val("snow_depth_max"),
|
||||
"wind": val("windspeed_10m_max"),
|
||||
"sunshine": sunshine_h,
|
||||
"weathercode": val("weathercode"),
|
||||
}
|
||||
)
|
||||
records_upserted += 1
|
||||
|
||||
db.commit()
|
||||
|
||||
db.execute(
|
||||
text("""
|
||||
UPDATE sync_log
|
||||
SET completed_at = NOW(), status = 'success',
|
||||
records_fetched = :fetched, records_created = :created
|
||||
WHERE id = (
|
||||
SELECT id FROM sync_log
|
||||
WHERE source = 'open_meteo' AND status = 'running'
|
||||
ORDER BY started_at DESC LIMIT 1
|
||||
)
|
||||
"""),
|
||||
{"fetched": len(dates), "created": records_upserted}
|
||||
)
|
||||
db.commit()
|
||||
|
||||
logger.info(f"Weather sync completed: {records_upserted} days upserted")
|
||||
return {"records_fetched": len(dates), "records_upserted": records_upserted}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Weather sync failed: {e}")
|
||||
try:
|
||||
db.rollback()
|
||||
db.execute(
|
||||
text("""
|
||||
UPDATE sync_log
|
||||
SET completed_at = NOW(), status = 'failed', error_message = :error
|
||||
WHERE id = (
|
||||
SELECT id FROM sync_log
|
||||
WHERE source = 'open_meteo' AND status = 'running'
|
||||
ORDER BY started_at DESC LIMIT 1
|
||||
)
|
||||
"""),
|
||||
{"error": str(e)[:500]}
|
||||
)
|
||||
db.commit()
|
||||
except Exception as log_error:
|
||||
logger.error(f"Failed to update sync_log: {log_error}")
|
||||
raise
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
async def run_weather_sync(triggered_by: str = "scheduler"):
|
||||
"""Daily scheduled sync — last 7 days to catch ERA5 corrections."""
|
||||
from_date = date.today() - timedelta(days=7)
|
||||
to_date = date.today()
|
||||
await sync_weather_data(from_date, to_date, triggered_by=triggered_by)
|
||||
|
||||
|
||||
async def run_weather_backfill(
|
||||
from_date: date,
|
||||
to_date: date,
|
||||
triggered_by: str = "manual"
|
||||
) -> dict:
|
||||
"""Backfill in 6-month chunks so progress is logged incrementally."""
|
||||
logger.info(f"Starting weather backfill {from_date} to {to_date}")
|
||||
total_upserted = 0
|
||||
|
||||
chunk_start = from_date
|
||||
while chunk_start <= to_date:
|
||||
chunk_end = min(chunk_start + relativedelta(months=6) - timedelta(days=1), to_date)
|
||||
logger.info(f"Weather backfill chunk: {chunk_start} to {chunk_end}")
|
||||
result = await sync_weather_data(chunk_start, chunk_end, triggered_by=triggered_by)
|
||||
total_upserted += result.get("records_upserted", 0)
|
||||
chunk_start = chunk_end + timedelta(days=1)
|
||||
|
||||
logger.info(f"Weather backfill complete: {total_upserted} total days")
|
||||
return {"records_upserted": total_upserted}
|
||||
Loading…
Add table
Add a link
Reference in a new issue