Modelos de Detección y Seguimiento de Rostros

from ultralytics import YOLO
import cv2

# Cargar el modelo
model = YOLO('yolov8n-face.pt')

# Procesar una imagen única
results = model('path/to/your/image.jpg')

# Procesar stream de video
cap = cv2.VideoCapture(0)
while True:
    ret, frame = cap.read()
    if not ret:
        break
    
    results = model(frame)
    annotated_frame = results[0].plot()
    cv2.imshow('Face Detection', annotated_frame)
    
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

import mediapipe as mp
import cv2

# Inicializar MediaPipe Face Detection una vez
mp_face_detection = mp.solutions.face_detection
mp_drawing = mp.solutions.drawing_utils

cap = cv2.VideoCapture(0)

# Usar operador 'with' fuera del bucle para gestión adecuada de recursos
with mp_face_detection.FaceDetection(
    model_selection=0, min_detection_confidence=0.5) as face_detection:

    while cap.isOpened():
        success, image = cap.read()
        if not success:
            print("Ignorando cuadro vacío de cámara.")
            continue

        # Para mejorar rendimiento, opcionalmente marcar imagen como no escribible
        # para pasar por referencia.
        image.flags.writeable = False
        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        results = face_detection.process(image)

        # Dibujar anotaciones de detección facial en la imagen.
        image.flags.writeable = True
        image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
        if results.detections:
            for detection in results.detections:
                mp_drawing.draw_detection(image, detection)
        
        cv2.imshow('MediaPipe Face Detection', image)
        if cv2.waitKey(5) & 0xFF == 27:  # Tecla ESC
            break

cap.release()
cv2.destroyAllWindows()

from retinaface import RetinaFace
import cv2
import numpy as np

