81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
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.")
|