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