salto alto exercise

tasks.py

@@ -52,7 +52,11 @@ def process_video(file_name: str,vitpose: VitPose,user_id: str,player_id: str):
     logger.info(f"Video sent to {url}")
     
 
-def process_salto_alto(file_name: str, vitpose: VitPose, player_data: dict, repetitions: int, exercise_id: str):
+def process_salto_alto(file_name: str,
+                       vitpose: VitPose,
+                       player_data: dict,
+                       repetitions: int,
+                       exercise_id: str) -> dict:
     """
     Process a high jump exercise video using VitPose for pose estimation.
     
@@ -73,37 +77,80 @@ def process_salto_alto(file_name: str, vitpose: VitPose, player_data: dict, repe
     body_mass_kg = player_data.get('weight', 64)  # Peso corporal en kg
     
     # Generate output paths
-    output_video = file_name.replace('.mp4', '_analyzed.mp4')
-    output_json = output_video.replace('.mp4', '.json')
-    
+    output_video = file_name.replace('.mp4', '_analyzed.mp4')    
     # Process the video and get the jump metrics
-    results_dict = analyze_jump_video(
-        model=model,
-        input_video=file_name,
-        output_video=output_video,
-        reference_height=reference_height,
-        body_mass_kg=body_mass_kg
-    )
-    
-    # Save results to JSON
-    with open(output_json, 'w') as f:
-        json.dumps(results_dict, indent=4)
-    
-    # Print summary
-    print("\nResultados finales:")
-    print(f"Salto Relativo máximo: {results_dict['jump_metrics']['max_relative_jump']:.2f}m")
-    print(f"Salto Alto máximo: {results_dict['jump_metrics']['max_high_jump']:.2f}m")
-    print(f"Potencia Sayer (estimada): {results_dict['jump_metrics']['peak_power_sayer']:.2f} W")
-    
-    # Return results dictionary
-    return {
-        "output_video": output_video,
-        "output_json": output_json,
-        "metrics": results_dict
-    }
+    # print(f"reference_height: {reference_height}")
+    # results_dict = analyze_jump_video(
+    #     model=model,
+    #     input_video=file_name,
+    #     output_video=output_video,
+    #     player_height= float(reference_height) / 100, #cm to m
+    #     body_mass_kg= float(body_mass_kg),
+    #     repetitions=repetitions
+    # )
+    
+    results_dict = {'video_analysis': {'output_video': 'user_id_2_player_id_2_exercise_salto_alto_VIDEO-2025-05-19-18-55-47_analyzed.mp4'}, 'repetition_data': [{'repetition': 1, 'distancia_elevada': 0.47999998927116394, 'salto_alto': 2.180000066757202, 'potencia_sayer': 3768.719970703125}, {'repetition': 2, 'distancia_elevada': 0.49000000953674316, 'salto_alto': 2.190000057220459, 'potencia_sayer': 3827.929931640625}, {'repetition': 3, 'distancia_elevada': 0.5099999904632568, 'salto_alto': 2.2100000381469727, 'potencia_sayer': 3915.5}]}
+    
+    print(f"results_dict: {results_dict}")
+    
+    
+    response = send_results_api(results_dict,
+                     player_data["id"],
+                     exercise_id,
+                     file_name)
+    
+    # os.remove(file_name)
+    # os.remove(output_video)
+
+
+def send_results_api(results_dict: dict,
+                     player_id: str,
+                     exercise_id: str,
+                     video_path: str):
+    """
+    Updated function to send results to the new webhook endpoint
+    """
+    url = API_URL + "/excercises/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 
+
 
 
-def analyze_jump_video(model, input_video, output_video, reference_height=1.68, body_mass_kg=64):
+
+
+
+def analyze_jump_video(model: VitPose,
+                       input_video: str,
+                       output_video: str,
+                       player_height: float,
+                       body_mass_kg: float,
+                       repetitions: int) -> dict | None:
     """
     Analyze a jump video to calculate various jump metrics.
     
