import cv2 import numpy as np import mediapipe as mp from collections import deque from pathlib import Path import tempfile import os import yt_dlp import json import requests import math # MediaPipe pose landmarks LHIP, RHIP = 23, 24 LKNEE, RKNEE = 25, 26 LANKLE, RANKLE = 27, 28 LSHOULDER, RSHOULDER = 11, 12 NOSE = 0 POSE_CONNECTIONS = mp.solutions.pose.POSE_CONNECTIONS # Jump performance standards (in cm) based on demographics JUMP_STANDARDS = { "Male": { "average": 45, # Average jump height for males "pro": 75 # Professional/elite level for males }, "Female": { "average": 35, # Average jump height for females "pro": 65 # Professional/elite level for females } } def smooth_moving_avg(series, k=5): """Simple causal moving average; ignores None values.""" out = [] q = deque() s = 0.0 cnt = 0 for v in series: if v is not None: q.append(v) s += v cnt += 1 else: q.append(None) if len(q) > k: old = q.popleft() if old is not None: s -= old cnt -= 1 out.append((s / max(cnt, 1)) if cnt > 0 else None) return out def calculate_angle(point1, point2, point3): """Calculate angle between three points (point2 is the vertex).""" # Calculate vectors vec1 = np.array([point1.x - point2.x, point1.y - point2.y]) vec2 = np.array([point3.x - point2.x, point3.y - point2.y]) # Calculate angle using dot product cos_angle = np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2)) cos_angle = np.clip(cos_angle, -1.0, 1.0) # Handle numerical errors angle = np.arccos(cos_angle) return math.degrees(angle) def analyze_knee_strain(landmarks, frame_height): """Analyze knee angles to detect strain and provide recommendations.""" if not landmarks: return {"left_knee": None, "right_knee": None, "strain_detected": False} lms = landmarks.landmark # Calculate knee angles (hip-knee-ankle) left_angle = None right_angle = None strain_detected = False try: # Left knee angle left_angle = calculate_angle(lms[LHIP], lms[LKNEE], lms[LANKLE]) # Right knee angle right_angle = calculate_angle(lms[RHIP], lms[RKNEE], lms[RANKLE]) # Check for strain (angles too acute indicate over-bending) # Healthy knee angle during jumping should be > 90 degrees # Angles < 70 degrees indicate potential strain left_strain = left_angle < 70 if left_angle else False right_strain = right_angle < 70 if right_angle else False strain_detected = left_strain or right_strain except (AttributeError, ZeroDivisionError): pass return { "left_knee": { "angle": left_angle, "strain": left_angle < 70 if left_angle else False, "optimal_angle": 90 # Recommended minimum angle }, "right_knee": { "angle": right_angle, "strain": right_angle < 70 if right_angle else False, "optimal_angle": 90 }, "strain_detected": strain_detected } def get_jump_reference_heights(gender, user_height_cm): """Get average and professional jump height references based on demographics.""" base_avg = JUMP_STANDARDS.get(gender, JUMP_STANDARDS["Male"])["average"] base_pro = JUMP_STANDARDS.get(gender, JUMP_STANDARDS["Male"])["pro"] # Adjust for height (taller people generally jump higher) height_factor = user_height_cm / 175.0 # Normalize to average height avg_height = base_avg * height_factor pro_height = base_pro * height_factor return { "average": avg_height, "professional": pro_height, "gender": gender, "height_adjusted": True } def draw_pose_landmarks(frame, landmarks, knee_analysis=None): """Draw pose landmarks and connections on the frame.""" if not landmarks: return frame h, w, _ = frame.shape # Draw pose connections mp_drawing = mp.solutions.drawing_utils mp_pose = mp.solutions.pose # Draw all pose landmarks mp_drawing.draw_landmarks( frame, landmarks, mp_pose.POSE_CONNECTIONS, mp_drawing.DrawingSpec(color=(0, 255, 0), thickness=2, circle_radius=2), mp_drawing.DrawingSpec(color=(0, 255, 255), thickness=2) ) # Highlight knees with strain indicators if knee_analysis and knee_analysis["strain_detected"]: lms = landmarks.landmark # Left knee if knee_analysis["left_knee"]["strain"]: left_knee_x = int(lms[LKNEE].x * w) left_knee_y = int(lms[LKNEE].y * h) cv2.circle(frame, (left_knee_x, left_knee_y), 8, (0, 0, 255), -1) # Show angle and recommendation angle_text = f"L: {knee_analysis['left_knee']['angle']:.0f}°" cv2.putText(frame, angle_text, (left_knee_x - 30, left_knee_y - 15), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) cv2.putText(frame, "STRAIN!", (left_knee_x - 25, left_knee_y + 25), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 0, 255), 2) # Right knee if knee_analysis["right_knee"]["strain"]: right_knee_x = int(lms[RKNEE].x * w) right_knee_y = int(lms[RKNEE].y * h) cv2.circle(frame, (right_knee_x, right_knee_y), 8, (0, 0, 255), -1) # Show angle and recommendation angle_text = f"R: {knee_analysis['right_knee']['angle']:.0f}°" cv2.putText(frame, angle_text, (right_knee_x + 10, right_knee_y - 15), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2) cv2.putText(frame, "STRAIN!", (right_knee_x + 5, right_knee_y + 25), cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 0, 255), 2) return frame def draw_jump_reference_lines(frame, references, current_jump_height, user_height_cm): """Draw average and professional jump height reference lines.""" h, w, _ = frame.shape # Calculate line positions (relative to frame height) # Assume the person's height spans about 70% of frame height person_height_pixels = int(h * 0.7) pixels_per_cm = person_height_pixels / user_height_cm # Base line (ground level) - bottom 10% of frame ground_y = int(h * 0.9) # Reference lines avg_jump_pixels = int(references["average"] * pixels_per_cm) pro_jump_pixels = int(references["professional"] * pixels_per_cm) current_jump_pixels = int(current_jump_height * pixels_per_cm) avg_line_y = ground_y - avg_jump_pixels pro_line_y = ground_y - pro_jump_pixels current_line_y = ground_y - current_jump_pixels # Draw ground line cv2.line(frame, (0, ground_y), (w, ground_y), (100, 100, 100), 2) cv2.putText(frame, "Ground", (10, ground_y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (100, 100, 100), 2) # Draw average line if avg_line_y > 0: cv2.line(frame, (0, avg_line_y), (w, avg_line_y), (255, 255, 0), 2) cv2.putText(frame, f"Avg: {references['average']:.0f}cm", (10, avg_line_y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 0), 2) # Draw professional line if pro_line_y > 0: cv2.line(frame, (0, pro_line_y), (w, pro_line_y), (0, 255, 0), 2) cv2.putText(frame, f"Pro: {references['professional']:.0f}cm", (10, pro_line_y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 0), 2) # Draw current jump line if current_line_y > 0 and current_jump_height > 0: cv2.line(frame, (0, current_line_y), (w, current_line_y), (0, 0, 255), 3) cv2.putText(frame, f"Your Jump: {current_jump_height:.0f}cm", (w - 200, current_line_y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2) return frame def calculate_peak_power_output(jump_height_m, body_mass_kg, flight_time_s): """Calculate peak power output using biomechanical models.""" if jump_height_m <= 0 or flight_time_s <= 0: return None # Using the equation: Power = (body_mass * gravity * jump_height) / flight_time # This is a simplified model - in reality, peak power occurs during takeoff phase gravity = 9.81 # m/s² # Average power during flight avg_power = (body_mass_kg * gravity * jump_height_m) / (flight_time_s / 2) # Peak power is typically 2-3x average power during explosive movements peak_power = avg_power * 2.5 return peak_power # Watts def calculate_rate_of_force_development(hip_y_series, fps, takeoff_start_idx, takeoff_end_idx): """Calculate Rate of Force Development from hip trajectory.""" if takeoff_end_idx <= takeoff_start_idx or len(hip_y_series) <= takeoff_end_idx: return None # Extract takeoff phase takeoff_phase = hip_y_series[takeoff_start_idx:takeoff_end_idx + 1] # Calculate velocity and acceleration dt = 1.0 / fps velocities = np.diff(takeoff_phase) / dt accelerations = np.diff(velocities) / dt if len(accelerations) == 0: return None # RFD is the maximum rate of change of force (approximated by acceleration) # Convert to relative units (normalized by body position change) max_acceleration = np.max(np.abs(accelerations)) # Normalize to get RFD index (higher values indicate faster force development) rfd_index = max_acceleration * fps # per second return rfd_index def detect_ground_contact_phases(hip_y_series, fps, threshold_velocity=0.01): """Detect ground contact and flight phases from hip trajectory.""" if len(hip_y_series) < 5: return [], [] # Calculate vertical velocity velocities = np.diff(hip_y_series) # Smooth velocities velocities = smooth_moving_avg(list(velocities), k=3) # Find phases where velocity is near zero (ground contact) ground_contact_frames = [] flight_frames = [] for i, vel in enumerate(velocities): if vel is not None and abs(vel) < threshold_velocity: ground_contact_frames.append(i) elif vel is not None: flight_frames.append(i) # Calculate ground contact time if ground_contact_frames: total_contact_frames = len(ground_contact_frames) ground_contact_time = total_contact_frames / fps else: ground_contact_time = 0.0 return ground_contact_frames, ground_contact_time def calculate_impulse_and_force(hip_y_series, fps, body_mass_kg, takeoff_start_idx, takeoff_end_idx): """Calculate impulse and force characteristics during takeoff.""" if takeoff_end_idx <= takeoff_start_idx or len(hip_y_series) <= takeoff_end_idx: return None, None # Extract takeoff phase takeoff_phase = hip_y_series[takeoff_start_idx:takeoff_end_idx + 1] dt = 1.0 / fps # Calculate velocity and acceleration velocities = np.diff(takeoff_phase) / dt accelerations = np.diff(velocities) / dt if len(accelerations) == 0: return None, None # Estimate force (F = ma, where a includes gravity) gravity = 9.81 forces = body_mass_kg * (np.array(accelerations) + gravity) # Calculate impulse (area under force-time curve) impulse = np.trapz(forces, dx=dt) # Peak force peak_force = np.max(forces) if len(forces) > 0 else None return impulse, peak_force def estimate_jump_metrics(hip_y_series, fps, body_mass_kg=75.0): """Return comprehensive jump metrics including advanced biomechanical parameters.""" # Remove None values hip_clean = [(i, h) for i, h in enumerate(hip_y_series) if h is not None] if len(hip_clean) < 5: return None indices, hip_values = zip(*hip_clean) hip = list(hip_values) # Smooth the data hip_smooth = smooth_moving_avg(hip, k=5) # Basic jump metrics min_y = min(hip_smooth) # apex (body highest) max_y = max(hip_smooth) # deepest crouch (body lowest) jump_height_norm = max(0.0, (max_y - min_y)) # Find key phase indices hip_arr = np.array(hip_smooth, dtype=float) vel = np.diff(hip_arr) if vel.size == 0: return None # Identify takeoff and landing phases takeoff_idx = int(np.argmin(vel)) # most negative velocity (takeoff) landing_idx = int(np.argmax(vel)) # most positive velocity (landing) # Flight time flight_frames = max(0, landing_idx - takeoff_idx) flight_time_s = flight_frames / float(fps or 30.0) # Find takeoff phase (from crouch to takeoff) crouch_idx = int(np.argmax(hip_smooth)) # deepest crouch takeoff_start_idx = max(0, crouch_idx - 10) # start of takeoff phase takeoff_end_idx = takeoff_idx # Advanced metrics jump_height_m = jump_height_norm * 2.0 # Rough conversion to meters # Peak Power Output peak_power = calculate_peak_power_output(jump_height_m, body_mass_kg, flight_time_s) # Rate of Force Development rfd = calculate_rate_of_force_development(hip_smooth, fps, takeoff_start_idx, takeoff_end_idx) # Ground Contact Time ground_contact_frames, ground_contact_time = detect_ground_contact_phases(hip_smooth, fps) # Impulse and Force impulse, peak_force = calculate_impulse_and_force(hip_smooth, fps, body_mass_kg, takeoff_start_idx, takeoff_end_idx) return { 'jump_height_norm': jump_height_norm, 'flight_time_s': flight_time_s, 'peak_power_watts': peak_power, 'rate_of_force_development': rfd, 'ground_contact_time_s': ground_contact_time, 'impulse_ns': impulse, 'peak_force_n': peak_force, 'takeoff_phase_duration_s': (takeoff_end_idx - takeoff_start_idx) / fps if takeoff_end_idx > takeoff_start_idx else 0 } def download_youtube_video(youtube_url, output_path): """Download YouTube video to specified path.""" ydl_opts = { 'format': 'best[height<=720]', # Limit quality for faster processing 'outtmpl': output_path, 'quiet': True, 'no_warnings': True, } with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([youtube_url]) return output_path def generate_annotated_video(video_path, user_height_cm, user_weight_kg, gender, output_path=None, progress_callback=None): """Generate annotated video with pose tracking, jump analysis, and knee strain detection.""" # Set up output path if output_path is None: video_name = Path(video_path).stem output_path = f"{video_name}_annotated.mp4" cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise Exception(f"Could not open video: {video_path}") # Get video properties w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = cap.get(cv2.CAP_PROP_FPS) or 30.0 total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) # Set up video writer fourcc = cv2.VideoWriter_fourcc(*'mp4v') out = cv2.VideoWriter(output_path, fourcc, fps, (w, h)) # Set up pose detection mp_pose = mp.solutions.pose pose = mp_pose.Pose(static_image_mode=False, model_complexity=1, enable_segmentation=False) # Get jump references jump_references = get_jump_reference_heights(gender, user_height_cm) # Track hip positions for jump height calculation hip_y_series = [] frame_idx = 0 print(f"Generating annotated video: {output_path}") print(f"Video dimensions: {w}x{h}, FPS: {fps}, Total frames: {total_frames}") while True: ret, frame = cap.read() if not ret: break # Process frame for pose detection rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) results = pose.process(rgb) # Track hip position hip_y = None if results.pose_landmarks: lms = results.pose_landmarks.landmark hip_y = (lms[LHIP].y + lms[RHIP].y) / 2.0 hip_y_series.append(hip_y) else: hip_y_series.append(None) # Calculate current jump height (rough estimate) current_jump_height = 0 if len(hip_y_series) > 10: # Need some history recent_hips = [h for h in hip_y_series[-20:] if h is not None] if recent_hips: min_hip = min(recent_hips) max_hip = max(recent_hips) normalized_jump = max_hip - min_hip current_jump_height = normalized_jump * user_height_cm # Analyze knee strain knee_analysis = analyze_knee_strain(results.pose_landmarks, h) # Draw pose landmarks with strain indicators annotated_frame = draw_pose_landmarks(frame, results.pose_landmarks, knee_analysis) # Draw jump reference lines annotated_frame = draw_jump_reference_lines( annotated_frame, jump_references, current_jump_height, user_height_cm ) # Add performance info overlay info_y = 30 cv2.putText(annotated_frame, f"Frame: {frame_idx}/{total_frames}", (10, info_y), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2) if current_jump_height > 0: cv2.putText(annotated_frame, f"Current Jump: {current_jump_height:.1f}cm", (10, info_y + 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2) # Add knee strain warnings if knee_analysis["strain_detected"]: warning_text = "⚠️ KNEE STRAIN DETECTED!" cv2.putText(annotated_frame, warning_text, (10, h - 60), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2) recommendations = "Keep knees above 90° angle" cv2.putText(annotated_frame, recommendations, (10, h - 30), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2) # Write frame to output video out.write(annotated_frame) frame_idx += 1 # Update progress if progress_callback and total_frames > 0: progress = min(frame_idx / total_frames, 1.0) progress_callback(progress, f"Processing frame {frame_idx}/{total_frames}") # Cleanup cap.release() out.release() # Calculate final jump metrics jump_metrics = estimate_jump_metrics(hip_y_series, fps, user_weight_kg) print(f"Annotated video saved: {output_path}") return { "output_video_path": output_path, "jump_metrics": jump_metrics, "jump_references": jump_references, "total_frames_processed": frame_idx, "knee_strain_detected": any(analyze_knee_strain(None, h)["strain_detected"] for _ in range(5)) # Simplified check } def process_video_analysis(video_path, user_height_cm, user_weight_kg=75.0, progress_callback=None): """Core video analysis function with progress tracking. Args: video_path (str): Path to the video file user_height_cm (float): User's height in centimeters user_weight_kg (float): User's weight in kilograms (default: 75kg) progress_callback (callable, optional): Function to call with progress updates Signature: progress_callback(progress_float, description_string) Returns: dict: Analysis results containing jump metrics and video info None: If analysis failed """ cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise Exception(f"Could not open video: {video_path}") w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) fps = cap.get(cv2.CAP_PROP_FPS) or 30.0 total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) mp_pose = mp.solutions.pose pose = mp_pose.Pose(static_image_mode=False, model_complexity=1, enable_segmentation=False) hip_y_series = [] frame_idx = 0 print(f"Processing video: {Path(video_path).name}") print(f"Video dimensions: {w}x{h}, FPS: {fps}, Total frames: {total_frames}") ok, frame = cap.read() while ok: rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) res = pose.process(rgb) if res.pose_landmarks: lms = res.pose_landmarks.landmark mid_hip_y = (lms[LHIP].y + lms[RHIP].y) / 2.0 hip_y_series.append(float(mid_hip_y)) else: hip_y_series.append(None) frame_idx += 1 # Update progress if progress_callback and total_frames > 0: progress = min(frame_idx / total_frames, 1.0) progress_callback(progress, f"Processing frame {frame_idx}/{total_frames}") ok, frame = cap.read() cap.release() print(f"Completed processing {frame_idx} frames") # Calculate comprehensive jump metrics metrics = estimate_jump_metrics(hip_y_series, fps, user_weight_kg) if metrics is None: return None # Convert normalized jump height to actual height in cm jump_height_cm = metrics['jump_height_norm'] * user_height_cm return { "video": Path(video_path).name, "frames": len(hip_y_series), "fps": fps, "jump_height_cm": jump_height_cm, "normalized_rise": metrics['jump_height_norm'], "flight_time_s": metrics['flight_time_s'], "peak_power_watts": metrics['peak_power_watts'], "rate_of_force_development": metrics['rate_of_force_development'], "ground_contact_time_s": metrics['ground_contact_time_s'], "impulse_ns": metrics['impulse_ns'], "peak_force_n": metrics['peak_force_n'], "takeoff_phase_duration_s": metrics['takeoff_phase_duration_s'], "user_weight_kg": user_weight_kg } def analyze_youtube_video(youtube_url, user_height_cm, user_weight_kg=75.0, progress_callback=None): """Analyze jump from YouTube video. Args: youtube_url (str): YouTube video URL user_height_cm (float): User's height in centimeters user_weight_kg (float): User's weight in kilograms (default: 75kg) progress_callback (callable, optional): Function to call with progress updates Returns: dict: Analysis results or error information """ # Validate inputs if not youtube_url or not youtube_url.strip(): return {"error": "Please provide a YouTube URL"} if not user_height_cm or user_height_cm <= 0: return {"error": "Please provide a valid height in centimeters"} try: if progress_callback: progress_callback(0.1, "Validating YouTube URL...") # Validate YouTube URL youtube_url = youtube_url.strip() if not any(domain in youtube_url for domain in ['youtube.com', 'youtu.be']): return {"error": "Please provide a valid YouTube URL"} # Create temporary directory for processing with tempfile.TemporaryDirectory() as temp_dir: if progress_callback: progress_callback(0.2, "Downloading video from YouTube...") # Download video video_filename = os.path.join(temp_dir, 'video.%(ext)s') try: download_youtube_video(youtube_url, video_filename) # Find the actual downloaded file video_files = [f for f in os.listdir(temp_dir) if f.startswith('video.')] if not video_files: return {"error": "Failed to download YouTube video. Please check the URL and try again."} video_path = os.path.join(temp_dir, video_files[0]) except Exception as e: return {"error": f"Failed to download YouTube video: {str(e)}"} if progress_callback: progress_callback(0.3, "Starting video analysis...") # Process the video with progress tracking def update_progress(prog, desc): if progress_callback: progress_callback(0.3 + (prog * 0.6), desc) result = process_video_analysis(video_path, user_height_cm, user_weight_kg, update_progress) if progress_callback: progress_callback(0.9, "Analysis complete!") return result except Exception as e: return {"error": f"Error during analysis: {str(e)}"} def analyze_video_file(video_path, user_height_cm, user_weight_kg=75.0, progress_callback=None): """Analyze jump from video file. Args: video_path (str): Path to video file user_height_cm (float): User's height in centimeters user_weight_kg (float): User's weight in kilograms (default: 75kg) progress_callback (callable, optional): Function to call with progress updates Returns: dict: Analysis results or error information """ # Validate inputs if not video_path: return {"error": "Please provide a video file"} if not user_height_cm or user_height_cm <= 0: return {"error": "Please provide a valid height in centimeters"} try: if progress_callback: progress_callback(0.1, "Processing video file...") # Process the video with progress tracking def update_progress(prog, desc): if progress_callback: progress_callback(0.1 + (prog * 0.8), desc) result = process_video_analysis(video_path, user_height_cm, user_weight_kg, update_progress) if progress_callback: progress_callback(1.0, "Analysis complete!") return result except Exception as e: return {"error": f"Error during analysis: {str(e)}"} def generate_annotated_video_from_youtube(youtube_url, user_height_cm, user_weight_kg, gender, progress_callback=None): """Generate annotated video from YouTube URL.""" # Validate inputs if not youtube_url or not youtube_url.strip(): return {"error": "Please provide a YouTube URL"} if not user_height_cm or user_height_cm <= 0: return {"error": "Please provide a valid height in centimeters"} try: if progress_callback: progress_callback(0.1, "Downloading YouTube video...") # Validate YouTube URL youtube_url = youtube_url.strip() if not any(domain in youtube_url for domain in ['youtube.com', 'youtu.be']): return {"error": "Please provide a valid YouTube URL"} # Create temporary directory for processing with tempfile.TemporaryDirectory() as temp_dir: if progress_callback: progress_callback(0.2, "Downloading video from YouTube...") # Download video video_filename = os.path.join(temp_dir, 'video.%(ext)s') try: download_youtube_video(youtube_url, video_filename) # Find the actual downloaded file video_files = [f for f in os.listdir(temp_dir) if f.startswith('video.')] if not video_files: return {"error": "Failed to download YouTube video. Please check the URL and try again."} video_path = os.path.join(temp_dir, video_files[0]) except Exception as e: return {"error": f"Failed to download YouTube video: {str(e)}"} if progress_callback: progress_callback(0.3, "Generating annotated video...") # Generate output path in temp directory output_path = os.path.join(temp_dir, "annotated_output.mp4") # Process the video with progress tracking def update_progress(prog, desc): if progress_callback: progress_callback(0.3 + (prog * 0.6), desc) result = generate_annotated_video( video_path, user_height_cm, user_weight_kg, gender, output_path, update_progress ) if progress_callback: progress_callback(0.95, "Finalizing annotated video...") # Move the output file to a permanent location final_output = f"annotated_jump_analysis_{Path(youtube_url).stem}.mp4" if os.path.exists(output_path): # In production, you'd save this to a proper storage location result["output_video_path"] = output_path result["download_ready"] = True if progress_callback: progress_callback(1.0, "Annotated video generation complete!") return result except Exception as e: return {"error": f"Error during video generation: {str(e)}"} def generate_annotated_video_from_file(video_file_path, user_height_cm, user_weight_kg, gender, progress_callback=None): """Generate annotated video from uploaded file.""" # Validate inputs if not video_file_path: return {"error": "Please provide a video file"} if not user_height_cm or user_height_cm <= 0: return {"error": "Please provide a valid height in centimeters"} try: if progress_callback: progress_callback(0.1, "Processing uploaded video...") # Generate output path video_name = Path(video_file_path).stem output_path = f"{video_name}_annotated.mp4" # Process the video with progress tracking def update_progress(prog, desc): if progress_callback: progress_callback(0.1 + (prog * 0.8), desc) result = generate_annotated_video( video_file_path, user_height_cm, user_weight_kg, gender, output_path, update_progress ) if progress_callback: progress_callback(1.0, "Annotated video generation complete!") return result except Exception as e: return {"error": f"Error during video generation: {str(e)}"} def get_performance_insights(result_dict): """Generate performance insights based on comprehensive jump metrics. Args: result_dict (dict): Dictionary containing all jump analysis results Returns: list: List of insight strings """ insights = [] jump_height_cm = result_dict.get('jump_height_cm', 0) flight_time_s = result_dict.get('flight_time_s', 0) peak_power_watts = result_dict.get('peak_power_watts') rfd = result_dict.get('rate_of_force_development') ground_contact_time = result_dict.get('ground_contact_time_s') peak_force = result_dict.get('peak_force_n') # Jump height insights if jump_height_cm > 60: insights.append("🔥 **Excellent jump height!** This is above average performance.") elif jump_height_cm > 40: insights.append("👍 **Good jump height!** Solid athletic performance.") elif jump_height_cm > 25: insights.append("📈 **Moderate jump height.** Room for improvement with training.") else: insights.append("🎯 **Starting point identified.** Focus on technique and strength training.") # Flight time insights if flight_time_s > 0.5: insights.append("⏱️ **Great flight time!** Shows good explosive power.") elif flight_time_s > 0.3: insights.append("⏱️ **Decent flight time.** Good coordination.") # Peak power insights if peak_power_watts and peak_power_watts > 3000: insights.append("⚡ **Outstanding power output!** Elite-level explosive strength.") elif peak_power_watts and peak_power_watts > 2000: insights.append("💪 **High power output!** Strong explosive capabilities.") elif peak_power_watts and peak_power_watts > 1000: insights.append("🏋️ **Moderate power output.** Good base strength to build on.") # Rate of Force Development insights if rfd and rfd > 15: insights.append("🚀 **Excellent RFD!** Very quick force generation - great for sprinting and jumping.") elif rfd and rfd > 10: insights.append("⚡ **Good RFD!** Solid ability to generate force quickly.") elif rfd and rfd > 5: insights.append("📊 **Moderate RFD.** Work on plyometric training to improve explosiveness.") # Ground contact time insights if ground_contact_time and ground_contact_time < 0.2: insights.append("🦘 **Excellent ground contact time!** Very efficient stretch-shortening cycle.") elif ground_contact_time and ground_contact_time < 0.3: insights.append("👟 **Good ground contact efficiency.** Solid reactive strength.") # Peak force insights if peak_force and peak_force > 2000: insights.append("🏋️‍♂️ **High peak force production!** Strong neuromuscular system.") elif peak_force and peak_force > 1500: insights.append("💪 **Good force production.** Solid strength foundation.") return insights def test_gemini_api_connection(api_key): """Test the Gemini API connection with a simple request.""" if not api_key: return {"error": "No API key provided"} url = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-pro:generateContent" headers = { 'Content-Type': 'application/json', 'X-goog-api-key': api_key.strip() } # Simple test data test_data = { "contents": [ { "parts": [ { "text": "Say hello in exactly 5 words." } ] } ], "generationConfig": { "temperature": 0.1, "maxOutputTokens": 20 } } try: response = requests.post(url, headers=headers, json=test_data, timeout=10) return { "status_code": response.status_code, "response_text": response.text[:500], "success": response.status_code == 200, "error": None if response.status_code == 200 else f"HTTP {response.status_code}" } except Exception as e: return { "status_code": None, "response_text": str(e), "success": False, "error": str(e) } def get_ai_sports_coaching_analysis(jump_height_cm, user_height_cm, gender, favorite_sports, peak_power_watts=None, flight_time_s=None, rfd=None, api_key=None): """Get AI-powered sports coaching analysis using Google Gemini API. Args: jump_height_cm (float): Jump height in centimeters user_height_cm (float): User's height in centimeters gender (str): User's gender ('Male' or 'Female') favorite_sports (list): List of user's favorite sports peak_power_watts (float, optional): Peak power output in watts flight_time_s (float, optional): Flight time in seconds rfd (float, optional): Rate of force development api_key (str): Google Gemini API key Returns: dict: AI analysis with percentile rankings and improvement suggestions """ if not api_key: return {"error": "Gemini API key is required"} # Calculate relative jump height relative_jump_height = (jump_height_cm / user_height_cm) * 100 if user_height_cm > 0 else 0 # Prepare the analysis prompt # Handle None values safely flight_time_str = f"{flight_time_s:.3f}" if flight_time_s is not None else "N/A" peak_power_str = f"{peak_power_watts:.0f}" if peak_power_watts is not None else "N/A" rfd_str = f"{rfd:.2f}" if rfd is not None else "N/A" # Format favorite sports list sports_list = ", ".join(favorite_sports) if favorite_sports else "None specified" prompt = f"""You are a sports coach. Based on this athlete's jump performance, provide analysis in JSON format. ATHLETE DATA: - Gender: {gender} - Height: {user_height_cm} cm - Jump Height: {jump_height_cm:.2f} cm - Sports: {sports_list} RESPOND WITH VALID JSON IN THIS EXACT FORMAT: {{ "sports": {{ "Basketball": 75, "Volleyball": 80 }}, "tips": [ "Focus on proper landing technique to reduce knee strain", "Add plyometric exercises to your training routine", "Strengthen your leg muscles with squats and lunges", "Practice explosive movements for better power development" ] }} RULES: - "sports" should contain percentile rankings (0-100) only for sports in their list: {sports_list} - "tips" must be exactly 4 practical improvement suggestions - Use simple English for beginner athletes - Return ONLY valid JSON, no extra text, no markdown formatting, no code blocks""" # Prepare the API request - try Gemini 1.5 Pro as fallback url = "https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-pro:generateContent" headers = { 'Content-Type': 'application/json', 'X-goog-api-key': api_key } data = { "contents": [ { "parts": [ { "text": prompt } ] } ], "generationConfig": { "temperature": 0.7, "topK": 40, "topP": 0.95, "maxOutputTokens": 1024 } } try: response = requests.post(url, headers=headers, json=data, timeout=30) # Enhanced error handling for debugging if response.status_code == 403: return { "error": f"API key authentication failed (403). Please verify:\n" f"1. API key is correct and active\n" f"2. Generative AI API is enabled in Google Cloud Console\n" f"3. Billing is set up for your Google Cloud project\n" f"4. API key has proper permissions\n" f"Response: {response.text[:200]}..." } elif response.status_code == 429: return {"error": "Rate limit exceeded. Please try again later."} elif response.status_code == 400: return {"error": f"Bad request (400). Response: {response.text[:200]}..."} response.raise_for_status() result = response.json() if 'candidates' in result and len(result['candidates']) > 0: ai_response_text = result['candidates'][0]['content']['parts'][0]['text'] try: # Clean the response text - remove markdown code blocks if present cleaned_response = ai_response_text.strip() # Remove ```json and ``` markers if they exist if cleaned_response.startswith('```json'): cleaned_response = cleaned_response[7:] # Remove ```json if cleaned_response.startswith('```'): cleaned_response = cleaned_response[3:] # Remove ``` if cleaned_response.endswith('```'): cleaned_response = cleaned_response[:-3] # Remove trailing ``` cleaned_response = cleaned_response.strip() # Parse the JSON response from AI ai_analysis_json = json.loads(cleaned_response) # Validate the expected structure if not isinstance(ai_analysis_json, dict): raise ValueError("Response is not a JSON object") # Extract required fields with defaults sports_percentiles = ai_analysis_json.get('sports', {}) tips = ai_analysis_json.get('tips', []) # Ensure tips is a list if not isinstance(tips, list): tips = [] # Ensure sports is a dict if not isinstance(sports_percentiles, dict): sports_percentiles = {} return { "success": True, "analysis": { "sports": sports_percentiles, "tips": tips }, "athlete_profile": { "gender": gender, "height_cm": user_height_cm, "jump_height_cm": jump_height_cm, "relative_jump_height": relative_jump_height, "flight_time_s": flight_time_s, "peak_power_watts": peak_power_watts, "rfd": rfd } } except json.JSONDecodeError as e: # Fallback: if JSON parsing fails, try to extract some information return { "error": f"AI returned invalid JSON. Raw response: {ai_response_text[:200]}...", "parsing_error": str(e) } except Exception as e: return { "error": f"Error processing AI response: {str(e)}", "raw_response": ai_response_text[:200] } else: return {"error": f"No response generated from AI. Response: {result}"} except requests.exceptions.RequestException as e: return {"error": f"API request failed: {str(e)}"} except json.JSONDecodeError as e: return {"error": f"Failed to parse API response: {str(e)}"} except Exception as e: return {"error": f"Unexpected error: {str(e)}"}