Upload 10 files

.gitignore

@@ -0,0 +1,41 @@
+# Virtual Environment
+.venv/
+venv/
+ENV/
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# OS
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db 

Dockerfile

@@ -0,0 +1,31 @@
+# Get a distribution that has uv already installed
+FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim
+
+# Add user - this is the user that will run the app
+# If you do not set user, the app will run as root (undesirable)
+RUN useradd -m -u 1000 user
+USER user
+
+# Set the home directory and path
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH        
+
+ENV UVICORN_WS_PROTOCOL=websockets
+
+
+# Set the working directory
+WORKDIR $HOME/app
+
+# Copy the app to the container
+COPY --chown=user . $HOME/app
+
+# Install the dependencies
+# RUN uv sync --frozen
+RUN uv sync
+
+# Expose the port
+EXPOSE 7860
+
+# Run the app
+CMD ["uv", "run", "chainlit", "run", "app.py", "--host", "0.0.0.0", "--port", "7860"]
+

README.md

@@ -1,11 +1,10 @@
----
-title: DEMO COMBINED
-emoji: 🐨
-colorFrom: blue
-colorTo: purple
-sdk: docker
-pinned: false
-license: mit
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: Project 2
+emoji: 📈
+colorFrom: pink
+colorTo: pink
+sdk: docker
+pinned: false
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,64 @@
+import chainlit as cl
+from pilates_evaluator import PilatesVideoEvaluator
+from rag_processor import RAGProcessor
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Initialize RAG processor
+rag_processor = RAGProcessor()
+
+# Sample Pilates knowledge base - you can replace this with your own documents
+PILATES_KNOWLEDGE = [
+    "The Hundred is a fundamental Pilates exercise that helps build core strength and breathing control.",
+    "Proper alignment is crucial in Pilates - maintain neutral spine position during exercises.",
+    "Pilates focuses on six principles: concentration, control, centering, flow, precision, and breathing.",
+    "The Pilates reformer is a versatile piece of equipment that provides resistance through springs.",
+    "Pilates exercises should be performed with controlled movements and proper breathing patterns.",
+    "The Pilates mat work series includes exercises like the Roll Up, Single Leg Stretch, and Double Leg Stretch.",
+    "Pilates helps improve posture, flexibility, and overall body awareness.",
+    "Joseph Pilates developed the method as a system of exercises to strengthen the mind and body.",
+    "Pilates exercises can be modified for different fitness levels and physical conditions.",
+    "Regular Pilates practice can help prevent injuries and improve athletic performance."
+]
+
+# Add documents to RAG system
+rag_processor.add_documents(PILATES_KNOWLEDGE)
+
+@cl.on_chat_start
+async def start():
+    await cl.Message(content="Welcome! You can chat with me about Pilates or upload a video to analyze your Pilates exercise.").send()
+
+@cl.on_message
+async def main(message: cl.Message):
+    # Handle video uploads
+    if message.elements:
+        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()
+        return
+
+    # Handle chat messages
+    if message.content:
+        try:
+            # Generate response using RAG system
+            response = rag_processor.generate_response(message.content)
+            await cl.Message(content=response).send()
+        except Exception as e:
+            await cl.Message(content=f"Error processing your message: {e}").send()
+        return
+
+    await cl.Message(content="Please either send a message or upload a video file.").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() 

pyproject.toml

@@ -0,0 +1,46 @@
+[project]
+name = "Certification_Challenge"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12,<3.13"
+dependencies = [
+  "accelerate>=1.3.0",
+  "arxiv>=2.2.0",
+  "beautifulsoup4>=4.13.3",
+  "chainlit>=2.5.5",
+  "datasets>=3.6.0",
+  "faiss-cpu>=1.11.0",
+  "ipykernel>=6.29.5",
+  "ipywidgets>=8.1.5",
+  "langchain>=0.3.25",
+  "langchain-cohere>=0.4.4",
+  "langchain-community>=0.3.24",
+  "langchain-core>=0.3.34",
+  "langchain-huggingface>=0.1.2",
+  "langchain-openai>=0.3.16",
+  "langchain-qdrant>=0.2.0",
+  "langchain-text-splitters>=0.3.8",
+  "langgraph>=0.4.3",
+  "libmagic>=1.0",
+  "lxml>=5.3.1",
+  "nltk==3.9.1",
+  "pyarrow>=20.0.0",
+  "pymupdf>=1.25.5",
+  "python-pptx==1.0.2",
+  "ragas>=0.2.15",
+  "sentence-transformers>=3.4.1",
+  "transformers[torch]>=4.48.3",
+  "unstructured>=0.17.2",
+  "wandb>=0.19.6",
+  "websockets==11.0.3",
+  "requests>=2.32.0",
+  "openai",
+  "torch",
+  "opencv-python==4.9.0.80",
+  "scikit-image==0.22.0",
+  "matplotlib==3.8.4",
+  "Pillow==10.3.0",
+  "numpy==1.26.4"
+]
+

rag_processor.py

@@ -0,0 +1,61 @@
+import os
+from typing import List
+import openai
+from sentence_transformers import SentenceTransformer
+import faiss
+import numpy as np
+from dotenv import load_dotenv
+
+load_dotenv()
+
+class RAGProcessor:
+    def __init__(self, model_name: str = "bsmith3715/legal-ft-demo_final"):
+        self.model = SentenceTransformer(model_name)
+        self.index = None
+        self.documents = []
+        self.openai_client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+    def add_documents(self, documents: List[str]):
+        """Add documents to the RAG system."""
+        self.documents = documents
+        embeddings = self.model.encode(documents)
+        
+        # Create FAISS index
+        dimension = embeddings.shape[1]
+        self.index = faiss.IndexFlatL2(dimension)
+        self.index.add(embeddings.astype('float32'))
+
+    def retrieve_relevant_context(self, query: str, k: int = 3) -> List[str]:
+        """Retrieve relevant documents for a given query."""
+        if not self.index:
+            return []
+        
+        query_embedding = self.model.encode([query])
+        distances, indices = self.index.search(query_embedding.astype('float32'), k)
+        
+        return [self.documents[i] for i in indices[0]]
+
+    def generate_response(self, query: str) -> str:
+        """Generate a response using OpenAI API with retrieved context."""
+        relevant_docs = self.retrieve_relevant_context(query)
+        context = "\n".join(relevant_docs)
+        
+        prompt = f"""Context information is below.
+---------------------
+{context}
+---------------------
+Given the context information, please answer the following question. If the context doesn't contain relevant information, say so.
+Question: {query}
+Answer:"""
+
+        response = self.openai_client.chat.completions.create(
+            model="gpt-3.5-turbo",
+            messages=[
+                {"role": "system", "content": "You are a helpful Pilates instructor assistant. Use the provided context to answer questions accurately."},
+                {"role": "user", "content": prompt}
+            ],
+            temperature=0.7,
+            max_tokens=500
+        )
+        
+        return response.choices[0].message.content 

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()