@@ -156,19 +203,13 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
     # Process first frame to calibrate
     output = model(frame)  # Detect pose in first frame
     keypoints = output.keypoints_xy.float().cpu().numpy()
-    print(f"keypoints {keypoints}")
     labels = model.pose_estimator_config.label2id
-    print(labels)
+
     nose_keypoint = labels["Nose"]
     L_ankle_keypoint = labels["L_Ankle"]
     R_ankle_keypoint = labels["R_Ankle"]
     L_shoulder_keypoint = labels["L_Shoulder"]
     R_shoulder_keypoint = labels["R_Shoulder"]
-    print(f"nose_keypoint {nose_keypoint}")
-    print(f"L_ankle_keypoint {L_ankle_keypoint}")
-    print(f"R_ankle_keypoint {R_ankle_keypoint}")
-    print(f"L_shoulder_keypoint {L_shoulder_keypoint}")
-    print(f"R_shoulder_keypoint {R_shoulder_keypoint}")
 
     if (
         keypoints is not None
@@ -178,9 +219,7 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
         kpts_first = keypoints[0]
         if len(kpts_first[nose_keypoint]) > 0 and len(kpts_first[L_ankle_keypoint]) > 0:  # Nose and ankles
             initial_person_height_px = min(kpts_first[L_ankle_keypoint][1], kpts_first[R_ankle_keypoint][1]) - kpts_first[nose_keypoint][1]
-            print(f"initial_person_height_px {initial_person_height_px}")
-            PX_PER_METER = float(initial_person_height_px) / float(reference_height)
-            print(f"Escala calculada: {PX_PER_METER:.2f} px/m")
+            PX_PER_METER = initial_person_height_px / player_height
         if len(kpts_first[L_shoulder_keypoint]) > 0 and len(kpts_first[R_shoulder_keypoint]) > 0:  # Left (5) and right (6) shoulders
             initial_left_shoulder_x = int(kpts_first[L_shoulder_keypoint][0])
             initial_right_shoulder_x = int(kpts_first[R_shoulder_keypoint][0])
@@ -188,8 +227,8 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
     if PX_PER_METER is None or initial_left_shoulder_x is None or initial_right_shoulder_x is None:
         print("No se pudo calibrar la escala o detectar los hombros en el primer frame.")
         cap.release()
-        return {}
-    
+        return None
+        
     # Reset video for processing
     cap.release()
     cap = cv2.VideoCapture(input_video)
@@ -210,10 +249,9 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
     head_y_buffer = []
     velocity_vertical = 0.0
     peak_power_sayer = 0.0  # Initialize Sayer power
-    person_detected = False  # Flag to indicate if person was detected in any frame
     current_power = 0.0
     repetition_count = 0
-    jump_in_air = False
+    jump_peak_power = 0.0  # Peak power for current jump only
     
     # Process each frame
     while cap.isOpened():
@@ -222,6 +260,8 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
             break
         
         annotated_frame = frame.copy()
+        if repetition_count == repetitions:
+            continue
         
         # Add try-except block around the model inference to catch any model errors
         try:
@@ -233,8 +273,6 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
                 len(keypoints) > 0 and 
                 len(keypoints[0]) > 0 and
                 keypoints.size > 0):  # Check if array is not empty
-                
-                person_detected = True
                 kpts = keypoints[0]
                 
                 # Make sure all required keypoints are detected
@@ -257,8 +295,6 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
                         current_ankle_y = min(a[1] for a in ankles)
                         last_detected_ankles_y = current_ankle_y
                         current_head_y = nose[1]
-                        current_left_shoulder_x = int(left_shoulder[0])
-                        current_right_shoulder_x = int(right_shoulder[0])
                         
                         # Smooth ankle and head positions
                         ankle_y_history.append(current_ankle_y)
@@ -294,16 +330,17 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
                             takeoff_head_y = smoothed_head_y
                             max_jump_height = 0
                             max_head_height_px = smoothed_head_y
+                            jump_peak_power = 0.0  # Reset for this jump
                         
                         # Detect jump end
                         if jump_started and relative_ankle_change <= JUMP_THRESHOLD_PERCENT:
                             # Add to repetition data
