From 762699f41e383586e797199660cbc33bc8d7d768 Mon Sep 17 00:00:00 2001 From: abraham Date: Thu, 1 Oct 2026 18:32:06 +0200 Subject: [PATCH] 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. --- .gitignore | 3 + Readme.md | 45 ++++++++-- pet_alerts.py | 139 +++++++++++++++++++++++++++++ pet_common.py | 5 ++ pet_speed.py | 75 ++++++++++++++-- pet_telegram.py | 182 ++++++++++++++++++++++++++++++++++++++ scripts/telegram_check.py | 102 +++++++++++++++++++++ 7 files changed, 537 insertions(+), 14 deletions(-) create mode 100644 pet_alerts.py create mode 100644 pet_telegram.py create mode 100644 scripts/telegram_check.py diff --git a/.gitignore b/.gitignore index 7136d0c..e27dff2 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,6 @@ hailort.1.log # Modelos grandes: se descargan con scripts/fetch_models.sh models/*.onnx + +# Capturas de las alertas +capturas/ diff --git a/Readme.md b/Readme.md index a5cedb8..7e8bab3 100644 --- a/Readme.md +++ b/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: diff --git a/pet_alerts.py b/pet_alerts.py new file mode 100644 index 0000000..2811fbf --- /dev/null +++ b/pet_alerts.py @@ -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 diff --git a/pet_common.py b/pet_common.py index 33e4ff5..9bd2d01 100644 --- a/pet_common.py +++ b/pet_common.py @@ -16,6 +16,11 @@ ENV_KEYS = ( "PET_AUDIO", "BARK_THRESHOLD", "BARK_COOLDOWN", + "TELEGRAM_BOT_TOKEN", + "TELEGRAM_CHAT_ID", + "ALERT_COOLDOWN", + "SNAPSHOTS", + "SNAPSHOT_DIR", ) diff --git a/pet_speed.py b/pet_speed.py index b8b21fb..4de4c12 100644 --- a/pet_speed.py +++ b/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.") diff --git a/pet_telegram.py b/pet_telegram.py new file mode 100644 index 0000000..f11d421 --- /dev/null +++ b/pet_telegram.py @@ -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 diff --git a/scripts/telegram_check.py b/scripts/telegram_check.py new file mode 100644 index 0000000..da0524d --- /dev/null +++ b/scripts/telegram_check.py @@ -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())