def detect_faces_retinaface(image_path):
    # Detección RetinaFace
    faces = RetinaFace.detect_faces(image_path)
    
    if isinstance(faces, dict):
        image = cv2.imread(image_path)
        
        for key in faces.keys():
            face = faces[key]
            facial_area = face["facial_area"]
            
            # Dibujar rectángulo delimitador
            cv2.rectangle(image, 
                         (facial_area[0], facial_area[1]), 
                         (facial_area[2], facial_area[3]), 
                         (0, 255, 0), 2)
            
            # Agregar puntuación de confianza
            confidence = face.get("score", 0)
            cv2.putText(image, f'{confidence:.2f}', 
                       (facial_area[0], facial_area[1]-10),
                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
        
        return image
    
    return None

# Uso
result_image = detect_faces_retinaface('path/to/image.jpg')
if result_image is not None:
    cv2.imshow('RetinaFace Detection', result_image)
    cv2.waitKey(0)
    cv2.destroyAllWindows()

import cv2
from yolox.tracker.byte_tracker import BYTETracker
from ultralytics import YOLO

class VideoTracker:
    def __init__(self):
        self.model = YOLO('yolov8n-face.pt')
        self.tracker = BYTETracker(frame_rate=30)
    
    def track_video(self, video_path):
        cap = cv2.VideoCapture(video_path)
        frame_id = 0
        
        while True:
            ret, frame = cap.read()
            if not ret:
                break
            
            # Detección
            results = self.model(frame)
            
            # Convertir a formato de tracking
            detections = []
            if len(results[0].boxes) > 0:
                boxes = results[0].boxes.xyxy.cpu().numpy()
                scores = results[0].boxes.conf.cpu().numpy()
                
                for box, score in zip(boxes, scores):
                    detections.append([*box, score])
            
            # Actualizar tracker
            tracks = self.tracker.update(
                np.array(detections), 
                frame.shape[:2], 
                frame.shape[:2]
            )
            
            # Dibujar tracks
            for track in tracks:
                x1, y1, x2, y2, track_id = track[:5]
                cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (0, 255, 0), 2)
                cv2.putText(frame, f'ID: {int(track_id)}', 
                           (int(x1), int(y1)-10),
                           cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2)
            
            cv2.imshow('Face Tracking', frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
            
            frame_id += 1
        
        cap.release()
        cv2.destroyAllWindows()

# Uso
tracker = VideoTracker()
tracker.track_video('path/to/video.mp4')

# Implementación de detector facial para dispositivos móviles.
import mediapipe as mp
import cv2
import numpy as np
import time

class MobileFaceDetector:
    """
    Clase para detección facial en stream de video, optimizada para
    rendimiento en dispositivos móviles.

    Incluye salto de cuadros para lograr FPS objetivo y renderizado
    suave de marcos para evitar "parpadeo".
    """
    def __init__(self, model_selection=0, min_detection_confidence=0.7):
        """
        Inicializa detector facial.

        Args:
            model_selection (int): Selección de modelo. 0 para radio corto (hasta 2 metros),
                                   1 para largo (hasta 5 metros).
            min_detection_confidence (float): Umbral mínimo de confianza para detección.
        """
        # --- Inicialización MediaPipe Face Detection ---
        # Crear instancia con parámetros dados.
        self.face_detection = mp.solutions.face_detection.FaceDetection(
            model_selection=model_selection,
            min_detection_confidence=min_detection_confidence,
        )
        
        # --- Variables para renderizado y FPS ---
        # Almacenar últimas detecciones exitosas para dibujarlas en cuadros saltados.
        self.last_results = None
        # Guardar dimensiones originales del cuadro para escalado correcto.
        self.original_width = 0
        self.original_height = 0

    def _draw_detections(self, frame, detections):
        """
        Función auxiliar para dibujar marcos en cuadro.
        
        Args:
            frame (np.ndarray): Cuadro en el que dibujar.
            detections: Resultados de detección de MediaPipe.
        """
        if detections:
            for detection in detections:
                # Obtener coordenadas relativas de marco (de 0.0 a 1.0).
                bboxC = detection.location_data.relative_bounding_box
                
                # Escalar coordenadas de vuelta al tamaño original del cuadro.
                x = int(bboxC.xmin * self.original_width)
                y = int(bboxC.ymin * self.original_height)
                width = int(bboxC.width * self.original_width)
                height = int(bboxC.height * self.original_height)
                
                # Dibujar rectángulo en cuadro original.
                cv2.rectangle(frame, (x, y), (x + width, y + height), (0, 255, 0), 2)
        return frame

    def process_frame(self, frame, target_width=320):
        """
        Procesa un cuadro del stream de video.

        Args:
            frame (np.ndarray): Cuadro de entrada en formato BGR.
            target_width (int): Ancho objetivo para procesamiento. El cuadro será
                                proporcionalmente reducido a este ancho.

        Returns:
            np.ndarray: Cuadro con marcos dibujados.
        """
        self.original_height, self.original_width = frame.shape[:2]

        # --- Optimización 1: Reducción de resolución ---
        # Preservar relación de aspecto.
        scale = target_width / self.original_width
        small_frame = cv2.resize(frame, (0, 0), fx=scale, fy=scale, interpolation=cv2.INTER_AREA)

        # --- Optimización 2: Conversión de color y mejora de rendimiento ---
        # MediaPipe requiere RGB. Convertir una vez.
        rgb_small_frame = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB)
        # Marcar cuadro como "solo lectura" para aceleración.
        rgb_small_frame.flags.writeable = False

        # --- Detección facial ---
        results = self.face_detection.process(rgb_small_frame)
        
        # Restaurar flag si necesitas trabajar más con el cuadro.
        rgb_small_frame.flags.writeable = True

        # --- Renderizado ---
        # Si detección fue exitosa, guardar resultados.
        if results.detections:
            self.last_results = results.detections
        
        # Dibujar marcos en cuadro original. Si no se encontraron rostros en cuadro actual
        # pero estaban en previos, usar datos antiguos para suavidad.
        processed_frame = self._draw_detections(frame, self.last_results)
        
        return processed_frame
        
    def close(self):
        """Liberar recursos MediaPipe."""
        self.face_detection.close()

