191 lines
6.2 KiB
Python
191 lines
6.2 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 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
|
|
|
|
# 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"
|
|
|
|
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=10)
|
|
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,weathercode",
|
|
"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, "freezing_active": False}
|
|
|
|
def save_state(state):
|
|
try:
|
|
with open(STATE_FILE, 'w') as f: json.dump(state, f)
|
|
except: pass
|
|
|
|
def analyze():
|
|
print("--- Controllo Meteo Severo (Wind/Rain/Ice) ---")
|
|
data = get_forecast()
|
|
if not data: return
|
|
|
|
hourly = data.get("hourly", {})
|
|
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()
|
|
|
|
# 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
|
|
|
|
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')
|
|
|
|
# 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 ---
|
|
alerts = []
|
|
should_notify = False
|
|
|
|
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:
|
|
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()
|