Files changed (2) hide show
  1. app.py +107 -261
  2. athletic_performance.py +430 -0
app.py CHANGED
@@ -1,186 +1,26 @@
1
- import cv2
2
- import numpy as np
3
- import mediapipe as mp
4
- from collections import deque
5
- from pathlib import Path
6
- import json
7
- import tempfile
8
- import os
9
- import yt_dlp
10
  import gradio as gr
11
  import pandas as pd
12
-
13
- LHIP, RHIP = 23, 24
14
- POSE_CONNECTIONS = mp.solutions.pose.POSE_CONNECTIONS
15
-
16
- def smooth_moving_avg(series, k=5):
17
- """Simple causal moving average; ignores None values."""
18
- out = []
19
- q = deque()
20
- s = 0.0
21
- cnt = 0
22
- for v in series:
23
- if v is not None:
24
- q.append(v); s += v; cnt += 1
25
- else:
26
- q.append(None)
27
- if len(q) > k:
28
- old = q.popleft()
29
- if old is not None:
30
- s -= old; cnt -= 1
31
- out.append((s / max(cnt, 1)) if cnt > 0 else None)
32
- return out
33
-
34
- def estimate_jump_metrics(hip_y_series, fps):
35
- """Return jump_height_norm (0..1), flight_time_s using hip trajectory."""
36
- # Remove None
37
- hip = [h for h in hip_y_series if h is not None]
38
- if len(hip) < 3:
39
- return None, None
40
-
41
- # Smooth
42
- hip = smooth_moving_avg(hip, k=5)
43
-
44
- # Jump height (normalized): deepest crouch (max y) to apex (min y)
45
- min_y = min(hip) # apex (body highest)
46
- max_y = max(hip) # deepest crouch (body lowest)
47
- jump_height_norm = max(0.0, (max_y - min_y))
48
-
49
- # Flight time heuristic using vertical velocity pattern
50
- hip_arr = np.array(hip, dtype=float)
51
- vel = np.diff(hip_arr)
52
- if vel.size == 0:
53
- flight_time_s = 0.0
54
- else:
55
- takeoff_idx = int(np.argmin(vel)) # most negative velocity
56
- landing_idx = int(np.argmax(vel)) # most positive velocity
57
- flight_frames = max(0, landing_idx - takeoff_idx)
58
- flight_time_s = flight_frames / float(fps or 30.0)
59
-
60
- return jump_height_norm, flight_time_s
61
-
62
- def download_youtube_video(youtube_url, output_path):
63
- """Download YouTube video to specified path."""
64
- ydl_opts = {
65
- 'format': 'best[height<=720]', # Limit quality for faster processing
66
- 'outtmpl': output_path,
67
- 'quiet': True,
68
- 'no_warnings': True,
69
- }
70
-
71
- with yt_dlp.YoutubeDL(ydl_opts) as ydl:
72
- ydl.download([youtube_url])
73
- return output_path
74
-
75
- def process_video_analysis(video_path, user_height_cm, progress_callback=None):
76
- """Core video analysis function with progress tracking."""
77
- cap = cv2.VideoCapture(video_path)
78
- if not cap.isOpened():
79
- raise Exception(f"Could not open video: {video_path}")
80
-
81
- w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
82
- h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
83
- fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
84
- total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
85
-
86
- mp_pose = mp.solutions.pose
87
- pose = mp_pose.Pose(static_image_mode=False, model_complexity=1, enable_segmentation=False)
88
-
89
- hip_y_series = []
90
- frame_idx = 0
91
-
92
- print(f"Processing video: {Path(video_path).name}")
93
- print(f"Video dimensions: {w}x{h}, FPS: {fps}, Total frames: {total_frames}")
94
-
95
- ok, frame = cap.read()
96
- while ok:
97
- rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
98
- res = pose.process(rgb)
99
-
100
- if res.pose_landmarks:
101
- lms = res.pose_landmarks.landmark
102
- mid_hip_y = (lms[LHIP].y + lms[RHIP].y) / 2.0
103
- hip_y_series.append(float(mid_hip_y))
104
- else:
105
- hip_y_series.append(None)
106
-
107
- frame_idx += 1
108
-
109
- # Update progress
110
- if progress_callback and total_frames > 0:
111
- progress = min(frame_idx / total_frames, 1.0)
112
- progress_callback(progress, f"Processing frame {frame_idx}/{total_frames}")
113
-
114
- ok, frame = cap.read()
115
-
116
- cap.release()
117
- print(f"Completed processing {frame_idx} frames")
118
-
119
- jump_norm, flight_time_s = estimate_jump_metrics(hip_y_series, fps)
120
-
121
- if jump_norm is None:
122
- return None
123
-
124
- jump_height_cm = jump_norm * user_height_cm
125
-
126
- return {
127
- "video": Path(video_path).name,
128
- "frames": len(hip_y_series),
129
- "fps": fps,
130
- "jump_height_cm": jump_height_cm,
131
- "normalized_rise": jump_norm,
132
- "flight_time_s": flight_time_s
133
- }
134
 
