Integração do MediaPipe e YOLOv8: Aplicação Avançada de Visão Computacional

Guia completo para criar uma aplicação que combina detecção de pose com MediaPipe e detecção de objetos com YOLOv8.

1. Configuração Inicial

# Instalação das dependências
pip install mediapipe ultralytics opencv-python numpy

# Import das 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. Arquitetura da Aplicação

2.1 Classes 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):
        # Conversão para RGB
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        
        # Detecção da pose
        results = self.pose.process(image_rgb)
        
        # Desenhar os pontos de referência
        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):
        # Detecção de objetos
        results = self.model(image)
        
        # Processamento dos resultados
        for result in results:
            boxes = result.boxes
            for box in boxes:
                # Coordenadas da caixa
                x1, y1, x2, y2 = box.xyxy[0]
                x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
                
                # Confiança
                confidence = float(box.conf[0])
                
                # Classe
                class_id = int(box.cls[0])
                class_name = self.model.names[class_id]
                
                # Desenhar a caixa
                cv2.rectangle(image, (x1, y1), (x2, y2), (0, 255, 0), 2)
                
                # Adicionar o 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. Processamento Multi-thread

3.1 Gerenciador 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:
                # Detecção 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:
                # Detecção de objetos
                processed_frame = self.detector.detect(frame)
                self.frame_manager.add_result((processed_frame, None))

4. Aplicação Principal

4.1 Configuração

class Application:
    def __init__(self):
        # Inicialização dos detectores
        self.pose_detector = PoseDetector()
        self.object_detector = ObjectDetector()
        
        # Inicialização do gerenciador de frames
        self.frame_manager = FrameManager()
        
        # Inicialização dos workers
        self.pose_worker = PoseWorker(self.frame_manager, self.pose_detector)
        self.object_worker = ObjectWorker(self.frame_manager, self.object_detector)
        
        # Configuração da câmera
        self.cap = cv2.VideoCapture(0)
        
        # Variáveis de controle
        self.running = True
        self.current_mode = 'both'  # 'pose', 'object', 'both'

4.2 Loop Principal

    def run(self):
        # Início dos workers
        self.pose_worker.start()
        self.object_worker.start()
        
        while self.running:
            success, frame = self.cap.read()
            if not success:
                continue
                
            # Adição do frame ao gerenciador
            self.frame_manager.add_frame(frame.copy())
            
            # Recuperação dos resultados
            pose_result = self.frame_manager.get_result()
            object_result = self.frame_manager.get_result()
            
            # Exibição dos resultados
            if pose_result is not None and object_result is not None:
                pose_frame, pose_landmarks = pose_result
                object_frame, _ = object_result
                
                # Combinação dos resultados
                if self.current_mode == 'both':
                    # Sobreposição dos 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)
            
            # Gerenciamento das 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'
        
        # Limpeza
        self.cleanup()

4.3 Limpeza

    def cleanup(self):
        # Parada dos workers
        self.pose_worker.running = False
        self.object_worker.running = False
        self.pose_worker.join()
        self.object_worker.join()
        
        # Liberação dos recursos
        self.cap.release()
        cv2.destroyAllWindows()

5. Funcionalidades Avançadas

5.1 Análise das Interações

def analyze_interaction(pose_landmarks, object_detections):
    if pose_landmarks is None or object_detections is None:
        return None
        
    # Extração dos pontos-chave da 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álise das interações com os objetos detectados
    interactions = []
    for detection in object_detections:
        box_center = np.array([
            (detection[0] + detection[2]) / 2,
            (detection[1] + detection[3]) / 2
        ])
        
        # Cálculo das distâncias
        left_distance = np.linalg.norm(left_hand - box_center)
        right_distance = np.linalg.norm(right_hand - box_center)
        
        # Detecção das interações
        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 Visualização Avançada

def visualize_results(frame, pose_landmarks, object_detections, interactions):
    # Desenhar os pontos de referência da pose
    if pose_landmarks:
        mp_draw.draw_landmarks(
            frame,
            pose_landmarks,
            mp_pose.POSE_CONNECTIONS
        )
    
    # Desenhar as caixas de detecção 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 as interações
    if interactions:
        for interaction in interactions:
            # Desenhar uma linha entre a mão e o 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. Otimização de Desempenho

6.1 Gerenciamento de Memória

def optimize_memory():
    # Configuração para uso otimizado da memória
    pose_detector = PoseDetector()
    object_detector = ObjectDetector()
    
    # Redução do tamanho dos frames
    def resize_frame(frame):
        return cv2.resize(frame, (640, 480))
    
    # Limpeza periódica das filas
    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 Aceleração por GPU

def enable_gpu_acceleration():
    # Configuração para uso da GPU
    import torch
    
    if torch.cuda.is_available():
        # Configuração do YOLO para a GPU
        object_detector.model.to('cuda')
        
        # Configuração do MediaPipe para a GPU
        pose_detector.pose = mp_pose.Pose(
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5,
            model_complexity=1,
            enable_segmentation=False
        )

7. Boas Práticas

1. Gerenciamento de Recursos

   • Usar multi-threading para as detecções
   • Implementar um sistema de cache
   • Limpar a memória regularmente

2. Otimização

   • Reduzir o tamanho dos frames
   • Usar a aceleração por GPU
   • Ajustar as frequências de detecção

3. Interface do Usuário

   • Fornecer controles intuitivos
   • Exibir as métricas de desempenho
   • Permitir a troca de modo em tempo real

Conclusão

Pontos-chave:

• Integração eficiente do MediaPipe e do YOLOv8
• Processamento multi-thread para melhor desempenho
• Análise das interações entre pose e objetos
• Interface do usuário flexível

Recomendações:

• Otimizar o uso de memória
• Usar a aceleração por GPU
• Implementar um tratamento robusto de erros
• Adicionar funcionalidades de personalização