Integración de MediaPipe y YOLOv8: aplicación avanzada de visión por ordenador

Guía completa para crear una aplicación que combine la detección de pose con MediaPipe y la detección de objetos con YOLOv8.

1. Configuración inicial

# Instalación de las dependencias
pip install mediapipe ultralytics opencv-python numpy

# Importación de las bibliotecas
import mediapipe as mp
from ultralytics import YOLO
import cv2
import numpy as np
import time
from threading import Thread
from queue import Queue

2. Arquitectura de la aplicación

2.1 Clases base

class PoseDetector:
    def __init__(self):
        self.mp_pose = mp.solutions.pose
        self.pose = self.mp_pose.Pose(
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5
        )
        self.mp_draw = mp.solutions.drawing_utils
        
    def detect(self, image):
        # Conversión a RGB
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        
        # Detección de la pose
        results = self.pose.process(image_rgb)
        
        # Dibujar los puntos de referencia
        if results.pose_landmarks:
            self.mp_draw.draw_landmarks(
                image,
                results.pose_landmarks,
                self.mp_pose.POSE_CONNECTIONS
            )
        
        return image, results.pose_landmarks

class ObjectDetector:
    def __init__(self):
        self.model = YOLO('yolov8n.pt')
        
    def detect(self, image):
        # Detección de objetos
        results = self.model(image)
        
        # Procesamiento de los resultados
        for result in results:
            boxes = result.boxes
            for box in boxes:
                # Coordenadas del recuadro
                x1, y1, x2, y2 = box.xyxy[0]
                x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
                
                # Confianza
                confidence = float(box.conf[0])
                
                # Clase
                class_id = int(box.cls[0])
                class_name = self.model.names[class_id]
                
                # Dibujar el recuadro
                cv2.rectangle(image, (x1, y1), (x2, y2), (0, 255, 0), 2)
                
                # Añadir el texto
                label = f"{class_name} {confidence:.2f}"
                cv2.putText(image, label, (x1, y1 - 10),
                           cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
        
        return image

3. Procesamiento multihilo

3.1 Gestor de frames

class FrameManager:
    def __init__(self, max_queue_size=2):
        self.frame_queue = Queue(maxsize=max_queue_size)
        self.result_queue = Queue(maxsize=max_queue_size)
        
    def add_frame(self, frame):
        if not self.frame_queue.full():
            self.frame_queue.put(frame)
            
    def get_frame(self):
        if not self.frame_queue.empty():
            return self.frame_queue.get()
        return None
        
    def add_result(self, result):
        if not self.result_queue.full():
            self.result_queue.put(result)
            
    def get_result(self):
        if not self.result_queue.empty():
            return self.result_queue.get()
        return None

3.2 Workers

class PoseWorker(Thread):
    def __init__(self, frame_manager, detector):
        super().__init__()
        self.frame_manager = frame_manager
        self.detector = detector
        self.running = True
        
    def run(self):
        while self.running:
            frame = self.frame_manager.get_frame()
            if frame is not None:
                # Detección de pose
                processed_frame, landmarks = self.detector.detect(frame)
                self.frame_manager.add_result((processed_frame, landmarks))

class ObjectWorker(Thread):
    def __init__(self, frame_manager, detector):
        super().__init__()
        self.frame_manager = frame_manager
        self.detector = detector
        self.running = True
        
    def run(self):
        while self.running:
            frame = self.frame_manager.get_frame()
            if frame is not None:
                # Detección de objetos
                processed_frame = self.detector.detect(frame)
                self.frame_manager.add_result((processed_frame, None))

4. Aplicación principal

4.1 Configuración

class Application:
    def __init__(self):
        # Inicialización de los detectores
        self.pose_detector = PoseDetector()
        self.object_detector = ObjectDetector()
        
        # Inicialización del gestor de frames
        self.frame_manager = FrameManager()
        
        # Inicialización de los workers
        self.pose_worker = PoseWorker(self.frame_manager, self.pose_detector)
        self.object_worker = ObjectWorker(self.frame_manager, self.object_detector)
        
        # Configuración de la cámara
        self.cap = cv2.VideoCapture(0)
        
        # Variables de control
        self.running = True
        self.current_mode = 'both'  # 'pose', 'object', 'both'

4.2 Bucle principal

    def run(self):
        # Arranque de los workers
        self.pose_worker.start()
        self.object_worker.start()
        
        while self.running:
            success, frame = self.cap.read()
            if not success:
                continue
                
            # Añadir el frame al gestor
            self.frame_manager.add_frame(frame.copy())
            
            # Recuperación de los resultados
            pose_result = self.frame_manager.get_result()
            object_result = self.frame_manager.get_result()
            
            # Visualización de los resultados
            if pose_result is not None and object_result is not None:
                pose_frame, pose_landmarks = pose_result
                object_frame, _ = object_result
                
                # Combinación de los resultados
                if self.current_mode == 'both':
                    # Superposición de los resultados
                    combined_frame = cv2.addWeighted(
                        pose_frame, 0.5, object_frame, 0.5, 0
                    )
                    cv2.imshow('Combined Detection', combined_frame)
                elif self.current_mode == 'pose':
                    cv2.imshow('Pose Detection', pose_frame)
                else:
                    cv2.imshow('Object Detection', object_frame)
            
            # Gestión de las teclas
            key = cv2.waitKey(1) & 0xFF
            if key == ord('q'):
                self.running = False
            elif key == ord('p'):
                self.current_mode = 'pose'
            elif key == ord('o'):
                self.current_mode = 'object'
            elif key == ord('b'):
                self.current_mode = 'both'
        
        # Limpieza
        self.cleanup()

4.3 Limpieza

    def cleanup(self):
        # Parada de los workers
        self.pose_worker.running = False
        self.object_worker.running = False
        self.pose_worker.join()
        self.object_worker.join()
        
        # Liberación de los recursos
        self.cap.release()
        cv2.destroyAllWindows()

5. Funcionalidades avanzadas

5.1 Análisis de las interacciones

def analyze_interaction(pose_landmarks, object_detections):
    if pose_landmarks is None or object_detections is None:
        return None
        
    # Extracción de los puntos clave de la pose
    left_hand = np.array([
        pose_landmarks.landmark[mp_pose.PoseLandmark.LEFT_WRIST.value].x,
        pose_landmarks.landmark[mp_pose.PoseLandmark.LEFT_WRIST.value].y
    ])
    
    right_hand = np.array([
        pose_landmarks.landmark[mp_pose.PoseLandmark.RIGHT_WRIST.value].x,
        pose_landmarks.landmark[mp_pose.PoseLandmark.RIGHT_WRIST.value].y
    ])
    
    # Análisis de las interacciones con los objetos detectados
    interactions = []
    for detection in object_detections:
        box_center = np.array([
            (detection[0] + detection[2]) / 2,
            (detection[1] + detection[3]) / 2
        ])
        
        # Cálculo de las distancias
        left_distance = np.linalg.norm(left_hand - box_center)
        right_distance = np.linalg.norm(right_hand - box_center)
        
        # Detección de las interacciones
        if left_distance < 0.1 or right_distance < 0.1:
            interactions.append({
                'object': detection[4],
                'hand': 'left' if left_distance < right_distance else 'right',
                'distance': min(left_distance, right_distance)
            })
    
    return interactions

5.2 Visualización avanzada

def visualize_results(frame, pose_landmarks, object_detections, interactions):
    # Dibujar los puntos de referencia de la pose
    if pose_landmarks:
        mp_draw.draw_landmarks(
            frame,
            pose_landmarks,
            mp_pose.POSE_CONNECTIONS
        )
    
    # Dibujar los recuadros de detección de objetos
    for detection in object_detections:
        x1, y1, x2, y2 = detection[:4]
        cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (0, 255, 0), 2)
    
    # Visualizar las interacciones
    if interactions:
        for interaction in interactions:
            # Dibujar una línea entre la mano y el objeto
            hand_point = left_hand if interaction['hand'] == 'left' else right_hand
            cv2.line(
                frame,
                (int(hand_point[0] * frame.shape[1]), int(hand_point[1] * frame.shape[0])),
                (int(box_center[0] * frame.shape[1]), int(box_center[1] * frame.shape[0])),
                (0, 0, 255),
                2
            )
    
    return frame

