Update pilates_evaluator.py

pilates_evaluator.py

@@ -1,226 +1,228 @@
-import cv2
-import numpy as np
-import json
-from datetime import datetime
-import matplotlib.pyplot as plt
-from pathlib import Path
-import os
-import urllib.request
-
-class PilatesVideoEvaluator:
-    def __init__(self):
-        # Initialize OpenCV pose detection
-        self.BODY_PARTS = {
-            "Neck": 0, "RShoulder": 1, "RElbow": 2, "RWrist": 3,
-            "LShoulder": 4, "LElbow": 5, "LWrist": 6, "RHip": 7, "RKnee": 8,
-            "RAnkle": 9, "LHip": 10, "LKnee": 11, "LAnkle": 12
-        }
-
-        # Download the model if it doesn't exist
-        if not os.path.exists('pose_model.caffemodel') or not os.path.exists('pose_deploy.prototxt'):
-            print("Downloading pose estimation model...")
-            model_url = "https://github.com/opencv/opencv_3rdparty/raw/dnn_samples_face_detector_20170830/res10_300x300_ssd_iter_140000.caffemodel"
-            proto_url = "https://raw.githubusercontent.com/opencv/opencv/master/samples/dnn/face_detector/deploy.prototxt"
-            
-            urllib.request.urlretrieve(model_url, "pose_model.caffemodel")
-            urllib.request.urlretrieve(proto_url, "pose_deploy.prototxt")
-            print("Model downloaded successfully!")
-
-        # Load the model
-        self.net = cv2.dnn.readNetFromCaffe("pose_deploy.prototxt", "pose_model.caffemodel")
-
-        # Evaluation metrics
-        self.metrics = {
-            'total_frames': 0,
-            'pose_detected_frames': 0,
-            'movement_consistency': [],
-            'balance_scores': [],
-            'posture_alignment': [],
-            'video_quality_score': 0,
-            'exercise_duration': 0,
-            'detected_exercises': []
-        }
-
-    def analyze_posture(self, frame):
-        """Analyze posture using OpenCV pose estimation"""
-        height, width = frame.shape[:2]
-        blob = cv2.dnn.blobFromImage(frame, 1.0/255, (368, 368), (0, 0, 0), swapRB=False, crop=False)
-        self.net.setInput(blob)
-        output = self.net.forward()
-        
-        # Process the output to get keypoints
-        points = []
-        for i in range(len(self.BODY_PARTS)):
-            # Confidence map for the current keypoint
-            probMap = output[0, i, :, :]
-            probMap = cv2.resize(probMap, (width, height))
-            
-            # Find global maxima of the probMap
-            minVal, prob, minLoc, point = cv2.minMaxLoc(probMap)
-            
-            if prob > 0.1:  # Confidence threshold
-                points.append((int(point[0]), int(point[1])))
-            else:
-                points.append(None)
-        
-        return points
-
-    def detect_exercise_type(self, points):
-        """Detect exercise type based on keypoint positions"""
-        if not points or len(points) < 18:
-            return "Unknown"
-            
-        # Example: Detect plank position
-        if (points[self.BODY_PARTS["RShoulder"]] and points[self.BODY_PARTS["RElbow"]] and
-            points[self.BODY_PARTS["LShoulder"]] and points[self.BODY_PARTS["LElbow"]]):
-            
-            r_shoulder = points[self.BODY_PARTS["RShoulder"]]
-            r_elbow = points[self.BODY_PARTS["RElbow"]]
-            l_shoulder = points[self.BODY_PARTS["LShoulder"]]
-            l_elbow = points[self.BODY_PARTS["LElbow"]]
-            
-            # Check if arms are straight (plank position)
-            r_arm_angle = self.calculate_angle(r_shoulder, r_elbow)
-            l_arm_angle = self.calculate_angle(l_shoulder, l_elbow)
-            
-            if 150 < r_arm_angle < 180 and 150 < l_arm_angle < 180:
-                return "Plank"
-        
-        return "Unknown"
-
-    def calculate_angle(self, point1, point2):
-        """Calculate angle between two points"""
-        if not point1 or not point2:
-            return 0
-        return np.degrees(np.arctan2(point2[1] - point1[1], point2[0] - point1[0]))
-
-    def process_video(self, video_path):
-        """Process video and analyze exercises"""
-        cap = cv2.VideoCapture(video_path)
-        if not cap.isOpened():
-            raise ValueError("Could not open video file")
-
-        while cap.isOpened():
-            ret, frame = cap.read()
-            if not ret:
-                break
-
-            self.metrics['total_frames'] += 1
-            
-            # Analyze posture
-            points = self.analyze_posture(frame)
-            if points:
-                self.metrics['pose_detected_frames'] += 1
-                
-                # Detect exercise type
-                exercise_type = self.detect_exercise_type(points)
-                if exercise_type != "Unknown":
-                    self.metrics['detected_exercises'].append(exercise_type)
-                
-                # Calculate metrics
-                self.metrics['movement_consistency'].append(self.calculate_movement_consistency(points))
-                self.metrics['balance_scores'].append(self.calculate_balance_score(points))
-                self.metrics['posture_alignment'].append(self.calculate_posture_alignment(points))
-
-        cap.release()
-        self.calculate_final_metrics()
-
-    def calculate_movement_consistency(self, points):
-        """Calculate movement consistency score"""
-        # Implement movement consistency calculation
-        return 0.8  # Placeholder
-
-    def calculate_balance_score(self, points):
-        """Calculate balance score"""
-        # Implement balance score calculation
-        return 0.7  # Placeholder
-
-    def calculate_posture_alignment(self, points):
-        """Calculate posture alignment score"""
-        # Implement posture alignment calculation
-        return 0.9  # Placeholder
-
-    def calculate_final_metrics(self):
-        """Calculate final metrics"""
-        if self.metrics['total_frames'] > 0:
-            self.metrics['video_quality_score'] = (
-                self.metrics['pose_detected_frames'] / self.metrics['total_frames']
-            ) * 100
-
-    def generate_report(self, output_path):
-        """Generate evaluation report"""
-        report = {
-            'timestamp': datetime.now().isoformat(),
-            'metrics': self.metrics,
-            'summary': {
-                'video_quality': f"{self.metrics['video_quality_score']:.2f}%",
-                'detected_exercises': list(set(self.metrics['detected_exercises'])),
-                'average_movement_consistency': np.mean(self.metrics['movement_consistency']),
-                'average_balance_score': np.mean(self.metrics['balance_scores']),
-                'average_posture_alignment': np.mean(self.metrics['posture_alignment'])
-            }
-        }
-        
-        with open(output_path, 'w') as f:
-            json.dump(report, f, indent=4)
-
-    def visualize_results(self, output_path):
-        """Visualize evaluation results"""
-        plt.figure(figsize=(12, 8))
-        
-        # Plot metrics over time
-        plt.subplot(2, 2, 1)
-        plt.plot(self.metrics['movement_consistency'], label='Movement Consistency')
-        plt.title('Movement Consistency Over Time')
-        plt.legend()
-        
-        plt.subplot(2, 2, 2)
-        plt.plot(self.metrics['balance_scores'], label='Balance Score')
-        plt.title('Balance Score Over Time')
-        plt.legend()
-        
-        plt.subplot(2, 2, 3)
-        plt.plot(self.metrics['posture_alignment'], label='Posture Alignment')
-        plt.title('Posture Alignment Over Time')
-        plt.legend()
-        
-        plt.tight_layout()
-        plt.savefig(output_path)
-        plt.close()
-
-def main():
-    """Example usage of the Pilates Video Evaluator"""
-    evaluator = PilatesVideoEvaluator()
-
-    # Replace with your video path
-    video_path = "pilates_workout.mp4"
-    output_video_path = "analyzed_pilates_workout.mp4"
-    report_path = "pilates_evaluation_report.json"
-
-    try:
-        # Process the video
-        print("Starting video analysis...")
-        evaluator.process_video(video_path)
-
-        # Print report
-        print("\n" + "="*50)
-        print("PILATES VIDEO EVALUATION REPORT")
-        print("="*50)
-
-        print(f"Video Quality: {evaluator.metrics['video_quality_score']:.2f}%")
-        print(f"Detected Exercises: {', '.join(evaluator.metrics['detected_exercises'])}")
-        print(f"Average Movement Consistency: {evaluator.metrics['average_movement_consistency']:.2f}")
-        print(f"Average Balance Score: {evaluator.metrics['average_balance_score']:.2f}")
-        print(f"Average Posture Alignment: {evaluator.metrics['average_posture_alignment']:.2f}")
-
-        # Save report and visualization
-        evaluator.generate_report(report_path)
-        evaluator.visualize_results(output_video_path)
-
-    except Exception as e:
-        print(f"Error processing video: {e}")
-        print("Make sure you have the required dependencies installed:")
-        print("pip install opencv-python numpy matplotlib")
-
-if __name__ == "__main__":
+import cv2
+import numpy as np
+import json
+from datetime import datetime
+import matplotlib.pyplot as plt
+from pathlib import Path
+import os
+import urllib.request
+
+class PilatesVideoEvaluator:
+    def __init__(self):
+        # Initialize OpenCV pose detection
+        self.BODY_PARTS = {
+            "Neck": 0, "RShoulder": 1, "RElbow": 2, "RWrist": 3,
+            "LShoulder": 4, "LElbow": 5, "LWrist": 6, "RHip": 7, "RKnee": 8,
+            "RAnkle": 9, "LHip": 10, "LKnee": 11, "LAnkle": 12
+        }
+
+        # Download the model if it doesn't exist
+        if not os.path.exists('pose_model.caffemodel') or not os.path.exists('pose_deploy.prototxt'):
+            print("Downloading pose estimation model...")
+                model_url = "https://github.com/CMU-Perceptual-Computing-Lab/openpose/blob/master/models/pose/coco/pose_iter_440000.caffemodel?raw=true"
+                proto_url = "https://raw.githubusercontent.com/opencv/opencv_extra/master/testdata/dnn/openpose_pose_coco.prototxt"
+
+            
+            urllib.request.urlretrieve(model_url, "pose_model.caffemodel")
+            urllib.request.urlretrieve(proto_url, "pose_deploy.prototxt")
+
+            print("Model downloaded successfully!")
+
+        # Load the model
+        self.net = cv2.dnn.readNetFromCaffe("pose_deploy.prototxt", "pose_model.caffemodel")
+
+        # Evaluation metrics
+        self.metrics = {
+            'total_frames': 0,
+            'pose_detected_frames': 0,
+            'movement_consistency': [],
+            'balance_scores': [],
+            'posture_alignment': [],
+            'video_quality_score': 0,
+            'exercise_duration': 0,
+            'detected_exercises': []
+        }
+
+    def analyze_posture(self, frame):
+        """Analyze posture using OpenCV pose estimation"""
+        height, width = frame.shape[:2]
+        blob = cv2.dnn.blobFromImage(frame, 1.0/255, (368, 368), (0, 0, 0), swapRB=False, crop=False)
+        self.net.setInput(blob)
+        output = self.net.forward()
+        
+        # Process the output to get keypoints
+        points = []
+        for i in range(len(self.BODY_PARTS)):
+            # Confidence map for the current keypoint
+            probMap = output[0, i, :, :]
+            probMap = cv2.resize(probMap, (width, height))
+            
+            # Find global maxima of the probMap
+            minVal, prob, minLoc, point = cv2.minMaxLoc(probMap)
+            
+            if prob > 0.1:  # Confidence threshold
+                points.append((int(point[0]), int(point[1])))
+            else:
+                points.append(None)
+        
+        return points
+
+    def detect_exercise_type(self, points):
+        """Detect exercise type based on keypoint positions"""
+        if not points or len(points) < 18:
+            return "Unknown"
+            
+        # Example: Detect plank position
+        if (points[self.BODY_PARTS["RShoulder"]] and points[self.BODY_PARTS["RElbow"]] and
+            points[self.BODY_PARTS["LShoulder"]] and points[self.BODY_PARTS["LElbow"]]):
+            
+            r_shoulder = points[self.BODY_PARTS["RShoulder"]]
+            r_elbow = points[self.BODY_PARTS["RElbow"]]
+            l_shoulder = points[self.BODY_PARTS["LShoulder"]]
+            l_elbow = points[self.BODY_PARTS["LElbow"]]
+            
+            # Check if arms are straight (plank position)
+            r_arm_angle = self.calculate_angle(r_shoulder, r_elbow)
+            l_arm_angle = self.calculate_angle(l_shoulder, l_elbow)
+            
+            if 150 < r_arm_angle < 180 and 150 < l_arm_angle < 180:
+                return "Plank"
+        
+        return "Unknown"
+
+    def calculate_angle(self, point1, point2):
+        """Calculate angle between two points"""
+        if not point1 or not point2:
+            return 0
+        return np.degrees(np.arctan2(point2[1] - point1[1], point2[0] - point1[0]))
+
+    def process_video(self, video_path):
+        """Process video and analyze exercises"""
+        cap = cv2.VideoCapture(video_path)
+        if not cap.isOpened():
+            raise ValueError("Could not open video file")
+
+        while cap.isOpened():
+            ret, frame = cap.read()
+            if not ret:
+                break
+
+            self.metrics['total_frames'] += 1
+            
+            # Analyze posture
+            points = self.analyze_posture(frame)
+            if points:
+                self.metrics['pose_detected_frames'] += 1
+                
+                # Detect exercise type
+                exercise_type = self.detect_exercise_type(points)
+                if exercise_type != "Unknown":
+                    self.metrics['detected_exercises'].append(exercise_type)
+                
+                # Calculate metrics
+                self.metrics['movement_consistency'].append(self.calculate_movement_consistency(points))
+                self.metrics['balance_scores'].append(self.calculate_balance_score(points))
+                self.metrics['posture_alignment'].append(self.calculate_posture_alignment(points))
+
+        cap.release()
+        self.calculate_final_metrics()
+
+    def calculate_movement_consistency(self, points):
+        """Calculate movement consistency score"""
+        # Implement movement consistency calculation
+        return 0.8  # Placeholder
+
+    def calculate_balance_score(self, points):
+        """Calculate balance score"""
+        # Implement balance score calculation
+        return 0.7  # Placeholder
+
+    def calculate_posture_alignment(self, points):
+        """Calculate posture alignment score"""
+        # Implement posture alignment calculation
+        return 0.9  # Placeholder
+
+    def calculate_final_metrics(self):
+        """Calculate final metrics"""
+        if self.metrics['total_frames'] > 0:
+            self.metrics['video_quality_score'] = (
+                self.metrics['pose_detected_frames'] / self.metrics['total_frames']
+            ) * 100
+
+    def generate_report(self, output_path):
+        """Generate evaluation report"""
+        report = {
+            'timestamp': datetime.now().isoformat(),
+            'metrics': self.metrics,
+            'summary': {
+                'video_quality': f"{self.metrics['video_quality_score']:.2f}%",
+                'detected_exercises': list(set(self.metrics['detected_exercises'])),
+                'average_movement_consistency': np.mean(self.metrics['movement_consistency']),
+                'average_balance_score': np.mean(self.metrics['balance_scores']),
+                'average_posture_alignment': np.mean(self.metrics['posture_alignment'])
+            }
+        }
+        
+        with open(output_path, 'w') as f:
+            json.dump(report, f, indent=4)
+
+    def visualize_results(self, output_path):
+        """Visualize evaluation results"""
+        plt.figure(figsize=(12, 8))
+        
+        # Plot metrics over time
+        plt.subplot(2, 2, 1)
+        plt.plot(self.metrics['movement_consistency'], label='Movement Consistency')
+        plt.title('Movement Consistency Over Time')
+        plt.legend()
+        
+        plt.subplot(2, 2, 2)
+        plt.plot(self.metrics['balance_scores'], label='Balance Score')
+        plt.title('Balance Score Over Time')
+        plt.legend()
+        
+        plt.subplot(2, 2, 3)
+        plt.plot(self.metrics['posture_alignment'], label='Posture Alignment')
+        plt.title('Posture Alignment Over Time')
+        plt.legend()
+        
+        plt.tight_layout()
+        plt.savefig(output_path)
+        plt.close()
+
+def main():
+    """Example usage of the Pilates Video Evaluator"""
+    evaluator = PilatesVideoEvaluator()
+
+    # Replace with your video path
+    video_path = "pilates_workout.mp4"
+    output_video_path = "analyzed_pilates_workout.mp4"
+    report_path = "pilates_evaluation_report.json"
+
+    try:
+        # Process the video
+        print("Starting video analysis...")
+        evaluator.process_video(video_path)
+
+        # Print report
+        print("\n" + "="*50)
+        print("PILATES VIDEO EVALUATION REPORT")
+        print("="*50)
+
+        print(f"Video Quality: {evaluator.metrics['video_quality_score']:.2f}%")
+        print(f"Detected Exercises: {', '.join(evaluator.metrics['detected_exercises'])}")
+        print(f"Average Movement Consistency: {evaluator.metrics['average_movement_consistency']:.2f}")
+        print(f"Average Balance Score: {evaluator.metrics['average_balance_score']:.2f}")
+        print(f"Average Posture Alignment: {evaluator.metrics['average_posture_alignment']:.2f}")
+
+        # Save report and visualization
+        evaluator.generate_report(report_path)
+        evaluator.visualize_results(output_video_path)
+
+    except Exception as e:
+        print(f"Error processing video: {e}")
+        print("Make sure you have the required dependencies installed:")
+        print("pip install opencv-python numpy matplotlib")
+
+if __name__ == "__main__":
     main()