|
|
@@ -0,0 +1,281 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+"""Välistemperatuuri koguja (Open-Meteo).
|
|
|
+
|
|
|
+Toob Open-Meteo API-st tunniandmed (temperatuur, niiskus, tuul, sademed)
|
|
|
+asukoha jaoks (vaikimisi Suuremõisa, Hiiumaa) ja kirjutab need ClickHouse'i
|
|
|
+tabelisse ``uus67.valisilm``.
|
|
|
+
|
|
|
+- Käivitusel backfill alates ``BACKFILL_FROM`` (vaikimisi 2026-09-17).
|
|
|
+- Edasi iga ``POLL_INTERVAL`` sekundi järel uuendab viimaseid tunde
|
|
|
+ (Open-Meteo uuendab hiljutisi väärtusi) ja täidab vahelejäänud tunnid.
|
|
|
+- Auke ei jäta: kui mõne tunni andmed puuduvad, kasutab eelmise teadaoleva
|
|
|
+ tunni väärtust ja märgib ``taidetud = true``.
|
|
|
+
|
|
|
+Tabel on ReplacingMergeTree (ORDER BY hour) — sama tunni uuesti kirjutamine
|
|
|
+asendab varasema rea. Päringud peaksid kasutama ``FINAL``.
|
|
|
+"""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import json
|
|
|
+import logging
|
|
|
+import os
|
|
|
+import signal
|
|
|
+import time
|
|
|
+from datetime import date, datetime, timedelta
|
|
|
+from typing import Any
|
|
|
+from zoneinfo import ZoneInfo
|
|
|
+
|
|
|
+import requests
|
|
|
+
|
|
|
+logging.basicConfig(
|
|
|
+ level=logging.INFO,
|
|
|
+ format="%(asctime)s %(levelname)s %(message)s",
|
|
|
+)
|
|
|
+LOG = logging.getLogger("weather-collector")
|
|
|
+
|
|
|
+TZ = ZoneInfo(os.environ.get("TZ_NAME", "Europe/Tallinn"))
|
|
|
+
|
|
|
+LATITUDE = float(os.environ.get("LATITUDE", "58.87009"))
|
|
|
+LONGITUDE = float(os.environ.get("LONGITUDE", "22.9479"))
|
|
|
+BACKFILL_FROM = os.environ.get("BACKFILL_FROM", "2026-09-17")
|
|
|
+POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", "1800"))
|
|
|
+
|
|
|
+FORECAST_URL = "https://api.open-meteo.com/v1/forecast"
|
|
|
+ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive"
|
|
|
+HOURLY_VARS = "temperature_2m,relative_humidity_2m,wind_speed_10m,precipitation"
|
|
|
+
|
|
|
+CH_URL = os.environ.get("CH_URL", "http://clickhouse:8123").rstrip("/")
|
|
|
+CH_USER = os.environ.get("CH_USER", "zigbee")
|
|
|
+CH_PASS = os.environ.get("CH_PASS", "")
|
|
|
+CH_DB = os.environ.get("CH_DB", "uus67")
|
|
|
+CH_TABLE = os.environ.get("CH_WEATHER_TABLE", "valisilm")
|
|
|
+
|
|
|
+MAX_PAST_DAYS = 92
|
|
|
+
|
|
|
+
|
|
|
+def current_hour(now: datetime | None = None) -> datetime:
|
|
|
+ now = now or datetime.now(TZ)
|
|
|
+ return now.replace(minute=0, second=0, microsecond=0)
|
|
|
+
|
|
|
+
|
|
|
+def parse_dt(value: str) -> datetime:
|
|
|
+ return datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=TZ)
|
|
|
+
|
|
|
+
|
|
|
+def _num(value: Any) -> float | None:
|
|
|
+ if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
|
+ return None
|
|
|
+ return float(value)
|
|
|
+
|
|
|
+
|
|
|
+def _parse_hourly(payload: dict) -> dict[datetime, dict[str, float | None]]:
|
|
|
+ hourly = payload.get("hourly") or {}
|
|
|
+ times = hourly.get("time") or []
|
|
|
+ temps = hourly.get("temperature_2m") or []
|
|
|
+ hums = hourly.get("relative_humidity_2m") or []
|
|
|
+ winds = hourly.get("wind_speed_10m") or []
|
|
|
+ rains = hourly.get("precipitation") or []
|
|
|
+ out: dict[datetime, dict[str, float | None]] = {}
|
|
|
+ for i, ts in enumerate(times):
|
|
|
+ dt = datetime.strptime(ts, "%Y-%m-%dT%H:%M").replace(tzinfo=TZ)
|
|
|
+ out[dt] = {
|
|
|
+ "temperature": _num(temps[i]) if i < len(temps) else None,
|
|
|
+ "humidity": _num(hums[i]) if i < len(hums) else None,
|
|
|
+ "wind": _num(winds[i]) if i < len(winds) else None,
|
|
|
+ "rain": _num(rains[i]) if i < len(rains) else None,
|
|
|
+ }
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def fetch_forecast(past_days: int) -> dict[datetime, dict[str, float | None]]:
|
|
|
+ params = {
|
|
|
+ "latitude": LATITUDE,
|
|
|
+ "longitude": LONGITUDE,
|
|
|
+ "hourly": HOURLY_VARS,
|
|
|
+ "timezone": "Europe/Tallinn",
|
|
|
+ "past_days": max(1, min(past_days, MAX_PAST_DAYS)),
|
|
|
+ "forecast_days": 1,
|
|
|
+ "wind_speed_unit": "kmh",
|
|
|
+ }
|
|
|
+ resp = requests.get(FORECAST_URL, params=params, timeout=30)
|
|
|
+ resp.raise_for_status()
|
|
|
+ return _parse_hourly(resp.json())
|
|
|
+
|
|
|
+
|
|
|
+def fetch_archive(start: date, end: date) -> dict[datetime, dict[str, float | None]]:
|
|
|
+ params = {
|
|
|
+ "latitude": LATITUDE,
|
|
|
+ "longitude": LONGITUDE,
|
|
|
+ "start_date": start.isoformat(),
|
|
|
+ "end_date": end.isoformat(),
|
|
|
+ "hourly": HOURLY_VARS,
|
|
|
+ "timezone": "Europe/Tallinn",
|
|
|
+ "wind_speed_unit": "kmh",
|
|
|
+ }
|
|
|
+ resp = requests.get(ARCHIVE_URL, params=params, timeout=30)
|
|
|
+ resp.raise_for_status()
|
|
|
+ return _parse_hourly(resp.json())
|
|
|
+
|
|
|
+
|
|
|
+def gather(start: datetime, now: datetime) -> dict[datetime, dict[str, float | None]]:
|
|
|
+ """Kogu API-andmed vahemikus [start, now] (lõpp kaasatud)."""
|
|
|
+ today = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
+ api: dict[datetime, dict[str, float | None]] = {}
|
|
|
+ if start < today:
|
|
|
+ archive_end = today - timedelta(days=1)
|
|
|
+ try:
|
|
|
+ api.update(fetch_archive(start.date(), archive_end.date()))
|
|
|
+ except Exception:
|
|
|
+ LOG.exception("Open-Meteo archive päring ebaõnnestus")
|
|
|
+ past_days = max(1, (today - start).days + 1)
|
|
|
+ try:
|
|
|
+ api.update(fetch_forecast(past_days))
|
|
|
+ except Exception:
|
|
|
+ LOG.exception("Open-Meteo forecast päring ebaõnnestus")
|
|
|
+ return {h: v for h, v in api.items() if start <= h <= now}
|
|
|
+
|
|
|
+
|
|
|
+def build_rows(
|
|
|
+ api: dict[datetime, dict[str, float | None]],
|
|
|
+ start: datetime,
|
|
|
+ end: datetime,
|
|
|
+) -> list[dict[str, Any]]:
|
|
|
+ """Ehita tunniread, täites puuduvad tunnid eelmise väärtusega."""
|
|
|
+ rows: list[dict[str, Any]] = []
|
|
|
+ last: dict[str, float | None] | None = None
|
|
|
+ hour = start
|
|
|
+ while hour <= end:
|
|
|
+ values = api.get(hour)
|
|
|
+ has_data = values is not None and values.get("temperature") is not None
|
|
|
+ if has_data:
|
|
|
+ last = values
|
|
|
+ rows.append(_row(hour, values, False))
|
|
|
+ elif last is not None:
|
|
|
+ rows.append(_row(hour, last, True))
|
|
|
+ hour += timedelta(hours=1)
|
|
|
+ return rows
|
|
|
+
|
|
|
+
|
|
|
+def _row(hour: datetime, values: dict[str, float | None], taidetud: bool) -> dict[str, Any]:
|
|
|
+ return {
|
|
|
+ "hour": hour.strftime("%Y-%m-%d %H:%M:%S"),
|
|
|
+ "temperature": values.get("temperature"),
|
|
|
+ "humidity": values.get("humidity"),
|
|
|
+ "wind": values.get("wind"),
|
|
|
+ "rain": values.get("rain"),
|
|
|
+ "taidetud": taidetud,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def ch_query(query: str) -> str:
|
|
|
+ resp = requests.post(
|
|
|
+ f"{CH_URL}/",
|
|
|
+ params={"database": CH_DB, "query": query},
|
|
|
+ auth=(CH_USER, CH_PASS),
|
|
|
+ timeout=20,
|
|
|
+ )
|
|
|
+ resp.raise_for_status()
|
|
|
+ return resp.text
|
|
|
+
|
|
|
+
|
|
|
+def load_last_hour() -> datetime | None:
|
|
|
+ try:
|
|
|
+ text = ch_query(
|
|
|
+ f"SELECT toString(max(hour)) AS mh FROM {CH_DB}.{CH_TABLE} FORMAT JSONEachRow"
|
|
|
+ ).strip()
|
|
|
+ if text:
|
|
|
+ obj = json.loads(text.splitlines()[0])
|
|
|
+ if obj.get("mh"):
|
|
|
+ return datetime.strptime(obj["mh"], "%Y-%m-%d %H:%M:%S").replace(tzinfo=TZ)
|
|
|
+ except Exception:
|
|
|
+ LOG.exception("Viimase tunni laadimine ebaõnnestus")
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def insert_rows(rows: list[dict[str, Any]]) -> None:
|
|
|
+ if not rows:
|
|
|
+ return
|
|
|
+ body = "\n".join(json.dumps(r, ensure_ascii=False) for r in rows)
|
|
|
+ resp = requests.post(
|
|
|
+ f"{CH_URL}/",
|
|
|
+ params={
|
|
|
+ "database": CH_DB,
|
|
|
+ "query": f"INSERT INTO {CH_DB}.{CH_TABLE} FORMAT JSONEachRow",
|
|
|
+ },
|
|
|
+ data=body.encode("utf-8"),
|
|
|
+ auth=(CH_USER, CH_PASS),
|
|
|
+ timeout=30,
|
|
|
+ )
|
|
|
+ resp.raise_for_status()
|
|
|
+ LOG.info("ClickHouse'i (%s) kirjutatud %d rida", CH_TABLE, len(rows))
|
|
|
+
|
|
|
+
|
|
|
+def run_once(backfill_start: datetime, full: bool) -> None:
|
|
|
+ now = current_hour()
|
|
|
+ if full:
|
|
|
+ start = backfill_start
|
|
|
+ else:
|
|
|
+ last = load_last_hour()
|
|
|
+ start = backfill_start if last is None else max(backfill_start, last - timedelta(days=2))
|
|
|
+ if start > now:
|
|
|
+ start = now
|
|
|
+ api = gather(start, now)
|
|
|
+ rows = build_rows(api, start, now)
|
|
|
+ insert_rows(rows)
|
|
|
+ LOG.info(
|
|
|
+ "Vahemik %s .. %s, API ridu %d, kirjutatud %d",
|
|
|
+ start.strftime("%Y-%m-%d %H:%M"),
|
|
|
+ now.strftime("%Y-%m-%d %H:%M"),
|
|
|
+ len(api),
|
|
|
+ len(rows),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def main() -> int:
|
|
|
+ running = {"stop": False}
|
|
|
+
|
|
|
+ def shutdown(signum, frame):
|
|
|
+ LOG.info("Lõpetamissignaal (%s), sulgen", signum)
|
|
|
+ running["stop"] = True
|
|
|
+
|
|
|
+ signal.signal(signal.SIGTERM, shutdown)
|
|
|
+ signal.signal(signal.SIGINT, shutdown)
|
|
|
+
|
|
|
+ backfill_start = parse_dt(BACKFILL_FROM).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
+ LOG.info(
|
|
|
+ "Välistemperatuuri koguja: lat=%.5f lon=%.5f, backfill alates %s, intervall %ds",
|
|
|
+ LATITUDE,
|
|
|
+ LONGITUDE,
|
|
|
+ backfill_start.strftime("%Y-%m-%d"),
|
|
|
+ POLL_INTERVAL,
|
|
|
+ )
|
|
|
+
|
|
|
+ # Käivitusel täisbackfill (kuni aegunud aukudeni)
|
|
|
+ try:
|
|
|
+ run_once(backfill_start, full=True)
|
|
|
+ except Exception:
|
|
|
+ LOG.exception("Backfill ebaõnnestus, jätkan pollimisega")
|
|
|
+
|
|
|
+ while not running["stop"]:
|
|
|
+ now = datetime.now(TZ)
|
|
|
+ nxt = now.replace(minute=5, second=0, microsecond=0)
|
|
|
+ if nxt <= now:
|
|
|
+ nxt += timedelta(hours=1)
|
|
|
+ wait_s = min(POLL_INTERVAL, max(30.0, (nxt - now).total_seconds()))
|
|
|
+ waited = 0.0
|
|
|
+ while waited < wait_s and not running["stop"]:
|
|
|
+ time.sleep(min(5.0, wait_s - waited))
|
|
|
+ waited += 5.0
|
|
|
+ if running["stop"]:
|
|
|
+ break
|
|
|
+ try:
|
|
|
+ run_once(backfill_start, full=False)
|
|
|
+ except Exception:
|
|
|
+ LOG.exception("Tunnipoll ebaõnnestus, proovin hiljem uuesti")
|
|
|
+
|
|
|
+ return 0
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ raise SystemExit(main())
|
ardo