Update app.py

app.py

@@ -11,12 +11,21 @@ import gradio as gr
 
 
 # ----------------------------
-# Settings (same as Colab)
+# Settings (optimized for speed, still robust)
 # ----------------------------
 UP_ANGLE = 155
 DOWN_ANGLE = 105
-MIN_REP_FRAMES = 8
-FRAME_STRIDE = 1
+
+# Aggressive sampling target (effective inference rate)
+# 6 fps usually gives ~5x fewer YOLO calls on 30fps videos.
+TARGET_FPS = 6.0
+
+# Minimum rep duration in seconds (keeps behavior stable when stride changes)
+MIN_REP_SECONDS = 0.33
+
+# Downscale only (no crop) for YOLO inference
+# 640 is a safe default across varying camera angles.
+MAX_INFER_SIDE = 640
 
 
 # ----------------------------
@@ -43,7 +52,7 @@ def load_pose_model():
 
 
 # ----------------------------
-# Helpers (from your script)
+# Helpers
 # ----------------------------
 def angle_deg(a, b, c):
     a = np.asarray(a, dtype=np.float32)
@@ -56,8 +65,8 @@ def angle_deg(a, b, c):
     return float(math.degrees(math.acos(cosv)))
 
 def pick_best_side(kxy, kconf):
-    left = [5, 7, 9]   # L shoulder, L elbow, L wrist (YOLO COCO indices)
-    right = [6, 8, 10] # R shoulder, R elbow, R wrist
+    left = [5, 7, 9]    # L shoulder, L elbow, L wrist (YOLO COCO indices)
+    right = [6, 8, 10]  # R shoulder, R elbow, R wrist
     if float(np.mean(kconf[right])) >= float(np.mean(kconf[left])):
         return right, float(np.mean(kconf[right]))
     return left, float(np.mean(kconf[left]))
@@ -93,6 +102,17 @@ def likelihood_to_score(p):
             return int(round(s_lo + t * (s_hi - s_lo)))
     return 0
 
+def resize_for_inference(frame, max_side=640):
+    """Downscale only (no crop) to speed YOLO. Keeps aspect ratio."""
+    h, w = frame.shape[:2]
+    m = max(h, w)
+    if m <= max_side:
+        return frame
+    scale = max_side / float(m)
+    new_w = int(round(w * scale))
+    new_h = int(round(h * scale))
+    return cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_AREA)
+
 
 # ----------------------------
 # Core pipeline
@@ -108,6 +128,21 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
     w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) or 0
     h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) or 0
 
+    # Compute stride to hit TARGET_FPS (effective inference rate)
+    frame_stride = max(1, int(round(float(fps) / float(TARGET_FPS))))
+    effective_fps = float(fps) / float(frame_stride)
+
+    # Convert time-based minimum rep duration to sampled frames
+    min_rep_frames = int(math.ceil(MIN_REP_SECONDS * effective_fps))
+    # Clamp so we don't reject real reps at low fps
+    min_rep_frames = max(2, min_rep_frames)
+
+    print(
+        f"[speed] video_fps={fps:.2f} target_fps={TARGET_FPS:.2f} "
+        f"stride={frame_stride} effective_fps={effective_fps:.2f} "
+        f"min_rep_frames={min_rep_frames} infer_max_side={MAX_INFER_SIDE}"
+    )
+
     # 1) First pass: compute angles + confs per sampled frame
     angles, confs, frame_ids = [], [], []
     frame_i = 0
@@ -117,11 +152,13 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
         if not ok:
             break
 
-        if frame_i % FRAME_STRIDE != 0:
+        if frame_i % frame_stride != 0:
             frame_i += 1
             continue
 
-        res = model(frame, verbose=False)[0]
+        infer_frame = resize_for_inference(frame, MAX_INFER_SIDE)
+
+        res = model(infer_frame, verbose=False)[0]
         if res.keypoints is None or len(res.keypoints.xy) == 0:
             angles.append(np.nan)
             confs.append(0.0)
@@ -160,14 +197,16 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
     if len(angles) < 5:
         raise RuntimeError("Video too short or no usable frames detected.")
 
+    # Interpolate missing angles
     mask = np.isfinite(angles)
     if np.any(mask) and not np.all(mask):
         angles[~mask] = np.interp(frame_ids[~mask], frame_ids[mask], angles[mask])
     elif not np.any(mask):
         raise RuntimeError("No valid pose angles detected.")
 
+    # Smooth
     win = min(31, (len(angles) // 2) * 2 + 1)
-    win = max(win, 5)  # savgol requires >= 5 for polyorder=2 comfortably
+    win = max(win, 5)
     angles_smooth = savgol_filter(angles, win, 2)
 
     # 2) Rep detection on smoothed angles
@@ -192,7 +231,7 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
             rep_len += 1
 
             if ang >= UP_ANGLE:
-                if rep_len >= MIN_REP_FRAMES:
+                if rep_len >= min_rep_frames:
                     mean_cf = float(rep_conf_sum / rep_len)
                     likelihood = rep_likelihood(rep_min, rep_max, mean_cf)
                     score = likelihood_to_score(likelihood)
@@ -220,7 +259,7 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
     df = pd.DataFrame(reps)
     df.to_csv(csv_path, index=False)
 
-    # 4) Annotated video
+    # 4) Annotated video (kept original resolution)
     annotated_path = os.path.join(out_dir, "pushup_annotated.mp4")
     cap = cv2.VideoCapture(video_path)
     fourcc = cv2.VideoWriter_fourcc(*"mp4v")
@@ -260,13 +299,21 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
         "avg_score": int(round(float(np.mean([r["pushup_score"] for r in reps])))) if reps else 0,
         "avg_likelihood": float(np.mean([r["pushup_likelihood"] for r in reps])) if reps else 0.0,
         "rep_events": reps,
+        "speed_settings": {
+            "video_fps": float(fps),
+            "target_fps": float(TARGET_FPS),
+            "frame_stride": int(frame_stride),
+            "effective_fps": float(effective_fps),
+            "min_rep_frames": int(min_rep_frames),
+            "max_infer_side": int(MAX_INFER_SIDE),
+        }
     }
 
     return summary, annotated_path, csv_path
 
 
 # ----------------------------
-# API wrapper (robust file handling like your old one)
+# API wrapper
 # ----------------------------
 def api_analyze(uploaded_file):
     if uploaded_file is None:
@@ -286,7 +333,6 @@ def api_analyze(uploaded_file):
     else:
         src_path = str(uploaded_file)
 
-    # Optional extension check (same idea as your old code)
     ext = os.path.splitext(src_path)[1].lower()
     allowed = {".mp4", ".mov", ".webm", ".mkv"}
     if ext and ext not in allowed:
@@ -307,7 +353,7 @@ def api_analyze(uploaded_file):
 with gr.Blocks(title="Pushup API (YOLO)") as demo:
     gr.Markdown("# Pushup Analyzer API (YOLO)\nUpload a video, get rep scores + CSV + annotated video.\n")
 
-    # IMPORTANT: keep this as gr.File to avoid the “Invalid file type: ['video']” problem you hit before
+    # Keep gr.File to avoid Invalid file type issues
     video_file = gr.File(label="Upload video")
 
     btn = gr.Button("Analyze")
@@ -323,4 +369,4 @@ with gr.Blocks(title="Pushup API (YOLO)") as demo:
     )
 
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()