ohmu/dev/weather-collector/collector.py

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