135
  def analyze_jump_from_youtube(youtube_url, user_height_cm, progress=gr.Progress()):
136
  """Main analysis function for Gradio interface."""
137
 
138
- # Validate inputs
139
- if not youtube_url or not youtube_url.strip():
140
- return "❌ Please provide a YouTube URL", None, None
141
 
142
- if not user_height_cm or user_height_cm <= 0:
143
- return "❌ Please provide a valid height in centimeters", None, None
144
 
145
- try:
146
- progress(0.1, desc="Validating YouTube URL...")
147
-
148
- # Validate YouTube URL
149
- youtube_url = youtube_url.strip()
150
- if not any(domain in youtube_url for domain in ['youtube.com', 'youtu.be']):
151
- return "❌ Please provide a valid YouTube URL", None, None
152
-
153
- # Create temporary directory for processing
154
- with tempfile.TemporaryDirectory() as temp_dir:
155
- progress(0.2, desc="Downloading video from YouTube...")
156
-
157
- # Download video
158
- video_filename = os.path.join(temp_dir, 'video.%(ext)s')
159
- try:
160
- download_youtube_video(youtube_url, video_filename)
161
- # Find the actual downloaded file
162
- video_files = [f for f in os.listdir(temp_dir) if f.startswith('video.')]
163
- if not video_files:
164
- return "❌ Failed to download YouTube video. Please check the URL and try again.", None, None
165
- video_path = os.path.join(temp_dir, video_files[0])
166
- except Exception as e:
167
- return f"❌ Failed to download YouTube video: {str(e)}", None, None
168
-
169
- progress(0.3, desc="Starting video analysis...")
170
-
171
- # Process the video with progress tracking
172
- def update_progress(prog, desc):
173
- progress(0.3 + (prog * 0.6), desc=desc)
174
-
175
- result = process_video_analysis(video_path, user_height_cm, update_progress)
176
-
177
- progress(0.9, desc="Generating results...")
178
-
179
- if result is None:
180
- return "⚠️ Could not analyze jump. Make sure the video shows a person clearly performing a vertical jump.", None, None
181
-
182
- # Format results for display
183
- results_text = f"""
184
  ## πŸŽ‰ Jump Analysis Results
185
 
186
  ### πŸ“Š Performance Metrics
@@ -195,64 +35,54 @@ def analyze_jump_from_youtube(youtube_url, user_height_cm, progress=gr.Progress(
195
 
196
  ### πŸ“ˆ Performance Insights
197
  """
198
-
199
- # Add performance insights
200
- if result['jump_height_cm'] > 60:
201
- results_text += "πŸ”₯ **Excellent jump height!** This is above average performance.\n"
202
- elif result['jump_height_cm'] > 40:
203
- results_text += "πŸ‘ **Good jump height!** Solid athletic performance.\n"
204
- elif result['jump_height_cm'] > 25:
205
- results_text += "πŸ“ˆ **Moderate jump height.** Room for improvement with training.\n"
206
- else:
207
- results_text += "🎯 **Starting point identified.** Focus on technique and strength training.\n"
208
-
209
- if result['flight_time_s'] > 0.5:
210
- results_text += "⏱️ **Great flight time!** Shows good explosive power.\n"
211
- elif result['flight_time_s'] > 0.3:
212
- results_text += "⏱️ **Decent flight time.** Good coordination.\n"
213
-
214
- # Create a results dataframe for the table
215
- results_df = pd.DataFrame([
216
- ["Jump Height", f"{result['jump_height_cm']:.2f} cm"],
217
- ["Flight Time", f"{result['flight_time_s']:.3f} seconds"],
218
- ["Normalized Rise", f"{result['normalized_rise']*100:.1f}%"],
219
- ["Video Frames", f"{result['frames']}"],
220
- ["Frame Rate", f"{result['fps']:.2f} FPS"],
221
- ], columns=["Metric", "Value"])
222
-
223
- progress(1.0, desc="Analysis complete!")
224
-
225
- return results_text, results_df, "βœ… Analysis completed successfully!"
226
-
227
- except Exception as e:
228
- return f"❌ Error during analysis: {str(e)}", None, None
229
 
230
  def analyze_jump_from_file(video_file, user_height_cm, progress=gr.Progress()):
231
  """Analysis function for uploaded video files."""
232
 
233
- # Validate inputs
234
- if video_file is None:
235
- return "❌ Please upload a video file", None, None
236
 
237
- if not user_height_cm or user_height_cm <= 0:
238
- return "❌ Please provide a valid height in centimeters", None, None
 
239
 
