Add Open-Meteo hourly weather pipeline + queue/weather view
Backfills knoebels."LZ_open_meteo_hourly" from 2024-05-15 (start of queue
collection) and keeps it current. Imperial units; timestamps stored as naive
America/New_York to line up with the rest of the knoebels schema.
- weather_logger.py: --backfill uses the ERA5 Archive API; default mode does a
self-healing recent sync via the Forecast API (past_days=7, upsert on conflict)
- docker-compose.yml: add `weather` oneshot service (also align scraper to its
live `oneshot` profile)
- systemd/knb-weather.{service,timer}: nox user timer, every 6h (linger enabled)
- queries/dm_knb_queue_weather.sql: dm_knb_queue_weather view joining the full
queue-time series to hourly weather (hour-bucketed), with a WMO code decode
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,193 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user