-                            salto_alto = calculate_absolute_jump_height(reference_height, max_jump_height)
+                            high_jump = calculate_high_jump(player_height, max_jump_height)
                             repetition_data.append({
                                 "repetition": repetition_count + 1,
-                                "relative_jump_m": round(max_jump_height, 3),
-                                "absolute_jump_m": round(salto_alto, 3),
-                                "peak_power_watts": round(current_power, 1)
+                                "distancia_elevada": round(max_jump_height, 2),
+                                "salto_alto": round(high_jump, 2),
+                                "potencia_sayer": round(jump_peak_power, 2)  # Use jump-specific peak
                             })
                             repetition_count += 1
                             jump_started = False
@@ -317,7 +354,9 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
                                 max_head_height_px = smoothed_head_y
                             if relative_jump:
                                 current_power = calculate_peak_power_sayer(relative_jump, body_mass_kg)
-                                if current_power > peak_power_sayer:
+                                if current_power > jump_peak_power:  # Track peak for THIS jump
+                                    jump_peak_power = current_power
+                                if current_power > peak_power_sayer:  # Keep global peak too
                                     peak_power_sayer = current_power
                     else:
                         # Skip processing for this frame - invalid coordinates
@@ -342,13 +381,13 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
         
         # Calculate metrics and draw overlay even if keypoints weren't detected
         # This ensures video continues to show previous metrics