240
- try:
241
- progress(0.1, desc="Processing uploaded video...")
242
-
243
- # Process the video with progress tracking
244
- def update_progress(prog, desc):
245
- progress(0.1 + (prog * 0.8), desc=desc)
246
-
247
- result = process_video_analysis(video_file.name, user_height_cm, update_progress)
248
-
249
- progress(0.9, desc="Generating results...")
250
-
251
- if result is None:
252
- return "⚠️ Could not analyze jump. Make sure the video shows a person clearly performing a vertical jump.", None, None
253
-
254
- # Format results (same as YouTube function)
255
- results_text = f"""
256
  ## πŸŽ‰ Jump Analysis Results
257
 
258
  ### πŸ“Š Performance Metrics
@@ -267,37 +97,33 @@ def analyze_jump_from_file(video_file, user_height_cm, progress=gr.Progress()):
267
 
268
  ### πŸ“ˆ Performance Insights
269
  """
270
-
271
- # Add performance insights
272
- if result['jump_height_cm'] > 60:
273
- results_text += "πŸ”₯ **Excellent jump height!** This is above average performance.\n"
274
- elif result['jump_height_cm'] > 40:
275
- results_text += "πŸ‘ **Good jump height!** Solid athletic performance.\n"
276
- elif result['jump_height_cm'] > 25:
277
- results_text += "πŸ“ˆ **Moderate jump height.** Room for improvement with training.\n"
278
- else:
279
- results_text += "🎯 **Starting point identified.** Focus on technique and strength training.\n"
280
-
281
- if result['flight_time_s'] > 0.5:
282
- results_text += "⏱️ **Great flight time!** Shows good explosive power.\n"
283
- elif result['flight_time_s'] > 0.3:
284
- results_text += "⏱️ **Decent flight time.** Good coordination.\n"
285
-
286
- # Create a results dataframe for the table
287
- results_df = pd.DataFrame([
288
- ["Jump Height", f"{result['jump_height_cm']:.2f} cm"],
289
- ["Flight Time", f"{result['flight_time_s']:.3f} seconds"],
290
- ["Normalized Rise", f"{result['normalized_rise']*100:.1f}%"],
291
- ["Video Frames", f"{result['frames']}"],
292
- ["Frame Rate", f"{result['fps']:.2f} FPS"],
293
- ], columns=["Metric", "Value"])
294
-
295
- progress(1.0, desc="Analysis complete!")
296
-
297
- return results_text, results_df, "βœ… Analysis completed successfully!"
298
-
299
- except Exception as e:
300
- return f"❌ Error during analysis: {str(e)}", None, None
301
 
302
  # Create Gradio interface
303
  def create_interface():
@@ -308,11 +134,17 @@ def create_interface():
308
  Analyze jumping performance from videos using computer vision and pose estimation.
309
  Upload a video or provide a YouTube URL to get detailed metrics about jump height, flight time, and athletic performance.
310
 
 
 
 
 
 
 
311
  ## πŸ“‹ Instructions
312
  1. Enter your height in centimeters
313
  2. Choose either YouTube URL or file upload
314
  3. Wait for the analysis to complete
315
- 4. View your detailed jump performance results
316
  """)
317
 
318
  with gr.Row():
@@ -356,6 +188,16 @@ def create_interface():
356
  datatype=["str", "str"]
357
  )
358
 
 
 
 
 
 
 
 
 
 
 
359
  status_message = gr.Textbox(label="Status", interactive=False)
360
 
361
  # Video requirements
@@ -379,19 +221,23 @@ def create_interface():
379
  - Jump height relative to your body size
380
  - Flight time during the airborne phase
381
  - Normalized rise showing jump efficiency
 
 
 
 
382
  """)
383
 
384
  # Event handlers
385
  youtube_btn.click(
386
  fn=analyze_jump_from_youtube,
387
  inputs=[youtube_url, user_height],
388
- outputs=[results_text, results_table, status_message]
389
  )
390
 
391
  file_btn.click(
392
  fn=analyze_jump_from_file,
393
  inputs=[video_file, user_height],
394
- outputs=[results_text, results_table, status_message]
395
  )
396
 
397
  # Example section
 
 
 
 
 
 
 
 
 
 
1
  import gradio as gr
2
  import pandas as pd
3
+ from athletic_performance import analyze_youtube_video, analyze_video_file, get_performance_insights
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4
 
5
  def analyze_jump_from_youtube(youtube_url, user_height_cm, progress=gr.Progress()):
6
  """Main analysis function for Gradio interface."""
7
 
8
+ # Create progress callback for the athletic_performance module
9
+ def progress_callback(prog, desc):
10
+ progress(prog, desc=desc)
11
 
12
+ # Call the core analysis function
13
+ result = analyze_youtube_video(youtube_url, user_height_cm, progress_callback)
14
 
15
+ # Handle errors
16
+ if "error" in result:
17
+ return f"❌ {result['error']}", None, None, None
18
+
19
+ if result is None:
20
+ return "⚠️ Could not analyze jump. Make sure the video shows a person clearly performing a vertical jump.", None, None, None
21
+
22
+ # Format results for display
23
+ results_text = f"""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
24
  ## πŸŽ‰ Jump Analysis Results
25
 
26
  ### πŸ“Š Performance Metrics
 
35
 
36
  ### πŸ“ˆ Performance Insights
37
  """
