#!/usr/bin/env python3 """ Pet Monitor - Visión por computador con cámara IP Tapo (RTSP) + NPU Hailo-8. Sustituye la cámara CSI interna de la Raspberry Pi por la cámara Tapo de red: - Ingesta RTSP en un hilo aparte, que descarta los frames viejos y expone siempre el más reciente (evita la latencia acumulada típica de RTSP). - Reconexión automática con backoff si la cámara se cae o cambia de IP. - Inferencia YOLOv8s sobre la NPU Hailo (HEF del sistema). - Detección de gato/perro (clases COCO 15 y 16) y alerta de carrera según el desplazamiento del centroide entre frames consecutivos. Variables en .env (junto a este script): CAMERA_IP, CAMERA_USER, CAMERA_PW (obligatorias: cuenta de cámara Tapo) CAMERA_STREAM (opcional: stream2 = 720p, stream1 = 2K) CAMERA_RTSP_URL (opcional: URL RTSP completa, tiene prioridad) Ejecución: uv run pet_speed.py """ import math import os import signal import threading import time import urllib.parse import cv2 import numpy as np from picamera2.devices.hailo import Hailo, hailo_architecture # --- Configuración ------------------------------------------------------- # Clases COCO: 15 = Cat (Gato), 16 = Dog (Perro) TARGET_CLASSES = {15: "Gato", 16: "Perro"} CONF_THRESHOLD = 0.45 # Confianza mínima para considerar una detección SPEED_THRESHOLD = 0.30 # Velocidad normalizada por segundo (fracción del frame) DEFAULT_STREAM = "stream2" # stream1 = 2304x1296@15, stream2 = 1280x720@15 STATS_INTERVAL = 10.0 # Segundos entre líneas de estadísticas # 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() ENV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") LETTERBOX_FILL = (114, 114, 114) # Latencia baja en RTSP: forzar TCP evita el jitter y las pérdidas de UDP. # (No se usa 'stimeout': fue eliminado en FFmpeg 6, que es el que trae OpenCV 4.10) os.environ.setdefault("OPENCV_FFMPEG_CAPTURE_OPTIONS", "rtsp_transport;tcp") # --- Utilidades ---------------------------------------------------------- def load_env(path=ENV_PATH): """Lee un .env sencillo (CLAVE=valor) sin dependencias externas.""" env = {} if os.path.exists(path): with open(path) as fh: for line in fh: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) env[key.strip()] = value.strip().strip('"').strip("'") # Las variables de entorno reales tienen prioridad sobre el fichero, para # poder sobreescribirlas desde systemd o desde una prueba puntual. for key in ("CAMERA_IP", "CAMERA_USER", "CAMERA_PW", "CAMERA_STREAM", "CAMERA_RTSP_URL"): if os.environ.get(key): env[key] = os.environ[key] return env def mask_url(url): """Oculta las credenciales de la URL para poder mostrarla por consola.""" parts = urllib.parse.urlsplit(url) if not parts.hostname: return "***" port = f":{parts.port}" if parts.port else "" return f"{parts.scheme}://***@{parts.hostname}{port}{parts.path}" def build_rtsp_url(env, stream): """Construye la URL RTSP de la Tapo, escapando usuario y contraseña.""" if env.get("CAMERA_RTSP_URL"): return env["CAMERA_RTSP_URL"] ip = env.get("CAMERA_IP") user = env.get("CAMERA_USER") password = env.get("CAMERA_PW") missing = [k for k, v in (("CAMERA_IP", ip), ("CAMERA_USER", user), ("CAMERA_PW", password)) if not v] if missing: raise SystemExit(f"[ERROR] Faltan variables en {ENV_PATH}: {', '.join(missing)}") return "rtsp://%s:%s@%s:554/%s" % ( urllib.parse.quote(user, safe=""), urllib.parse.quote(password, safe=""), ip, stream, ) def get_hailo_model_path(): """Elige el HEF de YOLOv8s que corresponde a la arquitectura detectada.""" arch = hailo_architecture() if arch == "HAILO8": candidates = ["/usr/share/hailo-models/yolov8s_h8.hef", "/usr/share/hailo-models/yolov8s_h8l.hef"] else: candidates = ["/usr/share/hailo-models/yolov8s_h8l.hef", "/usr/share/hailo-models/yolov8s_h8.hef"] for path in candidates: if os.path.exists(path): return path raise FileNotFoundError("No se encontró ningún modelo YOLOv8 HEF en /usr/share/hailo-models/") def letterbox(frame, width, height): """ Reescala manteniendo la relación de aspecto y rellena hasta width x height. Devuelve (imagen, escala, pad_x, pad_y) para poder deshacer la transformación. """ src_h, src_w = frame.shape[:2] scale = min(width / src_w, height / src_h) new_w, new_h = int(round(src_w * scale)), int(round(src_h * scale)) resized = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_LINEAR) pad_x, pad_y = (width - new_w) // 2, (height - new_h) // 2 padded = cv2.copyMakeBorder( resized, pad_y, height - new_h - pad_y, pad_x, width - new_w - pad_x, cv2.BORDER_CONSTANT, value=LETTERBOX_FILL, ) return padded, scale, pad_x, pad_y def box_to_source(box, scale, pad_x, pad_y, src_w, src_h): """ Convierte una caja normalizada del modelo (letterbox) a píxeles del frame original. Hailo NMS devuelve [ymin, xmin, ymax, xmax, confidence]. """ ymin, xmin, ymax, xmax = box[:4] model_w = src_w * scale + 2 * pad_x model_h = src_h * scale + 2 * pad_y x1 = (xmin * model_w - pad_x) / scale x2 = (xmax * model_w - pad_x) / scale y1 = (ymin * model_h - pad_y) / scale y2 = (ymax * model_h - pad_y) / scale clip_x = lambda v: float(min(max(v, 0.0), src_w)) # noqa: E731 clip_y = lambda v: float(min(max(v, 0.0), src_h)) # noqa: E731 return clip_x(x1), clip_y(y1), clip_x(x2), clip_y(y2) # --- Ingesta de vídeo ---------------------------------------------------- class RtspVideoStream: """ Hilo lector de RTSP: mantiene siempre el último frame disponible. Si el bucle principal va más lento que la cámara, los frames intermedios se descartan solos (no se encolan), así la detección trabaja sobre imagen en vivo y no sobre un buffer con retardo creciente. """ def __init__(self, url, reconnect_delay=2.0, max_reconnect_delay=30.0): self._url = url self._reconnect_delay = reconnect_delay self._max_reconnect_delay = max_reconnect_delay self._lock = threading.Lock() self._new_frame = threading.Event() self._stop = threading.Event() self._thread = None self._seq = 0 self._frame = None self._timestamp = 0.0 self.frames_read = 0 self.reconnects = 0 self.connected = False self.capture_fps = 0.0 def start(self): self._thread = threading.Thread(target=self._loop, name="rtsp-reader", daemon=True) self._thread.start() return self def _open_capture(self): cap = cv2.VideoCapture(self._url, cv2.CAP_FFMPEG) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) if not cap.isOpened(): cap.release() return None return cap def _loop(self): cap = None delay = self._reconnect_delay window_start = time.time() window_frames = 0 while not self._stop.is_set(): if cap is None: cap = self._open_capture() if cap is None: self.connected = False print(f"[WARN] No hay stream RTSP; reintentando en {delay:.0f}s", flush=True) self._stop.wait(delay) delay = min(delay * 2, self._max_reconnect_delay) self.reconnects += 1 continue self.connected = True delay = self._reconnect_delay window_start, window_frames = time.time(), 0 print("[INFO] Stream RTSP conectado.", flush=True) ok, frame = cap.read() if not ok or frame is None: print("[WARN] Lectura RTSP interrumpida; reconectando...", flush=True) cap.release() cap = None self.connected = False self.reconnects += 1 self._stop.wait(delay) continue now = time.time() window_frames += 1 elapsed = now - window_start if elapsed >= 2.0: self.capture_fps = window_frames / elapsed window_start, window_frames = now, 0 with self._lock: self._seq += 1 self._frame = frame self._timestamp = now self.frames_read += 1 self._new_frame.set() if cap is not None: cap.release() def wait_for_frame(self, last_seq, timeout=1.0): """ Bloquea hasta que haya un frame más nuevo que last_seq. Devuelve (seq, frame, timestamp) o (last_seq, None, None) si expira. """ deadline = time.time() + timeout while not self._stop.is_set() and time.time() < deadline: with self._lock: seq, frame, ts = self._seq, self._frame, self._timestamp if seq > last_seq and frame is not None: return seq, frame, ts self._new_frame.wait(0.05) self._new_frame.clear() return last_seq, None, None def stop(self): self._stop.set() self._new_frame.set() if self._thread is not None: self._thread.join(timeout=5.0) # --- Bucle principal ----------------------------------------------------- def main(): env = load_env() stream_name = env.get("CAMERA_STREAM", DEFAULT_STREAM) url = build_rtsp_url(env, stream_name) model_path = get_hailo_model_path() print(f"[INFO] Cargando modelo Hailo: {model_path} ...") hailo = Hailo(model_path) model_h, model_w, _ = hailo.get_input_shape() print(f"[INFO] Modelo cargado. Resolución de entrada: {model_w}x{model_h}") # Ocultamos la contraseña al mostrarla por consola print(f"[INFO] Cámara Tapo: {mask_url(url)}") stream = RtspVideoStream(url).start() print("[INFO] Monitorizando mascotas (Gatos y Perros)... Presiona Ctrl+C para salir.\n") running = threading.Event() running.set() def handle_signal(signum, _frame): print(f"\n[INFO] Señal {signum} recibida; deteniendo...", flush=True) running.clear() signal.signal(signal.SIGINT, handle_signal) signal.signal(signal.SIGTERM, handle_signal) last_seq = 0 last_centroid = None last_frame_time = 0.0 last_stats = time.time() frames_processed = 0 inference_time = 0.0 try: while running.is_set(): seq, frame, frame_time = stream.wait_for_frame(last_seq) if frame is None: continue last_seq = seq letterboxed, scale, pad_x, pad_y = letterbox(frame, model_w, model_h) if COLOR_ORDER == "bgr": model_input = np.ascontiguousarray(letterboxed) else: model_input = np.ascontiguousarray(cv2.cvtColor(letterboxed, cv2.COLOR_BGR2RGB)) started = time.time() detections_by_class = hailo.run(model_input) inference_time += time.time() - started frames_processed += 1 src_h, src_w = frame.shape[:2] dt = frame_time - last_frame_time if last_frame_time else 0.0 pet_found = False for class_id, label in TARGET_CLASSES.items(): if class_id >= len(detections_by_class): continue for det in detections_by_class[class_id]: confidence = det[4] if confidence <= CONF_THRESHOLD: continue pet_found = True 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) 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})") else: print(f"🐾 [OK] {label} detectado (calmado). Vel: {speed:.2f} norm/s (Conf: {confidence:.2f})") last_centroid = centroid break # Evaluamos la mascota principal detectada if pet_found: break if not pet_found: last_centroid = None last_frame_time = frame_time now = time.time() if now - last_stats >= STATS_INTERVAL: avg_inf = inference_time / frames_processed * 1000 if frames_processed else 0.0 print( f"[STATS] captura {stream.capture_fps:.1f} fps | análisis " f"{frames_processed / (now - last_stats):.1f} fps | inferencia {avg_inf:.1f} ms " f"| reconexiones {stream.reconnects}" ) last_stats = now frames_processed = 0 inference_time = 0.0 finally: stream.stop() hailo.close() print("[INFO] Stream RTSP cerrado y acelerador Hailo liberado correctamente.") if __name__ == "__main__": main()