Anadir avisos por Telegram con captura anotada (Fase B)
- pet_telegram.py: cliente de Telegram con cola y hilo de envio propio, para que la red nunca bloquee la vision ni la escucha de audio. Reintentos con retroceso exponencial, respeto del retry_after ante limites de frecuencia y descarte de los avisos mas viejos si la cola se llena. - pet_alerts.py: punto unico de reparto de avisos. Toma el fotograma mas reciente, dibuja la caja de la mascota detectada y la marca de tiempo, guarda la captura en capturas/ y la manda a Telegram. Anti-repeticion de 10 s en consola y de ALERT_COOLDOWN (60 s) para capturas y Telegram. - scripts/telegram_check.py: valida el token, deduce el chat_id del ultimo mensaje recibido por el bot y manda mensaje y foto de prueba. - pet_speed.py: integra el reparto de avisos y anade capturas y estado de Telegram a las estadisticas. - Readme.md: seccion de Telegram y estado de la Fase B.
This commit is contained in:
parent
937bbd91ac
commit
762699f41e
7 changed files with 537 additions and 14 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -16,3 +16,6 @@ hailort.1.log
|
|||
|
||||
# Modelos grandes: se descargan con scripts/fetch_models.sh
|
||||
models/*.onnx
|
||||
|
||||
# Capturas de las alertas
|
||||
capturas/
|
||||
|
|
|
|||
45
Readme.md
45
Readme.md
|
|
@ -72,11 +72,14 @@ uv pip install sounddevice numpy scipy requests onnxruntime
|
|||
> **Nota:** `uv` vive en `~/.local/bin/uv` y no está en el PATH de sesiones SSH no interactivas; usa `bash -lc 'uv run ...'` o `~/.local/bin/uv run ...`.
|
||||
---
|
||||
5. Estado Actual del Software
|
||||
El proyecto se reparte en tres módulos, más el script de descarga del modelo:
|
||||
El proyecto se reparte en módulos, más los scripts de apoyo:
|
||||
- `pet_common.py`: carga del `.env`, construcción de la URL RTSP y enmascarado de credenciales.
|
||||
- `pet_speed.py`: visión por computador (vídeo + NPU Hailo). Es el punto de entrada.
|
||||
- `pet_audio.py`: detección de ladridos sobre el audio de la cámara.
|
||||
- `pet_alerts.py`: decide qué sale por consola, qué se guarda como captura y qué va a Telegram.
|
||||
- `pet_telegram.py`: cliente de Telegram (cola + hilo de envío, reintentos).
|
||||
- `scripts/fetch_models.sh`: descarga y verifica (SHA-256) el modelo YAMNet.
|
||||
- `scripts/telegram_check.py`: puesta en marcha de Telegram (valida el token y deduce el chat_id).
|
||||
|
||||
`pet_speed.py`: Script de visión artificial optimizado con aceleración por hardware que:
|
||||
- Lee el vídeo de la cámara Tapo por **RTSP en un hilo aparte** (`RtspVideoStream`), que se queda siempre con el fotograma más reciente y descarta los antiguos. Así no se acumula latencia si el análisis va más lento que la cámara.
|
||||
|
|
@ -124,15 +127,45 @@ El recorte de banda a 4 kHz cuesta un 14% de puntuación, pero la separación en
|
|||
> **Pendiente de comprobar in situ:** el experimento aísla la pérdida de códec y banda, pero no la calidad del micrófono de la cámara, la distancia al animal ni la acústica de la sala. Conviene validarlo con un ladrido real delante de la cámara.
|
||||
|
||||
---
|
||||
7. Próximos Pasos (Hoja de Ruta)
|
||||
7. Avisos por Telegram
|
||||
Cuando salta una alerta (carrera o ladrido), el monitor manda un mensaje **con la foto de la cámara anotada**: la caja de la mascota detectada, su confianza y la fecha y hora. Las capturas se guardan además en `capturas/` (ignorado por git).
|
||||
|
||||
Puesta en marcha (una sola vez):
|
||||
1. En Telegram, habla con **@BotFather** y envía `/newbot`. Te devolverá un token.
|
||||
2. **Escríbele cualquier mensaje a tu bot** (si no, el bot no puede ver tu chat).
|
||||
3. Añade el token a `~/pet_monitor/.env`:
|
||||
```
|
||||
TELEGRAM_BOT_TOKEN=123456:ABC...
|
||||
```
|
||||
4. Comprueba que todo funciona:
|
||||
```bash
|
||||
uv run scripts/telegram_check.py
|
||||
```
|
||||
Ese script valida el token, **deduce el `chat_id` solo** a partir del último mensaje que recibió el bot, manda un mensaje y una foto de prueba, y te dice qué línea añadir al `.env` (`TELEGRAM_CHAT_ID`) para no tener que deducirlo cada vez.
|
||||
|
||||
Cómo se comporta:
|
||||
- Los envíos van por una **cola con su propio hilo**: la red nunca bloquea ni la visión ni la escucha de audio. Si Telegram va lento, se encolan.
|
||||
- Si falla, reintenta 3 veces con retroceso exponencial, y respeta el `retry_after` que pide Telegram ante un límite de frecuencia.
|
||||
- Si la cola se llena (Telegram caído mucho rato), **descarta los avisos viejos** en favor de los nuevos: uno de hace cinco minutos ya no sirve.
|
||||
- **Anti-repetición en dos niveles**: la consola agrupa avisos cada 10 s (antes un perro corriendo imprimía 14 líneas por segundo y tapaba las estadísticas) y las capturas/Telegram como mucho una cada `ALERT_COOLDOWN` (60 s por defecto) y por tipo de aviso.
|
||||
- El **token nunca se imprime**: todo texto de error pasa por el enmascarador, así que en `journalctl` aparecería `.../bot***/sendMessage`.
|
||||
|
||||
Variables opcionales en el `.env`:
|
||||
```
|
||||
# TELEGRAM_CHAT_ID=123456789 # deducido solo si no se indica
|
||||
# ALERT_COOLDOWN=60 # segundos entre capturas del mismo tipo de aviso
|
||||
# SNAPSHOTS=0 # no guardar las capturas en disco
|
||||
# SNAPSHOT_DIR=capturas # dónde guardarlas
|
||||
```
|
||||
|
||||
---
|
||||
8. Próximos Pasos (Hoja de Ruta)
|
||||
Fase A: Integración de Audio (Ladridos y Sonidos) — *hecha*
|
||||
Descartado el micrófono USB: se usa el de la propia cámara.
|
||||
El modelo YAMNet ONNX y el diccionario de clases de AudioSet (`yamnet_classes.csv`) están en `models/`.
|
||||
Listener en hilo secundario analizando ventanas de 0.975 s en busca de la clase `Bark`: implementado en `pet_audio.py`.
|
||||
Fase B: Sistema de Alertas y Notificaciones
|
||||
Crear un bot en Telegram con `@BotFather` y obtener el `TELEGRAM_BOT_TOKEN` y `CHAT_ID`.
|
||||
Programar un despachador asíncrono con control de enfriamiento (cooldown de 30-60 s) para evitar spam.
|
||||
Guardar una captura JPEG con la caja delimitadora en el momento de la carrera o ladrido y enviarla a Telegram (las cajas ya se calculan en píxeles del fotograma original, solo falta dibujarlas y guardarlas).
|
||||
Fase B: Sistema de Alertas y Notificaciones — *hecha (falta poner el token)*
|
||||
Bot de Telegram, despachador asíncrono con control de enfriamiento y captura JPEG con la caja delimitadora: implementado en `pet_telegram.py` y `pet_alerts.py`.
|
||||
Fase C: Despliegue Autónomo (Systemd)
|
||||
Crear una unidad de servicio `/etc/systemd/system/pet_monitor.service`.
|
||||
Habilitar arranque automático al encender la Raspberry Pi:
|
||||
|
|
|
|||
139
pet_alerts.py
Normal file
139
pet_alerts.py
Normal file
|
|
@ -0,0 +1,139 @@
|
|||
"""
|
||||
Reparto de avisos: qué sale por consola, qué se guarda como captura y qué se
|
||||
manda a Telegram.
|
||||
|
||||
Las capturas se toman del fotograma más reciente del stream de vídeo en el
|
||||
instante del aviso (no del que se esté analizando), y se anotan con las cajas
|
||||
de la última detección más una marca de tiempo.
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
|
||||
import cv2
|
||||
|
||||
# Tipos de aviso que merecen foto (y por tanto Telegram)
|
||||
TIPOS_CON_CAPTURA = ("carrera", "ladrido")
|
||||
|
||||
# Anti-repetición por consola: sin esto, un perro corriendo imprime 14 líneas
|
||||
# por segundo y tapa las estadísticas.
|
||||
CONSOLA_POR_DEFECTO = {"carrera": 10.0, "calmado": 10.0, "ladrido": 0.0}
|
||||
|
||||
|
||||
class AlertDispatcher:
|
||||
"""
|
||||
Punto único por el que pasan todos los avisos del monitor.
|
||||
|
||||
`alert(kind, mensaje)` hace tres cosas, cada una con su propia cadencia:
|
||||
- imprime por consola (anti-repetición corta),
|
||||
- guarda una captura anotada en disco,
|
||||
- la envía a Telegram.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
stream,
|
||||
notifier=None,
|
||||
snapshot_dir=None,
|
||||
save_snapshots=True,
|
||||
capture_cooldown=60.0,
|
||||
console_cooldowns=None,
|
||||
jpeg_quality=85,
|
||||
):
|
||||
self._stream = stream
|
||||
self._notifier = notifier
|
||||
self.snapshot_dir = snapshot_dir
|
||||
self.save_snapshots = save_snapshots
|
||||
self.capture_cooldown = capture_cooldown
|
||||
self.console_cooldowns = dict(CONSOLA_POR_DEFECTO)
|
||||
if console_cooldowns:
|
||||
self.console_cooldowns.update(console_cooldowns)
|
||||
self.jpeg_quality = jpeg_quality
|
||||
|
||||
self._cajas = []
|
||||
self._ultimo_consola = {}
|
||||
self._ultima_captura = {}
|
||||
self.snapshots = 0
|
||||
self.last_snapshot = None
|
||||
|
||||
def set_pet_boxes(self, cajas):
|
||||
"""Guarda las cajas del último fotograma analizado, para dibujarlas."""
|
||||
self._cajas = list(cajas)
|
||||
|
||||
def alert(self, kind, message):
|
||||
"""
|
||||
Emite un aviso. Devuelve True si hizo algo (lo imprimiera o no).
|
||||
`kind` es "carrera", "ladrido" o "calmado".
|
||||
"""
|
||||
ahora = time.time()
|
||||
consola = ahora - self._ultimo_consola.get(kind, 0.0) >= self.console_cooldowns.get(kind, 10.0)
|
||||
captura = (
|
||||
kind in TIPOS_CON_CAPTURA
|
||||
and ahora - self._ultima_captura.get(kind, 0.0) >= self.capture_cooldown
|
||||
)
|
||||
|
||||
if consola:
|
||||
self._ultimo_consola[kind] = ahora
|
||||
print(message, flush=True)
|
||||
if captura:
|
||||
self._ultima_captura[kind] = ahora
|
||||
self._captura(message)
|
||||
|
||||
return consola or captura
|
||||
|
||||
def stats(self):
|
||||
"""Resumen para la línea de estadísticas."""
|
||||
partes = [f"capturas {self.snapshots}"]
|
||||
if self._notifier is not None:
|
||||
partes.append(
|
||||
f"telegram {self._notifier.sent} enviados"
|
||||
+ (f", {self._notifier.errors} fallos" if self._notifier.errors else "")
|
||||
+ (f", {self._notifier.dropped} descartados" if self._notifier.dropped else "")
|
||||
)
|
||||
return " | ".join(partes)
|
||||
|
||||
# --- interno ---------------------------------------------------------
|
||||
def _captura(self, message):
|
||||
frame = self._stream.latest_frame()
|
||||
anotado = self._anota(frame, message) if frame is not None else None
|
||||
|
||||
if anotado is not None and self.save_snapshots and self.snapshot_dir:
|
||||
self._guarda(anotado, message)
|
||||
|
||||
if self._notifier is None:
|
||||
return
|
||||
if anotado is None:
|
||||
self._notifier.send_message(message)
|
||||
return
|
||||
|
||||
ok, buffer = cv2.imencode(".jpg", anotado, [int(cv2.IMWRITE_JPEG_QUALITY), self.jpeg_quality])
|
||||
if ok:
|
||||
self._notifier.send_photo(buffer.tobytes(), message)
|
||||
else:
|
||||
self._notifier.send_message(message)
|
||||
|
||||
def _anota(self, frame, message):
|
||||
"""Copia el fotograma con las cajas detectadas y una marca de tiempo."""
|
||||
imagen = frame.copy()
|
||||
for etiqueta, confianza, (x1, y1, x2, y2) in self._cajas:
|
||||
color = (0, 0, 255) if "corriendo" in message or "Ladrido" in message else (0, 200, 0)
|
||||
cv2.rectangle(imagen, (int(x1), int(y1)), (int(x2), int(y2)), color, 2)
|
||||
texto = f"{etiqueta} {confianza:.2f}"
|
||||
(ancho, alto), _ = cv2.getTextSize(texto, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 2)
|
||||
cv2.rectangle(imagen, (int(x1), int(y1) - alto - 8), (int(x1) + ancho + 6, int(y1)), color, -1)
|
||||
cv2.putText(imagen, texto, (int(x1) + 3, int(y1) - 5),
|
||||
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
|
||||
|
||||
sello = time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
cv2.putText(imagen, sello, (12, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 4)
|
||||
cv2.putText(imagen, sello, (12, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
|
||||
return imagen
|
||||
|
||||
def _guarda(self, imagen, message):
|
||||
os.makedirs(self.snapshot_dir, exist_ok=True)
|
||||
tipo = "ladrido" if "Ladrido" in message else "carrera" if "corriendo" in message else "aviso"
|
||||
nombre = time.strftime("%Y-%m-%d_%H-%M-%S") + f"_{tipo}.jpg"
|
||||
ruta = os.path.join(self.snapshot_dir, nombre)
|
||||
if cv2.imwrite(ruta, imagen):
|
||||
self.snapshots += 1
|
||||
self.last_snapshot = ruta
|
||||
|
|
@ -16,6 +16,11 @@ ENV_KEYS = (
|
|||
"PET_AUDIO",
|
||||
"BARK_THRESHOLD",
|
||||
"BARK_COOLDOWN",
|
||||
"TELEGRAM_BOT_TOKEN",
|
||||
"TELEGRAM_CHAT_ID",
|
||||
"ALERT_COOLDOWN",
|
||||
"SNAPSHOTS",
|
||||
"SNAPSHOT_DIR",
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
75
pet_speed.py
75
pet_speed.py
|
|
@ -8,7 +8,8 @@ Pet Monitor - Visión por computador con cámara IP Tapo (RTSP) + NPU Hailo.
|
|||
y alerta de carrera según el desplazamiento del centroide entre fotogramas.
|
||||
- Escucha de ladridos sobre el audio de la propia cámara (ver pet_audio.py),
|
||||
en un segundo hilo y con su propia conexión RTSP.
|
||||
- Reconexión automática con backoff en ambos canales.
|
||||
- Avisos por consola y a Telegram con captura anotada (ver pet_alerts.py).
|
||||
- Reconexión automática con backoff en todos los canales.
|
||||
|
||||
Variables en .env (junto a este script):
|
||||
CAMERA_IP, CAMERA_USER, CAMERA_PW (obligatorias: cuenta de cámara Tapo)
|
||||
|
|
@ -16,6 +17,10 @@ Variables en .env (junto a este script):
|
|||
CAMERA_RTSP_URL (opcional: URL RTSP completa, tiene prioridad)
|
||||
PET_AUDIO (opcional: 0 desactiva la escucha de ladridos)
|
||||
BARK_THRESHOLD, BARK_COOLDOWN (opcional: umbral y anti-repetición del ladrido)
|
||||
TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID (opcional: avisos a Telegram; el chat se
|
||||
deduce solo si no se indica)
|
||||
ALERT_COOLDOWN (opcional: segundos entre capturas de un aviso)
|
||||
SNAPSHOTS, SNAPSHOT_DIR (opcional: guardar las capturas en disco)
|
||||
|
||||
Ejecución: uv run pet_speed.py
|
||||
"""
|
||||
|
|
@ -30,8 +35,10 @@ import cv2
|
|||
import numpy as np
|
||||
from picamera2.devices.hailo import Hailo, hailo_architecture
|
||||
|
||||
from pet_alerts import AlertDispatcher
|
||||
from pet_audio import AudioListener, BarkDetector
|
||||
from pet_common import build_rtsp_url, load_env, mask_url
|
||||
from pet_telegram import TelegramNotifier
|
||||
|
||||
# --- Configuración -------------------------------------------------------
|
||||
# Clases COCO: 15 = Cat (Gato), 16 = Dog (Perro)
|
||||
|
|
@ -44,6 +51,9 @@ STATS_INTERVAL = 10.0 # Segundos entre líneas de estadísticas
|
|||
BARK_THRESHOLD = 0.15 # Puntuación de Bark a partir de la cual se avisa
|
||||
BARK_COOLDOWN = 15.0 # Segundos mínimos entre dos avisos de ladrido
|
||||
|
||||
ALERT_COOLDOWN = 60.0 # Segundos mínimos entre capturas del mismo tipo de aviso
|
||||
SNAPSHOT_DIR = "capturas"
|
||||
|
||||
# Orden de color que espera el HEF. Los modelos de Hailo Model Zoo se compilan
|
||||
# para RGB; con PET_COLOR_ORDER=bgr se invierte (útil para comparar precisión).
|
||||
COLOR_ORDER = os.environ.get("PET_COLOR_ORDER", "rgb").lower()
|
||||
|
|
@ -208,6 +218,17 @@ class RtspVideoStream:
|
|||
self._new_frame.clear()
|
||||
return last_seq, None, None
|
||||
|
||||
def latest_frame(self):
|
||||
"""
|
||||
Copia del último fotograma recibido, para las capturas de alerta.
|
||||
|
||||
Se copia fuera del cerrojo porque el hilo lector reemplaza el array en
|
||||
lugar de modificarlo, así que la referencia no cambia bajo los pies.
|
||||
"""
|
||||
with self._lock:
|
||||
frame = self._frame
|
||||
return None if frame is None else frame.copy()
|
||||
|
||||
def stop(self):
|
||||
self._stop.set()
|
||||
self._new_frame.set()
|
||||
|
|
@ -215,8 +236,23 @@ class RtspVideoStream:
|
|||
self._thread.join(timeout=5.0)
|
||||
|
||||
|
||||
# --- Escucha de ladridos -------------------------------------------------
|
||||
def start_audio_listener(url, env):
|
||||
# --- Arranque de los canales de aviso ------------------------------------
|
||||
def start_telegram(env):
|
||||
"""Arranca el envío a Telegram, o explica por qué no se puede."""
|
||||
token = env.get("TELEGRAM_BOT_TOKEN")
|
||||
if not token:
|
||||
print("[INFO] Telegram no configurado (falta TELEGRAM_BOT_TOKEN); los avisos solo saldrán por consola.")
|
||||
return None
|
||||
try:
|
||||
notifier = TelegramNotifier(token, env.get("TELEGRAM_CHAT_ID") or None).start()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f"[WARN] Telegram desactivado: {exc}")
|
||||
return None
|
||||
print("[INFO] Telegram activo.")
|
||||
return notifier
|
||||
|
||||
|
||||
def start_audio_listener(url, env, alerts):
|
||||
"""
|
||||
Arranca el hilo de audio, o explica por qué no se puede.
|
||||
|
||||
|
|
@ -229,7 +265,7 @@ def start_audio_listener(url, env):
|
|||
|
||||
try:
|
||||
detector = BarkDetector()
|
||||
except (FileNotFoundError, Exception) as exc: # noqa: BLE001
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f"[WARN] Audio desactivado: {exc}")
|
||||
return None
|
||||
|
||||
|
|
@ -237,7 +273,7 @@ def start_audio_listener(url, env):
|
|||
cooldown = float(env.get("BARK_COOLDOWN", BARK_COOLDOWN))
|
||||
|
||||
def avisa(ladrido, grupo, _cuando):
|
||||
print(f"🔊 [ALERTA] ¡Ladrido detectado! Bark {ladrido:.2f} (grupo canino {grupo:.2f})", flush=True)
|
||||
alerts.alert("ladrido", f"🔊 [ALERTA] ¡Ladrido detectado! Bark {ladrido:.2f} (grupo canino {grupo:.2f})")
|
||||
|
||||
listener = AudioListener(url, detector, on_bark=avisa, threshold=threshold, cooldown=cooldown).start()
|
||||
print(f"[INFO] Escucha de ladridos activa (umbral Bark {threshold:.2f}, aviso cada {cooldown:.0f}s como mínimo).")
|
||||
|
|
@ -259,7 +295,17 @@ def main():
|
|||
print(f"[INFO] Cámara Tapo: {mask_url(url)}")
|
||||
|
||||
stream = RtspVideoStream(url).start()
|
||||
audio = start_audio_listener(url, env)
|
||||
|
||||
notifier = start_telegram(env)
|
||||
alerts = AlertDispatcher(
|
||||
stream,
|
||||
notifier=notifier,
|
||||
snapshot_dir=env.get("SNAPSHOT_DIR", SNAPSHOT_DIR),
|
||||
save_snapshots=env.get("SNAPSHOTS", "1") != "0",
|
||||
capture_cooldown=float(env.get("ALERT_COOLDOWN", ALERT_COOLDOWN)),
|
||||
)
|
||||
audio = start_audio_listener(url, env, alerts)
|
||||
|
||||
print("[INFO] Monitorizando mascotas (Gatos y Perros)... Presiona Ctrl+C para salir.\n")
|
||||
|
||||
running = threading.Event()
|
||||
|
|
@ -313,13 +359,22 @@ def main():
|
|||
x1, y1, x2, y2 = box_to_source(det, scale, pad_x, pad_y, src_w, src_h)
|
||||
centroid = ((x1 + x2) / 2.0 / src_w, (y1 + y2) / 2.0 / src_h)
|
||||
|
||||
# Cajas del fotograma actual, para anotar la captura del aviso
|
||||
alerts.set_pet_boxes([(label, float(confidence), (x1, y1, x2, y2))])
|
||||
|
||||
if last_centroid and dt > 0:
|
||||
dist = math.hypot(centroid[0] - last_centroid[0], centroid[1] - last_centroid[1])
|
||||
speed = dist / dt
|
||||
if speed > SPEED_THRESHOLD:
|
||||
print(f"⚠️ [ALERTA] ¡{label} corriendo! Vel: {speed:.2f} norm/s (Conf: {confidence:.2f})")
|
||||
alerts.alert(
|
||||
"carrera",
|
||||
f"⚠️ [ALERTA] ¡{label} corriendo! Vel: {speed:.2f} norm/s (Conf: {confidence:.2f})",
|
||||
)
|
||||
else:
|
||||
print(f"🐾 [OK] {label} detectado (calmado). Vel: {speed:.2f} norm/s (Conf: {confidence:.2f})")
|
||||
alerts.alert(
|
||||
"calmado",
|
||||
f"🐾 [OK] {label} detectado (calmado). Vel: {speed:.2f} norm/s (Conf: {confidence:.2f})",
|
||||
)
|
||||
|
||||
last_centroid = centroid
|
||||
break # Evaluamos la mascota principal detectada
|
||||
|
|
@ -328,6 +383,7 @@ def main():
|
|||
|
||||
if not pet_found:
|
||||
last_centroid = None
|
||||
alerts.set_pet_boxes([])
|
||||
|
||||
last_frame_time = frame_time
|
||||
|
||||
|
|
@ -344,6 +400,7 @@ def main():
|
|||
f" || audio: Bark máx {audio.take_max_bark():.2f} (umbral {audio.threshold:.2f}) "
|
||||
f"| ventanas {audio.windows} | ladridos {audio.alerts}"
|
||||
)
|
||||
linea += f" || {alerts.stats()}"
|
||||
print(linea)
|
||||
last_stats = now
|
||||
frames_processed = 0
|
||||
|
|
@ -353,6 +410,8 @@ def main():
|
|||
stream.stop()
|
||||
if audio is not None:
|
||||
audio.stop()
|
||||
if notifier is not None:
|
||||
notifier.stop()
|
||||
hailo.close()
|
||||
print("[INFO] Streams cerrados y acelerador Hailo liberado correctamente.")
|
||||
|
||||
|
|
|
|||
182
pet_telegram.py
Normal file
182
pet_telegram.py
Normal file
|
|
@ -0,0 +1,182 @@
|
|||
"""
|
||||
Envío de avisos a Telegram.
|
||||
|
||||
Las llamadas HTTP nunca bloquean los bucles de detección: se encolan y un hilo
|
||||
aparte las envía por orden. Si la red falla, reintenta con retroceso exponencial;
|
||||
si la cola se llena (Telegram caído mucho rato), se descartan los avisos viejos
|
||||
en favor de los nuevos, porque un aviso de hace cinco minutos ya no sirve.
|
||||
|
||||
El token se trata como secreto: jamás se imprime, y todo texto de error pasa
|
||||
por el enmascarador (la URL de la API lo lleva dentro).
|
||||
"""
|
||||
|
||||
import json
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
from pet_common import mask_secrets
|
||||
|
||||
API = "https://api.telegram.org/bot%s/%s"
|
||||
|
||||
|
||||
def descubrir_chat_id(token, timeout=(5, 15)):
|
||||
"""
|
||||
Deduce el chat_id a partir del último mensaje recibido por el bot.
|
||||
|
||||
Evita tener que buscarlo a mano: basta con escribirle algo al bot y lanzar
|
||||
esto. Devuelve (chat_id, descripcion) o (None, motivo).
|
||||
"""
|
||||
try:
|
||||
respuesta = requests.get(API % (token, "getUpdates"), timeout=timeout)
|
||||
respuesta.raise_for_status()
|
||||
datos = respuesta.json()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return None, mask_secrets(exc, [token])
|
||||
|
||||
if not datos.get("ok"):
|
||||
return None, mask_secrets(datos.get("description", "respuesta sin ok"), [token])
|
||||
|
||||
for actualizacion in reversed(datos.get("result", [])):
|
||||
mensaje = actualizacion.get("message") or actualizacion.get("edited_message") or {}
|
||||
chat = mensaje.get("chat") or {}
|
||||
if chat.get("id") is not None:
|
||||
nombre = chat.get("title") or chat.get("first_name") or chat.get("username") or "?"
|
||||
return chat["id"], nombre
|
||||
return None, "el bot no ha recibido ningún mensaje todavía"
|
||||
|
||||
|
||||
class TelegramNotifier:
|
||||
"""Cola de avisos con un hilo que los envía a Telegram."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
token,
|
||||
chat_id=None,
|
||||
queue_size=16,
|
||||
connect_timeout=5,
|
||||
read_timeout=30,
|
||||
max_retries=3,
|
||||
):
|
||||
self._token = token
|
||||
self._chat_id = chat_id
|
||||
self._timeout = (connect_timeout, read_timeout)
|
||||
self._max_retries = max_retries
|
||||
self._queue = queue.Queue(maxsize=queue_size)
|
||||
self._thread = None
|
||||
|
||||
self.sent = 0
|
||||
self.errors = 0
|
||||
self.dropped = 0
|
||||
self.last_error = None
|
||||
|
||||
# --- ciclo de vida ---------------------------------------------------
|
||||
def start(self):
|
||||
"""Comprueba la configuración y arranca el hilo de envío."""
|
||||
if self._chat_id is None:
|
||||
chat_id, detalle = descubrir_chat_id(self._token, self._timeout)
|
||||
if chat_id is None:
|
||||
raise RuntimeError(f"no se pudo deducir el chat_id: {detalle}")
|
||||
self._chat_id = chat_id
|
||||
print(f"[INFO] Telegram: chat detectado ({detalle}, id {chat_id}).")
|
||||
self._thread = threading.Thread(target=self._loop, name="telegram", daemon=True)
|
||||
self._thread.start()
|
||||
return self
|
||||
|
||||
def stop(self, timeout=15.0):
|
||||
"""Deja de aceptar avisos y espera a que se envíe lo que queda en cola."""
|
||||
self._queue.put(None)
|
||||
if self._thread is not None:
|
||||
self._thread.join(timeout=timeout)
|
||||
|
||||
# --- API pública -----------------------------------------------------
|
||||
def send_message(self, texto):
|
||||
return self._encola(("message", None, texto))
|
||||
|
||||
def send_photo(self, jpeg, caption=""):
|
||||
return self._encola(("photo", jpeg, caption))
|
||||
|
||||
# --- interno ---------------------------------------------------------
|
||||
def _encola(self, item):
|
||||
try:
|
||||
self._queue.put_nowait(item)
|
||||
return True
|
||||
except queue.Full:
|
||||
# Nos quedamos con el aviso nuevo y tiramos el más viejo
|
||||
try:
|
||||
self._queue.get_nowait()
|
||||
self._queue.task_done()
|
||||
self.dropped += 1
|
||||
except queue.Empty:
|
||||
pass
|
||||
try:
|
||||
self._queue.put_nowait(item)
|
||||
except queue.Full:
|
||||
self.dropped += 1
|
||||
return False
|
||||
return True
|
||||
|
||||
def _loop(self):
|
||||
while True:
|
||||
item = self._queue.get()
|
||||
try:
|
||||
if item is None:
|
||||
return
|
||||
self._envia(item)
|
||||
finally:
|
||||
self._queue.task_done()
|
||||
|
||||
def _envia(self, item):
|
||||
tipo, datos, texto = item
|
||||
url = API % (self._token, "sendPhoto" if tipo == "photo" else "sendMessage")
|
||||
|
||||
for intento in range(1, self._max_retries + 1):
|
||||
try:
|
||||
if tipo == "photo":
|
||||
respuesta = requests.post(
|
||||
url,
|
||||
data={"chat_id": self._chat_id, "caption": texto},
|
||||
files={"photo": ("captura.jpg", datos, "image/jpeg")},
|
||||
timeout=self._timeout,
|
||||
)
|
||||
else:
|
||||
respuesta = requests.post(
|
||||
url,
|
||||
data={"chat_id": self._chat_id, "text": texto},
|
||||
timeout=self._timeout,
|
||||
)
|
||||
|
||||
# Límite de velocidad: Telegram dice cuánto hay que esperar
|
||||
if respuesta.status_code == 429:
|
||||
espera = self._retry_after(respuesta)
|
||||
print(f"[WARN] Telegram limitó la frecuencia; esperando {espera:.0f}s", flush=True)
|
||||
time.sleep(espera)
|
||||
continue
|
||||
|
||||
respuesta.raise_for_status()
|
||||
cuerpo = respuesta.json()
|
||||
if not cuerpo.get("ok"):
|
||||
raise RuntimeError(cuerpo.get("description", "respuesta sin ok"))
|
||||
|
||||
self.sent += 1
|
||||
return True
|
||||
|
||||
except Exception as exc: # noqa: BLE001
|
||||
self.last_error = mask_secrets(exc, [self._token])
|
||||
if intento < self._max_retries:
|
||||
time.sleep(min(2 ** intento, 10))
|
||||
|
||||
self.errors += 1
|
||||
print(f"[WARN] No se pudo enviar el aviso a Telegram: {self.last_error}", flush=True)
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def _retry_after(respuesta):
|
||||
"""Segundos que pide Telegram ante un 429 (por defecto, 5)."""
|
||||
try:
|
||||
parametros = respuesta.json().get("parameters", {})
|
||||
return float(parametros.get("retry_after", 5))
|
||||
except (ValueError, json.JSONDecodeError):
|
||||
return 5.0
|
||||
102
scripts/telegram_check.py
Normal file
102
scripts/telegram_check.py
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Comprueba la configuración de Telegram del Pet Monitor.
|
||||
|
||||
Hace todo el trabajo pesado de la puesta en marcha:
|
||||
1. Valida el token contra la API (getMe) y te dice de qué bot se trata.
|
||||
2. Deduce el chat_id a partir del último mensaje que haya recibido el bot.
|
||||
3. Manda un mensaje de prueba y una foto de prueba.
|
||||
4. Te dice exactamente qué líneas añadir al .env.
|
||||
|
||||
Uso:
|
||||
uv run scripts/telegram_check.py
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from pet_common import load_env, mask_secrets # noqa: E402
|
||||
from pet_telegram import API, TelegramNotifier, descubrir_chat_id # noqa: E402
|
||||
import requests # noqa: E402
|
||||
|
||||
|
||||
def main():
|
||||
env = load_env()
|
||||
token = env.get("TELEGRAM_BOT_TOKEN")
|
||||
if not token:
|
||||
print("No hay TELEGRAM_BOT_TOKEN en el .env.\n")
|
||||
print("Para crearlo:")
|
||||
print(" 1. Abre Telegram y habla con @BotFather")
|
||||
print(" 2. Envía /newbot, ponle nombre y usuario al bot")
|
||||
print(" 3. Copia el token que te devuelve y añádelo a ~/pet_monitor/.env:")
|
||||
print(" TELEGRAM_BOT_TOKEN=123456:ABC...")
|
||||
print(" 4. Escríbele cualquier mensaje a tu bot (para que él te vea)")
|
||||
print(" 5. Vuelve a lanzar este script")
|
||||
return 1
|
||||
|
||||
# 1. ¿El token es válido y de quién es?
|
||||
try:
|
||||
respuesta = requests.get(API % (token, "getMe"), timeout=(5, 15))
|
||||
respuesta.raise_for_status()
|
||||
datos = respuesta.json()
|
||||
if not datos.get("ok"):
|
||||
raise RuntimeError(datos.get("description", "respuesta sin ok"))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
print(f"[ERROR] El token no funciona: {mask_secrets(exc, [token])}")
|
||||
return 1
|
||||
|
||||
bot = datos["result"]
|
||||
print(f"[OK] Token válido. Bot: @{bot.get('username')} ({bot.get('first_name')})")
|
||||
|
||||
# 2. chat_id: del .env o deducido del último mensaje recibido
|
||||
chat_id = env.get("TELEGRAM_CHAT_ID")
|
||||
if chat_id:
|
||||
print(f"[OK] TELEGRAM_CHAT_ID tomado del .env: {chat_id}")
|
||||
else:
|
||||
chat_id, detalle = descubrir_chat_id(token)
|
||||
if chat_id is None:
|
||||
print(f"[ERROR] No se pudo deducir el chat_id: {detalle}")
|
||||
print(" Escríbele un mensaje a tu bot desde Telegram y vuelve a intentarlo.")
|
||||
return 1
|
||||
print(f"[OK] chat_id deducido: {chat_id} ({detalle})")
|
||||
|
||||
# 3. Mensaje y foto de prueba
|
||||
notifier = TelegramNotifier(token, chat_id).start()
|
||||
notifier.send_message(
|
||||
"🐾 Pet Monitor: prueba de conexión correcta.\n"
|
||||
f"Enviado el {time.strftime('%Y-%m-%d a las %H:%M:%S')} desde la Raspberry Pi.\n"
|
||||
"Si ves esta foto, los avisos ya funcionan."
|
||||
)
|
||||
|
||||
imagen = np.full((360, 640, 3), 40, dtype=np.uint8)
|
||||
cv2.rectangle(imagen, (40, 40), (600, 320), (0, 200, 0), 3)
|
||||
cv2.putText(imagen, "PRUEBA DE CAPTURA", (70, 160), cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255, 255, 255), 2)
|
||||
cv2.putText(imagen, time.strftime("%Y-%m-%d %H:%M:%S"), (70, 220), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (200, 200, 200), 2)
|
||||
ok, buffer = cv2.imencode(".jpg", imagen)
|
||||
if ok:
|
||||
notifier.send_photo(
|
||||
buffer.tobytes(),
|
||||
"Captura de prueba (un aviso real llevará la foto de la cámara con las cajas).",
|
||||
)
|
||||
|
||||
notifier.stop(timeout=30)
|
||||
print(f"[OK] Enviados: {notifier.sent} | fallos: {notifier.errors}")
|
||||
if notifier.last_error:
|
||||
print(f"[WARN] Último error: {notifier.last_error}")
|
||||
|
||||
# 4. Qué falta por apuntar
|
||||
if not env.get("TELEGRAM_CHAT_ID"):
|
||||
print("\nAñade esta línea a ~/pet_monitor/.env para no deducirlo cada vez:")
|
||||
print(f" TELEGRAM_CHAT_ID={chat_id}")
|
||||
print("\nComprueba el móvil: deberías tener un mensaje y una foto del bot.")
|
||||
return 0 if notifier.errors else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
Add table
Reference in a new issue