38
+
39
+ # Add performance insights using the new function
40
+ insights = get_performance_insights(result['jump_height_cm'], result['flight_time_s'])
41
+ for insight in insights:
42
+ results_text += f"{insight}\n"
43
+
44
+ # Add technique analysis insights
45
+ results_text += f"""
46
+ ### 🎯 Technique Analysis
47
+ - **Knee Position**: Analyzed for valgus and injury prevention
48
+ - **Shoulder Position**: Evaluated overhead squat mechanics
49
+ - **Movement Quality**: Real-time feedback on form
50
+ """
51
+
52
+ # Create a results dataframe for the table
53
+ results_df = pd.DataFrame([
54
+ ["Jump Height", f"{result['jump_height_cm']:.2f} cm"],
55
+ ["Flight Time", f"{result['flight_time_s']:.3f} seconds"],
56
+ ["Normalized Rise", f"{result['normalized_rise']*100:.1f}%"],
57
+ ["Video Frames", f"{result['frames']}"],
58
+ ["Frame Rate", f"{result['fps']:.2f} FPS"],
59
+ ], columns=["Metric", "Value"])
60
+
61
+ # Return overlay video if available
62
+ overlay_video = result.get('overlay_video', None)
63
+
64
+ return results_text, results_df, overlay_video, "βœ… Analysis completed successfully!"
 
 
 
 
65
 
66
  def analyze_jump_from_file(video_file, user_height_cm, progress=gr.Progress()):
67
  """Analysis function for uploaded video files."""
68
 
69
+ # Create progress callback for the athletic_performance module
70
+ def progress_callback(prog, desc):
71
+ progress(prog, desc=desc)
72
 
73
+ # Call the core analysis function
74
+ video_path = video_file.name if video_file else None
75
+ result = analyze_video_file(video_path, user_height_cm, progress_callback)
76
 
77
+ # Handle errors
78
+ if "error" in result:
79
+ return f"❌ {result['error']}", None, None, None
80
+
81
+ if result is None:
82
+ return "⚠️ Could not analyze jump. Make sure the video shows a person clearly performing a vertical jump.", None, None, None
83
+
84
+ # Format results (same as YouTube function)
85
+ results_text = f"""
 
 
 
 
 
 
 
86
  ## πŸŽ‰ Jump Analysis Results
87
 
88
  ### πŸ“Š Performance Metrics
 
97
 
98
  ### πŸ“ˆ Performance Insights
99
  """
100
+
101
+ # Add performance insights using the new function
102
+ insights = get_performance_insights(result['jump_height_cm'], result['flight_time_s'])
103
+ for insight in insights:
104
+ results_text += f"{insight}\n"
105
+
106
+ # Add technique analysis insights
107
+ results_text += f"""
108
+ ### 🎯 Technique Analysis
109
+ - **Knee Position**: Analyzed for valgus and injury prevention
110
+ - **Shoulder Position**: Evaluated overhead squat mechanics
111
+ - **Movement Quality**: Real-time feedback on form
112
+ """
113
+
114
+ # Create a results dataframe for the table
115
+ results_df = pd.DataFrame([
116
+ ["Jump Height", f"{result['jump_height_cm']:.2f} cm"],
117
+ ["Flight Time", f"{result['flight_time_s']:.3f} seconds"],
118
+ ["Normalized Rise", f"{result['normalized_rise']*100:.1f}%"],
119
+ ["Video Frames", f"{result['frames']}"],
120
+ ["Frame Rate", f"{result['fps']:.2f} FPS"],
121
+ ], columns=["Metric", "Value"])
122
+
123
+ # Return overlay video if available
124
+ overlay_video = result.get('overlay_video', None)
125
+
126
+ return results_text, results_df, overlay_video, "βœ… Analysis completed successfully!"
 
 
 
 
127
 
128
  # Create Gradio interface
129
  def create_interface():
 
134
  Analyze jumping performance from videos using computer vision and pose estimation.
135
  Upload a video or provide a YouTube URL to get detailed metrics about jump height, flight time, and athletic performance.
136
 
137
+ ## πŸ†• New Features
138
+ - **πŸŽ₯ Video Overlays**: Watch your movement with real-time technique analysis
139
+ - **🦡 Knee Injury Prevention**: Detect knee valgus and movement patterns
140
+ - **πŸ’ͺ Shoulder Position Analysis**: Evaluate overhead squat mechanics
141
+ - **πŸ“Š Performance Insights**: Get personalized coaching feedback
142
+
143
  ## πŸ“‹ Instructions
144
  1. Enter your height in centimeters
145
  2. Choose either YouTube URL or file upload
146
  3. Wait for the analysis to complete
