دمج MediaPipe وYOLOv8: تطبيق متقدم للرؤية الحاسوبية

دليل شامل لإنشاء تطبيق يجمع بين اكتشاف وضعية الجسم باستخدام MediaPipe واكتشاف الكائنات باستخدام YOLOv8.

1. الإعداد الأولي

# تثبيت الاعتمادات
pip install mediapipe ultralytics opencv-python numpy

# استيراد المكتبات
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. بنية التطبيق

2.1 الفئات الأساسية

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):
        # التحويل إلى RGB
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        
        # اكتشاف الوضعية
        results = self.pose.process(image_rgb)
        
        # رسم نقاط المعالم
        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):
        # اكتشاف الكائنات
        results = self.model(image)
        
        # معالجة النتائج
        for result in results:
            boxes = result.boxes
            for box in boxes:
                # إحداثيات المربع
                x1, y1, x2, y2 = box.xyxy[0]
                x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
                
                # درجة الثقة
                confidence = float(box.conf[0])
                
                # الفئة
                class_id = int(box.cls[0])
                class_name = self.model.names[class_id]
                
                # رسم المربع
                cv2.rectangle(image, (x1, y1), (x2, y2), (0, 255, 0), 2)
                
                # إضافة النص
                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. المعالجة متعددة الخيوط

3.1 مدير الإطارات

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:
                # اكتشاف الوضعية
                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:
                # اكتشاف الكائنات
                processed_frame = self.detector.detect(frame)
                self.frame_manager.add_result((processed_frame, None))

4. التطبيق الرئيسي

4.1 الإعداد

class Application:
    def __init__(self):
        # تهيئة أجهزة الاكتشاف
        self.pose_detector = PoseDetector()
        self.object_detector = ObjectDetector()
        
        # تهيئة مدير الإطارات
        self.frame_manager = FrameManager()
        
        # تهيئة العمّال
        self.pose_worker = PoseWorker(self.frame_manager, self.pose_detector)
        self.object_worker = ObjectWorker(self.frame_manager, self.object_detector)
        
        # إعداد الكاميرا
        self.cap = cv2.VideoCapture(0)
        
        # متغيرات التحكم
        self.running = True
        self.current_mode = 'both'  # 'pose', 'object', 'both'

4.2 الحلقة الرئيسية

    def run(self):
        # تشغيل العمّال
        self.pose_worker.start()
        self.object_worker.start()
        
        while self.running:
            success, frame = self.cap.read()
            if not success:
                continue
                
            # إضافة الإطار إلى المدير
            self.frame_manager.add_frame(frame.copy())
            
            # استرجاع النتائج
            pose_result = self.frame_manager.get_result()
            object_result = self.frame_manager.get_result()
            
            # عرض النتائج
            if pose_result is not None and object_result is not None:
                pose_frame, pose_landmarks = pose_result
                object_frame, _ = object_result
                
                # دمج النتائج
                if self.current_mode == 'both':
                    # تراكب النتائج
                    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)
            
            # التعامل مع المفاتيح
            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'
        
        # التنظيف
        self.cleanup()

4.3 التنظيف

    def cleanup(self):
        # إيقاف العمّال
        self.pose_worker.running = False
        self.object_worker.running = False
        self.pose_worker.join()
        self.object_worker.join()
        
        # تحرير الموارد
        self.cap.release()
        cv2.destroyAllWindows()

5. ميزات متقدمة

5.1 تحليل التفاعلات

def analyze_interaction(pose_landmarks, object_detections):
    if pose_landmarks is None or object_detections is None:
        return None
        
    # استخراج النقاط الرئيسية للوضعية
    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
    ])
    
    # تحليل التفاعلات مع الكائنات المكتشفة
    interactions = []
    for detection in object_detections:
        box_center = np.array([
            (detection[0] + detection[2]) / 2,
            (detection[1] + detection[3]) / 2
        ])
        
        # حساب المسافات
        left_distance = np.linalg.norm(left_hand - box_center)
        right_distance = np.linalg.norm(right_hand - box_center)
        
        # اكتشاف التفاعلات
        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 تصور متقدم للنتائج

def visualize_results(frame, pose_landmarks, object_detections, interactions):
    # رسم نقاط المعالم للوضعية
    if pose_landmarks:
        mp_draw.draw_landmarks(
            frame,
            pose_landmarks,
            mp_pose.POSE_CONNECTIONS
        )
    
    # رسم مربعات اكتشاف الكائنات
    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)
    
    # تصور التفاعلات
    if interactions:
        for interaction in interactions:
            # رسم خط بين اليد والكائن
            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. تحسين الأداء

6.1 إدارة الذاكرة

def optimize_memory():
    # إعداد للاستخدام الأمثل للذاكرة
    pose_detector = PoseDetector()
    object_detector = ObjectDetector()
    
    # تقليل حجم الإطارات
    def resize_frame(frame):
        return cv2.resize(frame, (640, 480))
    
    # تنظيف دوري للطوابير
    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

def enable_gpu_acceleration():
    # إعداد لاستخدام GPU
    import torch
    
    if torch.cuda.is_available():
        # إعداد YOLO لاستخدام GPU
        object_detector.model.to('cuda')
        
        # إعداد MediaPipe لاستخدام GPU
        pose_detector.pose = mp_pose.Pose(
            min_detection_confidence=0.5,
            min_tracking_confidence=0.5,
            model_complexity=1,
            enable_segmentation=False
        )

7. أفضل الممارسات

1. إدارة الموارد

   • استخدام تعدد الخيوط (multi-threading) في عمليات الاكتشاف
   • تنفيذ نظام للتخزين المؤقت (cache)
   • تنظيف الذاكرة بانتظام

2. التحسين

   • تقليل حجم الإطارات
   • استخدام التسريع عبر GPU
   • ضبط تكرار عمليات الاكتشاف

3. واجهة المستخدم

   • توفير عناصر تحكم بديهية
   • عرض مقاييس الأداء
   • إتاحة تغيير الوضع في الوقت الفعلي

الخلاصة

النقاط الرئيسية:

• دمج فعّال بين MediaPipe وYOLOv8
• معالجة متعددة الخيوط لأداء أفضل
• تحليل التفاعلات بين الوضعية والكائنات
• واجهة مستخدم مرنة

التوصيات:

• تحسين استخدام الذاكرة
• استخدام التسريع عبر GPU
• تنفيذ إدارة قوية للأخطاء
• إضافة ميزات التخصيص