dead lift exercise

.gitignore

@@ -175,4 +175,5 @@ cython_debug/
 
 static/
 *.mp4
-*.MOV
+*.MOV
+*.json

app.py

@@ -14,6 +14,7 @@ from fastapi.responses import JSONResponse
 from config import AI_API_TOKEN
 import logging
 import json
+from src.exercises.dead_lift import analyze_dead_lift
 
 
 logging.basicConfig(level=logging.INFO)
@@ -73,7 +74,7 @@ async def upload(background_tasks: BackgroundTasks,
                  token: str = Header(...),
                  player_data: str = Body(...),
                  repetitions: int|str = Body(...),
-                 exercise_id: str = Body(...)
+                 exercise_id: str = Body(...),
                  ):
     
     
@@ -106,3 +107,46 @@ async def upload(background_tasks: BackgroundTasks,
     return JSONResponse(content={"message": "Video uploaded successfully",
                                  "status": 200})
 
+
+
+@app.post("/exercise/dead_lift")
+async def upload(background_tasks: BackgroundTasks,
+                 file: UploadFile = File(...), 
+                 token: str = Header(...),
+                 player_data: str = Body(...),
+                 repetitions: int|str = Body(...),
+                 exercise_id: str = Body(...),
+                 weight: int|str = Body(...)
+                 ):
+    
+    player_data = json.loads(player_data)
+
+    if token != AI_API_TOKEN:
+        
+        raise HTTPException(status_code=401, detail="Unauthorized")
+    
+    logger.info("reading contents")
+    contents = await file.read()
+    
+    # Save the file to the local directory
+    logger.info("saving file")
+    with open(file.filename, "wb") as f:
+        f.write(contents)
+    
+    logger.info(f"file saved {file.filename}")
+    
+    
+    
+    background_tasks.add_task(analyze_dead_lift, 
+                              file.filename,
+                              repetitions,
+                              weight,
+                              player_data["height"],
+                              vitpose, 
+                              player_data["id"],
+                              exercise_id)  
+    
+    return JSONResponse(content={"message": "Video uploaded successfully",
+                                 "status": 200})
+
+    

src/exercises/dead_lift.py

@@ -0,0 +1,316 @@
+import cv2
+import numpy as np
+from scipy.optimize import linear_sum_assignment
+import requests
+import json
+from fastapi.responses import JSONResponse
+from config import API_URL, API_KEY
+import logging
+
+logger = logging.getLogger(__name__)
+
+class RepCounter:
+    def __init__(self, fps, height_cm, mass_kg=0, target_reps=None):
+        self.count = 0
+        self.last_state = None
+        self.cooldown_frames = 15
+        self.cooldown = 0
+        self.rep_start_frame = None
+        self.start_wrist_y = None
+        self.rep_data = []
+        self.power_data = [] 
+        self.fps = fps
+        self.cm_per_pixel = None
+        self.real_distance_cm = height_cm * 0.2735
+        self.calibration_done = False
+        self.mass_kg = mass_kg
+        self.gravity = 9.81
+        self.target_reps = int(target_reps)
+        self.target_reached = False
+        self.final_speed = None
+        self.final_power = None
+        self.SKELETON = [
+             (5, 6), (5, 7), (7, 9), (6, 8), (8, 10),
+            (9, 10), (11, 12), (5, 11), (6, 12),
+            (11, 13), (13, 15), (12, 14), (14, 16), (13, 14)
+        ]
+
+    def update(self, wrist_y, knee_y, current_frame):
+        if self.target_reached or self.cooldown > 0:
+            self.cooldown = max(0, self.cooldown - 1)
+            return
+
+        current_state = 'above' if wrist_y < knee_y else 'below'
+
+        if self.last_state != current_state:
+            if current_state == 'below':
+                self.rep_start_frame = current_frame
+                self.start_wrist_y = wrist_y
+            elif current_state == 'above' and self.last_state == 'below':
+                if self.rep_start_frame is not None and self.cm_per_pixel is not None:
+                    end_frame = current_frame
+                    duration = (end_frame - self.rep_start_frame) / self.fps
+                    distance_pixels = self.start_wrist_y - wrist_y
+                    distance_cm = distance_pixels * self.cm_per_pixel
+
+                    if duration > 0:
+                        speed_cmps = abs(distance_cm) / duration
+                        self.rep_data.append(speed_cmps)
+
+                        if self.mass_kg > 0:
+                            speed_mps = speed_cmps / 100
+                            force = self.mass_kg * self.gravity
+                            power = force * speed_mps
+                            self.power_data.append(power)
+
+                        self.count += 1
+                        if self.target_reps and self.count >= self.target_reps:
+                            self.count = self.target_reps
+                            self.target_reached = True
+                            self.final_speed = np.mean(self.rep_data) if self.rep_data else 0
+                            self.final_power = np.mean(self.power_data) if self.power_data else 0
+
+                        self.cooldown = self.cooldown_frames
+
+        self.last_state = current_state
+
+# CentroidTracker class
+class CentroidTracker:
+    def __init__(self, max_disappeared=50, max_distance=100):
+        self.next_id = 0
+        self.objects = {}
+        self.max_disappeared = max_disappeared
+        self.max_distance = max_distance
+
+    def _update_missing(self):
+        to_delete = []
+        for obj_id in list(self.objects.keys()):
+            self.objects[obj_id]["missed"] += 1
+            if self.objects[obj_id]["missed"] > self.max_disappeared:
+                to_delete.append(obj_id)
+        for obj_id in to_delete:
+            del self.objects[obj_id]
+
+    def update(self, detections):
+        if len(detections) == 0:
+            self._update_missing()
+            return []
+
+        centroids = np.array([[(x1 + x2) / 2, (y1 + y2) / 2] for x1, y1, x2, y2 in detections])
+
+        if len(self.objects) == 0:
+            return self._register_new(centroids)
+
+        return self._match_existing(centroids, detections)
+
+    def _register_new(self, centroids):
+        new_ids = []
+        for centroid in centroids:
+            self.objects[self.next_id] = {"centroid": centroid, "missed": 0}
+            new_ids.append(self.next_id)
+            self.next_id += 1
+        return new_ids
+
+    def _match_existing(self, centroids, detections):
+        existing_ids = list(self.objects.keys())
+        existing_centroids = [self.objects[obj_id]["centroid"] for obj_id in existing_ids]
+
+        cost = np.linalg.norm(np.array(existing_centroids)[:, np.newaxis] - centroids, axis=2)
+        row_ind, col_ind = linear_sum_assignment(cost)
+
+        used_rows = set()
+        used_cols = set()
+        matches = {}
+
+        for (row, col) in zip(row_ind, col_ind):
+            if cost[row, col] <= self.max_distance:
+                obj_id = existing_ids[row]
+                matches[obj_id] = centroids[col]
+                used_rows.add(row)
+                used_cols.add(col)
+
+        for obj_id in existing_ids:
+            if obj_id not in matches:
+                self.objects[obj_id]["missed"] += 1
+                if self.objects[obj_id]["missed"] > self.max_disappeared:
+                    del self.objects[obj_id]
+
+        new_ids = []
+        for col in range(len(centroids)):
+            if col not in used_cols:
+                self.objects[self.next_id] = {"centroid": centroids[col], "missed": 0}
+                new_ids.append(self.next_id)
+                self.next_id += 1
+
+        for obj_id, centroid in matches.items():
+            self.objects[obj_id]["centroid"] = centroid
+            self.objects[obj_id]["missed"] = 0
+
+        all_ids = []
+        for detection in detections:
+            centroid = np.array([(detection[0] + detection[2]) / 2, (detection[1] + detection[3]) / 2])
+            min_id = None
+            min_dist = float('inf')
+            for obj_id, data in self.objects.items():
+                dist = np.linalg.norm(centroid - data["centroid"])
+                if dist < min_dist and dist <= self.max_distance:
+                    min_dist = dist
+                    min_id = obj_id
+            if min_id is not None:
+                all_ids.append(min_id)
+                self.objects[min_id]["centroid"] = centroid
+            else:
+                all_ids.append(self.next_id)
+                self.objects[self.next_id] = {"centroid": centroid, "missed": 0}
+                self.next_id += 1
+
+        return all_ids
+
+# Función de procesamiento optimizada
+def process_frame_for_counting(frame, tracker, rep_counter, frame_number,vitpose):
+
+    pose_results = vitpose.pipeline(frame)
+
+
+    keypoints = pose_results.keypoints_xy.float().cpu().numpy()[0]
+    scores = pose_results.scores.float().cpu().numpy()[0]
+    valid_points = {}
+    wrist_midpoint = None
+    knee_line_y = None
+    
+    
+    print(keypoints)
+    print(scores)
+
+    # Procesar puntos clave
+    for i, (kp, conf) in enumerate(zip(keypoints, scores)):
+        if conf > 0.3 and 5 <= i <= 16:
+            x, y = map(int, kp[:2])
+            valid_points[i] = (x, y)
+
+    # Calibración usando keypoints de rodilla (14) y pie (16)
+    if not rep_counter.calibration_done and 14 in valid_points and 16 in valid_points:
+        knee = valid_points[14]
+        ankle = valid_points[16]
+        pixel_distance = np.sqrt((knee[0] - ankle[0])**2 + (knee[1] - ankle[1])**2)
+        if pixel_distance > 0:
+            rep_counter.cm_per_pixel = rep_counter.real_distance_cm / pixel_distance
+            rep_counter.calibration_done = True
+
+    # Calcular puntos de referencia para conteo
+    if 9 in valid_points and 10 in valid_points:
+        wrist_midpoint = (
+            (valid_points[9][0] + valid_points[10][0]) // 2,
+            (valid_points[9][1] + valid_points[10][1]) // 2
+        )
+    if 13 in valid_points and 14 in valid_points:
+        pt1 = np.array(valid_points[13])
+        pt2 = np.array(valid_points[14])
+        direction = pt2 - pt1
+        extension = 0.2
+        new_pt1 = pt1 - direction * extension
+        new_pt2 = pt2 + direction * extension
+        knee_line_y = (new_pt1[1] + new_pt2[1]) // 2
+
+    # Actualizar contador
+    if wrist_midpoint and knee_line_y:
+        rep_counter.update(wrist_midpoint[1], knee_line_y, frame_number)
+            
+            
+            
+# Función principal de Gradio
+def analyze_dead_lift(input_video, reps, weight, height,vitpose,player_id,exercise_id):
+    cap = cv2.VideoCapture(input_video)
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    
+    
+    rep_counter = RepCounter(fps, int(height), int(weight), int(reps))
+    tracker = CentroidTracker(max_distance=150)
+
+    frame_number = 0
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        process_frame_for_counting(frame, tracker, rep_counter, frame_number,vitpose)
+        frame_number += 1
+
+    cap.release()
+
+    # Preparar payload para webhook
+    if rep_counter.mass_kg > 0:
+        power_data = rep_counter.power_data
+    else:
+        # Si no hay masa, usar ceros para potencia
+        power_data = [0] * len(rep_counter.rep_data) if rep_counter.rep_data else []
+
+    # Asegurar que tenemos datos para enviar
+    if rep_counter.rep_data:
+        payload = {"repetition_data": [
+            {"repetition": i, "velocidad": round(s,1), "potencia": round(p,1)}
+            for i, (s, p) in enumerate(zip(rep_counter.rep_data, power_data), start=1)
+        ]}
+    else:
+        # En caso de no detectar repeticiones
+        payload = {"repetition_data": []}
+    
+    
+    send_results_api(payload, player_id, exercise_id, input_video)
+
+
+def send_results_api(results_dict: dict,
+                     player_id: str,
+                     exercise_id: str,
+                     video_path: str) -> JSONResponse:
+    """
+    Send video analysis results to the API webhook endpoint.
+    
+    This function uploads the analyzed video file along with the computed metrics
+    to the API's webhook endpoint for processing and storage.
+    
+    Args:
+        results_dict (dict): Dictionary containing analysis results including:
+            - video_analysis: Information about the processed video
+            - repetition_data: List of metrics for each jump repetition
+        player_id (str): Unique identifier for the player
+        exercise_id (str): Unique identifier for the exercise
+        video_path (str): Path to the video file to upload
+        
+    Returns:
+        JSONResponse: HTTP response from the API endpoint
+        
+    Raises:
+        FileNotFoundError: If the video file doesn't exist
+        requests.RequestException: If the API request fails
+        json.JSONEncodeError: If results_dict cannot be serialized to JSON
+    """
+    url = API_URL + "/exercises/webhooks/video-processed-results"
+    logger.info(f"Sending video results to {url}")
+    
+    # Open the video file
+    with open(video_path, 'rb') as video_file:
+        # Prepare the files dictionary for file upload
+        files = {
+            'file': (video_path.split('/')[-1], video_file, 'video/mp4')
+        }
+        
+        # Prepare the form data
+        data = {
+            'player_id': player_id,
+            'exercise_id': exercise_id,
+            'results': json.dumps(results_dict)  # Convert dict to JSON string
+        }
+        
+        # Send the request with both files and data
+        response = requests.post(
+            url, 
+            headers={"token": API_KEY},
+            files=files,
+            data=data,
+            stream=True
+        )
+    
+    logger.info(f"Response: {response.status_code}")
+    logger.info(f"Response: {response.text}")
+    return response 

tasks.py

@@ -97,6 +97,8 @@ class FramePosition:
     y: int
     width: int
     height: int
+    
+
 def process_video(file_name: str,vitpose: VitPose,user_id: str,player_id: str):
     """
     Process a video file using VitPose for pose estimation and send results to webhook.