#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import datetime
import html
import json
import logging
import os
import tempfile
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
# =============================================================================
# arome_snow_alert.py
#
# Scopo:
# Monitorare neve prevista nelle prossime 48 ore su più punti (Casa/Titano/Dogana/Carpegna)
# e notificare su Telegram se:
# - esiste almeno 1 ora nelle prossime 48h con snowfall > 0.2 cm
# (nessuna persistenza richiesta)
#
# Modello meteo:
# meteofrance_seamless (fornisce snowfall, rain, weathercode)
# Comparazione con italia_meteo_arpae_icon_2i quando scostamento >30%
#
# Token Telegram:
# Nessun token in chiaro. Lettura in ordine:
# 1) env TELEGRAM_BOT_TOKEN
# 2) ~/.telegram_dpc_bot_token
# 3) /etc/telegram_dpc_bot_token
#
# Debug:
# DEBUG=1 python3 arome_snow_alert.py
#
# Log:
# ./arome_snow_alert.log (stessa cartella dello script)
# =============================================================================
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"
# ----------------- PUNTI DI MONITORAGGIO -----------------
POINTS = [
{"name": "🏠 Casa", "lat": 43.9356, "lon": 12.4296},
{"name": "⛰️ Titano", "lat": 43.9360, "lon": 12.4460},
{"name": "🏢 Dogana", "lat": 43.9800, "lon": 12.4900},
{"name": "🏔️ Carpegna", "lat": 43.7819, "lon": 12.3346},
]
# ----------------- LOGICA ALLERTA -----------------
TZ = "Europe/Berlin"
TZINFO = ZoneInfo(TZ)
HOURS_AHEAD = 48
# Soglia precipitazione neve oraria (cm/h): NOTIFICA se qualsiasi ora > soglia
SNOW_HOURLY_THRESHOLD_CM = 0.2
# Codici meteo che indicano neve (WMO)
SNOW_WEATHER_CODES = [71, 73, 75, 77, 85, 86] # Neve leggera, moderata, forte, granelli, rovesci
# Soglie per notifiche incrementali (cambiamenti significativi)
SNOW_QUANTITY_CHANGE_THRESHOLD_PCT = 20.0 # 20% variazione
SNOW_QUANTITY_CHANGE_THRESHOLD_CM = 1.0 # O almeno 1 cm
START_TIME_CHANGE_THRESHOLD_HOURS = 2.0 # ±2 ore
# Aggiornamenti periodici per eventi prolungati
PERIODIC_UPDATE_INTERVAL_HOURS = 6.0 # Invia aggiornamento ogni 6 ore anche senza cambiamenti significativi
# Stagione invernale: 1 Nov -> 15 Apr
WINTER_START_MONTH = 11
WINTER_END_MONTH = 4
WINTER_END_DAY = 15
# File di stato
STATE_FILE = "/home/daniely/docker/telegram-bot/snow_state.json"
# ----------------- OPEN-METEO -----------------
OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
HTTP_HEADERS = {"User-Agent": "rpi-arome-snow-alert/3.0"}
MODEL_AROME = "meteofrance_seamless"
MODEL_ICON_IT = "italia_meteo_arpae_icon_2i"
COMPARISON_THRESHOLD = 0.30 # 30% scostamento per comparazione
# ----------------- LOG FILE -----------------
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
LOG_FILE = os.path.join(BASE_DIR, "arome_snow_alert.log")
def setup_logger() -> logging.Logger:
logger = logging.getLogger("arome_snow_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 is_winter_season() -> bool:
"""True se oggi è tra 1 Novembre e 15 Aprile (in TZ locale)."""
now = now_local()
m = now.month
d = now.day
if m >= WINTER_START_MONTH:
return True
if m <= 3:
return True
if m == WINTER_END_MONTH and d <= WINTER_END_DAY:
return True
return False
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")
def ddmmyy_hhmm(dt: datetime.datetime) -> str:
"""Formato: DD/MM HH:MM"""
return dt.strftime("%d/%m %H:%M")
# =============================================================================
# Telegram
# =============================================================================
def telegram_send_html(message_html: str, chat_ids: Optional[List[str]] = None) -> bool:
"""Invia solo in caso di allerta (mai per errori).
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
# Telegram HTML non supporta
come tag self-closing
# Sostituiamo
con \n e usiamo Markdown
message_md = message_html.replace("
", "\n").replace("
", "\n")
# Manteniamo i tag HTML base (b, i, code, pre) convertendoli in Markdown
message_md = message_md.replace("", "*").replace("", "*")
message_md = message_md.replace("", "_").replace("", "_")
message_md = message_md.replace("", "`").replace("", "`")
message_md = message_md.replace(">", ">").replace("<", "<")
message_md = message_md.replace("&", "&")
# rimane come codice monospazio in Markdown
message_md = message_md.replace("", "```\n").replace("", "\n```")
# Correggi doppio %% (HTML entity) -> % (Markdown non richiede escape)
message_md = message_md.replace("%%", "%")
if chat_ids is None:
chat_ids = TELEGRAM_CHAT_IDS
url = f"https://api.telegram.org/bot{token}/sendMessage"
base_payload = {
"text": message_md,
"parse_mode": "Markdown",
"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
def telegram_send_photo(photo_path: str, caption: str, chat_ids: Optional[List[str]] = None) -> bool:
"""
Invia foto via Telegram API.
Args:
photo_path: Percorso file immagine
caption: Didascalia foto (max 1024 caratteri)
chat_ids: Lista chat IDs (default: TELEGRAM_CHAT_IDS)
Returns:
True se inviata con successo, False altrimenti
"""
token = load_bot_token()
if not token:
LOGGER.warning("Telegram token missing: photo not sent.")
return False
if not os.path.exists(photo_path):
LOGGER.error("File foto non trovato: %s", photo_path)
return False
if chat_ids is None:
chat_ids = TELEGRAM_CHAT_IDS
url = f"https://api.telegram.org/bot{token}/sendPhoto"
# Limite Telegram per caption: 1024 caratteri
if len(caption) > 1024:
caption = caption[:1021] + "..."
sent_ok = False
with requests.Session() as s:
for chat_id in chat_ids:
try:
with open(photo_path, 'rb') as photo_file:
files = {'photo': photo_file}
data = {
'chat_id': chat_id,
'caption': caption,
'parse_mode': 'HTML'
}
resp = s.post(url, files=files, data=data, timeout=30)
if resp.status_code == 200:
sent_ok = True
LOGGER.info("Foto inviata a chat_id=%s", chat_id)
else:
LOGGER.error("Telegram error chat_id=%s status=%s body=%s",
chat_id, resp.status_code, resp.text[:500])
time.sleep(0.5)
except Exception as e:
LOGGER.exception("Telegram exception chat_id=%s err=%s", chat_id, e)
return sent_ok
# =============================================================================
# State
# =============================================================================
def load_state() -> Dict:
default = {
"alert_active": False,
"signature": "",
"updated": "",
"last_notification_utc": "", # Timestamp UTC dell'ultima notifica inviata
"casa_snow_24h": 0.0,
"casa_peak_hourly": 0.0,
"casa_first_thr_time": "",
"casa_duration_hours": 0.0,
}
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, casa_data: Optional[Dict] = None, last_notification_utc: Optional[str] = None) -> None:
try:
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
state_data = {
"alert_active": alert_active,
"signature": signature,
"updated": now_local().isoformat(),
}
if last_notification_utc:
state_data["last_notification_utc"] = last_notification_utc
if casa_data:
state_data.update({
"casa_snow_24h": casa_data.get("snow_24h", 0.0),
"casa_peak_hourly": casa_data.get("peak_hourly", 0.0),
"casa_first_thr_time": casa_data.get("first_thr_time", ""),
"casa_duration_hours": casa_data.get("duration_hours", 0.0),
})
with open(STATE_FILE, "w", encoding="utf-8") as f:
json.dump(state_data, f, ensure_ascii=False, indent=2)
except Exception as e:
LOGGER.exception("State write error: %s", e)
# =============================================================================
# Open-Meteo
# =============================================================================
def get_forecast(session: requests.Session, lat: float, lon: float, model: str, extended: bool = False) -> Optional[Dict]:
"""
Recupera previsioni meteo.
Args:
session: Session HTTP
lat: Latitudine
lon: Longitudine
model: Modello meteo
extended: Se True, richiede dati estesi (fino a 10 giorni) senza minutely_15 per analisi a lungo termine
"""
# Parametri hourly: aggiungi rain e snow_depth quando necessario
hourly_params = "snowfall,weathercode"
if model == MODEL_AROME:
hourly_params += ",rain" # AROME fornisce rain
elif model == MODEL_ICON_IT:
hourly_params += ",rain,snow_depth" # ICON Italia fornisce rain e snow_depth
params = {
"latitude": lat,
"longitude": lon,
"hourly": hourly_params,
"daily": "snowfall_sum", # Aggiungi daily per colpo d'occhio 24/48h
"timezone": TZ,
"models": model,
}
if extended:
# Per analisi estesa, richiedi fino a 10 giorni (solo hourly, no minutely_15)
params["forecast_days"] = 10
else:
params["forecast_days"] = 2
# Aggiungi minutely_15 per AROME Seamless (dettaglio 15 minuti per inizio preciso)
# Se fallisce, riprova senza minutely_15
if model == MODEL_AROME:
params["minutely_15"] = "snowfall,precipitation_probability,precipitation,rain,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
try:
j = r.json()
LOGGER.error("Open-Meteo 400 (model=%s lat=%.4f lon=%.4f): %s", model, lat, lon, j.get("reason", j))
except Exception:
LOGGER.error("Open-Meteo 400 (model=%s lat=%.4f lon=%.4f): %s", model, lat, lon, r.text[:500])
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
LOGGER.error("Open-Meteo 504 Gateway Timeout (model=%s lat=%.4f lon=%.4f)", model, lat, lon)
return None
r.raise_for_status()
data = r.json()
# Converti snow_depth da metri a cm per ICON Italia (Open-Meteo restituisce in metri)
hourly_data = data.get("hourly", {})
if hourly_data and "snow_depth" in hourly_data and model == MODEL_ICON_IT:
snow_depth_values = hourly_data.get("snow_depth", [])
# Converti da metri a cm (moltiplica per 100)
snow_depth_cm = []
for sd in snow_depth_values:
if sd is not None:
try:
val_m = float(sd)
val_cm = val_m * 100.0 # Converti da metri a cm
snow_depth_cm.append(val_cm)
except (ValueError, TypeError):
snow_depth_cm.append(None)
else:
snow_depth_cm.append(None)
hourly_data["snow_depth"] = snow_depth_cm
data["hourly"] = hourly_data
# 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 lat=%.4f lon=%.4f)", model, lat, lon)
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 lat=%.4f lon=%.4f): %s", model, lat, lon, e)
return None
def compare_models(arome_snow_24h: float, icon_snow_24h: float) -> Optional[Dict]:
"""Confronta i due modelli e ritorna info se scostamento >30%"""
if arome_snow_24h == 0 and icon_snow_24h == 0:
return None
# Calcola scostamento percentuale
if arome_snow_24h > 0:
diff_pct = abs(icon_snow_24h - arome_snow_24h) / arome_snow_24h
elif icon_snow_24h > 0:
diff_pct = abs(arome_snow_24h - icon_snow_24h) / icon_snow_24h
else:
return None
if diff_pct > COMPARISON_THRESHOLD:
return {
"diff_pct": diff_pct * 100,
"arome": arome_snow_24h,
"icon": icon_snow_24h
}
return None
def find_extended_end_time(session: requests.Session, lat: float, lon: float, model: str, first_thr_dt: datetime.datetime, now: datetime.datetime) -> Optional[str]:
"""
Trova la fine prevista del fenomeno nevoso usando dati estesi (hourly fino a 10 giorni).
Args:
session: Session HTTP
lat: Latitudine
lon: Longitudine
model: Modello meteo
first_thr_dt: Data/ora di inizio del fenomeno
now: Data/ora corrente
Returns:
Stringa con data/ora di fine prevista (formato DD/MM HH:MM) o None se non trovata
"""
try:
# Richiedi dati estesi (hourly, no minutely_15, fino a 10 giorni)
data_extended = get_forecast(session, lat, lon, model, extended=True)
if not data_extended:
return None
hourly = data_extended.get("hourly", {}) or {}
times_ext = hourly.get("time", []) or []
snow_ext = hourly.get("snowfall", []) or []
if not times_ext or not snow_ext:
return None
# Cerca l'ultima ora sopra soglia nei dati estesi
last_thr_dt_ext = None
for i in range(len(times_ext) - 1, -1, -1):
try:
dt_ext = parse_time_to_local(times_ext[i])
if dt_ext < first_thr_dt:
break # Non andare prima dell'inizio
if dt_ext < now:
continue # Salta il passato
s_val = float(snow_ext[i]) if snow_ext[i] is not None else 0.0
if s_val > SNOW_HOURLY_THRESHOLD_CM:
last_thr_dt_ext = dt_ext
break
except Exception:
continue
if last_thr_dt_ext:
return ddmmyy_hhmm(last_thr_dt_ext)
return None
except Exception as e:
LOGGER.exception("Errore nel calcolo fine estesa: %s", e)
return None
# =============================================================================
# Analytics
# =============================================================================
def compute_snow_stats(data: Dict) -> Optional[Dict]:
hourly = data.get("hourly", {}) or {}
daily = data.get("daily", {}) or {}
times = hourly.get("time", []) or []
snow = hourly.get("snowfall", []) or []
weathercode = hourly.get("weathercode", []) or [] # Per rilevare neve anche quando snowfall è basso
# Recupera snowfall_sum dai dati daily (solo per riferimento, non usato per calcoli)
# NOTA: I calcoli di accumulo (3h, 6h, 12h, 24h, 48h) vengono fatti da dati hourly
# perché daily rappresenta l'accumulo del giorno solare, non delle prossime N ore
n = min(len(times), len(snow))
if n == 0:
return None
times = times[:n]
snow = snow[:n]
now = now_local()
start_idx = -1
for i, t in enumerate(times):
try:
if parse_time_to_local(t) >= now:
start_idx = i
break
except Exception:
continue
if start_idx == -1:
return None
# Analizza sempre le prossime 48 ore
end_idx = min(start_idx + HOURS_AHEAD, n)
if end_idx <= start_idx:
return None
times_w = times[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]
# Recupera snow_depth se disponibile (solo per ICON Italia)
snow_depth_w = []
snow_depth_max = 0.0
if "snow_depth" in hourly:
snow_depth_raw = hourly.get("snow_depth", []) or []
if len(snow_depth_raw) > start_idx:
snow_depth_w = [float(x) if x is not None else None for x in snow_depth_raw[start_idx:end_idx]]
# Calcola massimo snow_depth nelle 48 ore (solo valori validi)
valid_sd = [sd for sd in snow_depth_w if sd is not None and sd >= 0]
if valid_sd:
snow_depth_max = max(valid_sd)
# Verifica se c'è un fenomeno nevoso nelle 48 ore
# Considera neve se: snowfall > soglia OPPURE weather_code indica neve
has_snow_event = any(
(s > SNOW_HOURLY_THRESHOLD_CM) or
(i < len(weathercode_w) and weathercode_w[i] in SNOW_WEATHER_CODES)
for i, s in enumerate(snow_w)
)
# Accumuli informativi (solo ore con neve effettiva)
def sum_h(h: int) -> float:
upto = min(h, len(snow_w))
# Somma solo snowfall > 0 (accumulo totale previsto)
return float(sum(s for s in snow_w[:upto] if s > 0.0))
s3 = sum_h(3)
s6 = sum_h(6)
s12 = sum_h(12)
s24 = sum_h(24) # Sempre calcolato da hourly (prossime 24h)
s48 = sum_h(48) # Calcolato da hourly (prossime 48h)
# Accumulo totale previsto (somma di tutti i snowfall orari > 0 nelle 48h)
total_accumulation_cm = float(sum(s for s in snow_w if s > 0.0))
# NOTA: NON usiamo snow_24h_daily[0] perché è l'accumulo di OGGI, non delle prossime 24h
# snow_48h viene calcolato correttamente da hourly (prossime 48h), non da daily[1]
# Picco orario e prima occorrenza di neve (snowfall > 0 OPPURE weathercode neve)
peak = max(snow_w) if snow_w else 0.0
peak_time = ""
first_snow_time = "" # Inizio nevicata (snowfall > 0 o weathercode)
first_snow_val = 0.0
first_thr_time = "" # Prima occorrenza sopra soglia (per compatibilità)
first_thr_val = 0.0
# Usa minutely_15 per trovare inizio preciso se disponibile
minutely = data.get("minutely_15", {}) or {}
minutely_times = minutely.get("time", []) or []
minutely_snow = minutely.get("snowfall", []) or []
minutely_available = bool(minutely_times) and len(minutely_times) > 0
if minutely_available:
# Trova prima occorrenza precisa (risoluzione 15 minuti)
# Analizza sempre le 48 ore
minutely_weathercode = hourly.get("weathercode", []) or [] # Minutely non ha weathercode, usiamo hourly come fallback
for i, (t_str, s_val) in enumerate(zip(minutely_times, minutely_snow)):
try:
dt_min = parse_time_to_local(t_str)
if dt_min < now or dt_min > (now + datetime.timedelta(hours=HOURS_AHEAD)):
continue
s_float = float(s_val) if s_val is not None else 0.0
# Rileva inizio neve: snowfall > 0 (anche se sotto soglia)
if s_float > 0.0:
if not first_snow_time:
first_snow_time = ddmmyy_hhmm(dt_min)
first_snow_val = s_float
first_thr_time = first_snow_time # Per compatibilità
first_thr_val = s_float
# Rileva picco sopra soglia
if s_float > SNOW_HOURLY_THRESHOLD_CM:
if s_float > peak:
peak = s_float
peak_time = hhmm(dt_min)
except Exception:
continue
# Fallback a dati hourly se minutely non disponibile o non ha trovato nulla
if not first_snow_time:
for i, (v, code) in enumerate(zip(snow_w, weathercode_w if len(weathercode_w) == len(snow_w) else [None] * len(snow_w))):
# Rileva inizio neve: snowfall > 0 OPPURE weathercode indica neve
is_snow = (v > 0.0) or (code is not None and code in SNOW_WEATHER_CODES)
if is_snow and not first_snow_time:
first_snow_time = ddmmyy_hhmm(dt_w[i])
first_snow_val = v if v > 0.0 else 0.0
first_thr_time = first_snow_time # Per compatibilità
first_thr_val = first_snow_val
# Rileva picco
if v > peak:
peak = v
peak_time = hhmm(dt_w[i])
if not peak_time and peak > 0 and dt_w:
try:
peak_i = snow_w.index(peak)
peak_time = hhmm(dt_w[peak_i])
except Exception:
peak_time = ""
# Calcola durata del fenomeno e andamento
duration_hours = 0.0
snow_timeline = [] # Lista di (datetime, valore) per grafico
first_snow_dt = None
last_snow_dt = None
# Usa first_snow_time se disponibile, altrimenti first_thr_time
start_time_to_use = first_snow_time if first_snow_time else first_thr_time
if start_time_to_use:
# Trova datetime di inizio (prima occorrenza di neve)
for i, dt in enumerate(dt_w):
snow_val = snow_w[i]
code = weathercode_w[i] if i < len(weathercode_w) else None
is_snow = (snow_val > 0.0) or (code is not None and code in SNOW_WEATHER_CODES)
if ddmmyy_hhmm(dt) == start_time_to_use or (not first_snow_dt and is_snow):
first_snow_dt = dt
break
# Trova fine del fenomeno (ultima ora con neve: snowfall > 0 OPPURE weathercode)
if first_snow_dt:
for i in range(len(dt_w) - 1, -1, -1):
snow_val = snow_w[i]
code = weathercode_w[i] if i < len(weathercode_w) else None
is_snow = (snow_val > 0.0) or (code is not None and code in SNOW_WEATHER_CODES)
if is_snow:
last_snow_dt = dt_w[i]
break
if last_snow_dt:
duration_hours = (last_snow_dt - first_snow_dt).total_seconds() / 3600.0
# Costruisci timeline per grafico (usa minutely se disponibile, altrimenti hourly)
# La timeline include tutto il fenomeno nelle 48 ore
if minutely_available:
for i, (t_str, s_val) in enumerate(zip(minutely_times, minutely_snow)):
try:
dt_min = parse_time_to_local(t_str)
if dt_min < first_snow_dt or dt_min > last_snow_dt:
continue
# Limita la timeline alle 48 ore
if dt_min > (now + datetime.timedelta(hours=HOURS_AHEAD)):
continue
s_float = float(s_val) if s_val is not None else 0.0
snow_timeline.append((dt_min, s_float))
except Exception:
continue
else:
# Usa hourly - include tutto il fenomeno fino alla fine
for i, dt in enumerate(dt_w):
if dt >= first_snow_dt and dt <= last_snow_dt:
snow_timeline.append((dt, snow_w[i]))
# Calcola orario di fine precipitazioni
last_thr_time = ""
is_ongoing = False # Indica se il fenomeno continua oltre la finestra analizzata
extended_end_time = "" # Fine prevista se estesa oltre 48h
if last_snow_dt:
last_thr_time = ddmmyy_hhmm(last_snow_dt)
# Verifica se il fenomeno continua oltre le 48 ore analizzate
# Questo accade se l'ultimo punto analizzato (dopo 48 ore) ha ancora neve
if has_snow_event:
# Controlla se siamo al limite delle 48 ore e l'ultimo punto ha ancora neve
hours_analyzed = (end_idx - start_idx)
if hours_analyzed >= HOURS_AHEAD and len(snow_w) > 0:
last_snow_val = snow_w[-1]
last_code = weathercode_w[-1] if len(weathercode_w) > 0 else None
last_has_snow = (last_snow_val > 0.0) or (last_code is not None and last_code in SNOW_WEATHER_CODES)
if last_has_snow:
is_ongoing = True
# L'ultimo punto ha ancora neve, quindi il fenomeno continua oltre le 48h
# La fine effettiva sarà calcolata con dati estesi se disponibili
return {
"snow_3h": s3,
"snow_6h": s6,
"snow_12h": s12,
"snow_24h": s24,
"snow_48h": s48, # Calcolato da hourly (prossime 48h), non da daily
"total_accumulation_cm": total_accumulation_cm, # Accumulo totale previsto (somma di tutti i snowfall > 0)
"peak_hourly": float(peak),
"peak_time": peak_time,
"first_thr_time": first_thr_time, # Per compatibilità
"first_thr_val": float(first_thr_val), # Per compatibilità
"first_snow_time": first_snow_time if first_snow_time else first_thr_time, # Inizio nevicata
"first_snow_val": float(first_snow_val),
"last_thr_time": last_thr_time,
"duration_hours": duration_hours,
"snow_timeline": snow_timeline,
"triggered": bool(first_snow_time or first_thr_time),
"is_ongoing": is_ongoing, # Indica se continua oltre 48h
"extended_end_time": extended_end_time, # Fine prevista se estesa
"snow_depth_max": float(snow_depth_max), # Massimo snow_depth nelle 48 ore (cm)
}
def point_summary(name: str, st: Dict) -> Dict:
return {
"name": name,
"triggered": bool(st["triggered"]),
"snow_3h": st["snow_3h"],
"snow_6h": st["snow_6h"],
"snow_12h": st["snow_12h"],
"snow_24h": st["snow_24h"],
"snow_48h": st.get("snow_48h"),
"snow_depth_max": st.get("snow_depth_max", 0.0),
"peak_hourly": st["peak_hourly"],
"peak_time": st["peak_time"],
"first_thr_time": st["first_thr_time"],
"first_thr_val": st["first_thr_val"],
"last_thr_time": st.get("last_thr_time", ""),
"duration_hours": st.get("duration_hours", 0.0),
"snow_timeline": st.get("snow_timeline", []),
"is_ongoing": st.get("is_ongoing", False),
"extended_end_time": st.get("extended_end_time", ""),
}
def build_signature(summaries: List[Dict]) -> str:
# Firma per evitare spam: arrotondiamo a 0.1 cm
parts = []
for s in summaries:
parts.append(
f"{s['name']}:t{int(s['triggered'])}"
f":24={s['snow_24h']:.1f}"
f":pk={s['peak_hourly']:.1f}"
f":ft={s['first_thr_time'] or '-'}"
)
return "|".join(parts)
def generate_snow_chart(timeline: List[Tuple[datetime.datetime, float]], max_width: int = 50) -> str:
"""Genera grafico ASCII orizzontale dell'andamento nevoso (barre affiancate)"""
if not timeline:
return ""
if len(timeline) == 0:
return ""
# Estrai valori
values = [v for _, v in timeline]
if not values or max(values) == 0:
return ""
max_val = max(values)
# Caratteri per barre verticali (8 livelli)
bars = "▁▂▃▄▅▆▇█"
# Determina se i dati sono a 15 minuti o orari
is_15min = len(timeline) > 20
# Frequenza etichette temporali: ogni ora (4 barre se 15min, 1 barra se hourly)
label_freq = 4 if is_15min else 1
# Crea grafico orizzontale: barre affiancate
# Ogni riga può contenere max_width barre, poi va a capo
result_lines = []
# Processa timeline in blocchi di max_width
for start_idx in range(0, len(timeline), max_width):
end_idx = min(start_idx + max_width, len(timeline))
block = timeline[start_idx:end_idx]
# Crea riga etichette temporali (solo ogni ora)
# Usa una lista di caratteri per allineare perfettamente con le barre
# Ogni carattere corrisponde esattamente a una barra
time_line_chars = [" "] * len(block)
bar_line_parts = []
for i, (dt, val) in enumerate(block):
global_idx = start_idx + i
# Calcola livello barra
if max_val > 0:
level = int((val / max_val) * 7)
level = min(level, 7)
else:
level = 0
bar = bars[level]
bar_line_parts.append(bar)
# Aggiungi etichetta temporale solo ogni ora, posizionata sopra la prima barra dell'ora
# Mostra anche l'etichetta se siamo all'ultimo punto e corrisponde a un'ora intera
is_hour_start = (global_idx == 0 or global_idx % label_freq == 0)
is_last_point = (global_idx == len(timeline) - 1)
# Se è l'ultimo punto e l'ora è intera (minuti == 0), mostra l'etichetta
if is_hour_start or (is_last_point and dt.minute == 0):
time_str = dt.strftime("%H:%M")
hour_str = time_str[:2] # Prendi solo "HH" (es. "07")
# Posiziona l'etichetta sopra le prime due barre dell'ora
# Ogni carattere dell'etichetta corrisponde esattamente a una barra
# IMPORTANTE: se il blocco ha solo 1 elemento, estendi time_line_chars per accogliere entrambe le cifre
if i + 1 >= len(time_line_chars) and len(hour_str) > 1:
# Estendi time_line_chars se necessario per la seconda cifra
time_line_chars.extend([" "] * (i + 2 - len(time_line_chars)))
if i < len(time_line_chars):
time_line_chars[i] = hour_str[0] if len(hour_str) > 0 else " "
if i + 1 < len(time_line_chars) and len(hour_str) > 1:
time_line_chars[i + 1] = hour_str[1]
# Le posizioni i+2 e i+3 (se esistono) rimangono spazi per le altre barre dell'ora
# Tronca time_line_chars alla lunghezza effettiva delle barre per allineamento
# Ma assicurati che sia almeno lunga quanto le barre
time_line_chars = time_line_chars[:max(len(bar_line_parts), len(time_line_chars))]
# Crea le righe - ogni carattere dell'etichetta corrisponde a una barra
time_line = "".join(time_line_chars)
bar_line = "".join(bar_line_parts)
# Assicurati che le righe abbiano la stessa lunghezza
max_len = max(len(time_line), len(bar_line))
time_line = time_line.ljust(max_len)
bar_line = bar_line.ljust(max_len)
# Aggiungi al risultato (etichetta sopra barre)
result_lines.append(time_line)
result_lines.append(bar_line)
# Aggiungi riga vuota tra i blocchi (tranne l'ultimo)
if end_idx < len(timeline):
result_lines.append("")
return "\n".join(result_lines)
def generate_snow_chart_image(
data_arome: Dict,
data_icon: Optional[Dict],
output_path: str,
location_name: str = "Casa"
) -> bool:
"""
Genera un grafico orario di 48 ore con linee parallele per:
- snow_depth da ICON Italia (☃️)
- snowfall da AROME seamless (❄️)
- rain da AROME seamless (💧)
- snowfall da ICON Italia (❄️)
- rain da ICON Italia (💧)
Args:
data_arome: Dati AROME seamless
data_icon: Dati ICON Italia (opzionale)
output_path: Percorso file immagine output
location_name: Nome località per titolo
Returns:
True se generato con successo, False altrimenti
"""
try:
import matplotlib
matplotlib.use('Agg') # Backend non-interattivo
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.lines import Line2D
# Configura font per supportare emoji
# Nota: matplotlib potrebbe non supportare completamente emoji colorati,
# ma proveremo a usare caratteri Unicode alternativi se il font emoji non è disponibile
try:
import matplotlib.font_manager as fm
import os
# Cerca direttamente il percorso del font Noto Color Emoji
possible_paths = [
'/usr/share/fonts/truetype/noto/NotoColorEmoji.ttf',
'/usr/share/fonts/opentype/noto/NotoColorEmoji.ttf',
'/usr/share/fonts/NotoColorEmoji.ttf',
]
emoji_font_path = None
for path in possible_paths:
if os.path.exists(path):
emoji_font_path = path
break
if emoji_font_path:
# Carica il font direttamente e registralo
try:
# Registra il font con matplotlib
fm.fontManager.addfont(emoji_font_path)
prop = fm.FontProperties(fname=emoji_font_path)
# Configura matplotlib per usare questo font
plt.rcParams['font.family'] = 'sans-serif'
# Aggiungi alla lista dei font di fallback
if 'Noto Color Emoji' not in plt.rcParams['font.sans-serif']:
plt.rcParams['font.sans-serif'].insert(0, 'Noto Color Emoji')
LOGGER.debug("Font emoji caricato: %s", emoji_font_path)
except Exception as e:
LOGGER.debug("Errore caricamento font emoji: %s", e)
except Exception as e:
LOGGER.debug("Errore configurazione font emoji: %s", e)
pass # Se fallisce, usa il font di default
except ImportError:
LOGGER.warning("matplotlib non disponibile: grafico non generato")
return False
try:
# Estrai dati hourly da AROME
hourly_arome = data_arome.get("hourly", {}) or {}
times_arome = hourly_arome.get("time", []) or []
snowfall_arome = hourly_arome.get("snowfall", []) or []
rain_arome = hourly_arome.get("rain", []) or []
# Estrai dati hourly da ICON Italia (se disponibile)
snowfall_icon = []
rain_icon = []
snow_depth_icon = []
times_icon = []
if data_icon:
hourly_icon = data_icon.get("hourly", {}) or {}
times_icon = hourly_icon.get("time", []) or []
snowfall_icon = hourly_icon.get("snowfall", []) or []
rain_icon = hourly_icon.get("rain", []) or []
# snow_depth è già convertito in cm da get_forecast
snow_depth_icon = hourly_icon.get("snow_depth", []) or []
# Prendi solo le prossime 48 ore dall'ora corrente
now = now_local()
times_parsed = []
for t_str in times_arome:
try:
dt = parse_time_to_local(t_str)
if dt >= now and (dt - now).total_seconds() <= HOURS_AHEAD * 3600:
times_parsed.append(dt)
except Exception:
continue
if not times_parsed:
LOGGER.warning("Nessun dato valido per grafico nelle prossime 48 ore")
return False
# Filtra i dati per le ore valide
start_idx_arome = len(times_arome) - len(times_parsed)
if start_idx_arome < 0:
start_idx_arome = 0
times_plot = times_parsed[:48] # Massimo 48 ore
snowfall_arome_plot = []
rain_arome_plot = []
for i, dt in enumerate(times_plot):
idx_arome = start_idx_arome + i
if idx_arome < len(snowfall_arome):
snowfall_arome_plot.append(float(snowfall_arome[idx_arome]) if snowfall_arome[idx_arome] is not None else 0.0)
else:
snowfall_arome_plot.append(0.0)
if idx_arome < len(rain_arome):
rain_arome_plot.append(float(rain_arome[idx_arome]) if rain_arome[idx_arome] is not None else 0.0)
else:
rain_arome_plot.append(0.0)
# Allinea dati ICON Italia ai timestamp di AROME
snowfall_icon_plot = []
rain_icon_plot = []
snow_depth_icon_plot = []
if times_icon and data_icon:
# Crea mappa timestamp -> valori per ICON (con tolleranza per allineamento)
icon_map = {}
for idx, t_str in enumerate(times_icon):
try:
dt = parse_time_to_local(t_str)
if dt >= now and (dt - now).total_seconds() <= HOURS_AHEAD * 3600:
# Converti snow_depth se necessario (dovrebbe già essere in cm da get_forecast)
sd_val = snow_depth_icon[idx] if idx < len(snow_depth_icon) else None
if sd_val is not None:
try:
sd_val = float(sd_val)
except (ValueError, TypeError):
sd_val = None
icon_map[dt] = {
'snowfall': float(snowfall_icon[idx]) if idx < len(snowfall_icon) and snowfall_icon[idx] is not None else 0.0,
'rain': float(rain_icon[idx]) if idx < len(rain_icon) and rain_icon[idx] is not None else 0.0,
'snow_depth': sd_val
}
except Exception:
continue
# Allinea ai timestamp di AROME (cerca corrispondenza esatta o più vicina entro 30 minuti)
for dt in times_plot:
# Cerca corrispondenza esatta
if dt in icon_map:
val = icon_map[dt]
snowfall_icon_plot.append(val['snowfall'])
rain_icon_plot.append(val['rain'])
snow_depth_icon_plot.append(val['snow_depth'])
else:
# Cerca corrispondenza più vicina (entro 30 minuti)
found = False
best_dt = None
best_diff = None
for icon_dt, val in icon_map.items():
diff_seconds = abs((icon_dt - dt).total_seconds())
if diff_seconds <= 1800: # Entro 30 minuti
if best_diff is None or diff_seconds < best_diff:
best_diff = diff_seconds
best_dt = icon_dt
found = True
if found and best_dt:
val = icon_map[best_dt]
snowfall_icon_plot.append(val['snowfall'])
rain_icon_plot.append(val['rain'])
snow_depth_icon_plot.append(val['snow_depth'])
else:
# Nessuna corrispondenza trovata: usa valori zero/None
snowfall_icon_plot.append(0.0)
rain_icon_plot.append(0.0)
snow_depth_icon_plot.append(None)
else:
# Se ICON non disponibile, riempi con None/0
snowfall_icon_plot = [0.0] * len(times_plot)
rain_icon_plot = [0.0] * len(times_plot)
snow_depth_icon_plot = [None] * len(times_plot)
# Crea figura e assi
fig, ax = plt.subplots(figsize=(14, 8), facecolor='white')
fig.patch.set_facecolor('white')
# Colori e stili per le linee (accattivanti e ben distinguibili)
colors = {
'snow_depth_icon': '#1E90FF', # Dodger blue per snow_depth (☃️)
'snowfall_arome': '#DC143C', # Crimson per snowfall AROME (❄️)
'rain_arome': '#00AA00', # Dark green per rain AROME (💧)
'snowfall_icon': '#FF69B4', # Hot pink per snowfall ICON (❄️)
'rain_icon': '#20B2AA' # Light sea green per rain ICON (💧)
}
markers = {
'snow_depth_icon': 'D', # Diamond (☃️)
'snowfall_arome': 's', # Square (❄️)
'rain_arome': 'o', # Circle (💧)
'snowfall_icon': '^', # Triangle up (❄️)
'rain_icon': 'v' # Triangle down (💧)
}
# Traccia le linee
lines_handles = []
lines_labels = []
# 1. snow_depth da ICON Italia - scala a sinistra (cm)
if any(sd is not None and sd > 0 for sd in snow_depth_icon_plot):
valid_data = [(t, sd) for t, sd in zip(times_plot, snow_depth_icon_plot) if sd is not None and sd >= 0]
if valid_data:
times_sd, values_sd = zip(*valid_data)
line_sd = ax.plot(times_sd, values_sd, color=colors['snow_depth_icon'],
marker=markers['snow_depth_icon'], markersize=7, linewidth=3.0,
label='Manto nevoso (ICON Italia) [cm]', linestyle='-', alpha=0.95,
markeredgecolor='white', markeredgewidth=1.5)
lines_handles.append(line_sd[0])
lines_labels.append('Manto nevoso (ICON Italia) [cm]')
# 2. snowfall da AROME seamless - scala a sinistra (cm)
if any(s > 0 for s in snowfall_arome_plot):
line_sf_arome = ax.plot(times_plot, snowfall_arome_plot, color=colors['snowfall_arome'],
marker=markers['snowfall_arome'], markersize=6, linewidth=2.5,
label='Neve (AROME) [cm]', linestyle='-', alpha=0.9,
markeredgecolor='white', markeredgewidth=1.2)
lines_handles.append(line_sf_arome[0])
lines_labels.append('Neve (AROME) [cm]')
# 3. rain da AROME seamless - scala a sinistra (mm)
if any(r > 0 for r in rain_arome_plot):
line_r_arome = ax.plot(times_plot, rain_arome_plot, color=colors['rain_arome'],
marker=markers['rain_arome'], markersize=6, linewidth=2.5,
label='Pioggia (AROME) [mm]', linestyle='-', alpha=0.9,
markeredgecolor='white', markeredgewidth=1.2)
lines_handles.append(line_r_arome[0])
lines_labels.append('Pioggia (AROME) [mm]')
# 4. snowfall da ICON Italia - scala a sinistra (cm)
if any(s > 0 for s in snowfall_icon_plot):
line_sf_icon = ax.plot(times_plot, snowfall_icon_plot, color=colors['snowfall_icon'],
marker=markers['snowfall_icon'], markersize=6, linewidth=2.5,
label='Neve (ICON Italia) [cm]', linestyle='--', alpha=0.9,
markeredgecolor='white', markeredgewidth=1.2, dashes=(5, 3))
lines_handles.append(line_sf_icon[0])
lines_labels.append('Neve (ICON Italia) [cm]')
# 5. rain da ICON Italia - scala a sinistra (mm)
if any(r > 0 for r in rain_icon_plot):
line_r_icon = ax.plot(times_plot, rain_icon_plot, color=colors['rain_icon'],
marker=markers['rain_icon'], markersize=6, linewidth=2.5,
label='Pioggia (ICON Italia) [mm]', linestyle='--', alpha=0.9,
markeredgecolor='white', markeredgewidth=1.2, dashes=(5, 3))
lines_handles.append(line_r_icon[0])
lines_labels.append('Pioggia (ICON Italia) [mm]')
if not lines_handles:
LOGGER.warning("Nessun dato da visualizzare nel grafico")
plt.close(fig)
return False
# Configurazione assi (grafica accattivante e ben leggibile)
ax.set_xlabel('Ora', fontsize=13, fontweight='bold', color='#333333')
ax.set_ylabel('Precipitazioni (cm per neve, mm per pioggia)', fontsize=13, fontweight='bold', color='#333333')
# Rimuovi emoji dal titolo per evitare problemi con font (emoji nella legenda sono OK)
location_clean = location_name.replace('🏠 ', '').replace('💧', '').replace('❄️', '').strip()
ax.set_title(f'Previsioni Neve e Precipitazioni - {location_clean}\n48 ore',
fontsize=15, fontweight='bold', pad=25, color='#1a1a1a')
ax.grid(True, alpha=0.4, linestyle='--', linewidth=0.8, color='#888888')
ax.set_facecolor('#F8F8F8')
# Colora gli assi
ax.spines['top'].set_visible(False)
ax.spines['right'].set_visible(False)
ax.spines['left'].set_color('#666666')
ax.spines['bottom'].set_color('#666666')
# Formatta asse X (date/ore)
ax.xaxis.set_major_formatter(mdates.DateFormatter('%d/%m %H:%M'))
ax.xaxis.set_major_locator(mdates.HourLocator(interval=6)) # Ogni 6 ore
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45, ha='right', fontsize=10)
plt.setp(ax.yaxis.get_majorticklabels(), fontsize=10)
# Legenda migliorata (con testo descrittivo e marker colorati distinti)
legend = ax.legend(handles=lines_handles, labels=lines_labels, loc='upper left',
framealpha=0.98, fontsize=11, ncol=1, frameon=True,
shadow=True, fancybox=True, edgecolor='#CCCCCC')
legend.get_frame().set_facecolor('#FFFFFF')
legend.get_frame().set_linewidth(1.5)
# Migliora layout (sopprimi warning emoji nel font)
import warnings
with warnings.catch_warnings():
warnings.filterwarnings('ignore', category=UserWarning, message='.*Glyph.*missing from font.*')
plt.tight_layout()
# Salva immagine
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with warnings.catch_warnings():
warnings.filterwarnings('ignore', category=UserWarning, message='.*Glyph.*missing from font.*')
plt.savefig(output_path, dpi=150, facecolor='white', bbox_inches='tight')
plt.close(fig)
LOGGER.info("Grafico generato: %s", output_path)
return True
except Exception as e:
LOGGER.exception("Errore generazione grafico: %s", e)
return False
def is_significant_change(current: Dict, previous: Dict) -> Tuple[bool, str]:
"""Verifica se ci sono cambiamenti significativi. Ritorna (is_significant, reason)"""
if not previous.get("alert_active", False):
return True, "nuova_allerta"
prev_snow_24h = previous.get("casa_snow_24h", 0.0)
prev_peak = previous.get("casa_peak_hourly", 0.0)
prev_first_time = previous.get("casa_first_thr_time", "")
curr_snow_24h = current.get("snow_24h", 0.0)
curr_peak = current.get("peak_hourly", 0.0)
curr_first_time = current.get("first_thr_time", "")
# Annullamento
if not current.get("triggered", False):
return True, "annullamento"
# Variazione quantità (20% o 1 cm)
if prev_snow_24h > 0:
pct_change = abs(curr_snow_24h - prev_snow_24h) / prev_snow_24h * 100
abs_change = abs(curr_snow_24h - prev_snow_24h)
if pct_change >= SNOW_QUANTITY_CHANGE_THRESHOLD_PCT or abs_change >= SNOW_QUANTITY_CHANGE_THRESHOLD_CM:
return True, f"variazione_quantita_{'+' if curr_snow_24h > prev_snow_24h else '-'}"
# Variazione picco
if prev_peak > 0:
pct_change = abs(curr_peak - prev_peak) / prev_peak * 100
if pct_change >= SNOW_QUANTITY_CHANGE_THRESHOLD_PCT:
return True, f"variazione_picco_{'+' if curr_peak > prev_peak else '-'}"
# Variazione orario inizio (±2 ore)
if prev_first_time and curr_first_time:
try:
# Parse date from "DD/MM HH:MM"
prev_parts = prev_first_time.split()
curr_parts = curr_first_time.split()
if len(prev_parts) == 2 and len(curr_parts) == 2:
prev_date_str, prev_time_str = prev_parts
curr_date_str, curr_time_str = curr_parts
# Determina l'anno prima del parsing per evitare warning di deprecazione
now = now_local()
# Estrai mese e giorno dalla stringa senza parsing
prev_day, prev_month = map(int, prev_date_str.split("/"))
curr_day, curr_month = map(int, curr_date_str.split("/"))
# Determina l'anno appropriato
if prev_month > now.month or (prev_month == now.month and prev_day > now.day):
prev_year = now.year - 1
else:
prev_year = now.year
if curr_month > now.month or (curr_month == now.month and curr_day > now.day):
curr_year = now.year - 1
else:
curr_year = now.year
# Parse completo con anno incluso
prev_dt = datetime.datetime.strptime(f"{prev_date_str}/{prev_year} {prev_time_str}", "%d/%m/%Y %H:%M")
curr_dt = datetime.datetime.strptime(f"{curr_date_str}/{curr_year} {curr_time_str}", "%d/%m/%Y %H:%M")
hours_diff = abs((curr_dt - prev_dt).total_seconds() / 3600.0)
if hours_diff >= START_TIME_CHANGE_THRESHOLD_HOURS:
return True, f"variazione_orario_{hours_diff:.1f}h"
except Exception:
pass
return False, ""
def is_periodic_update_due(previous: Dict) -> bool:
"""
Verifica se è il momento di inviare un aggiornamento periodico.
Ritorna True se:
- L'allerta è attiva
- È passato almeno PERIODIC_UPDATE_INTERVAL_HOURS dall'ultima notifica
"""
if not previous.get("alert_active", False):
return False
last_notification_str = previous.get("last_notification_utc", "")
if not last_notification_str:
# Se non c'è timestamp, considera che sia dovuto (primo aggiornamento periodico)
return True
try:
last_notification = datetime.datetime.fromisoformat(last_notification_str)
if last_notification.tzinfo is None:
last_notification = last_notification.replace(tzinfo=datetime.timezone.utc)
else:
last_notification = last_notification.astimezone(datetime.timezone.utc)
now_utc = datetime.datetime.now(datetime.timezone.utc)
hours_since_last = (now_utc - last_notification).total_seconds() / 3600.0
return hours_since_last >= PERIODIC_UPDATE_INTERVAL_HOURS
except Exception as e:
LOGGER.debug("Errore parsing last_notification_utc: %s", e)
return True # In caso di errore, considera dovuto
# =============================================================================
# Main
# =============================================================================
def analyze_snow(chat_ids: Optional[List[str]] = None, debug_mode: bool = False) -> None:
if not is_winter_season():
LOGGER.info("Fuori stagione invernale: nessuna verifica/notify.")
save_state(False, "", None)
return
now_str = now_local().strftime("%H:%M")
LOGGER.info("--- Check Neve AROME Seamless %s ---", now_str)
state = load_state()
was_active = bool(state.get("alert_active", False))
last_sig = state.get("signature", "")
summaries: List[Dict] = []
summaries_icon: Dict[str, Dict] = {} # point_name -> icon summary (solo se scostamento significativo)
comparisons: Dict[str, Dict] = {} # point_name -> comparison info
# Conserva dati raw per Casa per generare grafico
casa_data_arome_raw: Optional[Dict] = None
casa_data_icon_raw: Optional[Dict] = None
casa_location_name = "🏠 Casa"
with requests.Session() as session:
configure_open_meteo_session(session, headers=HTTP_HEADERS)
for p in POINTS:
# Recupera AROME seamless
data_arome = get_forecast(session, p["lat"], p["lon"], MODEL_AROME)
if not data_arome:
LOGGER.warning("Forecast AROME non disponibile per %s (skip).", p["name"])
continue
# Conserva dati raw per Casa per generare grafico
if p["name"] == "🏠 Casa":
casa_data_arome_raw = data_arome
st_arome = compute_snow_stats(data_arome)
if not st_arome:
LOGGER.warning("Statistiche AROME non calcolabili per %s (skip).", p["name"])
continue
# Se il fenomeno continua oltre le 48 ore, cerca la fine nei dati estesi
if st_arome.get("is_ongoing", False) and st_arome.get("first_thr_time"):
try:
# Parse first_thr_time per ottenere first_thr_dt
first_thr_parts = st_arome["first_thr_time"].split()
if len(first_thr_parts) == 2:
first_date_str, first_time_str = first_thr_parts
first_day, first_month = map(int, first_date_str.split("/"))
now = now_local()
if first_month > now.month or (first_month == now.month and first_day > now.day):
first_year = now.year - 1
else:
first_year = now.year
first_thr_dt = datetime.datetime.strptime(f"{first_date_str}/{first_year} {first_time_str}", "%d/%m/%Y %H:%M")
# Cerca fine estesa (prova prima AROME, poi ICON se AROME non disponibile)
extended_end = find_extended_end_time(session, p["lat"], p["lon"], MODEL_AROME, first_thr_dt, now)
if not extended_end:
# Fallback a ICON Italia se AROME non ha dati estesi
extended_end = find_extended_end_time(session, p["lat"], p["lon"], MODEL_ICON_IT, first_thr_dt, now)
if extended_end:
st_arome["extended_end_time"] = extended_end
LOGGER.info("Fine estesa trovata per %s: %s", p["name"], extended_end)
except Exception as e:
LOGGER.exception("Errore nel calcolo fine estesa per %s: %s", p["name"], e)
# Recupera ICON Italia per comparazione (sempre per Casa, opzionale per altri)
# Per Casa recuperiamo sempre ICON per mostrare nel grafico e nel messaggio
if p["name"] == "🏠 Casa":
data_icon = get_forecast(session, p["lat"], p["lon"], MODEL_ICON_IT)
if data_icon:
casa_data_icon_raw = data_icon
st_icon = compute_snow_stats(data_icon)
if st_icon:
summaries_icon[p["name"]] = point_summary(p["name"], st_icon)
# Calcola comparazione per logging (ma mostra sempre ICON nel messaggio per Casa)
comp = compare_models(st_arome["snow_24h"], st_icon["snow_24h"])
if comp:
comparisons[p["name"]] = comp
LOGGER.info("Scostamento >30%% per %s: AROME=%.1f cm, ICON=%.1f cm (diff=%.1f%%)",
p["name"], comp["arome"], comp["icon"], comp["diff_pct"])
else:
# Anche senza scostamento significativo, mostra ICON per Casa se ha dati interessanti
# (snow_depth > 0 o snow_48h > 0 anche se snow_24h non supera soglia)
if st_icon.get("snow_depth_max", 0) > 0 or st_icon.get("snow_48h", 0) > 0:
LOGGER.info("ICON disponibile per Casa: snow_depth_max=%.1f cm, snow_48h=%.1f cm",
st_icon.get("snow_depth_max", 0), st_icon.get("snow_48h", 0))
else:
# Per altri punti, recupera ICON solo se necessario per comparazione
data_icon = get_forecast(session, p["lat"], p["lon"], MODEL_ICON_IT)
if data_icon:
st_icon = compute_snow_stats(data_icon)
if st_icon:
comp = compare_models(st_arome["snow_24h"], st_icon["snow_24h"])
if comp:
comparisons[p["name"]] = comp
summaries_icon[p["name"]] = point_summary(p["name"], st_icon)
LOGGER.info("Scostamento >30%% per %s: AROME=%.1f cm, ICON=%.1f cm (diff=%.1f%%)",
p["name"], comp["arome"], comp["icon"], comp["diff_pct"])
summaries.append(point_summary(p["name"], st_arome))
time.sleep(0.2)
if not summaries:
LOGGER.error("Nessun punto ha restituito statistiche valide.")
return
# Calcola any_trigger escludendo Carpegna (solo Casa, Titano, Dogana possono attivare l'allerta)
# Carpegna viene comunque mostrato nel report se ha trigger, ma non attiva l'allerta
any_trigger = any(s["triggered"] for s in summaries if s["name"] != "🏔️ Carpegna")
# Verifica se solo Carpegna ha trigger (per logging)
carpegna_triggered = any(s["triggered"] for s in summaries if s["name"] == "🏔️ Carpegna")
if carpegna_triggered and not any_trigger:
LOGGER.info("Neve prevista solo a Carpegna: allerta non inviata (Carpegna escluso dal trigger principale)")
sig = build_signature(summaries)
# Prepara dati Casa per verifica cambiamenti
casa = next((s for s in summaries if s["name"] == "🏠 Casa"), None)
casa_data = None
if casa:
casa_data = {
"snow_24h": casa["snow_24h"],
"peak_hourly": casa["peak_hourly"],
"first_thr_time": casa["first_thr_time"],
"duration_hours": casa.get("duration_hours", 0.0),
"triggered": casa["triggered"],
}
# Verifica cambiamenti significativi
is_significant, change_reason = is_significant_change(
casa_data or {},
state
) if casa_data else (True, "nuova_allerta")
# Verifica se è il momento di un aggiornamento periodico
is_periodic_due = is_periodic_update_due(state)
# --- Scenario A: soglia superata ---
if any_trigger:
# Se change_reason è "annullamento" ma any_trigger è True, significa che Casa non ha più trigger
# ma altri punti sì. Non è un annullamento generale, quindi trattiamolo come aggiornamento normale.
if change_reason == "annullamento":
change_reason = "variazione_punti_monitorati"
is_significant = True
# Invia notifica se:
# - Prima allerta (not was_active)
# - Cambiamento significativo
# - Aggiornamento periodico dovuto (ogni 6 ore)
# - Modalità debug (bypassa controlli)
if debug_mode or (not was_active) or is_significant or is_periodic_due:
if debug_mode:
LOGGER.info("[DEBUG MODE] Bypass anti-spam: invio forzato")
msg: List[str] = []
# Titolo con indicazione tipo notifica
# Nota: "annullamento" non può verificarsi nello Scenario A (any_trigger=True)
if not was_active:
title = "❄️ ALLERTA NEVE"
elif is_periodic_due and not is_significant:
title = "⏱️ AGGIORNAMENTO PERIODICO ALLERTA NEVE"
change_reason = "aggiornamento_periodico"
elif "variazione_quantita" in change_reason:
direction = "📈" if "+" in change_reason else "📉"
title = f"{direction} AGGIORNAMENTO ALLERTA NEVE"
elif "variazione_orario" in change_reason:
title = "⏰ AGGIORNAMENTO ALLERTA NEVE"
else:
title = "🔄 AGGIORNAMENTO ALLERTA NEVE"
msg.append(title)
msg.append(f"🕒 Aggiornamento ore {html.escape(now_str)}")
if change_reason and change_reason != "nuova_allerta" and change_reason != "annullamento":
msg.append(f"📊 Motivo: {html.escape(change_reason.replace('_', ' '))}")
# La finestra di analisi è sempre 48 ore
window_text = f"prossime {HOURS_AHEAD} ore"
msg.append(f"⏱️ Finestra: {window_text} | Trigger: snowfall > {SNOW_HOURLY_THRESHOLD_CM:.1f} cm/h")
msg.append("")
# Sezione AROME Seamless
msg.append("🛰️ AROME SEAMLESS")
msg.append(f"Modello: {html.escape(MODEL_AROME)}")
msg.append("")
casa = next((s for s in summaries if s["name"] == "🏠 Casa"), None)
if casa:
msg.append("🏠 CASA")
msg.append(f"• 03h: {casa['snow_3h']:.1f} cm | 06h: {casa['snow_6h']:.1f} cm")
msg.append(f"• 12h: {casa['snow_12h']:.1f} cm | 24h: {casa['snow_24h']:.1f} cm")
if casa.get("snow_48h") is not None:
msg.append(f"• 📊 Totale previsto 48h: {casa['snow_48h']:.1f} cm")
if casa["triggered"]:
msg.append(
f"• Primo superamento soglia: {html.escape(casa['first_thr_time'] or '—')} "
f"({casa['first_thr_val']:.1f} cm/h)"
)
if casa["peak_hourly"] > 0:
msg.append(f"• Picco orario: {casa['peak_hourly']:.1f} cm/h (~{html.escape(casa['peak_time'] or '—')})")
if casa.get("duration_hours", 0) > 0:
msg.append(f"• Durata stimata: {casa['duration_hours']:.1f} ore")
if casa.get("last_thr_time"):
msg.append(f"• Fine precipitazioni nevose: {html.escape(casa['last_thr_time'])}")
# Se il fenomeno continua oltre le 48 ore, segnalalo
if casa.get("is_ongoing", False):
if casa.get("extended_end_time"):
msg.append(f"• ⚠️ Fenomeno in corso oltre 48h - Fine prevista: {html.escape(casa['extended_end_time'])}")
else:
msg.append(f"• ⚠️ Fenomeno in corso oltre 48h - Fine non ancora determinabile")
# Aggiungi grafico andamento
timeline = casa.get("snow_timeline", [])
if timeline:
chart = generate_snow_chart(timeline, max_width=30)
if chart:
msg.append("")
msg.append("📊 Andamento previsto:")
msg.append("" + html.escape(chart) + "
")
msg.append("")
msg.append("🌍 NEL CIRCONDARIO (AROME)")
lines_arome = []
for s in summaries:
if not s["triggered"]:
continue
lines_arome.append(
f"{s['name']}: primo > soglia alle {s['first_thr_time'] or '—'} "
f"({s['first_thr_val']:.1f} cm/h), picco {s['peak_hourly']:.1f} cm/h, 24h {s['snow_24h']:.1f} cm"
)
if lines_arome:
msg.append("" + html.escape("\n".join(lines_arome)) + "")
else:
msg.append("Nessun punto ha superato la soglia (anomalia).")
# Sezione ICON Italia (sempre mostrata per Casa se disponibile, altrimenti solo se scostamento significativo)
casa_icon_available = "🏠 Casa" in summaries_icon
if casa_icon_available or (comparisons and summaries_icon):
msg.append("")
msg.append("─" * 15) # Linea corta per iPhone
msg.append("")
msg.append("🇮🇹 ICON ITALIA (ARPAE 2i)")
msg.append(f"Modello: {html.escape(MODEL_ICON_IT)}")
msg.append("")
# Mostra avviso scostamento solo se c'è effettivo scostamento (non per Casa se mostrato sempre)
if comparisons and ("🏠 Casa" not in comparisons or not casa_icon_available):
msg.append("⚠️ Scostamento significativo rilevato (>30%%)")
msg.append("")
# Casa ICON
casa_icon = summaries_icon.get("🏠 Casa")
if casa_icon:
msg.append("🏠 CASA")
msg.append(f"• 03h: {casa_icon['snow_3h']:.1f} cm | 06h: {casa_icon['snow_6h']:.1f} cm")
msg.append(f"• 12h: {casa_icon['snow_12h']:.1f} cm | 24h: {casa_icon['snow_24h']:.1f} cm")
if casa_icon.get("snow_48h") is not None:
msg.append(f"• 📊 Totale previsto 48h: {casa_icon['snow_48h']:.1f} cm")
# Mostra snow_depth se > 0 (manto nevoso persistente)
if casa_icon.get("snow_depth_max", 0) > 0:
msg.append(f"• ⛄ Spessore manto nevoso massimo: {casa_icon['snow_depth_max']:.1f} cm")
if casa_icon["triggered"]:
msg.append(
f"• Primo superamento soglia: {html.escape(casa_icon['first_thr_time'] or '—')} "
f"({casa_icon['first_thr_val']:.1f} cm/h)"
)
if casa_icon["peak_hourly"] > 0:
msg.append(f"• Picco orario: {casa_icon['peak_hourly']:.1f} cm/h (~{html.escape(casa_icon['peak_time'] or '—')})")
msg.append("")
# Circondario ICON
msg.append("🌍 NEL CIRCONDARIO (ICON ITALIA)")
lines_icon = []
for point_name, comp in comparisons.items():
s_icon = summaries_icon.get(point_name)
if s_icon and s_icon["triggered"]:
lines_icon.append(
f"{point_name}: primo > soglia alle {s_icon['first_thr_time'] or '—'} "
f"({s_icon['first_thr_val']:.1f} cm/h), picco {s_icon['peak_hourly']:.1f} cm/h, 24h {s_icon['snow_24h']:.1f} cm"
)
if lines_icon:
msg.append("" + html.escape("\n".join(lines_icon)) + "")
else:
# Mostra comunque i punti con scostamento anche se non hanno trigger
for point_name, comp in comparisons.items():
s_icon = summaries_icon.get(point_name)
if s_icon:
msg.append(
f"• {html.escape(point_name)}: 24h {s_icon['snow_24h']:.1f} cm "
f"(scostamento {comp['diff_pct']:.0f}%%)"
)
msg.append("")
msg.append("Fonte dati: Open-Meteo")
# Unisci con
(sarà convertito in \n in telegram_send_html)
ok = telegram_send_html("
".join(msg), chat_ids=chat_ids)
# Genera e invia grafico (solo se abbiamo dati per Casa)
chart_generated = False
if casa_data_arome_raw:
try:
# Crea file temporaneo per il grafico
chart_path = os.path.join(tempfile.gettempdir(), f"snow_chart_{int(time.time())}.png")
chart_ok = generate_snow_chart_image(
casa_data_arome_raw,
casa_data_icon_raw,
chart_path,
location_name=casa_location_name
)
if chart_ok:
chart_caption = f"📊 Grafico Precipitazioni 48h\n🏠 {casa_location_name}\n🕒 Aggiornamento ore {html.escape(now_str)}"
photo_ok = telegram_send_photo(chart_path, chart_caption, chat_ids=chat_ids)
if photo_ok:
chart_generated = True
LOGGER.info("Grafico inviato con successo")
# Rimuovi file temporaneo dopo l'invio
try:
if os.path.exists(chart_path):
os.remove(chart_path)
except Exception:
pass
else:
LOGGER.warning("Generazione grafico fallita")
except Exception as e:
LOGGER.exception("Errore generazione/invio grafico: %s", e)
if ok:
reason_text = change_reason if change_reason else "aggiornamento_periodico" if is_periodic_due else "sconosciuto"
LOGGER.info("Notifica neve inviata. Motivo: %s", reason_text)
# Salva timestamp dell'ultima notifica
now_utc = datetime.datetime.now(datetime.timezone.utc)
save_state(True, sig, casa_data, last_notification_utc=now_utc.isoformat(timespec="seconds"))
else:
LOGGER.warning("Notifica neve NON inviata (token mancante o errore Telegram).")
# Salva comunque lo state (senza aggiornare last_notification_utc)
save_state(True, sig, casa_data, last_notification_utc=state.get("last_notification_utc", ""))
else:
LOGGER.info("Allerta attiva ma nessun cambiamento significativo. Motivo: %s", change_reason)
# Salva lo state anche se non inviamo (per mantenere alert_active e last_notification_utc)
save_state(True, sig, casa_data, last_notification_utc=state.get("last_notification_utc", ""))
return
# --- Scenario B: rientro (nessun superamento) ---
# Invia notifica di annullamento SOLO se:
# 1. C'era un'allerta attiva (was_active)
# 2. È stata effettivamente inviata una notifica precedente (last_notification_utc presente)
# Questo evita di inviare "annullamento" quando lo stato è attivo ma non è mai stata inviata una notifica
if was_active and not any_trigger:
last_notification_str = state.get("last_notification_utc", "")
has_previous_notification = bool(last_notification_str and last_notification_str.strip())
if has_previous_notification:
# C'è stata una notifica precedente, quindi possiamo inviare l'annullamento
msg = (
"🟢 PREVISIONE NEVE ANNULLATA
"
f"🕒 Aggiornamento ore {html.escape(now_str)}
"
f"Nelle prossime {HOURS_AHEAD} ore non risultano ore con snowfall > {SNOW_HOURLY_THRESHOLD_CM:.1f} cm/h.
"
"Fonte dati: Open-Meteo"
)
ok = telegram_send_html(msg, chat_ids=chat_ids)
if ok:
LOGGER.info("Rientro neve notificato.")
else:
LOGGER.warning("Rientro neve NON inviato (token mancante o errore Telegram).")
save_state(False, "", None)
else:
# Non c'è stata una notifica precedente, quindi non inviare l'annullamento
# ma resetta comunque lo stato per evitare loop
LOGGER.info("Allerta attiva ma nessuna notifica precedente trovata. Resetto stato senza inviare annullamento.")
save_state(False, "", None)
return
# --- Scenario C: tranquillo ---
save_state(False, "", None)
top = sorted(summaries, key=lambda x: x["snow_24h"], reverse=True)[:3]
LOGGER.info(
"Nessuna neve sopra soglia. Top accumuli 24h: %s",
" | ".join(f"{t['name']}={t['snow_24h']:.1f}cm (pk {t['peak_hourly']:.1f}cm/h)" for t in top)
)
if __name__ == "__main__":
arg_parser = argparse.ArgumentParser(description="Monitor neve AROME Seamless")
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
analyze_snow(chat_ids=chat_ids, debug_mode=args.debug)