""" 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}