147
+ 4. View your detailed jump performance results and technique analysis video
148
  """)
149
 
150
  with gr.Row():
 
188
  datatype=["str", "str"]
189
  )
190
 
191
+ # Video output section
192
+ gr.Markdown("## πŸŽ₯ Technique Analysis Video")
193
+ gr.Markdown("πŸ“Ή *Watch your movement with real-time technique feedback overlays*")
194
+
195
+ overlay_video = gr.Video(
196
+ label="Analysis Video with Overlays",
197
+ interactive=False,
198
+ info="Video showing pose landmarks and technique analysis"
199
+ )
200
+
201
  status_message = gr.Textbox(label="Status", interactive=False)
202
 
203
  # Video requirements
 
221
  - Jump height relative to your body size
222
  - Flight time during the airborne phase
223
  - Normalized rise showing jump efficiency
224
+ 4. **Technique Analysis**: Real-time movement assessment:
225
+ - Knee position analysis for injury prevention
226
+ - Shoulder position for overhead squat mechanics
227
+ - Visual overlays with color-coded feedback
228
  """)
229
 
230
  # Event handlers
231
  youtube_btn.click(
232
  fn=analyze_jump_from_youtube,
233
  inputs=[youtube_url, user_height],
234
+ outputs=[results_text, results_table, overlay_video, status_message]
235
  )
236
 
237
  file_btn.click(
238
  fn=analyze_jump_from_file,
239
  inputs=[video_file, user_height],
240
+ outputs=[results_text, results_table, overlay_video, status_message]
241
  )
242
 
243
  # Example section
