diff --git a/vigilancia/pi_bot.py b/vigilancia/pi_bot.py
new file mode 100644
index 0000000..55e5af9
--- /dev/null
+++ b/vigilancia/pi_bot.py
@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+import os
+import subprocess
+import threading
+import time
+import urllib.parse
+
+import requests
+
+BASE = os.path.dirname(os.path.abspath(__file__))
+LOG = os.path.join(BASE, "bot.log")
+TV_API = "http://127.0.0.1:8600/tv/mensaje"
+
+
+def log(linea):
+ try:
+ with open(LOG, "a") as f:
+ f.write(f"{time.strftime('%H:%M:%S')} {linea}\n")
+ except Exception:
+ pass
+
+
+def load_env():
+ env = {}
+ try:
+ for line in open(os.path.join(BASE, ".env")):
+ line = line.strip()
+ if line and not line.startswith("#") and "=" in line:
+ k, v = line.split("=", 1)
+ env[k.strip()] = v.strip()
+ except Exception:
+ pass
+ return env
+
+
+ENV = load_env()
+TOKEN = ENV.get("TELEGRAM_TOKEN", "")
+CHAT_ID = ENV.get("TELEGRAM_CHAT_ID", "")
+ALERTA_TEMP = float(ENV.get("ALERTA_TEMP", "80"))
+
+
+def sh(cmd):
+ try:
+ return subprocess.run(cmd, shell=True, capture_output=True,
+ text=True, timeout=10).stdout.strip()
+ except Exception:
+ return "n/a"
+
+
+def estado():
+ temp = sh("vcgencmd measure_temp")
+ upt = sh("uptime -p")
+ load = open("/proc/loadavg").read().split()[:3]
+ free = sh("free -h | awk 'NR==2{print $2\" total, \"$7\" disponible\"}'")
+ disk = sh("df -h / | awk 'NR==2{print $3\" usados de \"$2\" (\"$5\")\"}'")
+ ips = sh("hostname -I")
+ temp_c = None
+ try:
+ temp_c = float(temp.replace("'C", "").replace("temp=", ""))
+ except Exception:
+ pass
+ msg = (
+ "π Estado de minipc:\n"
+ f"π‘οΈ Temperatura: {temp}\n"
+ f"π Uptime: {upt}\n"
+ f"βοΈ Carga: {', '.join(load)}\n"
+ f"π§ RAM: {free}\n"
+ f"πΎ Disco /: {disk}\n"
+ f"π IPs: {ips}"
+ )
+ return msg, temp_c
+
+
+def enviar(texto, chat_id):
+ try:
+ requests.post(
+ f"https://api.telegram.org/bot{TOKEN}/sendMessage",
+ data={"chat_id": chat_id, "text": texto, "parse_mode": "HTML"},
+ timeout=10,
+ )
+ except Exception as e:
+ print("Error enviando:", e)
+
+
+def mandar_tv(mensaje, simple):
+ try:
+ url = f"{TV_API}?texto={urllib.parse.quote(mensaje)}"
+ if simple:
+ url += "&simple=1"
+ r = requests.get(url, timeout=8)
+ return r.status_code == 200
+ except Exception as e:
+ log(f"tv error: {e}")
+ return False
+
+
+def procesar(texto, chat_id):
+ t = texto.lower().strip()
+ log(f"RECIBIDO chat={chat_id} texto={texto!r}")
+ if t in ("estado", "/estado", "status", "/status"):
+ msg, _ = estado()
+ enviar(msg, chat_id)
+ elif "temperatura" in t or "temp" in t:
+ temp = sh("vcgencmd measure_temp")
+ enviar(f"π‘οΈ Temperatura de la Pi: {temp}", chat_id)
+ elif t.startswith("tv ") or t.startswith("/tv "):
+ msj = texto.split(" ", 1)[1].strip()
+ if not msj:
+ enviar("Uso: tv <mensaje> (mensaje rapido a la tele)", chat_id)
+ else:
+ ok = mandar_tv(msj, simple=True)
+ enviar("πΊ Mensaje enviado a la tele β
" if ok
+ else "β No se pudo enviar a la tele", chat_id)
+ elif t.startswith("alarma ") or t.startswith("/alarma "):
+ msj = texto.split(" ", 1)[1].strip()
+ if not msj:
+ enviar("Uso: alarma <mensaje> (secuencia completa: sirena x3)", chat_id)
+ else:
+ ok = mandar_tv(msj, simple=False)
+ enviar("π¨ Alarma lanzada en la tele β
" if ok
+ else "β No se pudo lanzar la alarma", chat_id)
+ elif t in ("ayuda", "/ayuda", "help", "/help"):
+ enviar("Comandos:\n"
+ "estado β estado de la Pi\n"
+ "temperatura β temperatura\n"
+ "tv <mensaje> β mensaje rapido a la tele\n"
+ "alarma <mensaje> β secuencia completa (sirena x3)",
+ chat_id)
+
+
+def monitor_temperatura():
+ while True:
+ time.sleep(300)
+ msg, temp_c = estado()
+ if temp_c is not None and temp_c > ALERTA_TEMP:
+ enviar(f"π¨ ALERTA: la Pi estΓ‘ a {temp_c:.1f}Β°C "
+ f"(lΓmite {ALERTA_TEMP}Β°C)\n{msg}", CHAT_ID)
+
+
+def main():
+ if not TOKEN:
+ print("Falta TELEGRAM_TOKEN en .env")
+ return
+ threading.Thread(target=monitor_temperatura, daemon=True).start()
+ offset = None
+ log("Bot iniciado")
+ print("Bot iniciado. Esperando mensajes...")
+ while True:
+ try:
+ params = {"timeout": 25}
+ if offset:
+ params["offset"] = offset
+ r = requests.get(
+ f"https://api.telegram.org/bot{TOKEN}/getUpdates",
+ params=params, timeout=35,
+ ).json()
+ for upd in r.get("result", []):
+ offset = upd["update_id"] + 1
+ log(f"UPDATE {upd.get('update_id')}")
+ msg = upd.get("message") or upd.get("channel_post")
+ if not msg:
+ continue
+ texto = msg.get("text")
+ chat_id = msg["chat"]["id"]
+ if texto:
+ procesar(texto, chat_id)
+ except Exception as e:
+ log(f"ERROR {e}")
+ print("Error:", e)
+ time.sleep(3)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/vigilancia/tv_api.py b/vigilancia/tv_api.py
index ffe74ad..8083674 100644
--- a/vigilancia/tv_api.py
+++ b/vigilancia/tv_api.py
@@ -9,7 +9,7 @@ import sys
sys.path.insert(0, sys_dir)
import config
-from voz import anuncio_en_teles
+from voz import anuncio_en_teles, anuncio_simple
class Handler(BaseHTTPRequestHandler):
@@ -19,9 +19,11 @@ class Handler(BaseHTTPRequestHandler):
if not texto:
self._resp(400, {"error": "falta parametro texto"})
return
- threading.Thread(target=anuncio_en_teles, args=(texto,), daemon=True).start()
- print(f"[TV-API] Mensaje lanzado: {texto[:80]}")
- self._resp(200, {"ok": True, "mensaje": texto})
+ simple = (q.get("simple") or ["0"])[0] == "1"
+ fn = anuncio_simple if simple else anuncio_en_teles
+ threading.Thread(target=fn, args=(texto,), daemon=True).start()
+ print(f"[TV-API] Mensaje lanzado ({'simple' if simple else 'alarma'}): {texto[:80]}")
+ self._resp(200, {"ok": True, "mensaje": texto, "modo": "simple" if simple else "alarma"})
def _resp(self, code, obj):
body = json.dumps(obj).encode("utf-8")
diff --git a/vigilancia/voz.py b/vigilancia/voz.py
index 77130b2..2e42913 100644
--- a/vigilancia/voz.py
+++ b/vigilancia/voz.py
@@ -320,3 +320,38 @@ def anuncio_en_teles(texto):
def reproducir_en_teles(texto):
anuncio_en_teles(texto)
+
+
+def anuncio_simple(texto):
+ """Mensaje rapido: enciende, reproduce el mensaje una vez y apaga."""
+ if not config.GOOGLE_TV_ACTIVO:
+ return
+ url_mensaje, tipo = _url_audio(texto)
+ if url_mensaje is None:
+ return
+ try:
+ import pychromecast
+
+ chromecasts, browser = pychromecast.get_chromecasts(timeout=8)
+ try:
+ if config.GOOGLE_TV_NOMBRES:
+ objetivos = [c for c in chromecasts
+ if c.cast_info.friendly_name in config.GOOGLE_TV_NOMBRES]
+ else:
+ objetivos = chromecasts
+ for c in objetivos:
+ try:
+ c.wait(timeout=10)
+ c.set_volume(config.GOOGLE_TV_VOLUMEN / 100.0)
+ _reproducir_y_esperar(
+ c, url_mensaje, tipo, texto[:80],
+ f"Mensaje simple en {c.cast_info.friendly_name}")
+ if config.GOOGLE_TV_APAGAR_VIA_ADB:
+ from tv_adb import apagar_tv
+ apagar_tv()
+ except Exception as e:
+ print(f"[Voz] Error en {c.cast_info.friendly_name}: {e}")
+ finally:
+ pychromecast.discovery.stop_discovery(browser)
+ except Exception as e:
+ print(f"[Voz] Error de casting: {e}")