-        salto_alto = calculate_absolute_jump_height(reference_height, max_jump_height)
+        high_jump = calculate_high_jump(player_height, max_jump_height)
         
         # Draw floating metric boxes
         annotated_frame = draw_metrics_overlay(
             frame=annotated_frame,
             max_jump_height=max_jump_height,
-            salto_alto=salto_alto,
+            salto_alto=high_jump,
             velocity_vertical=velocity_vertical,
             peak_power_sayer=peak_power_sayer,
             repetition_count=repetition_count,
@@ -370,31 +409,70 @@ def analyze_jump_video(model, input_video, output_video, reference_height=1.68,
             horizontal_offset_factor=HORIZONTAL_OFFSET_FACTOR
         )
         
+        # Draw person skeleton keypoints
+        try:
+            if keypoints is not None and len(keypoints) > 0 and len(keypoints[0]) > 0:
+                # Use the exact keypoint indices
+                keypoint_indices = {
+                    'L_Ankle': 15, 'L_Ear': 3, 'L_Elbow': 7, 'L_Eye': 1, 'L_Hip': 11, 
+                    'L_Knee': 13, 'L_Shoulder': 5, 'L_Wrist': 9, 'Nose': 0, 'R_Ankle': 16, 
+                    'R_Ear': 4, 'R_Elbow': 8, 'R_Eye': 2, 'R_Hip': 12, 'R_Knee': 14, 
+                    'R_Shoulder': 6, 'R_Wrist': 10
+                }
+                
+                # Define skeleton connections (pairs of keypoints that should be connected)
+                skeleton_connections = [
+                    (keypoint_indices["Nose"], keypoint_indices["L_Eye"]), 
+                    (keypoint_indices["Nose"], keypoint_indices["R_Eye"]),
+                    (keypoint_indices["L_Eye"], keypoint_indices["L_Ear"]), 
+                    (keypoint_indices["R_Eye"], keypoint_indices["R_Ear"]),
+                    (keypoint_indices["Nose"], keypoint_indices["L_Shoulder"]),
+                    (keypoint_indices["Nose"], keypoint_indices["R_Shoulder"]),
+                    (keypoint_indices["L_Shoulder"], keypoint_indices["R_Shoulder"]),
+                    (keypoint_indices["L_Shoulder"], keypoint_indices["L_Elbow"]), 
+                    (keypoint_indices["R_Shoulder"], keypoint_indices["R_Elbow"]),
+                    (keypoint_indices["L_Elbow"], keypoint_indices["L_Wrist"]), 
+                    (keypoint_indices["R_Elbow"], keypoint_indices["R_Wrist"]),
+                    (keypoint_indices["L_Shoulder"], keypoint_indices["L_Hip"]), 
+                    (keypoint_indices["R_Shoulder"], keypoint_indices["R_Hip"]),
+                    (keypoint_indices["L_Hip"], keypoint_indices["R_Hip"]),
+                    (keypoint_indices["L_Hip"], keypoint_indices["L_Knee"]), 
+                    (keypoint_indices["R_Hip"], keypoint_indices["R_Knee"]),
+                    (keypoint_indices["L_Knee"], keypoint_indices["L_Ankle"]), 
+                    (keypoint_indices["R_Knee"], keypoint_indices["R_Ankle"])
+                ]
+                
+                kpts = keypoints[0]
+                # Draw points
+                for i, point in enumerate(kpts):
+                    if point[0] > 0 and point[1] > 0:  # Only draw if keypoint is valid
+                        cv2.circle(annotated_frame, (int(point[0]), int(point[1])), 5, GREEN, -1)
+                        
+                # Draw connections
+                for connection in skeleton_connections:
+                    start_idx, end_idx = connection
+                    if (start_idx < len(kpts) and end_idx < len(kpts) and 
+                        kpts[start_idx][0] > 0 and kpts[start_idx][1] > 0 and 
+                        kpts[end_idx][0] > 0 and kpts[end_idx][1] > 0):
+                        start_point = (int(kpts[start_idx][0]), int(kpts[start_idx][1]))
+                        end_point = (int(kpts[end_idx][0]), int(kpts[end_idx][1]))
+                        cv2.line(annotated_frame, start_point, end_point, YELLOW, 2)
+        except Exception as e:
+            print(f"Error drawing skeleton: {e}")
+        
         out.write(annotated_frame)
     
     # Prepare results dictionary
     results_dict = {
-        "jump_metrics": {
-            "max_relative_jump": float(max(0, max_jump_height)),
-            "max_high_jump": float(max(0, salto_alto)),
-            "peak_power_sayer": float(peak_power_sayer),
-            "repetitions": int(repetition_count),
-            "reference_height": float(reference_height),
-            "body_mass_kg": float(body_mass_kg),
-            "px_per_meter": float(PX_PER_METER) if PX_PER_METER is not None else 0.0
-        },
         "video_analysis": {
-            "input_video": str(input_video),
             "output_video": str(output_video),
-            "fps": float(fps),
-            "resolution": f"{int(width)}x{int(height)}"
         },
         "repetition_data": [
             {
                 "repetition": int(rep["repetition"]),
-                "relative_jump_m": float(rep["relative_jump_m"]),
-                "absolute_jump_m": float(rep["absolute_jump_m"]),
-                "peak_power_watts": float(rep["peak_power_watts"])
+                "distancia_elevada": float(rep["distancia_elevada"]),
+                "salto_alto": float(rep["salto_alto"]),
+                "potencia_sayer": float(rep["potencia_sayer"])
             } for rep in repetition_data
         ]
     }
@@ -420,23 +498,19 @@ def calculate_peak_power_sayer(jump_height_m, body_mass_kg):
     return (60.7 * jump_height_cm) + (45.3 * body_mass_kg) - 2055
 
 
-def calculate_absolute_jump_height(reference_height, relative_jump):
+def calculate_high_jump(player_height:float, max_jump_height:float) -> float:
     """
-    Calculate absolute jump height based on reference height and relative jump.
+    Calculate the high jump height based on the player height and the max jump height.
     
     Args:
-        reference_height: Reference height in meters
-        relative_jump: Relative jump height in meters
+        player_height: Player height in meters
+        max_jump_height: Relative jump height in meters
         
     Returns:
-        Absolute jump height in meters
+        the high jump height in meters
+        
     """
-    absolute_jump = float(reference_height) + float(relative_jump)
-    # Apply validation rule
-    if absolute_jump > 1.72:
-        return absolute_jump
-    else:
-        return 0
+    return player_height + max_jump_height
 
 
 def draw_metrics_overlay(frame, max_jump_height, salto_alto, velocity_vertical, peak_power_sayer,