896 lines
37 KiB
Python
896 lines
37 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
|
||
from open_meteo_client import configure_open_meteo_session
|
||
|
||
# =============================================================================
|
||
# 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")
|
||
|
||
|
||
_WEEKDAYS_IT = ("lun", "mar", "mer", "gio", "ven", "sab", "dom")
|
||
|
||
|
||
def format_clock(dt: datetime.datetime, ref: Optional[datetime.datetime] = None) -> str:
|
||
"""HH:MM, con giorno corto se diverso da ref (default: oggi locale)."""
|
||
ref = ref or now_local()
|
||
label = hhmm(dt)
|
||
if dt.date() != ref.date():
|
||
return f"{_WEEKDAYS_IT[dt.weekday()]} {label}"
|
||
return label
|
||
|
||
|
||
def format_fascia(
|
||
start: Optional[datetime.datetime],
|
||
end: Optional[datetime.datetime],
|
||
ref: Optional[datetime.datetime] = None,
|
||
) -> str:
|
||
"""Fascia leggibile ~HH:MM–~HH:MM (con giorno se serve)."""
|
||
if start is None:
|
||
return "—"
|
||
ref = ref or now_local()
|
||
a = format_clock(start, ref)
|
||
if end is None or end == start:
|
||
return f"~{a}"
|
||
b = format_clock(end, ref)
|
||
return f"~{a}–~{b}"
|
||
|
||
|
||
# =============================================================================
|
||
# 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)
|
||
"""
|
||
try:
|
||
from telegram_gate import telegram_alerts_enabled
|
||
except ImportError:
|
||
telegram_alerts_enabled = lambda: True # type: ignore
|
||
|
||
if not telegram_alerts_enabled():
|
||
LOGGER.info("Telegram sospeso: skip student_alert")
|
||
return False
|
||
|
||
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, detail: Optional[Dict] = None) -> None:
|
||
try:
|
||
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
||
payload: Dict = {
|
||
"alert_active": alert_active,
|
||
"signature": signature,
|
||
"updated": now_local().isoformat(),
|
||
}
|
||
if detail:
|
||
payload.update(detail)
|
||
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(payload, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
LOGGER.exception("State write error: %s", e)
|
||
|
||
|
||
def build_plain_summary(bo_alerts: Dict, route_alerts: List[Dict]) -> str:
|
||
"""Testo leggibile per WebApp (dove / eventi / quando)."""
|
||
any_snow = bool(bo_alerts.get("snow_alert")) or any(x.get("snow_alert") for x in route_alerts)
|
||
any_rain = bool(bo_alerts.get("rain_alert")) or any(x.get("rain_alert") for x in route_alerts)
|
||
soglie: List[str] = []
|
||
if any_snow:
|
||
soglie.append(f"neve ≥{PERSIST_HOURS}h consecutive")
|
||
if any_rain:
|
||
soglie.append(
|
||
f"pioggia 3h ≥{SOGLIA_PIOGGIA_3H_MM:.0f} mm per ≥{PERSIST_HOURS}h"
|
||
)
|
||
|
||
lines: List[str] = [
|
||
f"Percorso scuola Bologna ↔ rientro · prossime {HOURS_AHEAD}h",
|
||
]
|
||
if soglie:
|
||
lines.append("Soglie: " + " · ".join(soglie))
|
||
lines.extend([
|
||
"",
|
||
"A Bologna:",
|
||
])
|
||
if bo_alerts.get("snow_alert"):
|
||
fascia = bo_alerts.get("snow_fascia") or f"~{bo_alerts.get('snow_run_time') or '—'}"
|
||
lines.append(
|
||
f"• Neve {fascia} "
|
||
f"(12h {bo_alerts.get('snow_12h', 0):.1f} cm · 24h {bo_alerts.get('snow_24h', 0):.1f} cm)"
|
||
)
|
||
if bo_alerts.get("rain_alert"):
|
||
r3 = float(bo_alerts.get("rain3_max") or 0)
|
||
fascia = bo_alerts.get("rain_fascia") or f"~{bo_alerts.get('rain_persist_time') or '—'}"
|
||
lines.append(f"• Pioggia forte {fascia} (max 3h {r3:.1f} mm)")
|
||
if not bo_alerts.get("snow_alert") and not bo_alerts.get("rain_alert"):
|
||
lines.append("• Nessuna criticità persistente (trigger Bologna assente nel dettaglio punti)")
|
||
|
||
issues = [x for x in route_alerts if x.get("snow_alert") or x.get("rain_alert")]
|
||
lines.append("")
|
||
lines.append("Caselli A14 / tratto:")
|
||
if not issues:
|
||
lines.append("• Nessuna criticità lungo il percorso")
|
||
else:
|
||
for x in issues:
|
||
parts: List[str] = []
|
||
if x.get("snow_alert"):
|
||
fascia = x.get("snow_fascia") or f"~{x.get('snow_run_time') or '—'}"
|
||
parts.append(f"neve {fascia} (24h {x.get('snow_24h', 0):.1f} cm)")
|
||
if x.get("rain_alert"):
|
||
r3 = float(x.get("rain3_max") or 0)
|
||
fascia = x.get("rain_fascia") or f"~{x.get('rain_persist_time') or '—'}"
|
||
parts.append(f"pioggia {fascia} (max 3h {r3:.1f} mm)")
|
||
lines.append(f"• {x.get('name', '?')}: " + " | ".join(parts))
|
||
|
||
lines.append("")
|
||
lines.append("Fonte: Open-Meteo (AROME / confronto ICON IT)")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def first_event_time(bo_alerts: Dict, route_alerts: List[Dict]) -> Optional[str]:
|
||
candidates: List[str] = []
|
||
for src in [bo_alerts, *route_alerts]:
|
||
if src.get("snow_alert") and src.get("snow_run_time"):
|
||
candidates.append(str(src["snow_run_time"]))
|
||
if src.get("rain_alert") and src.get("rain_persist_time"):
|
||
candidates.append(str(src["rain_persist_time"]))
|
||
return min(candidates) if candidates else None
|
||
|
||
|
||
def notify_webapp(title: str, body: str, severity: str = "warning") -> None:
|
||
try:
|
||
sys_path = "/home/daniely/docker/shared"
|
||
if sys_path not in __import__("sys").path:
|
||
__import__("sys").path.insert(0, sys_path)
|
||
from loogle_core.alert_dispatcher import dispatch_alert
|
||
|
||
dispatch_alert(title, body, category="student", severity=severity)
|
||
except Exception as e:
|
||
LOGGER.warning("WebApp notify fallita: %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=(5, 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=(5, 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=(5, 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=(5, 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=(5, 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=(5, 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]:
|
||
"""Prima run continua con valori >= soglia e lunghezza >= persist.
|
||
|
||
Ritorna (ok, start_idx, end_idx inclusivo, run_max). La run viene
|
||
estesa fino alla fine (non si ferma al minimo persist).
|
||
"""
|
||
consec = 0
|
||
run_start = -1
|
||
run_max = 0.0
|
||
found_start = -1
|
||
found_end = -1
|
||
found_max = 0.0
|
||
|
||
def _flush() -> None:
|
||
nonlocal found_start, found_end, found_max
|
||
if consec >= persist and found_start < 0:
|
||
found_start = run_start
|
||
found_end = run_start + consec - 1
|
||
found_max = run_max
|
||
|
||
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
|
||
else:
|
||
_flush()
|
||
if found_start >= 0:
|
||
break
|
||
consec = 0
|
||
run_start = -1
|
||
run_max = 0.0
|
||
else:
|
||
_flush()
|
||
|
||
if found_start >= 0:
|
||
return True, found_start, found_end, found_max
|
||
return False, -1, -1, 0.0
|
||
|
||
|
||
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, rain3_start, rain3_end, rain_run_max = first_persistent_run(
|
||
rain3, SOGLIA_PIOGGIA_3H_MM, PERSIST_HOURS
|
||
)
|
||
rain_run_len = (rain3_end - rain3_start + 1) if rain_persist_ok else 0
|
||
rain_persist_dt_start: Optional[datetime.datetime] = None
|
||
rain_persist_dt_end: Optional[datetime.datetime] = None
|
||
rain_persist_time = ""
|
||
rain_persist_end_time = ""
|
||
rain_fascia = ""
|
||
if rain_persist_ok and 0 <= rain3_start < len(dt_w):
|
||
# Fascia = ore di calendario coperte dalle finestre rolling (ultima = end+2)
|
||
covered_end_idx = min(rain3_end + 2, len(dt_w) - 1)
|
||
rain_persist_dt_start = dt_w[rain3_start]
|
||
rain_persist_dt_end = dt_w[covered_end_idx]
|
||
rain_persist_time = format_clock(rain_persist_dt_start)
|
||
rain_persist_end_time = format_clock(rain_persist_dt_end)
|
||
rain_fascia = format_fascia(rain_persist_dt_start, rain_persist_dt_end)
|
||
|
||
# Flag orari neve: accumulo orario sopra eps OPPURE weathercode neve
|
||
snow_flags: List[float] = []
|
||
for i, s_val in enumerate(snow_w):
|
||
code = weathercode_w[i] if i < len(weathercode_w) else None
|
||
is_snow = (s_val > SNOW_HOURLY_EPS_CM) or (
|
||
code is not None and code in SNOW_WEATHER_CODES
|
||
)
|
||
snow_flags.append(1.0 if is_snow else 0.0)
|
||
|
||
snow_ok, snow_run_start, snow_run_end, _ = first_persistent_run(
|
||
snow_flags, 1.0, PERSIST_HOURS
|
||
)
|
||
snow_run_len = (snow_run_end - snow_run_start + 1) if snow_ok else 0
|
||
snow_persist_dt_start: Optional[datetime.datetime] = None
|
||
snow_persist_dt_end: Optional[datetime.datetime] = None
|
||
snow_run_time = ""
|
||
snow_run_end_time = ""
|
||
snow_fascia = ""
|
||
if snow_ok and 0 <= snow_run_start < len(dt_w) and 0 <= snow_run_end < len(dt_w):
|
||
snow_persist_dt_start = dt_w[snow_run_start]
|
||
snow_persist_dt_end = dt_w[snow_run_end]
|
||
snow_run_time = format_clock(snow_persist_dt_start)
|
||
snow_run_end_time = format_clock(snow_persist_dt_end)
|
||
snow_fascia = format_fascia(snow_persist_dt_start, snow_persist_dt_end)
|
||
|
||
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_end_time": rain_persist_end_time,
|
||
"rain_fascia": rain_fascia,
|
||
"rain_persist_run_max": float(rain_run_max),
|
||
"rain_persist_run_len": int(rain_run_len),
|
||
"snow_run_len": int(snow_run_len),
|
||
"snow_run_time": snow_run_time,
|
||
"snow_run_end_time": snow_run_end_time,
|
||
"snow_fascia": snow_fascia,
|
||
"snow_12h": float(snow_12h),
|
||
"snow_24h": float(snow_24h),
|
||
}
|
||
|
||
|
||
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"],
|
||
"snow_run_end_time": stats.get("snow_run_end_time", ""),
|
||
"snow_fascia": stats.get("snow_fascia", ""),
|
||
"rain3_max": stats["rain3_max"],
|
||
"rain3_max_time": stats["rain3_max_time"],
|
||
"rain_persist_time": stats["rain_persist_time"],
|
||
"rain_persist_end_time": stats.get("rain_persist_end_time", ""),
|
||
"rain_fascia": stats.get("rain_fascia", ""),
|
||
"rain_persist_run_max": stats["rain_persist_run_max"],
|
||
"rain_persist_run_len": stats["rain_persist_run_len"],
|
||
}
|
||
|
||
|
||
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:
|
||
configure_open_meteo_session(session, headers=HTTP_HEADERS)
|
||
# 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"]:
|
||
fascia = bo_alerts.get("snow_fascia") or f"~{bo_alerts.get('snow_run_time') or '—'}"
|
||
msg.append(
|
||
f"❄️ Neve (≥{PERSIST_HOURS}h) <b>{html.escape(fascia)}</b>."
|
||
)
|
||
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"]:
|
||
fascia = bo_alerts.get("rain_fascia") or f"~{bo_alerts.get('rain_persist_time') or '—'}"
|
||
msg.append(
|
||
f"🌧️ Pioggia molto forte (3h ≥ {SOGLIA_PIOGGIA_3H_MM:.0f} mm, ≥{PERSIST_HOURS}h) "
|
||
f"<b>{html.escape(fascia)}</b> "
|
||
f"(max 3h <b>{bo_alerts['rain3_max']:.1f} mm</b>)."
|
||
)
|
||
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"]:
|
||
fascia = x.get("snow_fascia") or f"~{x.get('snow_run_time') or '—'}"
|
||
parts.append(
|
||
f"❄️ neve (≥{PERSIST_HOURS}h) {html.escape(fascia)} "
|
||
f"(24h {x['snow_24h']:.1f} cm)"
|
||
)
|
||
if x["rain_alert"]:
|
||
fascia = x.get("rain_fascia") or f"~{x.get('rain_persist_time') or '—'}"
|
||
parts.append(
|
||
f"🌧️ pioggia forte {html.escape(fascia)} "
|
||
f"(max 3h {x['rain3_max']:.1f} mm)"
|
||
)
|
||
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/>
|
||
html_msg = "\n".join(msg)
|
||
plain = build_plain_summary(bo_alerts, route_alerts)
|
||
ok = telegram_send_html(html_msg, chat_ids=chat_ids)
|
||
if ok:
|
||
LOGGER.info("Notifica inviata su Telegram.")
|
||
else:
|
||
LOGGER.info("Telegram saltato/sospeso; consegna via WebApp.")
|
||
|
||
detail = {
|
||
"summary": plain,
|
||
"window_hours": HOURS_AHEAD,
|
||
"persist_hours": PERSIST_HOURS,
|
||
"rain_threshold_mm_3h": SOGLIA_PIOGGIA_3H_MM,
|
||
"first_event_time": first_event_time(bo_alerts, route_alerts),
|
||
"bologna": {
|
||
"snow_alert": bool(bo_alerts.get("snow_alert")),
|
||
"rain_alert": bool(bo_alerts.get("rain_alert")),
|
||
"snow_run_time": bo_alerts.get("snow_run_time"),
|
||
"snow_run_end_time": bo_alerts.get("snow_run_end_time"),
|
||
"snow_fascia": bo_alerts.get("snow_fascia"),
|
||
"rain_persist_time": bo_alerts.get("rain_persist_time"),
|
||
"rain_persist_end_time": bo_alerts.get("rain_persist_end_time"),
|
||
"rain_fascia": bo_alerts.get("rain_fascia"),
|
||
"snow_24h": bo_alerts.get("snow_24h"),
|
||
"rain3_max": bo_alerts.get("rain3_max"),
|
||
},
|
||
"route_issues": [
|
||
{
|
||
"name": x.get("name"),
|
||
"snow_alert": bool(x.get("snow_alert")),
|
||
"rain_alert": bool(x.get("rain_alert")),
|
||
"snow_run_time": x.get("snow_run_time"),
|
||
"snow_run_end_time": x.get("snow_run_end_time"),
|
||
"snow_fascia": x.get("snow_fascia"),
|
||
"rain_persist_time": x.get("rain_persist_time"),
|
||
"rain_persist_end_time": x.get("rain_persist_end_time"),
|
||
"rain_fascia": x.get("rain_fascia"),
|
||
"snow_24h": x.get("snow_24h"),
|
||
"rain3_max": x.get("rain3_max"),
|
||
"rain_persist_run_len": x.get("rain_persist_run_len"),
|
||
}
|
||
for x in route_alerts
|
||
if x.get("snow_alert") or x.get("rain_alert")
|
||
],
|
||
}
|
||
save_state(True, sig, detail)
|
||
notify_webapp("Allerta percorso scuola", plain, severity="warning")
|
||
else:
|
||
LOGGER.info("Allerta già notificata e invariata.")
|
||
# Aggiorna comunque summary nello state se manca (card WebApp)
|
||
if not state.get("summary"):
|
||
plain = build_plain_summary(bo_alerts, route_alerts)
|
||
save_state(True, sig, {
|
||
"summary": plain,
|
||
"window_hours": HOURS_AHEAD,
|
||
"persist_hours": PERSIST_HOURS,
|
||
"rain_threshold_mm_3h": SOGLIA_PIOGGIA_3H_MM,
|
||
"first_event_time": first_event_time(bo_alerts, route_alerts),
|
||
})
|
||
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>"
|
||
)
|
||
plain = (
|
||
f"Allerta rientrata (Bologna / percorso scuola).\n"
|
||
f"Nelle prossime {HOURS_AHEAD} ore non risultano più neve persistente "
|
||
f"(≥{PERSIST_HOURS}h) né pioggia 3h sopra soglia (≥{PERSIST_HOURS}h)."
|
||
)
|
||
ok = telegram_send_html(msg, chat_ids=chat_ids)
|
||
if ok:
|
||
LOGGER.info("Rientro notificato su Telegram.")
|
||
else:
|
||
LOGGER.info("Rientro Telegram saltato/sospeso; consegna via WebApp.")
|
||
save_state(False, "")
|
||
notify_webapp("Allerta percorso scuola rientrata", plain, severity="info")
|
||
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)
|