kuh-no-bowls/weather_logger.py

#!/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()