diff --git a/.gitignore b/.gitignore index 20f869e..497c146 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,5 @@ telegram_mensajes.log vigilancia/snapshots/ vigilancia/*.pt vigilancia/.venv/ +vigilancia/modelos/ +vigilancia/conocidos/ diff --git a/vigilancia/enroll.py b/vigilancia/enroll.py new file mode 100644 index 0000000..90746ed --- /dev/null +++ b/vigilancia/enroll.py @@ -0,0 +1,121 @@ +import os +import sys + +import cv2 +import numpy as np + +import reconocedor as rec + +CONOCIDOS = rec.CONOCIDOS +WEB_CAM = 0 + + +def escribir(texto, frame): + cv2.putText(frame, texto, (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2) + + +def capturar_cara(r, frame): + caras = r.detectar_caras(frame) + if not caras: + return None, None + mejor = max(caras, key=lambda c: c[-1]) + emb = r.embedding_cara(frame, mejor) + return emb, mejor + + +def capturar_cuerpo(r, frame): + personas = r._detectar_personas(frame) + if not personas: + return None, None + mayor = max(personas, key=lambda p: p[2] * p[3]) + x, y, w, h = mayor + recorte = frame[y:y + h, x:x + w] + emb = r.embedding_cuerpo(recorte) + return emb, mayor + + +def main(): + nombres = sys.argv[1:3] + while len(nombres) < 2: + nombres.append(f"Hija{len(nombres) + 1}") + print(f"Registrando: {nombres}") + + r = rec.cargar_reconocedor() + os.makedirs(CONOCIDOS, exist_ok=True) + + cap = cv2.VideoCapture(WEB_CAM) + if not cap.isOpened(): + print("ERROR: no se pudo abrir la webcam") + sys.exit(1) + + cv2.namedWindow("Registro", cv2.WINDOW_NORMAL) + + for nombre in nombres: + carpeta = os.path.join(CONOCIDOS, nombre) + os.makedirs(carpeta, exist_ok=True) + print(f"\n>>> Hija: {nombre}") + + fase = "cara" + while True: + ok, frame = cap.read() + if not ok: + continue + + caras = r.detectar_caras(frame) + personas = r._detectar_personas(frame) + + for c in caras: + x, y, w, h = (int(v) for v in c[:4]) + cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2) + for p in personas: + x, y, w, h = p + cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 0, 255), 2) + + if fase == "cara": + escribir("CARA: pon la cara mirando a la camara y pulsa ESPACIO", frame) + else: + escribir("CUERPO: alejate (cuerpo entero) y pulsa B", frame) + escribir(f"Persona: {nombre}", frame) + + cv2.imshow("Registro", frame) + tecla = cv2.waitKey(1) & 0xFF + + if tecla == ord("q"): + cap.release() + cv2.destroyAllWindows() + print("Abortado por el usuario") + sys.exit(0) + + if fase == "cara" and tecla == ord(" "): + emb, caja = capturar_cara(r, frame) + if emb is None: + print(" No se vio ninguna cara clara. Reintenta.") + continue + np.save(os.path.join(carpeta, "face.npy"), emb) + x, y, w, h = (int(v) for v in caja[:4]) + cv2.imwrite(os.path.join(carpeta, "cara_ref.jpg"), + frame[max(0, y - 20):y + h + 20, max(0, x - 20):x + w + 20]) + print(f" Cara de {nombre} registrada (score {caja[-1]:.2f})") + fase = "cuerpo" + + elif fase == "cuerpo" and tecla == ord("b"): + emb, caja = capturar_cuerpo(r, frame) + if emb is None: + print(" No se detecto ninguna persona. Reintenta.") + continue + np.save(os.path.join(carpeta, "body.npy"), emb) + x, y, w, h = caja + cv2.imwrite(os.path.join(carpeta, "cuerpo_ref.jpg"), + frame[max(0, y):y + h, max(0, x):x + w]) + print(f" Cuerpo de {nombre} registrado") + break + + cap.release() + cv2.destroyAllWindows() + print("\nRegistro completado. Conocidos guardados:") + for n in nombres: + print(f" - {n} (face.npy + body.npy)") + + +if __name__ == "__main__": + main() diff --git a/vigilancia/main.py b/vigilancia/main.py index 6ba9c91..af2bd9a 100644 --- a/vigilancia/main.py +++ b/vigilancia/main.py @@ -9,10 +9,45 @@ from camara import CamaraRTSP from detector_movimiento import DetectorMovimiento from notificador import disparar_webhook_alexa, enviar_telegram -USAR_YOLO = False +USAR_YOLO = True +identificador = None -if USAR_YOLO: - from detector_persona import DetectorPersona + +def cargar_identificador(): + global identificador + if identificador is None: + try: + from reconocedor import CONOCIDOS, Reconocedor, os as _os + if _os.path.isdir(CONOCIDOS) and any( + _os.path.isdir(_os.path.join(CONOCIDOS, n)) + for n in _os.listdir(CONOCIDOS) + ): + identificador = Reconocedor() + print(f"Identificacion activa. Conocidos: " + f"{list(identificador.conocidos.keys())}") + else: + print("Sin conocidos registrados (ejecuta enroll.py). " + "Modo deteccion simple.") + except Exception as e: + print(f"No se pudo cargar identificador: {e}") + return identificador + + +def componer_alerta(resultados, cam_nombre, ts): + conocidas = sorted({r["nombre"] for r in resultados if "nombre" in r}) + desconocidas = sum(1 for r in resultados if r.get("desconocido")) + + if conocidas and desconocidas: + return (f"👧 {', '.join(conocidas)} | " + f"⚠️ {desconocidas} persona(s) desconocida(s) " + f"en {cam_nombre} - {ts}") + if conocidas: + return (f"👧 {', '.join(conocidas)} detectada(s) " + f"en {cam_nombre} - {ts}") + if desconocidas: + return (f"⚠️ {desconocidas} persona(s) desconocida(s) " + f"en {cam_nombre} - {ts}") + return f"🚨 Movimiento en {cam_nombre} - {ts}" def dentro_de_franja_horaria(hora_actual=None) -> bool: @@ -37,9 +72,11 @@ def main(): ultima_alerta = {} for i, c in enumerate(config.CAMARAS): - detectores.append(DetectorPersona() if USAR_YOLO else DetectorMovimiento()) + detectores.append(DetectorMovimiento()) ultima_alerta[i] = 0 + ident = cargar_identificador() + print("Sistema de vigilancia iniciado. Franja activa: " f"{config.HORA_INICIO}:00 - {config.HORA_FIN}:00") @@ -54,10 +91,7 @@ def main(): continue detector = detectores[i] - detectado = ( - detector.hay_persona(frame) if USAR_YOLO - else detector.hay_movimiento(frame) - ) + detectado = detector.hay_movimiento(frame) ahora = time.time() if detectado and (ahora - ultima_alerta[i] > config.COOLDOWN): @@ -65,6 +99,16 @@ def main(): ruta = os.path.join( config.SNAPSHOT_DIR, f"snapshot_{cam.nombre}_{ts}.jpg" ) + + texto = f"🚨 Movimiento en {cam.nombre} - {ts}" + if ident is not None: + try: + resultados = ident.analizar_frame(frame) + if resultados: + texto = componer_alerta(resultados, cam.nombre, ts) + except Exception as e: + print(f"[{cam.nombre}] Error identificando: {e}") + cv2_ok = True try: cv2.imwrite(ruta, frame) @@ -72,9 +116,6 @@ def main(): cv2_ok = False print(f"[{cam.nombre}] Error guardando snapshot: {e}") - tipo = "persona" if USAR_YOLO else "movimiento" - texto = f"⚠️ Detección de {tipo} en {cam.nombre} - {ts}" - if cv2_ok: enviar_telegram(ruta, texto) disparar_webhook_alexa(texto) diff --git a/vigilancia/reconocedor.py b/vigilancia/reconocedor.py new file mode 100644 index 0000000..0ebe201 --- /dev/null +++ b/vigilancia/reconocedor.py @@ -0,0 +1,184 @@ +import os + +import cv2 +import numpy as np +import torch +import torchvision.models as models +from torchvision import transforms + +BASE = os.path.dirname(os.path.abspath(__file__)) +MODELOS = os.path.join(BASE, "modelos") +CONOCIDOS = os.path.join(BASE, "conocidos") +YUNET_PATH = os.path.join(MODELOS, "face_detection_yunet.onnx") +ARCFACE_PATH = os.path.join(MODELOS, "buffalo_s", "w600k_mbf.onnx") +YOLO_PATH = os.path.join(BASE, "yolov8n.pt") + +UMBRAL_CARA = 0.40 +UMBRAL_CUERPO = 0.55 + + +class Reconocedor: + def __init__(self): + self.yunet = cv2.FaceDetectorYN_create(YUNET_PATH, "", (320, 320)) + self.yunet.setScoreThreshold(0.6) + self.arcface = cv2.dnn.readNetFromONNX(ARCFACE_PATH) + self.body = self._cargar_body() + self.tf_body = transforms.Compose([ + transforms.ToPILImage(), + transforms.Resize((224, 224)), + transforms.ToTensor(), + transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]), + ]) + self.conocidos = self._cargar_conocidos() + + def _cargar_body(self): + m = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1) + m.fc = torch.nn.Identity() + m.eval() + return m + + def _cargar_conocidos(self): + conocidos = {} + if not os.path.isdir(CONOCIDOS): + return conocidos + for nombre in os.listdir(CONOCIDOS): + carpeta = os.path.join(CONOCIDOS, nombre) + if not os.path.isdir(carpeta): + continue + emb_cara = self._cargar_npy(carpeta, "face") + emb_cuerpo = self._cargar_npy(carpeta, "body") + if emb_cara is not None or emb_cuerpo is not None: + conocidos[nombre] = (emb_cara, emb_cuerpo) + return conocidos + + def _cargar_npy(self, carpeta, tipo): + ruta = os.path.join(carpeta, f"{tipo}.npy") + if os.path.exists(ruta): + return np.load(ruta) + return None + + # ------------------------------------------------------------------ + # Cara + # ------------------------------------------------------------------ + def detectar_caras(self, frame): + h, w = frame.shape[:2] + self.yunet.setInputSize((w, h)) + _, caras = self.yunet.detect(frame) + if caras is None: + return [] + return [c for c in caras if c[-1] >= 0.6] + + def embedding_cara(self, frame, caja): + x, y, w, h = (int(v) for v in caja[:4]) + x, y = max(0, x), max(0, y) + recorte = frame[y:y + h, x:x + w] + if recorte.size == 0: + return None + blob = cv2.dnn.blobFromImage( + recorte, 1.0 / 128, (112, 112), (127.5, 127.5, 127.5), swapRB=True + ) + self.arcface.setInput(blob) + emb = self.arcface.forward().flatten() + norm = np.linalg.norm(emb) + return emb / norm if norm > 0 else None + + # ------------------------------------------------------------------ + # Cuerpo + # ------------------------------------------------------------------ + def embedding_cuerpo(self, recorte_bgr): + if recorte_bgr.size == 0: + return None + rgb = cv2.cvtColor(recorte_bgr, cv2.COLOR_BGR2RGB) + tensor = self.tf_body(rgb).unsqueeze(0) + with torch.no_grad(): + emb = self.body(tensor).numpy().flatten() + norm = np.linalg.norm(emb) + return emb / norm if norm > 0 else None + + # ------------------------------------------------------------------ + # Identificacion + # ------------------------------------------------------------------ + def identificar_cara(self, emb): + if emb is None or not self.conocidos: + return None, 0.0 + mejor, score = None, -1.0 + for nombre, (emb_cara, _) in self.conocidos.items(): + if emb_cara is not None: + sim = float(np.dot(emb, emb_cara)) + if sim > score: + score = sim + mejor = nombre + return (mejor, score) if score >= UMBRAL_CARA else (None, score) + + def identificar_cuerpo(self, emb): + if emb is None or not self.conocidos: + return None, 0.0 + mejor, score = None, -1.0 + for nombre, (_, emb_cuerpo) in self.conocidos.items(): + if emb_cuerpo is not None: + sim = float(np.dot(emb, emb_cuerpo)) + if sim > score: + score = sim + mejor = nombre + return (mejor, score) if score >= UMBRAL_CUERPO else (None, score) + + def analizar_frame(self, frame): + """Devuelve [{nombre:'X'} | {'desconocido':True}] por cada persona del frame.""" + personas = self._detectar_personas(frame) + if not personas: + return [] + resultados = [] + for p in personas: + x, y, w, h = p + persona = self._identificar_recorte(frame, x, y, w, h) + resultados.append(persona) + return resultados + + def _detectar_personas(self, frame): + try: + from ultralytics import YOLO + model = self._modelo_yolo() + r = model.predict(frame, verbose=False, conf=0.4) + cajas = [] + for box in r[0].boxes: + if int(box.cls) == 0: + x1, y1, x2, y2 = box.xyxy[0].tolist() + cajas.append([int(x1), int(y1), int(x2 - x1), int(y2 - y1)]) + return cajas + except Exception: + return [] + + def _modelo_yolo(self): + if not hasattr(self, "_yolo"): + from ultralytics import YOLO + self._yolo = YOLO(YOLO_PATH) + return self._yolo + + def _identificar_recorte(self, frame, x, y, w, h): + x, y = max(0, x), max(0, y) + recorte = frame[y:y + h, x:x + w] + if recorte.size == 0: + return {"desconocido": True} + + mejor_nombre, mejor_score = None, 0.0 + for cara in self.detectar_caras(frame): + cx, cy, cw, ch = (int(v) for v in cara[:4]) + if cx < x + w and cx + cw > x and cy < y + h and cy + ch > y: + emb = self.embedding_cara(frame, cara) + nombre, score = self.identificar_cara(emb) + if nombre and score > mejor_score: + mejor_nombre, mejor_score = nombre, score + + if mejor_nombre: + return {"nombre": mejor_nombre, "score": mejor_score} + + emb_cuerpo = self.embedding_cuerpo(recorte) + nombre, score = self.identificar_cuerpo(emb_cuerpo) + if nombre: + return {"nombre": nombre, "score": score, "via": "cuerpo"} + + return {"desconocido": True} + + +def cargar_reconocedor(): + return Reconocedor()