From 1ccb77c7ae778852cee587b3e7d5d324d331f9c8 Mon Sep 17 00:00:00 2001 From: daniele Date: Wed, 10 Dec 2025 20:23:13 +0100 Subject: [PATCH] Backup automatico script del 2025-12-10 20:23 --- services/telegram-bot/severe_weather.py | 169 ++++++++++++------------ 1 file changed, 88 insertions(+), 81 deletions(-) diff --git a/services/telegram-bot/severe_weather.py b/services/telegram-bot/severe_weather.py index 60f451e..791fef2 100644 --- a/services/telegram-bot/severe_weather.py +++ b/services/telegram-bot/severe_weather.py @@ -14,14 +14,18 @@ TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"] LAT = 43.9356 LON = 12.4296 -# --- SOGLIE UFFICIALI EMILIA-ROMAGNA (Vento) --- -WIND_YELLOW = 62.0 # km/h (Burrasca Moderata) -WIND_ORANGE = 75.0 # km/h (Burrasca Forte) -WIND_RED = 88.0 # km/h (Tempesta) +# --- SOGLIE DI ALLARME --- +# Vento (Protezione Civile Emilia-Romagna) +WIND_YELLOW = 62.0 # km/h +WIND_ORANGE = 75.0 # km/h +WIND_RED = 88.0 # km/h -# Soglia Pioggia (Invariata) +# Pioggia RAIN_3H_LIMIT = 25.0 # mm in 3 ore +# Codici WMO per Gelicidio (Freezing Rain) +FREEZING_CODES = [56, 57, 66, 67] + # File di stato STATE_FILE = "/home/daniely/docker/telegram-bot/weather_state.json" @@ -33,7 +37,7 @@ def send_telegram_message(message): url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage" for chat_id in TELEGRAM_CHAT_IDS: try: - requests.post(url, json={"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}, timeout=5) + requests.post(url, json={"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}, timeout=10) time.sleep(0.2) except: pass @@ -41,7 +45,7 @@ def get_forecast(): url = "https://api.open-meteo.com/v1/forecast" params = { "latitude": LAT, "longitude": LON, - "hourly": "precipitation,windgusts_10m", + "hourly": "precipitation,windgusts_10m,weathercode", "models": "arome_france_hd", "timezone": "Europe/Rome", "forecast_days": 2 @@ -57,7 +61,7 @@ def load_state(): try: with open(STATE_FILE, 'r') as f: return json.load(f) except: pass - return {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0} + return {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0, "freezing_active": False} def save_state(state): try: @@ -65,7 +69,7 @@ def save_state(state): except: pass def analyze(): - print("--- Controllo Meteo Severo ---") + print("--- Controllo Meteo Severo (Wind/Rain/Ice) ---") data = get_forecast() if not data: return @@ -73,6 +77,7 @@ def analyze(): times = hourly.get("time", []) wind = hourly.get("windgusts_10m", []) rain = hourly.get("precipitation", []) + wcode = hourly.get("weathercode", []) now = datetime.datetime.now(ZoneInfo("Europe/Rome")) state = load_state() @@ -93,91 +98,93 @@ def analyze(): max_wind_time = "" sum_rain_3h = 0.0 - # Cerca picco vento + freezing_rain_detected = False + freezing_time = "" + + # Cerca picchi for i in range(start_idx, end_idx): + # Vento if wind[i] > max_wind: max_wind = wind[i] max_wind_time = parser.isoparse(times[i]).strftime('%H:%M') - - # Cerca picco pioggia (sliding window 3h) + + # Gelicidio + if wcode[i] in FREEZING_CODES: + freezing_rain_detected = True + if not freezing_time: # Prendi il primo orario + freezing_time = parser.isoparse(times[i]).strftime('%H:%M') + + # Pioggia (sliding window 3h) for i in range(start_idx, end_idx - 3): current_sum = sum(rain[i:i+3]) if current_sum > sum_rain_3h: sum_rain_3h = current_sum - # --- CLASSIFICAZIONE VENTO --- - wind_level_curr = 0 - wind_msg_header = "" - wind_msg_desc = "" - - if max_wind > WIND_RED: - wind_level_curr = 3 - wind_msg_header = "πŸ”΄ **TEMPESTA (Burrasca Fortissima)**" - wind_msg_desc = f"Previste raffiche > {WIND_RED:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\n_Possibili danni a tetti e alberi._" - elif max_wind > WIND_ORANGE: - wind_level_curr = 2 - wind_msg_header = "🟠 **VENTO MOLTO FORTE (Burrasca Forte)**" - wind_msg_desc = f"Previste raffiche > {WIND_ORANGE:.0f} km/h (Picco: **{max_wind:.0f}** km/h)." - elif max_wind > WIND_YELLOW: - wind_level_curr = 1 - wind_msg_header = "🟑 **VENTO FORTE (Burrasca Moderata)**" - wind_msg_desc = f"Previste raffiche > {WIND_YELLOW:.0f} km/h (Picco: **{max_wind:.0f}** km/h)." - - # --- LOGICA DI ALLARME --- - - is_rain_alarm = sum_rain_3h > RAIN_3H_LIMIT - IS_ALARM_NOW = (wind_level_curr > 0) or is_rain_alarm - - WAS_ALARM = state.get("alert_active", False) - LAST_WIND_LEVEL = state.get("wind_level", 0) - + # --- CLASSIFICAZIONE --- alerts = [] should_notify = False - # SCENARIO A: C'È UN ALLARME - if IS_ALARM_NOW: - # 1. Gestione Vento - if wind_level_curr > 0: - # Notifica se Γ¨ un NUOVO allarme o se il livello Γ¨ AUMENTATO (es. Giallo -> Arancione) - if not WAS_ALARM or wind_level_curr > LAST_WIND_LEVEL: - alerts.append(f"{wind_msg_header}\n{wind_msg_desc}\nπŸ•’ Orario picco: {max_wind_time}") - should_notify = True - state["wind_level"] = wind_level_curr # Aggiorna livello salvato - state["last_wind"] = max_wind - - # 2. Gestione Pioggia - if is_rain_alarm: - if not WAS_ALARM or sum_rain_3h > state.get("last_rain", 0) + 10: - alerts.append(f"🌧️ **PIOGGIA INTENSA**\nPrevisti **{sum_rain_3h:.1f} mm** in 3 ore.\n_Rischio disagi idraulici._") - state["last_rain"] = sum_rain_3h - should_notify = True - - if should_notify: - full_msg = f"⚠️ **AVVISO METEO SEVERO**\n\n" + "\n\n".join(alerts) - send_telegram_message(full_msg) - state["alert_active"] = True - save_state(state) - print("Allerta inviata.") - else: - print(f"Condizioni severe stabili (Livello {wind_level_curr}). GiΓ  notificato.") - - # SCENARIO B: ALLARME RIENTRATO - elif WAS_ALARM and not IS_ALARM_NOW: - msg = ( - f"🟒 **ALLERTA METEO RIENTRATA**\n" - f"πŸ“… _Aggiornamento ore {now.strftime('%H:%M')}_\n\n" - f"Vento e pioggia sono scesi sotto le soglie di guardia per le prossime 12 ore." - ) - send_telegram_message(msg) - - # Reset stato completo - state = {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0} - save_state(state) - print("Allarme rientrato inviato.") - + WAS_ALARM = state.get("alert_active", False) + + # 1. GELO (PrioritΓ  Massima) + if freezing_rain_detected: + if not state.get("freezing_active", False): + alerts.append(f"🧊 **ALLARME GELICIDIO**\nPrevista pioggia che gela (Freezing Rain).\nπŸ•’ Inizio previsto: {freezing_time}\n_Pericolo ghiaccio su strada!_") + should_notify = True + state["freezing_active"] = True else: - # Aggiorna lo stato per dire "tutto ok" ma non resetta se non necessario - print(f"Situazione tranquilla. Vento Max: {max_wind:.0f} km/h") + state["freezing_active"] = False + + # 2. VENTO + wind_level_curr = 0 + if max_wind > WIND_RED: + wind_level_curr = 3 + msg = f"πŸ”΄ **TEMPESTA (Burrasca Fortissima)**\nRaffiche > {WIND_RED:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\nπŸ•’ Ore {max_wind_time}" + elif max_wind > WIND_ORANGE: + wind_level_curr = 2 + msg = f"🟠 **VENTO MOLTO FORTE**\nRaffiche > {WIND_ORANGE:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\nπŸ•’ Ore {max_wind_time}" + elif max_wind > WIND_YELLOW: + wind_level_curr = 1 + msg = f"🟑 **VENTO FORTE**\nRaffiche > {WIND_YELLOW:.0f} km/h (Picco: **{max_wind:.0f}** km/h).\nπŸ•’ Ore {max_wind_time}" + + if wind_level_curr > 0: + if not WAS_ALARM or wind_level_curr > state.get("wind_level", 0): + alerts.append(msg) + should_notify = True + state["wind_level"] = wind_level_curr + state["last_wind"] = max_wind + + # 3. PIOGGIA + if sum_rain_3h > RAIN_3H_LIMIT: + if not WAS_ALARM or sum_rain_3h > state.get("last_rain", 0) + 10: + alerts.append(f"🌧️ **PIOGGIA INTENSA**\nPrevisti **{sum_rain_3h:.1f} mm** in 3 ore.") + state["last_rain"] = sum_rain_3h + should_notify = True + + # --- INVIO --- + IS_ALARM_NOW = (wind_level_curr > 0) or (sum_rain_3h > RAIN_3H_LIMIT) or freezing_rain_detected + + if should_notify and alerts: + full_msg = f"⚠️ **AVVISO METEO SEVERO**\n\n" + "\n\n".join(alerts) + send_telegram_message(full_msg) + state["alert_active"] = True + save_state(state) + print("Allerta inviata.") + + elif WAS_ALARM and not IS_ALARM_NOW: + # Allarme rientrato + msg = f"🟒 **ALLERTA METEO RIENTRATA**\nLe condizioni sono tornate sotto le soglie di guardia." + send_telegram_message(msg) + state = {"alert_active": False, "last_wind": 0, "last_rain": 0, "wind_level": 0, "freezing_active": False} + save_state(state) + print("Allarme rientrato.") + + else: + # Aggiorna stato parziale se serve (es. vento calato ma ancora presente) + if not IS_ALARM_NOW: + state["alert_active"] = False + save_state(state) + print(f"Nessun nuovo allarme. W:{max_wind:.0f} R:{sum_rain_3h:.1f} Ice:{freezing_rain_detected}") if __name__ == "__main__": analyze()