6. Optimización del rendimiento

6.1 Gestión de la memoria

def optimize_memory():
    # Configuración para un uso óptimo de la memoria
    pose_detector = PoseDetector()
    object_detector = ObjectDetector()
    
    # Reducción del tamaño de los frames
    def resize_frame(frame):
        return cv2.resize(frame, (640, 480))
    
    # Limpieza periódica de las colas
    def cleanup_queues():
        while not frame_manager.frame_queue.empty():
            frame_manager.frame_queue.get()
        while not frame_manager.result_queue.empty():
            frame_manager.result_queue.get()

6.2 Aceleración por GPU

def enable_gpu_acceleration():
    # Configuración para el uso de la GPU
    import torch
    
    if torch.cuda.is_available():
        # Configuración de YOLO para la GPU
        object_detector.model.to('cuda')
        
        # Configuración de MediaPipe para la GPU
        pose_detector.pose = mp_pose.Pose(
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5,
            model_complexity=1,
            enable_segmentation=False
        )

7. Buenas prácticas

1. Gestión de recursos

   • Utilizar multihilo para las detecciones
   • Implementar un sistema de caché
   • Limpiar la memoria con regularidad

2. Optimización

   • Reducir el tamaño de los frames
   • Utilizar la aceleración por GPU
   • Ajustar las frecuencias de detección

3. Interfaz de usuario

   • Ofrecer controles intuitivos
   • Mostrar las métricas de rendimiento
   • Permitir el cambio de modo en tiempo real

Conclusión

Puntos clave:

• Integración eficaz de MediaPipe y YOLOv8
• Procesamiento multihilo para un mejor rendimiento
• Análisis de las interacciones entre pose y objetos
• Interfaz de usuario flexible

Recomendaciones:

• Optimizar el uso de la memoria
• Utilizar la aceleración por GPU
• Implementar una gestión robusta de los errores
• Añadir funcionalidades de personalización