# --- Ejemplo de uso con webcam ---
if __name__ == '__main__':
    # Inicializar detector
    detector = MobileFaceDetector(min_detection_confidence=0.5)
    
    # Capturar video de webcam (0 - usualmente cámara integrada)
    cap = cv2.VideoCapture(0)
    
    if not cap.isOpened():
        print("Error: no se pudo abrir webcam.")
        exit()

    # Variables para cálculo de FPS
    prev_frame_time = 0
    new_frame_time = 0

    try:
        while True:
            # Leer cuadro de cámara
            success, frame = cap.read()
            if not success:
                print("No se pudo obtener cuadro. Terminando.")
                break

            # Voltear cuadro para efecto "espejo"
            frame = cv2.flip(frame, 1)

            # --- Saltar procesamiento para lograr FPS objetivo ---
            # Esto es menos importante con nueva lógica de renderizado, pero aún ahorra recursos
            # Aquí puedes implementar tu lógica de salto de cuadros si es necesario
            
            # Procesar cuadro
            processed_frame = detector.process_frame(frame)

            # --- Cálculo y visualización de FPS ---
            new_frame_time = time.time()
            fps = 1 / (new_frame_time - prev_frame_time)
            prev_frame_time = new_frame_time
            cv2.putText(processed_frame, f'FPS: {int(fps)}', (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
            
            # Mostrar resultado
            cv2.imshow('Mobile Face Detection', processed_frame)

            # Salir al presionar tecla 'q'
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
    finally:
        # Liberar recursos
        print("Liberando recursos...")
        cap.release()
        detector.close()
        cv2.destroyAllWindows()

# Sistema de seguridad con tracking, verificación,
# reconocimiento optimizado y visualización.
import cv2
import numpy as np
import time
import os
import pickle
from typing import Dict, List, Any, Optional, Tuple

# --- Se asume que estas librerías están instaladas ---
# pip install opencv-python numpy
# pip install insightface
# pip install retinaface-pytorch deep-sort-realtime
#
# Importante: InsightFace requiere onnxruntime-gpu. DeepSort también tiene dependencias.
# La instalación puede requerir pasos adicionales, incluyendo descarga de modelos.

from retinaface import RetinaFace
from deep_sort_realtime.deepsort_tracker import DeepSort
from insightface.app import FaceAnalysis

class AdvancedSecuritySystem:
    """
    Sistema de seguridad integral que combina detección, tracking y
    reconocimiento facial con optimizaciones para tiempo real.
    """

    def __init__(self, db_path: str, detection_thresh: float = 0.9, recognition_thresh: float = 0.5):
        """
        Inicialización de todos los componentes del sistema.

        Args:
            db_path (str): Ruta al archivo de base de datos de rostros (.pkl).
            detection_thresh (float): Umbral de confianza para detector facial.
            recognition_thresh (float): Umbral de similitud para reconocimiento facial.
        """
        print("Inicializando sistema de seguridad...")

        # --- Detector facial (RetinaFace) ---
        # RetinaFace funciona bien para detectar rostros "difíciles".
        # Para CPU puedes usar 'mobilenet', para GPU - 'resnet50'.
        self.detector = RetinaFace(gpu_id=0) # Especifica -1 para uso de CPU
        print("Detector facial (RetinaFace) inicializado.")

        # --- Tracker (DeepSort) ---
        # max_age: cuántos cuadros puede existir un track sin detección.
        # n_init: cuántas veces debe detectarse objeto para iniciar tracking.
        self.tracker = DeepSort(max_age=30, n_init=3)
        print("Tracker (DeepSort) inicializado.")

        # --- Reconocedor facial (InsightFace) ---
        # Usar CUDA para máximo rendimiento.
        self.recognizer = FaceAnalysis(name='buffalo_l', providers=['CUDAExecutionProvider'])
        self.recognizer.prepare(ctx_id=0) # ctx_id=0 para GPU, -1 para CPU
        print("Reconocedor facial (InsightFace) inicializado.")
        
        # --- Parámetros y base de datos ---
        self.detection_thresh = detection_thresh
        self.recognition_thresh = recognition_thresh
        self.face_db = self._load_face_database(db_path)
        
        # Diccionario para almacenar identidades ya reconocidas por track_id
        # {track_id: {"identity": "Name", "verified": True/False}}
        self.tracked_identities: Dict[str, Dict[str, Any]] = {}

    def _load_face_database(self, db_path: str) -> Optional[Dict[str, np.ndarray]]:
        """Carga base de datos de embeddings pre-creada."""
        if not os.path.exists(db_path):
            print(f"Error: Archivo de base de datos no encontrado en ruta: {db_path}")
            print("Por favor, primero crea la base de datos usando create_database_from_folder().")
            return None
        with open(db_path, 'rb') as f:
            db = pickle.load(f)
            print(f"Base de datos facial cargada exitosamente. Encontradas {len(db)} identidades.")
            return db

    def _extract_embedding(self, frame: np.ndarray, bbox: np.ndarray) -> Optional[np.ndarray]:
        """Extrae embedding (vector de características) del área facial."""
        # Recortar rostro de forma segura, recortando coordenadas a límites del cuadro
        h, w = frame.shape[:2]
        x1, y1, x2, y2 = np.maximum(0, bbox).astype(int)
        x2, y2 = min(x2, w), min(y2, h)
        
        face_img = frame[y1:y2, x1:x2]
        if face_img.size == 0:
            return None
            
        # Usar InsightFace para obtener embedding
        faces = self.recognizer.get(face_img)
        return faces[0].embedding if faces else None

    def _find_match(self, embedding: np.ndarray) -> Tuple[str, float]:
        """Encuentra rostro más similar en base de datos."""
        if embedding is None or self.face_db is None:
            return "Unknown", 0.0
            
        best_match = "Unknown"
        best_score = 0.0
        
        # Usar similitud coseno (producto punto de vectores normalizados)
        for name, db_embed in self.face_db.items():
            score = np.dot(embedding, db_embed) / (np.linalg.norm(embedding) * np.linalg.norm(db_embed))
            if score > self.recognition_thresh and score > best_score:
                best_score = score
                best_match = name
                
        return best_match, best_score

    def process_frame(self, frame: np.ndarray) -> np.ndarray:
        """
        Ciclo completo de procesamiento de un cuadro: detección, tracking, reconocimiento.

        Returns:
            np.ndarray: Cuadro con información dibujada (marcos, IDs, nombres).
        """
        if self.face_db is None:
            cv2.putText(frame, "Face DB not loaded!", (20, 40), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
            return frame
            
        # --- 1. Detección ---
        # RetinaFace devuelve diccionario que necesita conversión
        detections_raw = self.detector.detect(frame, threshold=self.detection_thresh)
        
        # Conversión a formato entendible por DeepSort: [[x1, y1, x2, y2], confidence]
        deepsort_detections = []
        for face_data in detections_raw:
            x1, y1, x2, y2, conf = face_data
            deepsort_detections.append(([x1, y1, x2, y2], conf, "face"))

        # --- 2. Tracking ---
        tracks = self.tracker.update_tracks(deepsort_detections, frame=frame)
        
        # --- 3. Reconocimiento y Visualización ---
        for track in tracks:
            if not track.is_confirmed():
                continue
            
            track_id = track.track_id
            bbox = track.to_tlbr() # (x1, y1, x2, y2)
            
            identity_info = self.tracked_identities.get(track_id)
            
            # OPTIMIZACIÓN: Reconocer solo tracks nuevos
            if identity_info is None:
                embedding = self._extract_embedding(frame, bbox)
                name, score = self._find_match(embedding)
                self.tracked_identities[track_id] = {"identity": name, "score": score}
                identity_info = self.tracked_identities[track_id]

            # --- Visualización ---
            name = identity_info.get("identity", "Unknown")
            score = identity_info.get("score", 0.0)
            
            color = (0, 255, 0) if name != "Unknown" else (0, 0, 255)
            # Dibujar marco
            cv2.rectangle(frame, (int(bbox[0]), int(bbox[1])), (int(bbox[2]), int(bbox[3])), color, 2)
            # Formar texto para etiqueta
            label = f"ID: {track_id} | {name} ({score:.2f})"
            # Dibujar etiqueta
            cv2.putText(frame, label, (int(bbox[0]), int(bbox[1]) - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
            
        return frame

    def create_database_from_folder(self, folder_path: str, output_db_path: str):
        """
        Crea base de datos de embeddings desde carpeta con imágenes.
        Estructura de carpeta:
        - folder_path/
          - persona_1/
            - imagen1.jpg
            - imagen2.png
          - persona_2/
            - foto.jpg
        """
        face_db = {}
        print(f"Creando base de datos desde carpeta: {folder_path}")
        for person_name in os.listdir(folder_path):
            person_folder = os.path.join(folder_path, person_name)
            if not os.path.isdir(person_folder):
                continue
            
            embeddings = []
            for image_name in os.listdir(person_folder):
                image_path = os.path.join(person_folder, image_name)
                img = cv2.imread(image_path)
                if img is None:
                    continue
                
                # Obtener embedding para primer rostro encontrado en foto
                faces = self.recognizer.get(img)
                if faces:
                    embeddings.append(faces[0].embedding)
            
            if embeddings:
                # Promediar embeddings para mayor estabilidad
                face_db[person_name] = np.mean(embeddings, axis=0)
                print(f"-> Rostro encontrado y procesado: {person_name}")

        # Guardar base en archivo
        with open(output_db_path, 'wb') as f:
            pickle.dump(face_db, f)
        print(f"Base de datos creada exitosamente y guardada en: {output_db_path}")

# --- Ejemplo de uso ---
if __name__ == '__main__':
    DB_FILE = "face_db.pkl"
    DB_SOURCE_FOLDER = "face_images" # Carpeta con fotos para crear BD
    
    # --- Paso 1: Crear base de datos (ejecutar una vez) ---
    # Para este paso necesitas crear carpeta face_images con subcarpetas
    # de nombres de personas conteniendo sus fotografías.
    if not os.path.exists(DB_FILE):
        print("Base de datos no encontrada. Iniciando proceso de creación...")
        # Crear carpeta temporal para ejemplo si no existe
        if not os.path.exists(DB_SOURCE_FOLDER):
            os.makedirs(os.path.join(DB_SOURCE_FOLDER, "persona_ejemplo"))
            print(f"Carpeta de ejemplo creada: {DB_SOURCE_FOLDER}/persona_ejemplo")
            print("Por favor coloca archivos .jpg con rostros ahí y reinicia el script.")
            exit()
            
        # Inicializar sistema solo para creación de BD
        temp_system = AdvancedSecuritySystem(db_path=DB_FILE)
        temp_system.create_database_from_folder(DB_SOURCE_FOLDER, DB_FILE)
        print("-" * 30)

    # --- Paso 2: Ejecutar sistema en stream de video ---
    system = AdvancedSecuritySystem(db_path=DB_FILE)
    cap = cv2.VideoCapture(0) # 0 para webcam o ruta a archivo de video

    if not cap.isOpened():
        print("Error: no se pudo abrir stream de video.")
        exit()

    prev_time = 0
    try:
        while True:
            success, frame = cap.read()
            if not success:
                break
            
            # Procesar cuadro
            processed_frame = system.process_frame(frame)

            # Calcular y mostrar FPS
            curr_time = time.time()
            fps = 1 / (curr_time - prev_time)
            prev_time = curr_time
            cv2.putText(processed_frame, f"FPS: {int(fps)}", (20, 80), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)
            
            cv2.imshow("Advanced Security System", processed_frame)
            
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
    finally:
        print("Terminando y liberando recursos.")
        cap.release()
        cv2.destroyAllWindows()

# Implementación de detección y tracking facial para dispositivos Edge
# usando TensorFlow Lite.

import cv2
import numpy as np
import time
import argparse
from typing import List, Dict, Any, Tuple, Optional

# Este script requiere tflite_runtime.
# Instalación: pip install tflite-runtime
try:
    from tflite_runtime.interpreter import Interpreter
except ImportError:
    print("Error: tflite_runtime no encontrado. Por favor instálalo:")
    print("pip install tflite-runtime")
    exit()

class EdgeFacePipeline:
    """
    Clase para detección y tracking facial en dispositivos Edge usando TFLite.
    Incluye optimizaciones hardware y tracker IOU ligero.
    """
    def __init__(self, model_path: str, device: str = 'cpu', conf_thresh: float = 0.7):
        """
        Inicialización del pipeline.

        Args:
            model_path (str): Ruta al archivo de modelo .tflite.
            device (str): Dispositivo para ejecución ('cpu' o 'npu').
            conf_thresh (float): Umbral de confianza para detección facial.
        """
        print(f"Cargando modelo TFLite desde: {model_path}")
        self.interpreter = Interpreter(model_path=model_path)
        self.interpreter.allocate_tensors()

        self.input_details = self.interpreter.get_input_details()
        self.output_details = self.interpreter.get_output_details()
        
        # Importante: obtener dimensiones exactas que requiere el modelo
        self.input_height = self.input_details[0]['shape'][1]
        self.input_width = self.input_details[0]['shape'][2]

        self.conf_thresh = conf_thresh
        self.tracked_faces: List[Dict[str, Any]] = []
        self.next_track_id = 0
        self.iou_thresh = 0.4 # Umbral IOU para tracking

        self._configure_device(device)
        print(f"Pipeline inicializado para dispositivo: {device}")

    def _configure_device(self, device: str):
        """Aplica optimizaciones específicas de hardware."""
        cv2.setUseOptimized(True)
        # Configuraciones NPU (Neural Processing Unit)
        if device == 'npu':
            # Aquí pueden ir delegados específicos para NPU, ej., 'libedgetpu.so.1'
            # self.interpreter = Interpreter(model_path=self.model_path,
            #                                experimental_delegates=[load_delegate('libedgetpu.so.1')])
            self.interpreter.set_num_threads(1)
            print("Optimizaciones NPU aplicadas (configuración simplificada).")
        # Optimizaciones CPU
        else:
            self.interpreter.set_num_threads(4)
            print("Optimizaciones CPU aplicadas (4 hilos).")

    def _preprocess(self, frame: np.ndarray) -> np.ndarray:
        """
        Preparar cuadro para red neuronal: redimensionar y normalizar.
        """
        # Redimensionar a dimensiones exactas de entrada del modelo ---
        resized_frame = cv2.resize(frame, (self.input_width, self.input_height))
        
        # Convertir a RGB si el modelo lo requiere
        rgb_frame = cv2.cvtColor(resized_frame, cv2.COLOR_BGR2RGB)
        
        # Normalización y agregar dimensión batch
        # Muchos modelos TFLite esperan float32 en rango [-1, 1] o [0, 1]
        # o uint8 [0, 255]. Revisa documentación de tu modelo.
        # Aquí asumimos float32 [0, 1].
        input_data = np.expand_dims(rgb_frame, axis=0).astype(np.float32) / 255.0
        return input_data

    def process_frame(self, frame: np.ndarray) -> List[Dict[str, Any]]:
        """
        Ciclo completo de procesamiento: detección y tracking.

        Args:
            frame (np.ndarray): Cuadro de entrada en formato BGR.

        Returns:
            List[Dict[str, Any]]: Lista de rostros rastreados con sus IDs y coordenadas.
        """
        original_h, original_w = frame.shape[:2]
        
        # 1. Preprocesamiento de cuadro
        input_data = self._preprocess(frame)
        
        # 2. Inferencia del modelo
        self.interpreter.set_tensor(self.input_details[0]['index'], input_data)
        self.interpreter.invoke()
        
        # 3. Obtener y postprocesar resultados
        # El formato de salida puede diferir para diferentes modelos.
        # Asumimos que el modelo devuelve [boxes, scores].
        boxes = self.interpreter.get_tensor(self.output_details[0]['index'])[0]
        scores = self.interpreter.get_tensor(self.output_details[1]['index'])[0]

        current_detections = []
        for i, score in enumerate(scores):
            if score > self.conf_thresh:
                # Coordenadas devueltas en forma normalizada [ymin, xmin, ymax, xmax]
                ymin, xmin, ymax, xmax = boxes[i]
                
                # Escalar coordenadas al tamaño original del cuadro
                abs_xmin = int(xmin * original_w)
                abs_ymin = int(ymin * original_h)
                abs_xmax = int(xmax * original_w)
                abs_ymax = int(ymax * original_h)
                
                current_detections.append({
                    'bbox': (abs_xmin, abs_ymin, abs_xmax, abs_ymax),
                    'score': float(score)
                })
        
        # 4. Actualizar tracker
        self._update_tracker(current_detections)
        return self.tracked_faces

    def _update_tracker(self, new_detections: List[Dict[str, Any]]):
        """Actualiza tracks basado en nuevas detecciones usando IOU."""
        if not self.tracked_faces:
            # Si no hay tracks, inicializarlos desde nuevas detecciones
            for det in new_detections:
                det['track_id'] = self.next_track_id
                det['age'] = 0 # Contador de "edad" del track
                self.tracked_faces.append(det)
                self.next_track_id += 1
            return

        matched_indices = set()
        
        # Intentar emparejar nuevas detecciones con tracks existentes
        for i, track in enumerate(self.tracked_faces):
            best_match_iou = 0
            best_match_idx = -1
            for j, det in enumerate(new_detections):
                if j in matched_indices:
                    continue
                iou = self._calculate_iou(track['bbox'], det['bbox'])
                if iou > self.iou_thresh and iou > best_match_iou:
                    best_match_iou = iou
                    best_match_idx = j
            
            if best_match_idx != -1:
                # Actualizar track
                track['bbox'] = new_detections[best_match_idx]['bbox']
                track['age'] = 0 # Reiniciar edad ya que track fue encontrado
                matched_indices.add(best_match_idx)
            else:
                # Aumentar edad si track no fue encontrado
                track['age'] += 1

        # Agregar nuevos tracks para detecciones no emparejadas
        for j, det in enumerate(new_detections):
            if j not in matched_indices:
                det['track_id'] = self.next_track_id
                det['age'] = 0
                self.tracked_faces.append(det)
                self.next_track_id += 1
        
        # Remover tracks antiguos que no han sido vistos por un tiempo
        self.tracked_faces = [t for t in self.tracked_faces if t['age'] < 15]

    @staticmethod
    def _calculate_iou(box1: Tuple, box2: Tuple) -> float:
        """Calcula Intersection over Union (IOU) para dos marcos."""
        x1, y1, x2, y2 = box1
        x3, y3, x4, y4 = box2
        
        xi1, yi1 = max(x1, x3), max(y1, y3)
        xi2, yi2 = min(x2, x4), min(y2, y4)
        
        inter_area = max(0, xi2 - xi1) * max(0, yi2 - yi1)
        box1_area = (x2 - x1) * (y2 - y1)
        box2_area = (x4 - x3) * (y4 - y3)
        
        union_area = box1_area + box2_area - inter_area
        return inter_area / union_area if union_area > 0 else 0.0

def draw_results(frame: np.ndarray, faces: List[Dict[str, Any]]):
    """Dibujar resultados en cuadro."""
    for face in faces:
        # Usar coordenadas para dibujar
        xmin, ymin, xmax, ymax = face['bbox']
        track_id = face['track_id']
        
        cv2.rectangle(frame, (xmin, ymin), (xmax, ymax), (0, 255, 0), 2)
        label = f"ID: {track_id}"
        cv2.putText(frame, label, (xmin, ymin - 10),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)

# Ejemplo de uso
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Detección Facial en Dispositivos Edge")
    parser.add_argument("--model", type=str, required=True, help="Ruta al archivo de modelo .tflite.")
    parser.add_argument("--device", type=str, default='cpu', choices=['cpu', 'npu'], help="Dispositivo para ejecutar inferencia.")
    parser.add_argument("--video", type=str, help="Ruta a archivo de video. Si no se proporciona, usa webcam.")
    args = parser.parse_args()

    pipeline = EdgeFacePipeline(model_path=args.model, device=args.device)
    
    cap = cv2.VideoCapture(args.video if args.video else 0)
    if not cap.isOpened():
        print(f"Error: no se pudo abrir fuente de video.")
        exit()
        
    prev_time = 0
    try:
        while True:
            ret, frame = cap.read()
            if not ret:
                break
                
            # Procesar cuadro
            tracked_faces = pipeline.process_frame(frame)
            
            # Visualizar resultados
            draw_results(frame, tracked_faces)
            
            # Mostrar FPS
            curr_time = time.time()
            fps = 1 / (curr_time - prev_time)
            prev_time = curr_time
            cv2.putText(frame, f"FPS: {int(fps)}", (10, 30),
                        cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2)
            
            cv2.imshow('Edge Face Detection', frame)
            
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
    finally:
        print("Terminando y liberando recursos.")
        cap.release()
        cv2.destroyAllWindows()

# Crear entorno virtual
python -m venv face_detection_env
source face_detection_env/bin/activate

# Dependencias básicas
pip install opencv-python mediapipe ultralytics

# Para modelos avanzados
pip install insightface retinaface-pytorch deepface

# Para tracking
pip install yolox filterpy scikit-image

import gc
import torch

def optimize_gpu_memory():
    """Limpieza de memoria GPU después del procesamiento"""
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
        gc.collect()

def batch_processing(image_list, model, batch_size=4):
    """Procesamiento por lotes para optimización de throughput"""
    results = []
    
    for i in range(0, len(image_list), batch_size):
        batch = image_list[i:i+batch_size]
        batch_results = model(batch)
        results.extend(batch_results)
        
        # Limpieza de memoria después de cada lote
        optimize_gpu_memory()
    
    return results

import threading
from queue import Queue
import time

class ThreadedVideoProcessor:
    def __init__(self, model):
        self.model = model
        self.frame_queue = Queue(maxsize=10)
        self.result_queue = Queue(maxsize=10)
        self.processing = True
        
    def frame_reader(self, video_source):
        cap = cv2.VideoCapture(video_source)
        while self.processing:
            ret, frame = cap.read()
            if not ret:
                break
            
            if not self.frame_queue.full():
                self.frame_queue.put(frame)
        cap.release()
    
    def frame_processor(self):
        while self.processing:
            if not self.frame_queue.empty():
                frame = self.frame_queue.get()
                result = self.model(frame)
                
                if not self.result_queue.full():
                    self.result_queue.put((frame, result))

# Ejemplo: framework de prueba de sesgo
def evaluate_model_bias(model, test_datasets):
    """
    Evaluar rendimiento del modelo por grupos demográficos
    """
    results = {}
    
    for group_name, dataset in test_datasets.items():
        detections = []
        for image_path in dataset['images']:
            result = model(image_path)
            detections.append(result)
        
        # Calcular métricas para este grupo demográfico
        accuracy = calculate_accuracy(detections, dataset['ground_truth'])
        results[group_name] = {
            'accuracy': accuracy,
            'sample_size': len(dataset['images'])
        }
    
    return results

# Ejemplo de uso
demographic_results = evaluate_model_bias(model, {
    'young': young_adult_dataset,
    'elderly': elderly_dataset,
    'diverse': diverse_dataset
})

class ConsentAwareFaceTracker:
    def __init__(self):
        self.user_consent = self.check_user_consent()
        self.tracking_enabled = False
        
    def check_user_consent(self):
        # Verificar preferencias guardadas del usuario
        # Devolver estado de consentimiento y timestamp
        pass
    
    def request_consent(self):
        """
        Presentar diálogo claro de consentimiento al usuario
        """
        consent_text = """
        Esta aplicación usa detección facial para:
        - Mejorar tu experiencia con filtros AR
        - Enfoque automático de cámara en rostros
        
        Tus datos faciales:
        - Se procesan completamente en tu dispositivo
        - Nunca se guardan ni transmiten
        - Pueden deshabilitarse en cualquier momento en configuración
        
        ¿Aceptas la detección facial? [Sí/No]
        """
        return self.present_consent_dialog(consent_text)
    
    def process_with_consent(self, frame):
        if not self.user_consent['granted']:
            return frame  # Devolver cuadro sin procesar
        
        if self.consent_expired():
            self.user_consent = self.request_consent()
        
        return self.detect_faces(frame) if self.user_consent['granted'] else frame