MediaPipe and YOLOv8 Integration: An Advanced Computer Vision Application

A complete guide to building an application that combines pose detection with MediaPipe and object detection with YOLOv8.

1. Initial Setup

# Installing dependencies
pip install mediapipe ultralytics opencv-python numpy

# Importing libraries
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. Application Architecture

2.1 Base Classes

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):
        # Convert to RGB
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        
        # Pose detection
        results = self.pose.process(image_rgb)
        
        # Draw landmarks
        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):
        # Object detection
        results = self.model(image)
        
        # Process results
        for result in results:
            boxes = result.boxes
            for box in boxes:
                # Box coordinates
                x1, y1, x2, y2 = box.xyxy[0]
                x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
                
                # Confidence
                confidence = float(box.conf[0])
                
                # Class
                class_id = int(box.cls[0])
                class_name = self.model.names[class_id]
                
                # Draw the box
                cv2.rectangle(image, (x1, y1), (x2, y2), (0, 255, 0), 2)
                
                # Add the text
                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. Multithreaded Processing

3.1 Frame Manager

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:
                # Pose detection
                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:
                # Object detection
                processed_frame = self.detector.detect(frame)
                self.frame_manager.add_result((processed_frame, None))

4. Main Application

4.1 Configuration

class Application:
    def __init__(self):
        # Initialize the detectors
        self.pose_detector = PoseDetector()
        self.object_detector = ObjectDetector()
        
        # Initialize the frame manager
        self.frame_manager = FrameManager()
        
        # Initialize the workers
        self.pose_worker = PoseWorker(self.frame_manager, self.pose_detector)
        self.object_worker = ObjectWorker(self.frame_manager, self.object_detector)
        
        # Camera configuration
        self.cap = cv2.VideoCapture(0)
        
        # Control variables
        self.running = True
        self.current_mode = 'both'  # 'pose', 'object', 'both'

4.2 Main Loop

    def run(self):
        # Start the workers
        self.pose_worker.start()
        self.object_worker.start()
        
        while self.running:
            success, frame = self.cap.read()
            if not success:
                continue
                
            # Add the frame to the manager
            self.frame_manager.add_frame(frame.copy())
            
            # Retrieve the results
            pose_result = self.frame_manager.get_result()
            object_result = self.frame_manager.get_result()
            
            # Display the results
            if pose_result is not None and object_result is not None:
                pose_frame, pose_landmarks = pose_result
                object_frame, _ = object_result
                
                # Combine the results
                if self.current_mode == 'both':
                    # Overlay the results
                    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)
            
            # Handle key presses
            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'
        
        # Cleanup
        self.cleanup()

4.3 Cleanup

    def cleanup(self):
        # Stop the workers
        self.pose_worker.running = False
        self.object_worker.running = False
        self.pose_worker.join()
        self.object_worker.join()
        
        # Release resources
        self.cap.release()
        cv2.destroyAllWindows()

5. Advanced Features

5.1 Interaction Analysis

def analyze_interaction(pose_landmarks, object_detections):
    if pose_landmarks is None or object_detections is None:
        return None
        
    # Extract the pose keypoints
    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
    ])
    
    # Analyze interactions with the detected objects
    interactions = []
    for detection in object_detections:
        box_center = np.array([
            (detection[0] + detection[2]) / 2,
            (detection[1] + detection[3]) / 2
        ])
        
        # Compute distances
        left_distance = np.linalg.norm(left_hand - box_center)
        right_distance = np.linalg.norm(right_hand - box_center)
        
        # Detect interactions
        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 Advanced Visualization

def visualize_results(frame, pose_landmarks, object_detections, interactions):
    # Draw the pose landmarks
    if pose_landmarks:
        mp_draw.draw_landmarks(
            frame,
            pose_landmarks,
            mp_pose.POSE_CONNECTIONS
        )
    
    # Draw the object detection boxes
    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)
    
    # Visualize the interactions
    if interactions:
        for interaction in interactions:
            # Draw a line between the hand and the object
            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. Performance Optimization

6.1 Memory Management

def optimize_memory():
    # Configuration for optimal memory usage
    pose_detector = PoseDetector()
    object_detector = ObjectDetector()
    
    # Reduce frame size
    def resize_frame(frame):
        return cv2.resize(frame, (640, 480))
    
    # Periodic queue cleanup
    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 GPU Acceleration

def enable_gpu_acceleration():
    # Configuration for GPU usage
    import torch
    
    if torch.cuda.is_available():
        # Configure YOLO for the GPU
        object_detector.model.to('cuda')
        
        # Configure MediaPipe for the GPU
        pose_detector.pose = mp_pose.Pose(
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5,
            model_complexity=1,
            enable_segmentation=False
        )

7. Best Practices

1. Resource Management

   • Use multithreading for detections
   • Implement a caching system
   • Clean up memory regularly

2. Optimization

   • Reduce frame size
   • Use GPU acceleration
   • Adjust detection frequencies

3. User Interface

   • Provide intuitive controls
   • Display performance metrics
   • Allow mode switching in real time

Conclusion

Key takeaways:

• Effective integration of MediaPipe and YOLOv8
• Multithreaded processing for better performance
• Analysis of interactions between pose and objects
• Flexible user interface

Recommendations:

• Optimize memory usage
• Use GPU acceleration
• Implement robust error handling
• Add customization features