#!/usr/bin/env python3 """Fetch hourly weather for Knoebels (Elysburg, PA) from Open-Meteo and upsert into knoebels."LZ_open_meteo_hourly". Two modes: (default) recent -- Forecast API with past_days; keeps the table current and self-heals recent gaps. This is what the daily cron runs. --backfill -- Archive (ERA5 reanalysis) API back to --start (default 2024-05-15, when collection began). Timestamps are stored as naive America/New_York to match how the rest of the knoebels schema records time. Upserts are idempotent via ON CONFLICT (time), so rerunning overwrites provisional values with better data as it arrives. """ import argparse import logging import os import sys from datetime import date, datetime import psycopg2 import requests from psycopg2.extras import execute_values # Knoebels Amusement Resort, Elysburg PA LATITUDE = 40.7906 LONGITUDE = -76.4847 COLLECTION_START = "2024-05-15" # first day of queue-time collection TIMEZONE = "America/New_York" ARCHIVE_ENDPOINT = "https://archive-api.open-meteo.com/v1/archive" FORECAST_ENDPOINT = "https://api.open-meteo.com/v1/forecast" # Hourly variables available identically in both the archive and forecast APIs. HOURLY_VARS = [ "temperature_2m", "relative_humidity_2m", "apparent_temperature", "precipitation", "rain", "weather_code", "cloud_cover", "wind_speed_10m", "wind_gusts_10m", ] COMMON_PARAMS = { "latitude": LATITUDE, "longitude": LONGITUDE, "hourly": ",".join(HOURLY_VARS), "temperature_unit": "fahrenheit", "precipitation_unit": "inch", "wind_speed_unit": "mph", "timezone": TIMEZONE, } LOG_PATH = os.environ.get("LOG_PATH") # unset -> log to stderr DB_PASSWORD = os.environ.get("DB_PASSWORD") if not DB_PASSWORD: sys.stderr.write("ERROR: DB_PASSWORD env var is required\n") sys.exit(1) db_params = { "dbname": os.environ.get("DB_NAME", "gp0"), "user": os.environ.get("DB_USER", "kuhnobowls"), "password": DB_PASSWORD, "host": os.environ.get("DB_HOST", "192.168.88.9"), "port": os.environ.get("DB_PORT", "5432"), "options": "-c search_path=knoebels", } CREATE_TABLE = """ CREATE TABLE IF NOT EXISTS knoebels."LZ_open_meteo_hourly" ( "time" timestamp without time zone PRIMARY KEY, temperature_2m real, relative_humidity_2m smallint, apparent_temperature real, precipitation real, rain real, weather_code smallint, cloud_cover smallint, wind_speed_10m real, wind_gusts_10m real, inserted_at timestamp without time zone DEFAULT now() ) """ UPSERT = """ INSERT INTO knoebels."LZ_open_meteo_hourly" ("time", temperature_2m, relative_humidity_2m, apparent_temperature, precipitation, rain, weather_code, cloud_cover, wind_speed_10m, wind_gusts_10m) VALUES %s ON CONFLICT ("time") DO UPDATE SET temperature_2m = EXCLUDED.temperature_2m, relative_humidity_2m = EXCLUDED.relative_humidity_2m, apparent_temperature = EXCLUDED.apparent_temperature, precipitation = EXCLUDED.precipitation, rain = EXCLUDED.rain, weather_code = EXCLUDED.weather_code, cloud_cover = EXCLUDED.cloud_cover, wind_speed_10m = EXCLUDED.wind_speed_10m, wind_gusts_10m = EXCLUDED.wind_gusts_10m, inserted_at = now() """ def fetch(endpoint, extra): params = dict(COMMON_PARAMS, **extra) resp = requests.get(endpoint, params=params, timeout=60) resp.raise_for_status() return resp.json() def rows_from_payload(payload): """Turn an Open-Meteo payload into row tuples, dropping all-null hours (the archive API pads the last few days with nulls until ERA5 lands).""" hourly = payload["hourly"] times = hourly["time"] cols = [hourly[var] for var in HOURLY_VARS] rows = [] for i, t in enumerate(times): values = [col[i] for col in cols] if all(v is None for v in values): continue rows.append((datetime.fromisoformat(t), *values)) return rows def upsert(conn, rows): if not rows: return 0 with conn.cursor() as cur: execute_values(cur, UPSERT, rows, page_size=1000) conn.commit() return len(rows) def run_recent(conn, past_days, forecast_days): payload = fetch(FORECAST_ENDPOINT, {"past_days": past_days, "forecast_days": forecast_days}) rows = rows_from_payload(payload) n = upsert(conn, rows) span = f"{rows[0][0]} .. {rows[-1][0]}" if rows else "(none)" logging.info("recent: upserted %d hours [%s]", n, span) return n def run_backfill(conn, start): end = date.today().isoformat() payload = fetch(ARCHIVE_ENDPOINT, {"start_date": start, "end_date": end}) rows = rows_from_payload(payload) n = upsert(conn, rows) span = f"{rows[0][0]} .. {rows[-1][0]}" if rows else "(none)" logging.info("backfill: upserted %d hours [%s]", n, span) return n def main(): parser = argparse.ArgumentParser(description="Open-Meteo -> knoebels weather logger") parser.add_argument("--backfill", action="store_true", help="pull the full ERA5 archive history before the recent sync") parser.add_argument("--start", default=COLLECTION_START, help="backfill start date (YYYY-MM-DD)") parser.add_argument("--past-days", type=int, default=7, help="forecast-API lookback for the recent sync (max 92)") parser.add_argument("--forecast-days", type=int, default=1, help="forecast-API lookahead for the recent sync (0-16)") args = parser.parse_args() logging.basicConfig( format="%(asctime)s %(levelname)s %(message)s", level=logging.INFO, filename=LOG_PATH, filemode="a" if LOG_PATH else None, ) logging.info("-" * 80) with psycopg2.connect(**db_params) as conn: with conn.cursor() as cur: cur.execute(CREATE_TABLE) conn.commit() if args.backfill: run_backfill(conn, args.start) # Always finish with a recent sync so the ERA5 gap (last ~5 days) is # filled and the table is current. run_recent(conn, args.past_days, args.forecast_days) if __name__ == "__main__": main()