210 lines
6.5 KiB
Python
210 lines
6.5 KiB
Python
import requests
|
|
import datetime
|
|
import time
|
|
import json
|
|
import os
|
|
from dateutil import parser
|
|
from zoneinfo import ZoneInfo
|
|
|
|
# --- CONFIGURAZIONE UTENTE ---
|
|
# 👇👇 INSERISCI QUI I TUOI DATI 👇👇
|
|
TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
|
|
TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
|
|
|
|
# --- PUNTI DI MONITORAGGIO ---
|
|
# Sostituito San Leo con Carpegna
|
|
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}
|
|
]
|
|
|
|
# Soglia notifica (cm)
|
|
SOGLIA_NOTIFICA = 0.0
|
|
|
|
# File di stato per ricordare l'ultima allerta
|
|
STATE_FILE = "/home/daniely/docker/telegram-bot/snow_state.json"
|
|
|
|
# --- FUNZIONI DI UTILITÀ ---
|
|
|
|
def is_winter_season():
|
|
"""Ritorna True se oggi è tra il 1 Novembre e il 15 Aprile"""
|
|
now = datetime.datetime.now()
|
|
month = now.month
|
|
day = now.day
|
|
|
|
if month >= 11: return True # Nov, Dic
|
|
if month <= 3: return True # Gen, Feb, Mar
|
|
if month == 4 and day <= 15: return True # Fino al 15 Apr
|
|
return False
|
|
|
|
def load_last_state():
|
|
"""Legge se c'era un allerta attiva"""
|
|
if not os.path.exists(STATE_FILE): return False
|
|
try:
|
|
with open(STATE_FILE, 'r') as f:
|
|
data = json.load(f)
|
|
return data.get("alert_active", False)
|
|
except: return False
|
|
|
|
def save_current_state(is_active):
|
|
"""Salva lo stato corrente"""
|
|
try:
|
|
with open(STATE_FILE, 'w') as f:
|
|
json.dump({"alert_active": is_active, "updated": str(datetime.datetime.now())}, f)
|
|
except Exception as e:
|
|
print(f"Errore salvataggio stato: {e}")
|
|
|
|
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:
|
|
payload = {"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}
|
|
try:
|
|
requests.post(url, json=payload, timeout=10)
|
|
time.sleep(0.2)
|
|
except Exception as e:
|
|
print(f"Errore invio a {chat_id}: {e}")
|
|
|
|
def get_forecast(lat, lon):
|
|
url = "https://api.open-meteo.com/v1/forecast"
|
|
params = {
|
|
"latitude": lat, "longitude": lon,
|
|
"hourly": "snowfall",
|
|
"models": "arome_france_hd",
|
|
"timezone": "Europe/Rome",
|
|
"forecast_days": 2
|
|
}
|
|
try:
|
|
response = requests.get(url, params=params, timeout=5)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except:
|
|
return None
|
|
|
|
def calculate_sums(data):
|
|
if not data: return None
|
|
|
|
hourly = data.get("hourly", {})
|
|
times = hourly.get("time", [])
|
|
snow = hourly.get("snowfall", [])
|
|
|
|
if not times or not snow: return None
|
|
|
|
now = datetime.datetime.now(ZoneInfo("Europe/Rome"))
|
|
|
|
start_idx = -1
|
|
for i, t_str in enumerate(times):
|
|
try:
|
|
t_obj = parser.isoparse(t_str).replace(tzinfo=ZoneInfo("Europe/Rome"))
|
|
if t_obj >= now.replace(minute=0, second=0, microsecond=0):
|
|
start_idx = i
|
|
break
|
|
except: continue
|
|
|
|
if start_idx == -1: return None
|
|
|
|
end = len(snow)
|
|
# Calcola somme sugli orizzonti temporali
|
|
def get_sum(hours):
|
|
return sum(x for x in snow[start_idx:min(start_idx+hours, end)] if x)
|
|
|
|
return {
|
|
"3h": get_sum(3),
|
|
"6h": get_sum(6),
|
|
"12h": get_sum(12),
|
|
"24h": get_sum(24)
|
|
}
|
|
|
|
# --- LOGICA PRINCIPALE ---
|
|
|
|
def analyze_snow():
|
|
# 1. Controllo Stagionale
|
|
if not is_winter_season():
|
|
print("Stagione estiva. Script in pausa.")
|
|
return
|
|
|
|
now_str = datetime.datetime.now(ZoneInfo("Europe/Rome")).strftime('%H:%M')
|
|
print(f"--- Check Meteo {now_str} ---")
|
|
|
|
home_stats = None
|
|
max_area_snow = 0.0
|
|
area_details = ""
|
|
|
|
# 2. Raccolta Dati
|
|
for p in POINTS:
|
|
data = get_forecast(p["lat"], p["lon"])
|
|
stats = calculate_sums(data)
|
|
|
|
if not stats: continue
|
|
|
|
if p["name"] == "🏠 Casa":
|
|
home_stats = stats
|
|
|
|
# Aggiorna il massimo rilevato in zona
|
|
if stats["24h"] > max_area_snow:
|
|
max_area_snow = stats["24h"]
|
|
|
|
# Costruisci dettaglio se c'è neve
|
|
if stats["24h"] > 0:
|
|
area_details += f"{p['name']}: {stats['24h']:.1f}cm (12h: {stats['12h']:.1f})\n"
|
|
|
|
time.sleep(1)
|
|
|
|
# 3. Decisione Alert
|
|
# C'è neve se a casa o nei dintorni l'accumulo è > soglia
|
|
home_max = home_stats["24h"] if home_stats else 0.0
|
|
SNOW_DETECTED = (home_max > SOGLIA_NOTIFICA or max_area_snow > SOGLIA_NOTIFICA)
|
|
|
|
# Leggi stato precedente
|
|
WAS_ACTIVE = load_last_state()
|
|
|
|
# --- SCENARIO A: C'È NEVE (Nuova o Continua) ---
|
|
if SNOW_DETECTED:
|
|
|
|
def f(v): return f"**{v:.1f}**" if v > 0 else f"{v:.1f}"
|
|
|
|
msg = f"❄️ **ALLERTA NEVE (AROME HD)**\n📅 _Aggiornamento ore {now_str}_\n\n"
|
|
|
|
if home_stats:
|
|
msg += f"🏠 **CASA:**\n"
|
|
msg += f"• 03h: {f(home_stats['3h'])} cm\n"
|
|
msg += f"• 06h: {f(home_stats['6h'])} cm\n"
|
|
msg += f"• 12h: {f(home_stats['12h'])} cm\n"
|
|
msg += f"• 24h: {f(home_stats['24h'])} cm\n\n"
|
|
|
|
if area_details:
|
|
msg += f"🌍 **NEL CIRCONDARIO (24h):**\n"
|
|
msg += f"`{area_details}`"
|
|
else:
|
|
msg += "🌍 Nessuna neve rilevante nei dintorni."
|
|
|
|
send_telegram_message(msg)
|
|
save_current_state(True) # Salva che l'allerta è attiva
|
|
print("Neve rilevata. Notifica inviata.")
|
|
|
|
# --- SCENARIO B: ALLARME RIENTRATO (Neve 0, ma prima c'era) ---
|
|
elif not SNOW_DETECTED and WAS_ACTIVE:
|
|
msg = (
|
|
f"🟢 **PREVISIONE NEVE ANNULLATA**\n"
|
|
f"📅 _Aggiornamento ore {now_str}_\n\n"
|
|
f"Le ultime previsioni AROME non indicano più accumuli nevosi rilevanti nelle prossime 24 ore.\n"
|
|
f"Situazione tornata alla normalità."
|
|
)
|
|
send_telegram_message(msg)
|
|
save_current_state(False) # Resetta lo stato
|
|
print("Allarme rientrato. Notifica inviata.")
|
|
|
|
# --- SCENARIO C: TUTTO TRANQUILLO (E lo era anche prima) ---
|
|
else:
|
|
# Aggiorna timestamp ma mantieni false
|
|
save_current_state(False)
|
|
print(f"Nessuna neve. Casa: {home_max}cm, Area: {max_area_snow}cm")
|
|
|
|
if __name__ == "__main__":
|
|
analyze_snow()
|