athletic_performance.py ADDED
@@ -0,0 +1,430 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import cv2
2
+ import numpy as np
3
+ import mediapipe as mp
4
+ from collections import deque
5
+ from pathlib import Path
6
+ import tempfile
7
+ import os
8
+ import yt_dlp
9
+
10
+ # MediaPipe pose landmarks
11
+ LHIP, RHIP = 23, 24
12
+ LANKLE, RANKLE = 27, 28
13
+ LKNEEL, RKNEEL = 25, 26
14
+ LSHOULDER, RSHOULDER = 11, 12
15
+ LELBOW, RELBOW = 13, 14
16
+ LWRIST, RWRIST = 15, 16
17
+ POSE_CONNECTIONS = mp.solutions.pose.POSE_CONNECTIONS
18
+
19
+
20
+ def smooth_moving_avg(series, k=5):
21
+ """Simple causal moving average; ignores None values."""
22
+ out = []
23
+ q = deque()
24
+ s = 0.0
25
+ cnt = 0
26
+ for v in series:
27
+ if v is not None:
28
+ q.append(v)
29
+ s += v
30
+ cnt += 1
31
+ else:
32
+ q.append(None)
33
+ if len(q) > k:
34
+ old = q.popleft()
35
+ if old is not None:
36
+ s -= old
37
+ cnt -= 1
38
+ out.append((s / max(cnt, 1)) if cnt > 0 else None)
39
+ return out
40
+
41
+
42
+ def estimate_jump_metrics(hip_y_series, fps):
43
+ """Return jump_height_norm (0..1), flight_time_s using hip trajectory."""
44
+ # Remove None
45
+ hip = [h for h in hip_y_series if h is not None]
46
+ if len(hip) < 3:
47
+ return None, None
48
+
49
+ # Smooth
50
+ hip = smooth_moving_avg(hip, k=5)
51
+
52
+ # Jump height (normalized): deepest crouch (max y) to apex (min y)
53
+ min_y = min(hip) # apex (body highest)
54
+ max_y = max(hip) # deepest crouch (body lowest)
55
+ jump_height_norm = max(0.0, (max_y - min_y))
56
+
57
+ # Flight time heuristic using vertical velocity pattern
58
+ hip_arr = np.array(hip, dtype=float)
59
+ vel = np.diff(hip_arr)
60
+ if vel.size == 0:
61
+ flight_time_s = 0.0
62
+ else:
63
+ takeoff_idx = int(np.argmin(vel)) # most negative velocity
64
+ landing_idx = int(np.argmax(vel)) # most positive velocity
65
+ flight_frames = max(0, landing_idx - takeoff_idx)
66
+ flight_time_s = flight_frames / float(fps or 30.0)
67
+
68
+ return jump_height_norm, flight_time_s
69
+
70
+
71
+ def download_youtube_video(youtube_url, output_path):
72
+ """Download YouTube video to specified path."""
73
+ ydl_opts = {
74
+ 'format': 'best[height<=720]', # Limit quality for faster processing
75
+ 'outtmpl': output_path,
76
+ 'quiet': True,
77
+ 'no_warnings': True,
78
+ }
79
+
80
+ with yt_dlp.YoutubeDL(ydl_opts) as ydl:
81
+ ydl.download([youtube_url])
82
+ return output_path
83
+
84
+
85
+ def process_video_analysis(video_path, user_height_cm, progress_callback=None, generate_overlay_video=True):
86
+ """Core video analysis function with progress tracking and overlay video generation.
87
+
88
+ Args:
89
+ video_path (str): Path to the video file
90
+ user_height_cm (float): User's height in centimeters
91
+ progress_callback (callable, optional): Function to call with progress updates
92
+ Signature: progress_callback(progress_float, description_string)
93
+ generate_overlay_video (bool): Whether to generate overlay video with technique analysis
94
+
95
+ Returns:
96
+ dict: Analysis results containing jump metrics, video info, and overlay video path
97
+ None: If analysis failed
98
+ """
99
+ cap = cv2.VideoCapture(video_path)
100
+ if not cap.isOpened():
101
+ raise Exception(f"Could not open video: {video_path}")
102
+
103
+ w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
104
+ h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
105
+ fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
106
+ total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
107
+
108
+ mp_pose = mp.solutions.pose
109
+ pose = mp_pose.Pose(static_image_mode=False, model_complexity=1, enable_segmentation=False)
110
+
111
+ hip_y_series = []
112
+ frame_idx = 0
113
+
114
+ # Setup video writer for overlay video
115
+ overlay_video_path = None
116
+ out = None
117
+ if generate_overlay_video:
118
+ overlay_video_path = str(Path(video_path).parent / f"overlay_{Path(video_path).name}")
119
+ fourcc = cv2.VideoWriter_fourcc(*'mp4v')
120
+ out = cv2.VideoWriter(overlay_video_path, fourcc, fps, (w, h))
121
+
122
+ print(f"Processing video: {Path(video_path).name}")
123
+ print(f"Video dimensions: {w}x{h}, FPS: {fps}, Total frames: {total_frames}")
124
+
125
+ ok, frame = cap.read()
126
+ while ok:
127
+ rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
128
+ res = pose.process(rgb)
129
+
130
+ if res.pose_landmarks:
131
+ lms = res.pose_landmarks.landmark
132
+ mid_hip_y = (lms[LHIP].y + lms[RHIP].y) / 2.0
133
+ hip_y_series.append(float(mid_hip_y))
134
+
135
+ # Generate overlay frame with technique analysis
136
+ if generate_overlay_video and out:
137
+ overlay_frame = draw_technique_overlay(frame.copy(), res.pose_landmarks, h, w)
138
+ out.write(overlay_frame)
139
+ else:
140
+ hip_y_series.append(None)
141
+
142
+ # Write original frame if no pose detected
143
+ if generate_overlay_video and out:
144
+ out.write(frame)
145
+
146
+ frame_idx += 1
147
+
148
+ # Update progress
149
+ if progress_callback and total_frames > 0:
150
+ progress = min(frame_idx / total_frames, 1.0)
151
+ progress_callback(progress, f"Processing frame {frame_idx}/{total_frames}")
152
+
153
+ ok, frame = cap.read()
154
+
155
+ cap.release()
156
+ if out:
157
+ out.release()
158
+
159
+ print(f"Completed processing {frame_idx} frames")
160
+
161
+ jump_norm, flight_time_s = estimate_jump_metrics(hip_y_series, fps)
162
+
163
+ if jump_norm is None:
164
+ return None
165
+
166
+ jump_height_cm = jump_norm * user_height_cm
167
+
168
+ result = {
169
+ "video": Path(video_path).name,
170
+ "frames": len(hip_y_series),
171
+ "fps": fps,
172
+ "jump_height_cm": jump_height_cm,
173
+ "normalized_rise": jump_norm,
174
+ "flight_time_s": flight_time_s
175
+ }
176
+
177
+ if generate_overlay_video and overlay_video_path:
178
+ result["overlay_video"] = overlay_video_path
179
+
180
+ return result
181
+
182
+
183
+ def analyze_youtube_video(youtube_url, user_height_cm, progress_callback=None):
184
+ """Analyze jump from YouTube video.
185
+
186
+ Args:
187
+ youtube_url (str): YouTube video URL
188
+ user_height_cm (float): User's height in centimeters
189
+ progress_callback (callable, optional): Function to call with progress updates
190
+
191
+ Returns:
192
+ dict: Analysis results or error information
193
+ """
194
+ # Validate inputs
195
+ if not youtube_url or not youtube_url.strip():
196
+ return {"error": "Please provide a YouTube URL"}
197
+
198
+ if not user_height_cm or user_height_cm <= 0:
199
+ return {"error": "Please provide a valid height in centimeters"}
200
+
201
+ try:
202
+ if progress_callback:
203
+ progress_callback(0.1, "Validating YouTube URL...")
204
+
205
+ # Validate YouTube URL
206
+ youtube_url = youtube_url.strip()
207
+ if not any(domain in youtube_url for domain in ['youtube.com', 'youtu.be']):
208
+ return {"error": "Please provide a valid YouTube URL"}
209
+
210
+ # Create temporary directory for processing
211
+ with tempfile.TemporaryDirectory() as temp_dir:
212
+ if progress_callback:
213
+ progress_callback(0.2, "Downloading video from YouTube...")
214
+
215
+ # Download video
216
+ video_filename = os.path.join(temp_dir, 'video.%(ext)s')
217
+ try:
218
+ download_youtube_video(youtube_url, video_filename)
219
+ # Find the actual downloaded file
220
+ video_files = [f for f in os.listdir(temp_dir) if f.startswith('video.')]
221
+ if not video_files:
222
+ return {"error": "Failed to download YouTube video. Please check the URL and try again."}
223
+ video_path = os.path.join(temp_dir, video_files[0])
224
+ except Exception as e:
225
+ return {"error": f"Failed to download YouTube video: {str(e)}"}
226
+
227
+ if progress_callback:
228
+ progress_callback(0.3, "Starting video analysis...")
229
+
230
+ # Process the video with progress tracking
231
+ def update_progress(prog, desc):
232
+ if progress_callback:
233
+ progress_callback(0.3 + (prog * 0.6), desc)
234
+
235
+ result = process_video_analysis(video_path, user_height_cm, update_progress)
236
+
237
+ if progress_callback:
238
+ progress_callback(0.9, "Analysis complete!")
239
+
240
+ return result
241
+
242
+ except Exception as e:
243
+ return {"error": f"Error during analysis: {str(e)}"}
244
+
245
+
246
+ def analyze_video_file(video_path, user_height_cm, progress_callback=None):
247
+ """Analyze jump from video file.
248
+
249
+ Args:
250
+ video_path (str): Path to video file
251
+ user_height_cm (float): User's height in centimeters
252
+ progress_callback (callable, optional): Function to call with progress updates
253
+
254
+ Returns:
255
+ dict: Analysis results or error information
256
+ """
257
+ # Validate inputs
258
+ if not video_path:
259
+ return {"error": "Please provide a video file"}
260
+
261
+ if not user_height_cm or user_height_cm <= 0:
262
+ return {"error": "Please provide a valid height in centimeters"}
263
+
264
+ try:
265
+ if progress_callback:
266
+ progress_callback(0.1, "Processing video file...")
267
+
268
+ # Process the video with progress tracking
269
+ def update_progress(prog, desc):
270
+ if progress_callback:
271
+ progress_callback(0.1 + (prog * 0.8), desc)
272
+
273
+ result = process_video_analysis(video_path, user_height_cm, update_progress)
274
+
275
+ if progress_callback:
276
+ progress_callback(1.0, "Analysis complete!")
277
+
278
+ return result
279
+
280
+ except Exception as e:
281
+ return {"error": f"Error during analysis: {str(e)}"}
282
+
283
+
284
+ def analyze_knee_position(landmarks, frame_height, frame_width):
285
+ """Analyze knee position for injury prevention."""
286
+ if not landmarks:
287
+ return None, "No pose detected"
288
+
289
+ lms = landmarks.landmark
290
+
291
+ # Get knee and ankle positions
292
+ lknee = (int(lms[LKNEEL].x * frame_width), int(lms[LKNEEL].y * frame_height))
293
+ rknee = (int(lms[RKNEEL].x * frame_width), int(lms[RKNEEL].y * frame_height))
294
+ lankle = (int(lms[LANKLE].x * frame_width), int(lms[LANKLE].y * frame_height))
295
+ rankle = (int(lms[RANKLE].x * frame_width), int(lms[RANKLE].y * frame_height))
296
+
297
+ # Calculate knee angle (simplified)
298
+ knee_angle_l = calculate_angle(lankle, lknee, lms[LHIP])
299
+ knee_angle_r = calculate_angle(rankle, rknee, lms[RHIP])
300
+
301
+ feedback = []
302
+ color = (0, 255, 0) # Green by default
303
+
304
+ # Check for knee valgus (knee caving in)
305
+ if knee_angle_l < 160 or knee_angle_r < 160:
306
+ feedback.append("⚠️ Knee valgus detected - risk of injury")
307
+ color = (0, 0, 255) # Red
308
+ elif knee_angle_l < 170 or knee_angle_r < 170:
309
+ feedback.append("πŸ’‘ Keep knees tracking over toes")
310
+ color = (0, 165, 255) # Orange
311
+
312
+ return color, feedback
313
+
314
+ def analyze_shoulder_position(landmarks, frame_height, frame_width):
315
+ """Analyze shoulder position and overhead squat mechanics."""
316
+ if not landmarks:
317
+ return None, "No pose detected"
318
+
319
+ lms = landmarks.landmark
320
+
321
+ # Get shoulder, elbow, and wrist positions
322
+ lshoulder = (int(lms[LSHOULDER].x * frame_width), int(lms[LSHOULDER].y * frame_height))
323
+ rshoulder = (int(lms[RSHOULDER].x * frame_width), int(lms[RSHOULDER].y * frame_height))
324
+ lwrist = (int(lms[LWRIST].x * frame_width), int(lms[LWRIST].y * frame_height))
325
+ rwrist = (int(lms[RWRIST].x * frame_width), int(lms[RWRIST].y * frame_height))
326
+
327
+ feedback = []
328
+ color = (0, 255, 0) # Green by default
329
+
330
+ # Check if arms are overhead (OHS position)
331
+ shoulder_y = (lshoulder[1] + rshoulder[1]) / 2
332
+ wrist_y = (lwrist[1] + rwrist[1]) / 2
333
+
334
+ if wrist_y < shoulder_y - 50: # Arms significantly overhead
335
+ feedback.append("βœ… Good overhead position")
336
+ color = (0, 255, 0) # Green
337
+ elif wrist_y < shoulder_y:
338
+ feedback.append("πŸ’‘ Arms overhead - good OHS position")
339
+ color = (0, 255, 255) # Yellow
340
+ else:
341
+ feedback.append("⚠️ Arms not overhead - improve shoulder mobility")
342
+ color = (0, 0, 255) # Red
343
+
344
+ return color, feedback
345
+
346
+ def calculate_angle(point1, point2, point3):
347
+ """Calculate angle between three points."""
348
+ # Convert MediaPipe landmark to tuple if needed
349
+ if hasattr(point3, 'x'):
350
+ point3 = (int(point3.x * 1000), int(point3.y * 1000)) # Scale for calculation
351
+
352
+ # Calculate vectors
353
+ v1 = np.array(point1) - np.array(point2)
354
+ v2 = np.array(point3) - np.array(point2)
355
+
356
+ # Calculate angle
357
+ cos_angle = np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))
358
+ cos_angle = np.clip(cos_angle, -1.0, 1.0)
359
+ angle = np.arccos(cos_angle)
360
+
361
+ return np.degrees(angle)
362
+
363
+ def draw_technique_overlay(frame, landmarks, frame_height, frame_width):
364
+ """Draw technique analysis overlay on frame."""
365
+ if not landmarks:
366
+ return frame
367
+
368
+ # Analyze knee position
369
+ knee_color, knee_feedback = analyze_knee_position(landmarks, frame_height, frame_width)
370
+
371
+ # Analyze shoulder position
372
+ shoulder_color, shoulder_feedback = analyze_shoulder_position(landmarks, frame_height, frame_width)
373
+
374
+ # Draw pose landmarks
375
+ mp_drawing = mp.solutions.drawing_utils
376
+ mp_drawing.draw_landmarks(frame, landmarks, mp.solutions.pose.POSE_CONNECTIONS)
377
+
378
+ # Draw knee analysis
379
+ lms = landmarks.landmark
380
+ lknee = (int(lms[LKNEEL].x * frame_width), int(lms[LKNEEL].y * frame_height))
381
+ rknee = (int(lms[RKNEEL].x * frame_width), int(lms[RKNEEL].y * frame_height))
382
+
383
+ cv2.circle(frame, lknee, 8, knee_color, -1)
384
+ cv2.circle(frame, rknee, 8, knee_color, -1)
385
+
386
+ # Draw shoulder analysis
387
+ lshoulder = (int(lms[LSHOULDER].x * frame_width), int(lms[LSHOULDER].y * frame_height))
388
+ rshoulder = (int(lms[RSHOULDER].x * frame_width), int(lms[RSHOULDER].y * frame_height))
389
+
390
+ cv2.circle(frame, lshoulder, 8, shoulder_color, -1)
391
+ cv2.circle(frame, rshoulder, 8, shoulder_color, -1)
392
+
393
+ # Add text feedback
394
+ y_offset = 30
395
+ for feedback in knee_feedback + shoulder_feedback:
396
+ cv2.putText(frame, feedback, (10, y_offset), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
397
+ cv2.putText(frame, feedback, (10, y_offset), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 1)
398
+ y_offset += 30
399
+
400
+ return frame
401
+
402
+ def get_performance_insights(jump_height_cm, flight_time_s):
403
+ """Generate performance insights based on jump metrics.
404
+
405
+ Args:
406
+ jump_height_cm (float): Jump height in centimeters
407
+ flight_time_s (float): Flight time in seconds
408
+
409
+ Returns:
410
+ list: List of insight strings
411
+ """
412
+ insights = []
413
+
414
+ # Jump height insights
415
+ if jump_height_cm > 60:
416
+ insights.append("πŸ”₯ **Excellent jump height!** This is above average performance.")
417
+ elif jump_height_cm > 40:
418
+ insights.append("πŸ‘ **Good jump height!** Solid athletic performance.")
419
+ elif jump_height_cm > 25:
420
+ insights.append("πŸ“ˆ **Moderate jump height.** Room for improvement with training.")
421
+ else:
422
+ insights.append("🎯 **Starting point identified.** Focus on technique and strength training.")
423
+
424
+ # Flight time insights
425
+ if flight_time_s > 0.5:
426
+ insights.append("⏱️ **Great flight time!** Shows good explosive power.")
427
+ elif flight_time_s > 0.3:
428
+ insights.append("⏱️ **Decent flight time.** Good coordination.")
429
+
430
+ return insights