Initial commit

.ipynb_checkpoints/main-checkpoint.py

@@ -0,0 +1,12 @@
+import os
+from dotenv import load_dotenv
+
+def main():
+    # Load environment variables
+    load_dotenv()
+    
+    print("Hello from main.py!")
+    # Add your main application logic here
+
+if __name__ == "__main__":
+    main() 

app.py

@@ -0,0 +1,28 @@
+import chainlit as cl
+from pilates_evaluator import PilatesVideoEvaluator
+
+@cl.on_chat_start
+async def start():
+    await cl.Message(content="Welcome! Upload a video to analyze your Pilates exercise.").send()
+
+@cl.on_message
+async def main(message: cl.Message):
+    if not message.elements:
+        await cl.Message(content="Please upload a video file.").send()
+        return
+
+    video_file = message.elements[0]
+    if not video_file.name.endswith(('.mp4', '.avi', '.mov')):
+        await cl.Message(content="Please upload a valid video file (mp4, avi, mov).").send()
+        return
+
+    await cl.Message(content=f"Analyzing video: {video_file.name}...").send()
+
+    evaluator = PilatesVideoEvaluator()
+    try:
+        evaluator.process_video(video_file.path)
+        report_path = "pilates_evaluation_report.json"
+        evaluator.generate_report(report_path)
+        await cl.Message(content=f"Analysis complete! Report saved to {report_path}.").send()
+    except Exception as e:
+        await cl.Message(content=f"Error analyzing video: {e}").send() 

download_model.py

@@ -0,0 +1,21 @@
+import urllib.request
+import os
+
+def download_model():
+    model_url = "https://github.com/CMU-Perceptual-Computing-Lab/openpose/raw/master/models/pose/coco/pose_iter_440000.caffemodel"
+    proto_url = "https://raw.githubusercontent.com/CMU-Perceptual-Computing-Lab/openpose/master/models/pose/coco/pose_deploy_linevec.prototxt"
+    
+    print("Downloading pose estimation model...")
+    
+    # Download the model file
+    urllib.request.urlretrieve(model_url, "pose_model.caffemodel")
+    print("Downloaded pose model")
+    
+    # Download the prototxt file
+    urllib.request.urlretrieve(proto_url, "pose_deploy_linevec.prototxt")
+    print("Downloaded prototxt file")
+    
+    print("Model files downloaded successfully!")
+
+if __name__ == "__main__":
+    download_model() 

pilates_evaluator.py

@@ -0,0 +1,226 @@
+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__":
+    main() 

project_2

@@ -0,0 +1 @@
+Subproject commit b67aa95aed8dcf03101d5647a2f7bfed9bef7c79

requirements.txt

@@ -0,0 +1,11 @@
+requests==2.31.0
+python-dotenv==1.0.0
+pytest==7.4.3
+black==23.11.0
+flake8==6.1.0
+opencv-python==4.9.0.80
+tensorflow==2.16.1
+scikit-image==0.22.0
+matplotlib==3.8.4
+Pillow==10.3.0
+numpy==1.26.4 

test_pilates.py

@@ -0,0 +1,10 @@
+from pilates_evaluator import PilatesVideoEvaluator
+
+def test_evaluator():
+    evaluator = PilatesVideoEvaluator()
+    print("PilatesVideoEvaluator initialized successfully!")
+    print("Body parts:", evaluator.BODY_PARTS)
+    print("Model loaded:", evaluator.net is not None)
+
+if __name__ == "__main__":
+    test_evaluator()