184 lines
6.1 KiB
Python
184 lines
6.1 KiB
Python
import requests
|
|
import datetime
|
|
import os
|
|
import json
|
|
import time
|
|
from dateutil import parser
|
|
from zoneinfo import ZoneInfo
|
|
|
|
# --- CONFIGURAZIONE UTENTE ---
|
|
TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
|
|
TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
|
|
|
|
# Coordinate (San Marino / Casa)
|
|
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)
|
|
|
|
# Soglia Pioggia (Invariata)
|
|
RAIN_3H_LIMIT = 25.0 # mm in 3 ore
|
|
|
|
# File di stato
|
|
STATE_FILE = "/home/daniely/docker/telegram-bot/weather_state.json"
|
|
|
|
def send_telegram_message(message):
|
|
if not TELEGRAM_BOT_TOKEN or "INSERISCI" in TELEGRAM_BOT_TOKEN:
|
|
print(f"[TEST OUT] {message}")
|
|
return
|
|
|
|
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)
|
|
time.sleep(0.2)
|
|
except: pass
|
|
|
|
def get_forecast():
|
|
url = "https://api.open-meteo.com/v1/forecast"
|
|
params = {
|
|
"latitude": LAT, "longitude": LON,
|
|
"hourly": "precipitation,windgusts_10m",
|
|
"models": "arome_france_hd",
|
|
"timezone": "Europe/Rome",
|
|
"forecast_days": 2
|
|
}
|
|
try:
|
|
r = requests.get(url, params=params, timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except: return None
|
|
|
|
def load_state():
|
|
if os.path.exists(STATE_FILE):
|
|
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}
|
|
|
|
def save_state(state):
|
|
try:
|
|
with open(STATE_FILE, 'w') as f: json.dump(state, f)
|
|
except: pass
|
|
|
|
def analyze():
|
|
print("--- Controllo Meteo Severo ---")
|
|
data = get_forecast()
|
|
if not data: return
|
|
|
|
hourly = data.get("hourly", {})
|
|
times = hourly.get("time", [])
|
|
wind = hourly.get("windgusts_10m", [])
|
|
rain = hourly.get("precipitation", [])
|
|
|
|
now = datetime.datetime.now(ZoneInfo("Europe/Rome"))
|
|
state = load_state()
|
|
|
|
# Trova indice ora corrente
|
|
start_idx = -1
|
|
for i, t in enumerate(times):
|
|
if parser.isoparse(t).replace(tzinfo=ZoneInfo("Europe/Rome")) >= now:
|
|
start_idx = i
|
|
break
|
|
|
|
if start_idx == -1: return
|
|
|
|
# Analisi prossime 12 ore
|
|
end_idx = min(start_idx + 12, len(times))
|
|
|
|
max_wind = 0.0
|
|
max_wind_time = ""
|
|
sum_rain_3h = 0.0
|
|
|
|
# Cerca picco vento
|
|
for i in range(start_idx, end_idx):
|
|
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)
|
|
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)
|
|
|
|
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.")
|
|
|
|
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")
|
|
|
|
if __name__ == "__main__":
|
|
analyze()
|