Pet-Monitor/pet_alerts.py

"""
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