394 lines
11 KiB
Python
394 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import datetime
|
||
import html
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
from logging.handlers import RotatingFileHandler
|
||
from typing import Dict, List, Optional, Tuple
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import requests
|
||
from dateutil import parser
|
||
|
||
# =============================================================================
|
||
# morning_weather_report.py
|
||
#
|
||
# Report meteo giornaliero (08:00) per Casa (San Marino):
|
||
# - prossime 24h e 48h: Tmin/Tmax, nuvolosità, vento/raffiche, precipitazioni, fenomeni
|
||
#
|
||
# Telegram:
|
||
# - nessun token in chiaro (env oppure ~/.telegram_dpc_bot_token oppure /etc/telegram_dpc_bot_token)
|
||
# - invia SOLO il report (niente notifiche errori)
|
||
#
|
||
# Log:
|
||
# - ./morning_weather_report.log
|
||
# - DEBUG=1 -> anche su stdout
|
||
# =============================================================================
|
||
|
||
DEBUG = os.environ.get("DEBUG", "0").strip() == "1"
|
||
|
||
# --- POSIZIONE CASA (Strada Cà Toro, 12 - San Marino) ---
|
||
LAT = 43.9356
|
||
LON = 12.4296
|
||
LOCATION_NAME = "🏠 Casa (Strada Cà Toro)"
|
||
|
||
# --- TIMEZONE ---
|
||
TZ = "Europe/Rome"
|
||
TZINFO = ZoneInfo(TZ)
|
||
|
||
# --- TELEGRAM (multi-chat) ---
|
||
TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
|
||
TOKEN_FILE_HOME = os.path.expanduser("~/.telegram_dpc_bot_token")
|
||
TOKEN_FILE_ETC = "/etc/telegram_dpc_bot_token"
|
||
|
||
# --- OPEN-METEO ---
|
||
OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
|
||
MODEL = "meteofrance_arome_france_hd"
|
||
HTTP_HEADERS = {"User-Agent": "rpi-morning-weather-report/1.0"}
|
||
|
||
# --- REPORT WINDOW ---
|
||
HOURS_1 = 24
|
||
HOURS_2 = 48 # includiamo fino a 48h, poi dividiamo 0–24 e 24–48
|
||
|
||
# --- LOG FILE ---
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
LOG_FILE = os.path.join(BASE_DIR, "morning_weather_report.log")
|
||
|
||
|
||
def setup_logger() -> logging.Logger:
|
||
logger = logging.getLogger("morning_weather_report")
|
||
logger.setLevel(logging.DEBUG if DEBUG else logging.INFO)
|
||
logger.handlers.clear()
|
||
|
||
fh = RotatingFileHandler(LOG_FILE, maxBytes=1_000_000, backupCount=5, encoding="utf-8")
|
||
fh.setLevel(logging.DEBUG)
|
||
fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
|
||
fh.setFormatter(fmt)
|
||
logger.addHandler(fh)
|
||
|
||
if DEBUG:
|
||
sh = logging.StreamHandler()
|
||
sh.setLevel(logging.DEBUG)
|
||
sh.setFormatter(fmt)
|
||
logger.addHandler(sh)
|
||
|
||
return logger
|
||
|
||
|
||
LOGGER = setup_logger()
|
||
|
||
|
||
def now_local() -> datetime.datetime:
|
||
return datetime.datetime.now(TZINFO)
|
||
|
||
|
||
def read_text_file(path: str) -> str:
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return f.read().strip()
|
||
except FileNotFoundError:
|
||
return ""
|
||
except PermissionError:
|
||
LOGGER.debug("Permission denied reading %s", path)
|
||
return ""
|
||
except Exception as e:
|
||
LOGGER.exception("Error reading %s: %s", path, e)
|
||
return ""
|
||
|
||
|
||
def load_bot_token() -> str:
|
||
tok = (os.environ.get("TELEGRAM_BOT_TOKEN") or "").strip()
|
||
if tok:
|
||
return tok
|
||
tok = (os.environ.get("BOT_TOKEN") or "").strip()
|
||
if tok:
|
||
return tok
|
||
tok = read_text_file(TOKEN_FILE_HOME)
|
||
if tok:
|
||
return tok
|
||
tok = read_text_file(TOKEN_FILE_ETC)
|
||
return tok.strip() if tok else ""
|
||
|
||
|
||
def parse_time_to_local(t: str) -> datetime.datetime:
|
||
dt = parser.isoparse(t)
|
||
if dt.tzinfo is None:
|
||
return dt.replace(tzinfo=TZINFO)
|
||
return dt.astimezone(TZINFO)
|
||
|
||
|
||
def telegram_send_markdown(message: str) -> bool:
|
||
"""
|
||
Invia il report. In caso di errori, SOLO log.
|
||
"""
|
||
token = load_bot_token()
|
||
if not token:
|
||
LOGGER.error("Telegram token missing: report NOT sent.")
|
||
return False
|
||
|
||
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||
payload_base = {
|
||
"text": message,
|
||
"parse_mode": "Markdown",
|
||
"disable_web_page_preview": True,
|
||
}
|
||
|
||
ok_any = False
|
||
with requests.Session() as s:
|
||
for chat_id in TELEGRAM_CHAT_IDS:
|
||
payload = dict(payload_base)
|
||
payload["chat_id"] = chat_id
|
||
try:
|
||
r = s.post(url, json=payload, timeout=20)
|
||
if r.status_code == 200:
|
||
ok_any = True
|
||
else:
|
||
LOGGER.error("Telegram HTTP %s chat_id=%s body=%s", r.status_code, chat_id, r.text[:500])
|
||
time.sleep(0.25)
|
||
except Exception as e:
|
||
LOGGER.exception("Telegram send exception chat_id=%s err=%s", chat_id, e)
|
||
|
||
return ok_any
|
||
|
||
|
||
# --- Weathercode categories (Open-Meteo / WMO-style codes) ---
|
||
def code_to_category(code: int) -> Optional[Tuple[str, str]]:
|
||
"""
|
||
Ritorna (emoji, label) per fenomeni "rilevanti" da elencare.
|
||
Non elenchiamo 'sereno/parzialmente nuvoloso' come fenomeno.
|
||
"""
|
||
if code is None:
|
||
return None
|
||
|
||
try:
|
||
c = int(code)
|
||
except Exception:
|
||
return None
|
||
|
||
# Nebbia
|
||
if c in (45, 48):
|
||
return ("🌫️", "nebbia")
|
||
|
||
# Pioviggine
|
||
if 51 <= c <= 57:
|
||
return ("🌦️", "pioviggine")
|
||
|
||
# Pioggia / gelicidio
|
||
if 61 <= c <= 65:
|
||
return ("🌧️", "pioggia")
|
||
if c in (66, 67):
|
||
return ("🧊", "gelicidio")
|
||
|
||
# Neve
|
||
if 71 <= c <= 77:
|
||
return ("❄️", "neve")
|
||
|
||
# Rovesci
|
||
if 80 <= c <= 82:
|
||
return ("🌧️", "rovesci")
|
||
if c in (85, 86):
|
||
return ("❄️", "rovesci di neve")
|
||
|
||
# Temporali / grandine
|
||
if c == 95:
|
||
return ("⛈️", "temporali")
|
||
if c in (96, 99):
|
||
return ("⛈️", "temporali con grandine")
|
||
|
||
return None
|
||
|
||
|
||
def get_forecast() -> Optional[Dict]:
|
||
params = {
|
||
"latitude": LAT,
|
||
"longitude": LON,
|
||
"timezone": TZ,
|
||
"forecast_days": 3, # copre bene 48h
|
||
"models": MODEL,
|
||
"wind_speed_unit": "kmh",
|
||
"precipitation_unit": "mm",
|
||
"hourly": ",".join([
|
||
"temperature_2m",
|
||
"cloudcover",
|
||
"windspeed_10m",
|
||
"windgusts_10m",
|
||
"precipitation",
|
||
"weathercode",
|
||
]),
|
||
}
|
||
|
||
try:
|
||
r = requests.get(OPEN_METEO_URL, params=params, headers=HTTP_HEADERS, timeout=25)
|
||
if r.status_code == 400:
|
||
try:
|
||
j = r.json()
|
||
LOGGER.error("Open-Meteo 400: %s", j.get("reason", j))
|
||
except Exception:
|
||
LOGGER.error("Open-Meteo 400: %s", r.text[:500])
|
||
return None
|
||
r.raise_for_status()
|
||
return r.json()
|
||
except Exception as e:
|
||
LOGGER.exception("Open-Meteo request error: %s", e)
|
||
return None
|
||
|
||
|
||
def window_indices(times: List[str], start: datetime.datetime, end: datetime.datetime) -> List[int]:
|
||
idx = []
|
||
for i, t in enumerate(times):
|
||
try:
|
||
dt = parse_time_to_local(t)
|
||
except Exception:
|
||
continue
|
||
if dt >= start and dt < end:
|
||
idx.append(i)
|
||
return idx
|
||
|
||
|
||
def summarize_window(hourly: Dict, idx: List[int]) -> Dict:
|
||
def safe_list(key: str) -> List[float]:
|
||
arr = hourly.get(key, []) or []
|
||
out = []
|
||
for i in idx:
|
||
try:
|
||
v = arr[i]
|
||
out.append(float(v) if v is not None else 0.0)
|
||
except Exception:
|
||
out.append(0.0)
|
||
return out
|
||
|
||
temps = safe_list("temperature_2m")
|
||
clouds = safe_list("cloudcover")
|
||
wind = safe_list("windspeed_10m")
|
||
gust = safe_list("windgusts_10m")
|
||
prec = safe_list("precipitation")
|
||
|
||
# weathercode: manteniamo int
|
||
wcodes_raw = (hourly.get("weathercode", []) or [])
|
||
wcodes = []
|
||
for i in idx:
|
||
try:
|
||
wcodes.append(int(wcodes_raw[i]))
|
||
except Exception:
|
||
wcodes.append(-1)
|
||
|
||
# Base stats
|
||
tmin = min(temps) if temps else None
|
||
tmax = max(temps) if temps else None
|
||
|
||
cavg = (sum(clouds) / len(clouds)) if clouds else None
|
||
cmax = max(clouds) if clouds else None
|
||
|
||
wmax = max(wind) if wind else None
|
||
gmax = max(gust) if gust else None
|
||
|
||
psum = sum(prec) if prec else 0.0
|
||
pmax = max(prec) if prec else 0.0
|
||
|
||
# Fenomeni (categorie uniche con prima occorrenza)
|
||
phenomena: List[str] = []
|
||
seen = set()
|
||
for c in wcodes:
|
||
cat = code_to_category(c)
|
||
if not cat:
|
||
continue
|
||
em, label = cat
|
||
key = label
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
phenomena.append(f"{em} {label}")
|
||
|
||
# Se nessuna categoria ma piove un po' (precip) e code non classificato
|
||
if not phenomena and psum > 0.0:
|
||
phenomena.append("🌧️ precipitazioni")
|
||
|
||
return {
|
||
"tmin": tmin, "tmax": tmax,
|
||
"cavg": cavg, "cmax": cmax,
|
||
"wmax": wmax, "gmax": gmax,
|
||
"psum": psum, "pmax": pmax,
|
||
"phenomena": phenomena,
|
||
}
|
||
|
||
|
||
def fmt_num(v: Optional[float], dec: int = 1) -> str:
|
||
if v is None:
|
||
return "—"
|
||
try:
|
||
return f"{v:.{dec}f}"
|
||
except Exception:
|
||
return "—"
|
||
|
||
|
||
def build_message(s0: Dict, s1: Dict, generated_at: datetime.datetime) -> str:
|
||
# Nota: Markdown. Manteniamo righe brevi.
|
||
dt_str = generated_at.strftime("%d/%m %H:%M")
|
||
|
||
def block(title: str, s: Dict) -> str:
|
||
phen = ", ".join(s["phenomena"]) if s["phenomena"] else "—"
|
||
return (
|
||
f"*{title}*\n"
|
||
f"🌡️ Tmin/Tmax: `{fmt_num(s['tmin'])}°C` / `{fmt_num(s['tmax'])}°C`\n"
|
||
f"☁️ Nuvolosità: avg `{fmt_num(s['cavg'],0)}%` max `{fmt_num(s['cmax'],0)}%`\n"
|
||
f"💨 Vento/Raff.: `{fmt_num(s['wmax'],0)} km/h` / `{fmt_num(s['gmax'],0)} km/h`\n"
|
||
f"🌧️ Pioggia: `{fmt_num(s['psum'])} mm` (max/h `{fmt_num(s['pmax'])} mm`)\n"
|
||
f"🔎 Fenomeni: {phen}"
|
||
)
|
||
|
||
msg = (
|
||
f"🌤️ *REPORT METEO*\n"
|
||
f"📍 {LOCATION_NAME}\n"
|
||
f"🕗 Agg. `{dt_str}` (modello: `{MODEL}`)\n\n"
|
||
f"{block('0–24h', s0)}\n\n"
|
||
f"{block('24–48h', s1)}\n"
|
||
)
|
||
return msg
|
||
|
||
|
||
def main() -> None:
|
||
LOGGER.info("--- Morning weather report ---")
|
||
|
||
data = get_forecast()
|
||
if not data:
|
||
# errori: solo log
|
||
return
|
||
|
||
hourly = data.get("hourly", {}) or {}
|
||
times = hourly.get("time", []) or []
|
||
if not times:
|
||
LOGGER.error("Open-Meteo: hourly.time missing/empty")
|
||
return
|
||
|
||
now = now_local()
|
||
start0 = now
|
||
end0 = now + datetime.timedelta(hours=HOURS_1)
|
||
start1 = end0
|
||
end1 = now + datetime.timedelta(hours=HOURS_2)
|
||
|
||
idx0 = window_indices(times, start0, end0)
|
||
idx1 = window_indices(times, start1, end1)
|
||
|
||
if not idx0 or not idx1:
|
||
LOGGER.error("Insufficient hourly coverage for windows: idx0=%s idx1=%s", len(idx0), len(idx1))
|
||
return
|
||
|
||
s0 = summarize_window(hourly, idx0)
|
||
s1 = summarize_window(hourly, idx1)
|
||
|
||
msg = build_message(s0, s1, now)
|
||
|
||
ok = telegram_send_markdown(msg)
|
||
if ok:
|
||
LOGGER.info("Weather report sent.")
|
||
else:
|
||
LOGGER.error("Weather report NOT sent (token/telegram error).")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|