jardin/station_meteo/update_station_db.py

#!/usr/bin/env python3
"""Mise a jour ponctuelle de la table meteostation depuis la station locale.

Script autonome (hors webapp) qui lit:
- le flux RSS courant de la station
- la page HTML de la station (donnees enrichies)
- le fichier NOAA mensuel pour une date cible

Puis ecrit dans la base SQLite:
- 1 ligne type="current" (heure observee arrondie a l'heure)
- 1 ligne type="veille" (date cible a T00:00), sauf si --current-only
"""

from __future__ import annotations

import argparse
import json
import os
import sqlite3
from datetime import datetime, timedelta
from email.utils import parsedate_to_datetime
from pathlib import Path
from typing import Any

from local_station_weather import (
    fetch_text,
    parse_current_from_rss,
    parse_daily_summary_from_rss,
    parse_station_page,
    parse_yesterday_from_noaa,
)


def _first_not_none(*values: Any) -> Any:
    for v in values:
        if v is not None:
            return v
    return None


def _to_kmh(value_m_s: float | None) -> float | None:
    if value_m_s is None:
        return None
    return round(value_m_s * 3.6, 1)


def _deg_to_dir(deg: int | float | None) -> str | None:
    if deg is None:
        return None
    dirs = ["N", "NE", "E", "SE", "S", "SO", "O", "NO"]
    idx = int((float(deg) + 22.5) // 45) % 8
    return dirs[idx]


def _parse_observed_hour(observed_at: str | None) -> str:
    if observed_at:
        try:
            dt = parsedate_to_datetime(observed_at)
            if dt.tzinfo:
                dt = dt.astimezone()
            dt = dt.replace(minute=0, second=0, microsecond=0)
            return dt.strftime("%Y-%m-%dT%H:00")
        except Exception:
            pass
    now = datetime.now().replace(minute=0, second=0, microsecond=0)
    return now.strftime("%Y-%m-%dT%H:00")


def _target_date(date_arg: str | None) -> datetime:
    if date_arg:
        return datetime.strptime(date_arg, "%Y-%m-%d")
    return datetime.now() - timedelta(days=1)


def _build_current_row(base_url: str) -> dict[str, Any]:
    base = base_url.rstrip("/") + "/"
    rss_url = f"{base}rss.xml"
    station_page_url = base

    rss_xml = fetch_text(rss_url)
    current = parse_current_from_rss(rss_xml)
    daily = parse_daily_summary_from_rss(rss_xml)

    station_html = fetch_text(station_page_url)
    station_extra = parse_station_page(station_html)
    ext = station_extra.get("current_extended", {})
    stats_today = station_extra.get("stats_today", {})

    wind_deg = _first_not_none(current.get("wind_dir_deg"), ext.get("wind_dir_deg"))
    vent_dir = _first_not_none(_deg_to_dir(wind_deg), ext.get("wind_dir_text"))

    wind_m_s = _first_not_none(current.get("wind_speed_m_s"), ext.get("wind_speed_m_s"))

    return {
        "date_heure": _parse_observed_hour(current.get("observed_at")),
        "type": "current",
        "temp_ext": _first_not_none(current.get("temperature_ext_c"), ext.get("temperature_ext_c")),
        "t_min": _first_not_none(daily.get("temp_ext_min_c"), stats_today.get("temp_ext_min_c")),
        "t_max": _first_not_none(daily.get("temp_ext_max_c"), stats_today.get("temp_ext_max_c")),
        "temp_int": _first_not_none(current.get("temperature_int_c"), ext.get("temperature_int_c")),
        "humidite": _first_not_none(current.get("humidity_ext_pct"), ext.get("humidity_ext_pct")),
        "pression": _first_not_none(current.get("pressure_mbar"), ext.get("pressure_mbar")),
        "pluie_mm": _first_not_none(current.get("rain_mm"), ext.get("rain_today_mm")),
        "vent_kmh": _to_kmh(wind_m_s),
        "vent_dir": vent_dir,
        "uv": ext.get("uv_index"),
        "solaire": ext.get("solar_radiation_w_m2"),
    }


def _build_day_row(base_url: str, target: datetime) -> dict[str, Any] | None:
    base = base_url.rstrip("/") + "/"
    noaa_url = f"{base}NOAA/NOAA-{target.year}-{target.month:02d}.txt"
    noaa_text = fetch_text(noaa_url)
    day_data = parse_yesterday_from_noaa(noaa_text, target.day)
    if "error" in day_data:
        return None

    return {
        "date_heure": target.strftime("%Y-%m-%dT00:00"),
        "type": "veille",
        "temp_ext": day_data.get("temp_mean_c"),
        "t_min": day_data.get("temp_min_c"),
        "t_max": day_data.get("temp_max_c"),
        "temp_int": None,
        "humidite": None,
        "pression": None,
        "pluie_mm": day_data.get("rain_mm"),
        "vent_kmh": _to_kmh(day_data.get("wind_max_m_s")),
        "vent_dir": _deg_to_dir(day_data.get("wind_dom_dir_deg")),
        "uv": None,
        "solaire": None,
    }


def _upsert_row(conn: sqlite3.Connection, row: dict[str, Any]) -> None:
    conn.execute(
        """
        INSERT INTO meteostation (
          date_heure, type, temp_ext, t_min, t_max, temp_int, humidite, pression,
          pluie_mm, vent_kmh, vent_dir, uv, solaire
        ) VALUES (
          :date_heure, :type, :temp_ext, :t_min, :t_max, :temp_int, :humidite, :pression,
          :pluie_mm, :vent_kmh, :vent_dir, :uv, :solaire
        )
        ON CONFLICT(date_heure) DO UPDATE SET
          type=excluded.type,
          temp_ext=excluded.temp_ext,
          t_min=excluded.t_min,
          t_max=excluded.t_max,
          temp_int=excluded.temp_int,
          humidite=excluded.humidite,
          pression=excluded.pression,
          pluie_mm=excluded.pluie_mm,
          vent_kmh=excluded.vent_kmh,
          vent_dir=excluded.vent_dir,
          uv=excluded.uv,
          solaire=excluded.solaire
        """,
        row,
    )


def _assert_db_writable(db_path: Path) -> None:
    if not db_path.exists():
        raise FileNotFoundError(f"Base introuvable: {db_path}")
    if not db_path.is_file():
        raise RuntimeError(f"Chemin de base invalide (pas un fichier): {db_path}")
    if not os.access(db_path, os.R_OK):
        raise PermissionError(f"Pas de lecture sur la base: {db_path}")
    if not os.access(db_path, os.W_OK):
        raise PermissionError(
            f"Pas d'ecriture sur la base: {db_path}. "
            "Lance le script avec un utilisateur qui a les droits (ou dans le conteneur backend)."
        )


def main() -> int:
    parser = argparse.ArgumentParser(description="Met a jour la table meteostation depuis la station locale.")
    parser.add_argument("--base", default="http://10.0.0.8:8081/", help="URL de base de la station locale")
    parser.add_argument("--db", default="data/jardin.db", help="Chemin SQLite (defaut: data/jardin.db)")
    parser.add_argument("--date", help="Date NOAA cible (YYYY-MM-DD). Defaut: veille")
    parser.add_argument("--current-only", action="store_true", help="Met a jour uniquement la ligne current")
    parser.add_argument("--dry-run", action="store_true", help="N ecrit pas en base, affiche seulement le payload")
    args = parser.parse_args()

    db_path = Path(args.db).expanduser().resolve()
    if not args.dry_run:
        _assert_db_writable(db_path)

    target = _target_date(args.date)
    current_row = _build_current_row(args.base)
    day_row = None if args.current_only else _build_day_row(args.base, target)

    payload = {"current": current_row, "day_data": day_row, "target_date": target.strftime("%Y-%m-%d")}
    print(json.dumps(payload, ensure_ascii=False, indent=2))

    if args.dry_run:
        return 0

    conn: sqlite3.Connection | None = None
    try:
        conn = sqlite3.connect(str(db_path))
        with conn:
            _upsert_row(conn, current_row)
            if day_row is not None:
                _upsert_row(conn, day_row)
    finally:
        if conn is not None:
            conn.close()

    print(f"\nOK: base mise a jour -> {db_path}")
    print(f"- current: {current_row['date_heure']}")
    if day_row is not None:
        print(f"- veille: {day_row['date_heure']}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())