Aggiornamento freeze_alert.py
This commit is contained in:
@@ -1,134 +1,376 @@
|
|||||||
import requests
|
#!/usr/bin/env python3
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
import datetime
|
import datetime
|
||||||
|
import html
|
||||||
import json
|
import json
|
||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
from dateutil import parser
|
from logging.handlers import RotatingFileHandler
|
||||||
|
from typing import Dict, Optional, Tuple
|
||||||
from zoneinfo import ZoneInfo
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
# --- CONFIGURAZIONE UTENTE ---
|
import requests
|
||||||
TELEGRAM_BOT_TOKEN = "8155587974:AAF9OekvBpixtk8ZH6KoIc0L8edbhdXt7A4"
|
from dateutil import parser
|
||||||
TELEGRAM_CHAT_IDS = ["64463169", "24827341", "132455422", "5405962012"]
|
|
||||||
|
|
||||||
# --- COORDINATE (Strada Cà Toro, 12 - San Marino) ---
|
# =============================================================================
|
||||||
|
# FREEZE ALERT (next 48h) - San Marino (Casa)
|
||||||
|
#
|
||||||
|
# Scopo:
|
||||||
|
# Notificare su Telegram se nelle prossime 48 ore è prevista una temperatura
|
||||||
|
# minima < SOGLIA_GELO.
|
||||||
|
#
|
||||||
|
# Requisiti operativi applicati (come gli altri script):
|
||||||
|
# - Nessun token in chiaro (legge da env o file /etc/telegram_dpc_bot_token)
|
||||||
|
# - Log su file + modalità DEBUG (DEBUG=1)
|
||||||
|
# - Timezone robusta (naive -> Europe/Rome; offset -> conversione)
|
||||||
|
# - Nessun Telegram in caso di errori (solo log)
|
||||||
|
# - Anti-spam: notifica solo su nuovo evento o variazione significativa
|
||||||
|
#
|
||||||
|
# Esecuzione:
|
||||||
|
# DEBUG=1 python3 freeze_alert.py
|
||||||
|
# tail -n 200 freeze_alert.log
|
||||||
|
# =============================================================================
|
||||||
|
|
||||||
|
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"
|
||||||
|
|
||||||
|
# ----------------- LOCATION -----------------
|
||||||
LAT = 43.9356
|
LAT = 43.9356
|
||||||
LON = 12.4296
|
LON = 12.4296
|
||||||
LOCATION_NAME = "🏠 Casa (Strada Cà Toro)"
|
LOCATION_NAME = "🏠 Casa (Strada Cà Toro)"
|
||||||
|
|
||||||
# Soglia Gelo (°C)
|
# ----------------- THRESHOLD -----------------
|
||||||
SOGLIA_GELO = 0.0
|
SOGLIA_GELO = 0.0 # °C (allerta se min < 0.0°C)
|
||||||
|
|
||||||
# File di stato
|
# ----------------- HORIZON -----------------
|
||||||
|
HOURS_AHEAD = 48
|
||||||
|
FORECAST_DAYS = 3 # per coprire bene 48h
|
||||||
|
|
||||||
|
# ----------------- TIMEZONE -----------------
|
||||||
|
TZ = "Europe/Rome"
|
||||||
|
TZINFO = ZoneInfo(TZ)
|
||||||
|
|
||||||
|
# ----------------- FILES -----------------
|
||||||
STATE_FILE = "/home/daniely/docker/telegram-bot/freeze_state.json"
|
STATE_FILE = "/home/daniely/docker/telegram-bot/freeze_state.json"
|
||||||
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||||
|
LOG_FILE = os.path.join(BASE_DIR, "freeze_alert.log")
|
||||||
|
|
||||||
def load_state():
|
# ----------------- OPEN-METEO -----------------
|
||||||
|
OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
|
||||||
|
HTTP_HEADERS = {"User-Agent": "rpi-freeze-alert/2.0"}
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# LOGGING
|
||||||
|
# =============================================================================
|
||||||
|
def setup_logger() -> logging.Logger:
|
||||||
|
logger = logging.getLogger("freeze_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()
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# UTILS
|
||||||
|
# =============================================================================
|
||||||
|
def now_local() -> datetime.datetime:
|
||||||
|
return datetime.datetime.now(TZINFO)
|
||||||
|
|
||||||
|
|
||||||
|
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:
|
||||||
|
"""
|
||||||
|
Parsing robusto:
|
||||||
|
- se naive (frequente con timezone=Europe/Rome) -> interpreta come Europe/Rome
|
||||||
|
- se con offset -> converte in Europe/Rome
|
||||||
|
"""
|
||||||
|
dt = parser.isoparse(t)
|
||||||
|
if dt.tzinfo is None:
|
||||||
|
return dt.replace(tzinfo=TZINFO)
|
||||||
|
return dt.astimezone(TZINFO)
|
||||||
|
|
||||||
|
|
||||||
|
def fmt_dt(dt: datetime.datetime) -> str:
|
||||||
|
return dt.strftime("%d/%m alle %H:%M")
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# TELEGRAM
|
||||||
|
# =============================================================================
|
||||||
|
def telegram_send_html(message_html: str) -> bool:
|
||||||
|
"""
|
||||||
|
Non solleva eccezioni. Ritorna True se almeno un invio ha successo.
|
||||||
|
IMPORTANTE: chiamare solo per allerte (mai per errori).
|
||||||
|
"""
|
||||||
|
token = load_bot_token()
|
||||||
|
if not token:
|
||||||
|
LOGGER.warning("Telegram token missing: message not sent.")
|
||||||
|
return False
|
||||||
|
|
||||||
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||||||
|
base_payload = {
|
||||||
|
"text": message_html,
|
||||||
|
"parse_mode": "HTML",
|
||||||
|
"disable_web_page_preview": True,
|
||||||
|
}
|
||||||
|
|
||||||
|
sent_ok = False
|
||||||
|
with requests.Session() as s:
|
||||||
|
for chat_id in TELEGRAM_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
|
||||||
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# STATE
|
||||||
|
# =============================================================================
|
||||||
|
def load_state() -> Dict:
|
||||||
|
default = {
|
||||||
|
"alert_active": False,
|
||||||
|
"min_temp": 100.0,
|
||||||
|
"min_time": "",
|
||||||
|
"signature": "",
|
||||||
|
"updated": "",
|
||||||
|
}
|
||||||
if os.path.exists(STATE_FILE):
|
if os.path.exists(STATE_FILE):
|
||||||
try:
|
try:
|
||||||
with open(STATE_FILE, 'r') as f: return json.load(f)
|
with open(STATE_FILE, "r", encoding="utf-8") as f:
|
||||||
except: pass
|
data = json.load(f) or {}
|
||||||
return {"alert_active": False, "min_temp": 100.0}
|
default.update(data)
|
||||||
|
except Exception as e:
|
||||||
|
LOGGER.exception("State read error: %s", e)
|
||||||
|
return default
|
||||||
|
|
||||||
def save_state(active, min_temp):
|
|
||||||
|
def save_state(state: Dict) -> None:
|
||||||
try:
|
try:
|
||||||
with open(STATE_FILE, 'w') as f:
|
os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
|
||||||
json.dump({"alert_active": active, "min_temp": min_temp, "updated": str(datetime.datetime.now())}, f)
|
state["updated"] = now_local().isoformat()
|
||||||
except: pass
|
with open(STATE_FILE, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
||||||
|
except Exception as e:
|
||||||
|
LOGGER.exception("State write error: %s", 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:
|
# OPEN-METEO
|
||||||
try:
|
# =============================================================================
|
||||||
requests.post(url, json={"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}, timeout=10)
|
def get_forecast() -> Optional[Dict]:
|
||||||
time.sleep(0.2)
|
|
||||||
except: pass
|
|
||||||
|
|
||||||
def get_forecast():
|
|
||||||
url = "https://api.open-meteo.com/v1/forecast"
|
|
||||||
params = {
|
params = {
|
||||||
"latitude": LAT, "longitude": LON,
|
"latitude": LAT,
|
||||||
|
"longitude": LON,
|
||||||
"hourly": "temperature_2m",
|
"hourly": "temperature_2m",
|
||||||
"timezone": "Europe/Rome",
|
"timezone": TZ,
|
||||||
"forecast_days": 3 # Prendiamo 3 giorni per coprire bene le 48h
|
"forecast_days": FORECAST_DAYS,
|
||||||
}
|
}
|
||||||
try:
|
try:
|
||||||
r = requests.get(url, params=params, timeout=10)
|
r = requests.get(OPEN_METEO_URL, params=params, headers=HTTP_HEADERS, timeout=25)
|
||||||
|
if r.status_code == 400:
|
||||||
|
try:
|
||||||
|
j = r.json()
|
||||||
|
LOGGER.error("Open-Meteo 400: %s", j.get("reason", j))
|
||||||
|
except Exception:
|
||||||
|
LOGGER.error("Open-Meteo 400: %s", r.text[:500])
|
||||||
|
return None
|
||||||
r.raise_for_status()
|
r.raise_for_status()
|
||||||
return r.json()
|
return r.json()
|
||||||
except: return None
|
except Exception as e:
|
||||||
|
LOGGER.exception("Open-Meteo request error: %s", e)
|
||||||
|
return None
|
||||||
|
|
||||||
def analyze_freeze():
|
|
||||||
print("--- Controllo Gelo ---")
|
|
||||||
data = get_forecast()
|
|
||||||
if not data: return
|
|
||||||
|
|
||||||
hourly = data.get("hourly", {})
|
def compute_min_next_48h(data: Dict) -> Optional[Tuple[float, datetime.datetime]]:
|
||||||
times = hourly.get("time", [])
|
hourly = data.get("hourly", {}) or {}
|
||||||
temps = hourly.get("temperature_2m", [])
|
times = hourly.get("time", []) or []
|
||||||
|
temps = hourly.get("temperature_2m", []) or []
|
||||||
|
|
||||||
now = datetime.datetime.now(ZoneInfo("Europe/Rome"))
|
n = min(len(times), len(temps))
|
||||||
limit_time = now + datetime.timedelta(hours=48)
|
if n == 0:
|
||||||
|
return None
|
||||||
|
|
||||||
|
now = now_local()
|
||||||
|
limit_time = now + datetime.timedelta(hours=HOURS_AHEAD)
|
||||||
|
|
||||||
min_temp_val = 100.0
|
min_temp_val = 100.0
|
||||||
min_temp_time = None
|
min_temp_time: Optional[datetime.datetime] = None
|
||||||
|
|
||||||
# Cerca la minima nelle prossime 48 ore
|
for i in range(n):
|
||||||
for i, t_str in enumerate(times):
|
try:
|
||||||
t_obj = parser.isoparse(t_str).replace(tzinfo=ZoneInfo("Europe/Rome"))
|
t_obj = parse_time_to_local(times[i])
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
|
||||||
|
# solo intervallo (now, now+48h]
|
||||||
|
if t_obj <= now or t_obj > limit_time:
|
||||||
|
continue
|
||||||
|
|
||||||
|
try:
|
||||||
|
temp = float(temps[i])
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
|
||||||
# Filtra solo futuro prossimo (da adesso a +48h)
|
|
||||||
if t_obj > now and t_obj <= limit_time:
|
|
||||||
temp = temps[i]
|
|
||||||
if temp < min_temp_val:
|
if temp < min_temp_val:
|
||||||
min_temp_val = temp
|
min_temp_val = temp
|
||||||
min_temp_time = t_obj
|
min_temp_time = t_obj
|
||||||
|
|
||||||
# --- LOGICA ALLARME ---
|
if min_temp_time is None:
|
||||||
state = load_state()
|
return None
|
||||||
was_active = state.get("alert_active", False)
|
|
||||||
|
|
||||||
# C'è rischio gelo?
|
return float(min_temp_val), min_temp_time
|
||||||
is_freezing = min_temp_val < SOGLIA_GELO
|
|
||||||
|
|
||||||
|
# =============================================================================
|
||||||
|
# MAIN
|
||||||
|
# =============================================================================
|
||||||
|
def analyze_freeze() -> None:
|
||||||
|
LOGGER.info("--- Controllo Gelo (next %sh) ---", HOURS_AHEAD)
|
||||||
|
|
||||||
|
data = get_forecast()
|
||||||
|
if not data:
|
||||||
|
# errori: solo log
|
||||||
|
return
|
||||||
|
|
||||||
|
result = compute_min_next_48h(data)
|
||||||
|
if not result:
|
||||||
|
LOGGER.error("Impossibile calcolare minima nelle prossime %s ore.", HOURS_AHEAD)
|
||||||
|
return
|
||||||
|
|
||||||
|
min_temp_val, min_temp_time = result
|
||||||
|
is_freezing = (min_temp_val < SOGLIA_GELO)
|
||||||
|
|
||||||
|
state = load_state()
|
||||||
|
was_active = bool(state.get("alert_active", False))
|
||||||
|
last_sig = str(state.get("signature", ""))
|
||||||
|
|
||||||
|
# firma per evitare spam: temp (0.1) + timestamp
|
||||||
|
sig = f"{min_temp_val:.1f}|{min_temp_time.isoformat()}"
|
||||||
|
|
||||||
if is_freezing:
|
if is_freezing:
|
||||||
# Formatta orario
|
# invia se:
|
||||||
time_str = min_temp_time.strftime('%d/%m alle %H:%M')
|
# - prima non era attivo, oppure
|
||||||
|
# - peggiora di almeno 2°C, oppure
|
||||||
|
# - cambia la firma (es. orario minima spostato o min diversa)
|
||||||
|
prev_min = float(state.get("min_temp", 100.0) or 100.0)
|
||||||
|
|
||||||
# SCENARIO A: NUOVO GELO (o peggioramento significativo di 2 gradi)
|
should_notify = (not was_active) or (min_temp_val < prev_min - 2.0) or (sig != last_sig)
|
||||||
if not was_active or min_temp_val < state.get("min_temp", 0) - 2.0:
|
|
||||||
|
if should_notify:
|
||||||
msg = (
|
msg = (
|
||||||
f"❄️ **ALLERTA GELO**\n"
|
"❄️ <b>ALLERTA GELO</b><br/>"
|
||||||
f"📍 {LOCATION_NAME}\n\n"
|
f"📍 {html.escape(LOCATION_NAME)}<br/><br/>"
|
||||||
f"Prevista temperatura minima di **{min_temp_val:.1f}°C**\n"
|
f"Minima prevista (entro {HOURS_AHEAD}h): <b>{min_temp_val:.1f}°C</b><br/>"
|
||||||
f"📅 Quando: {time_str}\n\n"
|
f"📅 Quando: <b>{html.escape(fmt_dt(min_temp_time))}</b><br/><br/>"
|
||||||
f"_Proteggere piante e tubature esterne._"
|
"<i>Proteggere piante e tubature esterne.</i>"
|
||||||
)
|
)
|
||||||
send_telegram_message(msg)
|
ok = telegram_send_html(msg)
|
||||||
save_state(True, min_temp_val)
|
if ok:
|
||||||
print(f"Allerta inviata: {min_temp_val}°C")
|
LOGGER.info("Allerta gelo inviata. Tmin=%.1f°C at %s", min_temp_val, min_temp_time.isoformat())
|
||||||
else:
|
else:
|
||||||
print(f"Gelo già notificato ({min_temp_val}°C).")
|
LOGGER.warning("Allerta gelo NON inviata (token mancante o errore Telegram).")
|
||||||
# Aggiorniamo comunque la minima registrata nel file
|
|
||||||
save_state(True, min(min_temp_val, state.get("min_temp", 100)))
|
|
||||||
|
|
||||||
# SCENARIO B: ALLARME RIENTRATO
|
|
||||||
elif was_active and not is_freezing:
|
|
||||||
msg = (
|
|
||||||
f"☀️ **RISCHIO GELO RIENTRATO**\n"
|
|
||||||
f"📍 {LOCATION_NAME}\n\n"
|
|
||||||
f"Le previsioni per le prossime 48 ore indicano temperature sopra lo zero.\n"
|
|
||||||
f"Minima prevista: {min_temp_val:.1f}°C."
|
|
||||||
)
|
|
||||||
send_telegram_message(msg)
|
|
||||||
save_state(False, min_temp_val)
|
|
||||||
print("Allarme rientrato.")
|
|
||||||
|
|
||||||
else:
|
else:
|
||||||
save_state(False, min_temp_val)
|
LOGGER.info("Gelo già notificato (invariato o peggioramento < 2°C). Tmin=%.1f°C", min_temp_val)
|
||||||
print(f"Nessun gelo. Minima: {min_temp_val}°C")
|
|
||||||
|
state.update({
|
||||||
|
"alert_active": True,
|
||||||
|
"min_temp": min_temp_val,
|
||||||
|
"min_time": min_temp_time.isoformat(),
|
||||||
|
"signature": sig,
|
||||||
|
})
|
||||||
|
save_state(state)
|
||||||
|
return
|
||||||
|
|
||||||
|
# --- RIENTRO ---
|
||||||
|
if was_active and not is_freezing:
|
||||||
|
msg = (
|
||||||
|
"☀️ <b>RISCHIO GELO RIENTRATO</b><br/>"
|
||||||
|
f"📍 {html.escape(LOCATION_NAME)}<br/><br/>"
|
||||||
|
f"Le previsioni nelle prossime {HOURS_AHEAD} ore indicano temperature sopra lo zero.<br/>"
|
||||||
|
f"Minima prevista: <b>{min_temp_val:.1f}°C</b> (alle {html.escape(fmt_dt(min_temp_time))})."
|
||||||
|
)
|
||||||
|
ok = telegram_send_html(msg)
|
||||||
|
if ok:
|
||||||
|
LOGGER.info("Rientro gelo notificato. Tmin=%.1f°C", min_temp_val)
|
||||||
|
else:
|
||||||
|
LOGGER.warning("Rientro gelo NON inviato (token mancante o errore Telegram).")
|
||||||
|
|
||||||
|
state.update({
|
||||||
|
"alert_active": False,
|
||||||
|
"min_temp": min_temp_val,
|
||||||
|
"min_time": min_temp_time.isoformat(),
|
||||||
|
"signature": "",
|
||||||
|
})
|
||||||
|
save_state(state)
|
||||||
|
return
|
||||||
|
|
||||||
|
# --- TRANQUILLO ---
|
||||||
|
state.update({
|
||||||
|
"alert_active": False,
|
||||||
|
"min_temp": min_temp_val,
|
||||||
|
"min_time": min_temp_time.isoformat(),
|
||||||
|
"signature": "",
|
||||||
|
})
|
||||||
|
save_state(state)
|
||||||
|
LOGGER.info("Nessun gelo. Tmin=%.1f°C at %s", min_temp_val, min_temp_time.isoformat())
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
analyze_freeze()
|
analyze_freeze()
|
||||||
|
|||||||
Reference in New Issue
Block a user