02-ago: [proyecto] vigilancia - script test_alerta.py (presencia real -> Telegram IDD + anuncio Alexa) probado con INES

vigilancia/test_alerta.py

@@ -0,0 +1,80 @@
+import os
+import sys
+import time
+from datetime import datetime
+
+import cv2
+
+sys.path.insert(0, r'C:\Users\juanm\Documents\GitHub\biblioteca_conocimiento_laboratorio\vigilancia')
+import config
+import reconocedor as rec
+from camara import CamaraRTSP
+from notificador import disparar_webhook_alexa, enviar_telegram
+
+ALEXA_TEXTO = "Inés, ¿qué haces? No entres ahí."
+NOMBRE_BUSCADO = "INES"
+DURACION = 90
+COOLDOWN = 30
+
+r = rec.cargar_reconocedor()
+print("Conocidos:", list(r.conocidos.keys()))
+if NOMBRE_BUSCADO not in r.conocidos:
+    print(f"{NOMBRE_BUSCADO} no esta registrado")
+    sys.exit(1)
+
+camaras = [CamaraRTSP(c["rtsp"], c["nombre"]) for c in config.CAMARAS]
+os.makedirs(config.SNAPSHOT_DIR, exist_ok=True)
+
+t0 = time.time()
+ultima_alerta = 0.0
+disparada = False
+
+print(f"Esperando a que {NOMBRE_BUSCADO} pase delante de la camara "
+      f"({DURACION}s)...")
+while time.time() - t0 < DURACION:
+    for cam in camaras:
+        frame = cam.leer_frame()
+        if frame is None:
+            continue
+
+        personas = r._detectar_personas(frame)
+        if not personas:
+            continue
+
+        for p in personas:
+            x, y, w, h = p
+            caras = r.detectar_caras(frame)
+            for c in caras:
+                cx, cy, cw, ch = (int(v) for v in c[:4])
+                if cx < x + w and cx + cw > x and cy < y + h and cy + ch > y:
+                    emb = r.embedding_cara(frame, c)
+                    nombre, score = r.identificar_cara(emb)
+                    if nombre == NOMBRE_BUSCADO and score >= rec.UMBRAL_CARA:
+                        ahora = time.time()
+                        if ahora - ultima_alerta < COOLDOWN:
+                            continue
+                        ultima_alerta = ahora
+                        disparada = True
+
+                        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+                        ruta = os.path.join(
+                            config.SNAPSHOT_DIR,
+                            f"alerta_{cam.nombre}_{ts}.jpg",
+                        )
+                        cv2.imwrite(ruta, frame)
+
+                        texto = f"👧 Es {NOMBRE_BUSCADO} ({score:.2f}) - {cam.nombre} - {ts}"
+                        print(f"[{cam.nombre}] IDENTIFICADA {NOMBRE_BUSCADO} "
+                              f"score {score:.2f}")
+                        print(f"  -> Telegram: {texto}")
+                        enviar_telegram(ruta, texto)
+                        print(f"  -> Alexa: {ALEXA_TEXTO}")
+                        disparar_webhook_alexa(ALEXA_TEXTO)
+                        print("  ALERTA ENVIADA")
+                        break
+    time.sleep(0.15)
+
+if not disparada:
+    print(f"NO se identifico a {NOMBRE_BUSCADO} en {DURACION}s")
+else:
+    print(f"\nAlerta enviada: Telegram + Alexa.")