761 lines
32 KiB
Python
761 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import argparse
|
|
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
|
|
|
|
# =============================================================================
|
|
# student_alert.py
|
|
#
|
|
# Scopo:
|
|
# Notificare lo studente (Bologna) se nelle prossime 24 ore sono previsti:
|
|
# - Neve persistente (>= 2 ore consecutive con snowfall > 0)
|
|
# - Pioggia molto forte persistente (>= 2 ore consecutive con 3h-rolling >= soglia)
|
|
# sia a Bologna (trigger) sia lungo il tragitto (caselli A14) fino a San Marino.
|
|
# =============================================================================
|
|
|
|
DEBUG = os.environ.get("DEBUG", "0").strip() == "1"
|
|
|
|
# ----------------- TELEGRAM -----------------
|
|
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"
|
|
|
|
# ----------------- SOGLIE E LOGICA -----------------
|
|
SOGLIA_PIOGGIA_3H_MM = 30.0 # mm in 3 ore (rolling)
|
|
PERSIST_HOURS = 2 # persistenza minima (ore)
|
|
HOURS_AHEAD = 24
|
|
SNOW_HOURLY_EPS_CM = 0.2 # Soglia minima neve cm/h
|
|
# Codici meteo che indicano neve (WMO)
|
|
SNOW_WEATHER_CODES = [71, 73, 75, 77, 85, 86] # Neve leggera, moderata, forte, granelli, rovesci
|
|
|
|
# File di stato
|
|
STATE_FILE = "/home/daniely/docker/telegram-bot/student_state.json"
|
|
|
|
# ----------------- PUNTI DEL PERCORSO (Caselli A14) -----------------
|
|
POINTS = [
|
|
{"name": "🎓 Bologna (V. Regnoli)", "lat": 44.4930, "lon": 11.3690, "type": "trigger"},
|
|
{"name": "🛣️ Casello Imola", "lat": 44.3798, "lon": 11.7397, "type": "route"},
|
|
{"name": "🛣️ Casello Faenza", "lat": 44.3223, "lon": 11.9040, "type": "route"},
|
|
{"name": "🛣️ Casello Forlì", "lat": 44.2502, "lon": 12.0910, "type": "route"},
|
|
{"name": "🛣️ Casello Cesena", "lat": 44.1675, "lon": 12.2835, "type": "route"},
|
|
{"name": "🛣️ Casello Rimini", "lat": 44.0362, "lon": 12.5659, "type": "route"},
|
|
{"name": "🏠 San Marino", "lat": 43.9356, "lon": 12.4296, "type": "end"},
|
|
]
|
|
|
|
# ----------------- OPEN-METEO -----------------
|
|
OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
|
|
TZ = "Europe/Berlin"
|
|
TZINFO = ZoneInfo(TZ)
|
|
HTTP_HEADERS = {"User-Agent": "rpi-student-alert/2.0"}
|
|
MODEL_AROME = "meteofrance_seamless"
|
|
MODEL_ICON_IT = "italia_meteo_arpae_icon_2i"
|
|
COMPARISON_THRESHOLD = 0.30 # 30% scostamento per comparazione
|
|
|
|
# ----------------- LOG -----------------
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
LOG_FILE = os.path.join(BASE_DIR, "student_alert.log")
|
|
|
|
|
|
def setup_logger() -> logging.Logger:
|
|
logger = logging.getLogger("student_alert")
|
|
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()
|
|
|
|
|
|
# =============================================================================
|
|
# Utility
|
|
# =============================================================================
|
|
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", "").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 hhmm(dt: datetime.datetime) -> str:
|
|
return dt.strftime("%H:%M")
|
|
|
|
|
|
# =============================================================================
|
|
# Telegram
|
|
# =============================================================================
|
|
def telegram_send_html(message_html: str, chat_ids: Optional[List[str]] = None) -> bool:
|
|
"""
|
|
Args:
|
|
message_html: Messaggio HTML da inviare
|
|
chat_ids: Lista di chat IDs (default: TELEGRAM_CHAT_IDS)
|
|
"""
|
|
token = load_bot_token()
|
|
if not token:
|
|
LOGGER.warning("Telegram token missing: message not sent.")
|
|
return False
|
|
|
|
if chat_ids is None:
|
|
chat_ids = TELEGRAM_CHAT_IDS
|
|
|
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
|
base_payload = {
|
|
"text": message_html,
|
|
"parse_mode": "HTML",
|
|
"disable_web_page_preview": True,
|
|
}
|
|
|
|
sent_ok = False
|
|
with requests.Session() as s:
|
|
for chat_id in chat_ids:
|
|
payload = dict(base_payload)
|
|
payload["chat_id"] = chat_id
|
|
try:
|
|
resp = s.post(url, json=payload, timeout=15)
|
|
if resp.status_code == 200:
|
|
sent_ok = True
|
|
else:
|
|
LOGGER.error("Telegram error chat_id=%s status=%s body=%s",
|
|
chat_id, resp.status_code, resp.text[:500])
|
|
time.sleep(0.25)
|
|
except Exception as e:
|
|
LOGGER.exception("Telegram exception chat_id=%s err=%s", chat_id, e)
|
|
|
|
return sent_ok
|
|
|
|
|
|
# =============================================================================
|
|
# State & Open-Meteo
|
|
# =============================================================================
|
|
def load_state() -> Dict:
|
|
default = {"alert_active": False, "signature": "", "updated": ""}
|
|
if os.path.exists(STATE_FILE):
|
|
try:
|
|
with open(STATE_FILE, "r", encoding="utf-8") as f:
|
|
data = json.load(f) or {}
|
|
default.update(data)
|
|
except Exception as e:
|
|
LOGGER.exception("State read error: %s", e)
|
|
return default
|
|
|
|
|
|
def save_state(alert_active: bool, signature: str) -> None:
|
|
try:
|
|
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
|
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(
|
|
{"alert_active": alert_active, "signature": signature, "updated": now_local().isoformat()},
|
|
f,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
except Exception as e:
|
|
LOGGER.exception("State write error: %s", e)
|
|
|
|
|
|
def get_forecast(session: requests.Session, lat: float, lon: float, model: str) -> Optional[Dict]:
|
|
params = {
|
|
"latitude": lat,
|
|
"longitude": lon,
|
|
"hourly": "precipitation,snowfall,weathercode", # Aggiunto weathercode per rilevare neve
|
|
"timezone": TZ,
|
|
"forecast_days": 2,
|
|
"precipitation_unit": "mm",
|
|
"models": model,
|
|
}
|
|
|
|
# Aggiungi minutely_15 per AROME Seamless (dettaglio 15 minuti per inizio preciso eventi)
|
|
# Se fallisce o ha buchi, riprova senza minutely_15
|
|
if model == MODEL_AROME:
|
|
params["minutely_15"] = "precipitation,rain,snowfall,precipitation_probability,temperature_2m"
|
|
try:
|
|
r = session.get(OPEN_METEO_URL, params=params, headers=HTTP_HEADERS, timeout=25)
|
|
if r.status_code == 400:
|
|
# Se 400 e abbiamo minutely_15, riprova senza
|
|
if "minutely_15" in params and model == MODEL_AROME:
|
|
LOGGER.warning("Open-Meteo 400 con minutely_15 (model=%s), riprovo senza minutely_15", model)
|
|
params_no_minutely = params.copy()
|
|
del params_no_minutely["minutely_15"]
|
|
try:
|
|
r2 = session.get(OPEN_METEO_URL, params=params_no_minutely, headers=HTTP_HEADERS, timeout=25)
|
|
if r2.status_code == 200:
|
|
return r2.json()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
elif r.status_code == 504:
|
|
# Gateway Timeout: se abbiamo minutely_15, riprova senza
|
|
if "minutely_15" in params and model == MODEL_AROME:
|
|
LOGGER.warning("Open-Meteo 504 Gateway Timeout con minutely_15 (model=%s), riprovo senza minutely_15", model)
|
|
params_no_minutely = params.copy()
|
|
del params_no_minutely["minutely_15"]
|
|
try:
|
|
r2 = session.get(OPEN_METEO_URL, params=params_no_minutely, headers=HTTP_HEADERS, timeout=25)
|
|
if r2.status_code == 200:
|
|
return r2.json()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
|
|
# Verifica se minutely_15 ha buchi (anche solo 1 None = fallback a hourly)
|
|
if "minutely_15" in params and model == MODEL_AROME:
|
|
minutely = data.get("minutely_15", {}) or {}
|
|
minutely_times = minutely.get("time", []) or []
|
|
minutely_precip = minutely.get("precipitation", []) or []
|
|
minutely_snow = minutely.get("snowfall", []) or []
|
|
|
|
# Controlla se ci sono buchi (anche solo 1 None)
|
|
if minutely_times:
|
|
# Controlla tutti i parametri principali per buchi
|
|
has_holes = False
|
|
# Controlla precipitation
|
|
if minutely_precip and any(v is None for v in minutely_precip):
|
|
has_holes = True
|
|
# Controlla snowfall
|
|
if minutely_snow and any(v is None for v in minutely_snow):
|
|
has_holes = True
|
|
|
|
if has_holes:
|
|
LOGGER.warning("minutely_15 ha buchi (valori None rilevati, model=%s), riprovo senza minutely_15", model)
|
|
params_no_minutely = params.copy()
|
|
del params_no_minutely["minutely_15"]
|
|
try:
|
|
r2 = session.get(OPEN_METEO_URL, params=params_no_minutely, headers=HTTP_HEADERS, timeout=25)
|
|
if r2.status_code == 200:
|
|
return r2.json()
|
|
except Exception:
|
|
pass
|
|
|
|
return data
|
|
except requests.exceptions.Timeout:
|
|
# Timeout: se abbiamo minutely_15, riprova senza
|
|
if "minutely_15" in params and model == MODEL_AROME:
|
|
LOGGER.warning("Open-Meteo Timeout con minutely_15 (model=%s), riprovo senza minutely_15", model)
|
|
params_no_minutely = params.copy()
|
|
del params_no_minutely["minutely_15"]
|
|
try:
|
|
r2 = session.get(OPEN_METEO_URL, params=params_no_minutely, headers=HTTP_HEADERS, timeout=25)
|
|
if r2.status_code == 200:
|
|
return r2.json()
|
|
except Exception:
|
|
pass
|
|
LOGGER.exception("Open-Meteo timeout (model=%s)", model)
|
|
return None
|
|
except Exception as e:
|
|
# Altri errori: se abbiamo minutely_15, riprova senza
|
|
if "minutely_15" in params and model == MODEL_AROME:
|
|
LOGGER.warning("Open-Meteo error con minutely_15 (model=%s): %s, riprovo senza minutely_15", model, str(e))
|
|
params_no_minutely = params.copy()
|
|
del params_no_minutely["minutely_15"]
|
|
try:
|
|
r2 = session.get(OPEN_METEO_URL, params=params_no_minutely, headers=HTTP_HEADERS, timeout=25)
|
|
if r2.status_code == 200:
|
|
return r2.json()
|
|
except Exception:
|
|
pass
|
|
LOGGER.exception("Open-Meteo request error (model=%s): %s", model, e)
|
|
return None
|
|
|
|
|
|
def compare_values(arome_val: float, icon_val: float) -> Optional[Dict]:
|
|
"""Confronta due valori e ritorna info se scostamento >30%"""
|
|
if arome_val == 0 and icon_val == 0:
|
|
return None
|
|
|
|
if arome_val > 0:
|
|
diff_pct = abs(icon_val - arome_val) / arome_val
|
|
elif icon_val > 0:
|
|
diff_pct = abs(arome_val - icon_val) / icon_val
|
|
else:
|
|
return None
|
|
|
|
if diff_pct > COMPARISON_THRESHOLD:
|
|
return {
|
|
"diff_pct": diff_pct * 100,
|
|
"arome": arome_val,
|
|
"icon": icon_val
|
|
}
|
|
return None
|
|
|
|
|
|
# =============================================================================
|
|
# Analytics
|
|
# =============================================================================
|
|
def rolling_sum_3h(values: List[float]) -> List[float]:
|
|
out = []
|
|
for i in range(0, max(0, len(values) - 2)):
|
|
try:
|
|
s = float(values[i]) + float(values[i + 1]) + float(values[i + 2])
|
|
except Exception:
|
|
s = 0.0
|
|
out.append(s)
|
|
return out
|
|
|
|
|
|
def first_persistent_run(values: List[float], threshold: float, persist: int) -> Tuple[bool, int, int, float]:
|
|
consec = 0
|
|
run_start = -1
|
|
run_max = 0.0
|
|
for i, v in enumerate(values):
|
|
vv = float(v) if v is not None else 0.0
|
|
if vv >= threshold:
|
|
if consec == 0:
|
|
run_start = i
|
|
run_max = vv
|
|
else:
|
|
run_max = max(run_max, vv)
|
|
consec += 1
|
|
if consec >= persist:
|
|
return True, run_start, consec, run_max
|
|
else:
|
|
consec = 0
|
|
return False, -1, 0, 0.0
|
|
|
|
|
|
def max_consecutive_gt(values: List[float], eps: float) -> Tuple[int, int]:
|
|
best_len = 0
|
|
best_start = -1
|
|
consec = 0
|
|
start = -1
|
|
for i, v in enumerate(values):
|
|
vv = float(v) if v is not None else 0.0
|
|
if vv > eps:
|
|
if consec == 0:
|
|
start = i
|
|
consec += 1
|
|
if consec > best_len:
|
|
best_len = consec
|
|
best_start = start
|
|
else:
|
|
consec = 0
|
|
return best_len, best_start
|
|
|
|
|
|
def compute_stats(data: Dict) -> Optional[Dict]:
|
|
hourly = data.get("hourly", {}) or {}
|
|
times = hourly.get("time", []) or []
|
|
precip = hourly.get("precipitation", []) or []
|
|
snow = hourly.get("snowfall", []) or []
|
|
weathercode = hourly.get("weathercode", []) or [] # Per rilevare neve anche quando snowfall è basso
|
|
|
|
n = min(len(times), len(precip), len(snow))
|
|
if n == 0: return None
|
|
|
|
now = now_local()
|
|
start_idx = -1
|
|
for i, t in enumerate(times[:n]):
|
|
if parse_time_to_local(t) >= now:
|
|
start_idx = i
|
|
break
|
|
if start_idx == -1: return None
|
|
|
|
end_idx = min(start_idx + HOURS_AHEAD, n)
|
|
if end_idx <= start_idx: return None
|
|
|
|
times_w = times[start_idx:end_idx]
|
|
precip_w = precip[start_idx:end_idx]
|
|
snow_w = [float(x) if x is not None else 0.0 for x in snow[start_idx:end_idx]]
|
|
weathercode_w = [int(x) if x is not None else None for x in weathercode[start_idx:end_idx]] if len(weathercode) > start_idx else []
|
|
dt_w = [parse_time_to_local(t) for t in times_w]
|
|
|
|
rain3 = rolling_sum_3h(precip_w)
|
|
rain3_max = max(rain3) if rain3 else 0.0
|
|
rain3_max_idx = rain3.index(rain3_max) if rain3 else -1
|
|
rain3_max_time = hhmm(dt_w[rain3_max_idx]) if (rain3_max_idx >= 0 and rain3_max_idx < len(dt_w)) else ""
|
|
|
|
rain_persist_ok, rain_run_start, rain_run_len, rain_run_max = first_persistent_run(
|
|
rain3, SOGLIA_PIOGGIA_3H_MM, PERSIST_HOURS
|
|
)
|
|
rain_persist_time = hhmm(dt_w[rain_run_start]) if (rain_persist_ok and rain_run_start < len(dt_w)) else ""
|
|
|
|
# Analizza evento pioggia completa (48h): rileva inizio e calcola durata e accumulo totale
|
|
rain_start_idx = None
|
|
rain_end_idx = None
|
|
total_rain_accumulation = 0.0
|
|
rain_duration_hours = 0.0
|
|
max_rain_intensity = 0.0
|
|
|
|
# Codici meteo che indicano pioggia (WMO)
|
|
RAIN_WEATHER_CODES = [61, 63, 65, 66, 67, 80, 81, 82]
|
|
|
|
# Trova inizio evento pioggia (prima occorrenza con precipitation > 0 OPPURE weathercode pioggia)
|
|
# Estendi l'analisi a 48 ore se disponibile
|
|
extended_end_idx = min(start_idx + 48, n) # Estendi a 48 ore
|
|
precip_extended = precip[start_idx:extended_end_idx]
|
|
weathercode_extended = [int(x) if x is not None else None for x in weathercode[start_idx:extended_end_idx]] if len(weathercode) > start_idx else []
|
|
|
|
for i, (p_val, code) in enumerate(zip(precip_extended, weathercode_extended if len(weathercode_extended) == len(precip_extended) else [None] * len(precip_extended))):
|
|
p_val_float = float(p_val) if p_val is not None else 0.0
|
|
is_rain = (p_val_float > 0.0) or (code is not None and code in RAIN_WEATHER_CODES)
|
|
if is_rain and rain_start_idx is None:
|
|
rain_start_idx = i
|
|
break
|
|
|
|
# Se trovato inizio, calcola durata e accumulo totale su 48 ore
|
|
if rain_start_idx is not None:
|
|
# Trova fine evento pioggia (ultima occorrenza con pioggia)
|
|
for i in range(len(precip_extended) - 1, rain_start_idx - 1, -1):
|
|
p_val = precip_extended[i] if i < len(precip_extended) else None
|
|
code = weathercode_extended[i] if i < len(weathercode_extended) else None
|
|
p_val_float = float(p_val) if p_val is not None else 0.0
|
|
is_rain = (p_val_float > 0.0) or (code is not None and code in RAIN_WEATHER_CODES)
|
|
if is_rain:
|
|
rain_end_idx = i
|
|
break
|
|
|
|
if rain_end_idx is not None:
|
|
# Calcola durata
|
|
times_extended = times[start_idx:extended_end_idx]
|
|
dt_extended = [parse_time_to_local(t) for t in times_extended]
|
|
if rain_end_idx < len(dt_extended) and rain_start_idx < len(dt_extended):
|
|
rain_duration_hours = (dt_extended[rain_end_idx] - dt_extended[rain_start_idx]).total_seconds() / 3600.0
|
|
# Calcola accumulo totale (somma di tutti i precipitation > 0)
|
|
total_rain_accumulation = sum(float(p) for p in precip_extended[rain_start_idx:rain_end_idx+1] if p is not None and float(p) > 0.0)
|
|
# Calcola intensità massima oraria
|
|
max_rain_intensity = max((float(p) for p in precip_extended[rain_start_idx:rain_end_idx+1] if p is not None), default=0.0)
|
|
|
|
# Analizza nevicata completa (48h): rileva inizio usando snowfall > 0 OPPURE weathercode
|
|
# Calcola durata e accumulo totale
|
|
snow_start_idx = None
|
|
snow_end_idx = None
|
|
total_snow_accumulation = 0.0
|
|
snow_duration_hours = 0.0
|
|
|
|
# Trova inizio nevicata (prima occorrenza con snowfall > 0 OPPURE weathercode neve)
|
|
for i, (s_val, code) in enumerate(zip(snow_w, weathercode_w if len(weathercode_w) == len(snow_w) else [None] * len(snow_w))):
|
|
is_snow = (s_val > 0.0) or (code is not None and code in SNOW_WEATHER_CODES)
|
|
if is_snow and snow_start_idx is None:
|
|
snow_start_idx = i
|
|
break
|
|
|
|
# Se trovato inizio, calcola durata e accumulo totale
|
|
if snow_start_idx is not None:
|
|
# Trova fine nevicata (ultima occorrenza con neve)
|
|
for i in range(len(snow_w) - 1, snow_start_idx - 1, -1):
|
|
s_val = snow_w[i]
|
|
code = weathercode_w[i] if i < len(weathercode_w) else None
|
|
is_snow = (s_val > 0.0) or (code is not None and code in SNOW_WEATHER_CODES)
|
|
if is_snow:
|
|
snow_end_idx = i
|
|
break
|
|
|
|
if snow_end_idx is not None:
|
|
# Calcola durata
|
|
snow_duration_hours = (dt_w[snow_end_idx] - dt_w[snow_start_idx]).total_seconds() / 3600.0
|
|
# Calcola accumulo totale (somma di tutti i snowfall > 0)
|
|
total_snow_accumulation = sum(s for s in snow_w[snow_start_idx:snow_end_idx+1] if s > 0.0)
|
|
|
|
# Per compatibilità con logica esistente
|
|
snow_run_len, snow_run_start = max_consecutive_gt(snow_w, eps=SNOW_HOURLY_EPS_CM)
|
|
snow_run_time = hhmm(dt_w[snow_run_start]) if (snow_run_start >= 0 and snow_run_start < len(dt_w)) else ""
|
|
|
|
# Se trovato inizio nevicata, usa quello invece del run
|
|
if snow_start_idx is not None:
|
|
snow_run_time = hhmm(dt_w[snow_start_idx])
|
|
# Durata minima per alert: almeno 2 ore
|
|
if snow_duration_hours >= PERSIST_HOURS:
|
|
snow_run_len = int(snow_duration_hours)
|
|
else:
|
|
snow_run_len = 0 # Durata troppo breve
|
|
|
|
snow_12h = sum(s for s in snow_w[:min(12, len(snow_w))] if s > 0.0)
|
|
snow_24h = sum(s for s in snow_w[:min(24, len(snow_w))] if s > 0.0)
|
|
|
|
return {
|
|
"rain3_max": float(rain3_max),
|
|
"rain3_max_time": rain3_max_time,
|
|
"rain_persist_ok": bool(rain_persist_ok),
|
|
"rain_persist_time": rain_persist_time,
|
|
"rain_persist_run_max": float(rain_run_max),
|
|
"rain_persist_run_len": int(rain_run_len),
|
|
"rain_duration_hours": float(rain_duration_hours),
|
|
"total_rain_accumulation_mm": float(total_rain_accumulation),
|
|
"max_rain_intensity_mm_h": float(max_rain_intensity),
|
|
"snow_run_len": int(snow_run_len),
|
|
"snow_run_time": snow_run_time,
|
|
"snow_12h": float(snow_12h),
|
|
"snow_24h": float(snow_24h),
|
|
"snow_duration_hours": float(snow_duration_hours),
|
|
"total_snow_accumulation_cm": float(total_snow_accumulation),
|
|
}
|
|
|
|
|
|
def point_alerts(point_name: str, stats: Dict) -> Dict:
|
|
snow_alert = (stats["snow_run_len"] >= PERSIST_HOURS) and (stats["snow_24h"] > 0.0)
|
|
rain_alert = bool(stats["rain_persist_ok"])
|
|
return {
|
|
"name": point_name,
|
|
"snow_alert": snow_alert,
|
|
"rain_alert": rain_alert,
|
|
"snow_12h": stats["snow_12h"],
|
|
"snow_24h": stats["snow_24h"],
|
|
"snow_run_len": stats["snow_run_len"],
|
|
"snow_run_time": stats["snow_run_time"],
|
|
"rain3_max": stats["rain3_max"],
|
|
"rain3_max_time": stats["rain3_max_time"],
|
|
"rain_persist_time": stats["rain_persist_time"],
|
|
"rain_persist_run_max": stats["rain_persist_run_max"],
|
|
"rain_persist_run_len": stats["rain_persist_run_len"],
|
|
"rain_duration_hours": stats.get("rain_duration_hours", 0.0),
|
|
"total_rain_accumulation_mm": stats.get("total_rain_accumulation_mm", 0.0),
|
|
"max_rain_intensity_mm_h": stats.get("max_rain_intensity_mm_h", 0.0),
|
|
}
|
|
|
|
|
|
def build_signature(bologna: Dict, route: List[Dict]) -> str:
|
|
parts = [
|
|
f"BO:snow={int(bologna['snow_alert'])},rain={int(bologna['rain_alert'])},"
|
|
f"s24={bologna['snow_24h']:.1f},r3max={bologna['rain3_max']:.1f}"
|
|
]
|
|
for r in route:
|
|
parts.append(
|
|
f"{r['name']}:s{int(r['snow_alert'])}r{int(r['rain_alert'])}"
|
|
f":s24={r['snow_24h']:.1f}:r3max={r['rain3_max']:.1f}"
|
|
)
|
|
return "|".join(parts)
|
|
|
|
|
|
def main(chat_ids: Optional[List[str]] = None, debug_mode: bool = False) -> None:
|
|
LOGGER.info("--- Student alert Bologna (Neve/Pioggia intensa) ---")
|
|
|
|
state = load_state()
|
|
was_active = bool(state.get("alert_active", False))
|
|
last_sig = state.get("signature", "")
|
|
|
|
comparisons: Dict[str, Dict] = {} # point_name -> comparison info
|
|
|
|
with requests.Session() as session:
|
|
# Trigger: Bologna
|
|
bo = POINTS[0]
|
|
bo_data_arome = get_forecast(session, bo["lat"], bo["lon"], MODEL_AROME)
|
|
if not bo_data_arome: return
|
|
bo_stats_arome = compute_stats(bo_data_arome)
|
|
if not bo_stats_arome:
|
|
LOGGER.error("Impossibile calcolare statistiche Bologna.")
|
|
return
|
|
bo_alerts = point_alerts(bo["name"], bo_stats_arome)
|
|
|
|
# Recupera ICON Italia per Bologna
|
|
bo_data_icon = get_forecast(session, bo["lat"], bo["lon"], MODEL_ICON_IT)
|
|
if bo_data_icon:
|
|
bo_stats_icon = compute_stats(bo_data_icon)
|
|
if bo_stats_icon:
|
|
comp_snow = compare_values(bo_stats_arome["snow_24h"], bo_stats_icon["snow_24h"])
|
|
comp_rain = compare_values(bo_stats_arome["rain3_max"], bo_stats_icon["rain3_max"])
|
|
if comp_snow or comp_rain:
|
|
comparisons[bo["name"]] = {"snow": comp_snow, "rain": comp_rain, "icon_stats": bo_stats_icon}
|
|
|
|
# Route points
|
|
route_alerts: List[Dict] = []
|
|
for p in POINTS[1:]:
|
|
d_arome = get_forecast(session, p["lat"], p["lon"], MODEL_AROME)
|
|
if not d_arome: continue
|
|
st_arome = compute_stats(d_arome)
|
|
if not st_arome: continue
|
|
route_alerts.append(point_alerts(p["name"], st_arome))
|
|
|
|
# Recupera ICON Italia per punto
|
|
d_icon = get_forecast(session, p["lat"], p["lon"], MODEL_ICON_IT)
|
|
if d_icon:
|
|
st_icon = compute_stats(d_icon)
|
|
if st_icon:
|
|
comp_snow = compare_values(st_arome["snow_24h"], st_icon["snow_24h"])
|
|
comp_rain = compare_values(st_arome["rain3_max"], st_icon["rain3_max"])
|
|
if comp_snow or comp_rain:
|
|
comparisons[p["name"]] = {"snow": comp_snow, "rain": comp_rain, "icon_stats": st_icon}
|
|
|
|
any_route_alert = any(x["snow_alert"] or x["rain_alert"] for x in route_alerts)
|
|
any_alert = (bo_alerts["snow_alert"] or bo_alerts["rain_alert"] or any_route_alert)
|
|
|
|
sig = build_signature(bo_alerts, route_alerts)
|
|
|
|
# --- Scenario A: Allerta ---
|
|
if any_alert:
|
|
# In modalità debug, bypassa controlli anti-spam
|
|
if debug_mode:
|
|
LOGGER.info("[DEBUG MODE] Bypass anti-spam: invio forzato")
|
|
if debug_mode or (not was_active) or (sig != last_sig):
|
|
now_str = now_local().strftime("%H:%M")
|
|
header_icon = "❄️" if (bo_alerts["snow_alert"] or any(x["snow_alert"] for x in route_alerts)) \
|
|
else "🌧️" if (bo_alerts["rain_alert"] or any(x["rain_alert"] for x in route_alerts)) \
|
|
else "⚠️"
|
|
|
|
msg: List[str] = []
|
|
msg.append(f"{header_icon} <b>ALLERTA METEO (Bologna / Rientro)</b>")
|
|
msg.append(f"🕒 <i>Aggiornamento ore {html.escape(now_str)}</i>")
|
|
model_info = MODEL_AROME
|
|
if comparisons:
|
|
model_info = f"{MODEL_AROME} + ICON Italia (discordanza rilevata)"
|
|
msg.append(f"🛰️ <code>Modello: {html.escape(model_info)}</code>")
|
|
msg.append(f"⏱️ <code>Finestra: {HOURS_AHEAD} ore | Persistenza: {PERSIST_HOURS} ore</code>")
|
|
msg.append("")
|
|
|
|
# Bologna
|
|
msg.append("🎓 <b>A BOLOGNA</b>")
|
|
bo_comp = comparisons.get(bo["name"])
|
|
if bo_alerts["snow_alert"]:
|
|
msg.append(f"❄️ Neve (≥{PERSIST_HOURS}h) da ~<b>{html.escape(bo_alerts['snow_run_time'] or '—')}</b> (run ~{bo_alerts['snow_run_len']}h).")
|
|
msg.append(f"• Accumulo: 12h <b>{bo_alerts['snow_12h']:.1f} cm</b> | 24h <b>{bo_alerts['snow_24h']:.1f} cm</b>")
|
|
if bo_comp and bo_comp.get("snow"):
|
|
comp = bo_comp["snow"]
|
|
icon_s24 = bo_comp["icon_stats"]["snow_24h"]
|
|
msg.append(f"⚠️ <b>Discordanza modelli</b>: AROME {comp['arome']:.1f} cm | ICON {icon_s24:.1f} cm (scostamento {comp['diff_pct']:.0f}%)")
|
|
else:
|
|
msg.append(f"❄️ Neve: nessuna persistenza ≥ {PERSIST_HOURS}h (24h {bo_alerts['snow_24h']:.1f} cm).")
|
|
|
|
if bo_alerts["rain_alert"]:
|
|
rain_duration = bo_alerts.get("rain_duration_hours", 0.0)
|
|
total_rain = bo_alerts.get("total_rain_accumulation_mm", 0.0)
|
|
max_intensity = bo_alerts.get("max_rain_intensity_mm_h", 0.0)
|
|
|
|
msg.append(f"🌧️ Pioggia molto forte (3h ≥ {SOGLIA_PIOGGIA_3H_MM:.0f} mm, ≥{PERSIST_HOURS}h) da ~<b>{html.escape(bo_alerts['rain_persist_time'] or '—')}</b>.")
|
|
if rain_duration > 0:
|
|
msg.append(f"⏱️ <b>Durata totale evento (48h):</b> ~{rain_duration:.0f} ore | <b>Accumulo totale:</b> ~{total_rain:.1f} mm | <b>Intensità max:</b> {max_intensity:.1f} mm/h")
|
|
if bo_comp and bo_comp.get("rain"):
|
|
comp = bo_comp["rain"]
|
|
icon_r3 = bo_comp["icon_stats"]["rain3_max"]
|
|
msg.append(f"⚠️ <b>Discordanza modelli</b>: AROME {comp['arome']:.1f} mm | ICON {icon_r3:.1f} mm (scostamento {comp['diff_pct']:.0f}%)")
|
|
else:
|
|
msg.append(f"🌧️ Pioggia: max 3h <b>{bo_alerts['rain3_max']:.1f} mm</b> (picco ~{html.escape(bo_alerts['rain3_max_time'] or '—')}).")
|
|
|
|
msg.append("")
|
|
msg.append("🚗 <b>CASELLI (A14) / TRATTO</b>")
|
|
|
|
issues = [x for x in route_alerts if x["snow_alert"] or x["rain_alert"]]
|
|
if not issues:
|
|
msg.append("✅ Nessuna criticità persistente rilevata lungo il percorso.")
|
|
else:
|
|
for x in issues:
|
|
line = f"• <b>{html.escape(x['name'])}</b>: "
|
|
parts: List[str] = []
|
|
if x["snow_alert"]:
|
|
parts.append(f"❄️ neve (≥{PERSIST_HOURS}h) da ~{html.escape(x['snow_run_time'] or '—')} (24h {x['snow_24h']:.1f} cm)")
|
|
if x["rain_alert"]:
|
|
rain_dur = x.get("rain_duration_hours", 0.0)
|
|
rain_tot = x.get("total_rain_accumulation_mm", 0.0)
|
|
if rain_dur > 0:
|
|
parts.append(f"🌧️ pioggia forte da ~{html.escape(x['rain_persist_time'] or '—')} (durata ~{rain_dur:.0f}h, totale ~{rain_tot:.1f}mm)")
|
|
else:
|
|
parts.append(f"🌧️ pioggia forte da ~{html.escape(x['rain_persist_time'] or '—')}")
|
|
line += " | ".join(parts)
|
|
msg.append(line)
|
|
|
|
# Aggiungi nota discordanza se presente
|
|
point_comp = comparisons.get(x["name"])
|
|
if point_comp:
|
|
disc_parts = []
|
|
if point_comp.get("snow"):
|
|
comp = point_comp["snow"]
|
|
icon_s24 = point_comp["icon_stats"]["snow_24h"]
|
|
disc_parts.append(f"Neve: AROME {comp['arome']:.1f} cm | ICON {icon_s24:.1f} cm ({comp['diff_pct']:.0f}%)")
|
|
if point_comp.get("rain"):
|
|
comp = point_comp["rain"]
|
|
icon_r3 = point_comp["icon_stats"]["rain3_max"]
|
|
disc_parts.append(f"Pioggia: AROME {comp['arome']:.1f} mm | ICON {icon_r3:.1f} mm ({comp['diff_pct']:.0f}%)")
|
|
if disc_parts:
|
|
msg.append(f" ⚠️ Discordanza: {' | '.join(disc_parts)}")
|
|
|
|
msg.append("")
|
|
msg.append("<i>Fonte dati: Open-Meteo</i>")
|
|
|
|
# FIX: usare \n invece di <br/>
|
|
ok = telegram_send_html("\n".join(msg), chat_ids=chat_ids)
|
|
if ok:
|
|
LOGGER.info("Notifica inviata.")
|
|
else:
|
|
LOGGER.warning("Notifica NON inviata.")
|
|
|
|
save_state(True, sig)
|
|
else:
|
|
LOGGER.info("Allerta già notificata e invariata.")
|
|
return
|
|
|
|
# --- Scenario B: Rientro ---
|
|
if was_active and not any_alert:
|
|
now_str = now_local().strftime("%H:%M")
|
|
# FIX: usare \n invece di <br/>
|
|
msg = (
|
|
"🟢 <b>ALLERTA RIENTRATA (Bologna / Rientro)</b>\n"
|
|
f"🕒 <i>Aggiornamento ore {html.escape(now_str)}</i>\n\n"
|
|
f"Nelle prossime {HOURS_AHEAD} ore non risultano più condizioni persistenti\n"
|
|
f"di neve (≥{PERSIST_HOURS}h) o pioggia 3h sopra soglia (≥{PERSIST_HOURS}h).\n"
|
|
"<i>Fonte dati: Open-Meteo</i>"
|
|
)
|
|
ok = telegram_send_html(msg, chat_ids=chat_ids)
|
|
if ok:
|
|
LOGGER.info("Rientro notificato.")
|
|
else:
|
|
LOGGER.warning("Rientro NON inviato.")
|
|
save_state(False, "")
|
|
return
|
|
|
|
# --- Scenario C: Tranquillo ---
|
|
save_state(False, "")
|
|
LOGGER.info("Nessuna allerta.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
arg_parser = argparse.ArgumentParser(description="Student alert Bologna")
|
|
arg_parser.add_argument("--debug", action="store_true", help="Invia messaggi solo all'admin (chat ID: %s)" % TELEGRAM_CHAT_IDS[0])
|
|
args = arg_parser.parse_args()
|
|
|
|
# In modalità debug, invia solo al primo chat ID (admin) e bypassa anti-spam
|
|
chat_ids = [TELEGRAM_CHAT_IDS[0]] if args.debug else None
|
|
|
|
main(chat_ids=chat_ids, debug_mode=args.debug)
|