Update app.py

app.py

@@ -11,12 +11,22 @@ import gradio as gr
 
 
 # ----------------------------
-# Settings (same as Colab)
+# Settings
 # ----------------------------
 UP_ANGLE = 155
 DOWN_ANGLE = 105
-MIN_REP_FRAMES = 8
-FRAME_STRIDE = 1
+
+# Target processing FPS (how often we run YOLO)
+# 10 is a good balance for speed vs accuracy.
+TARGET_FPS = 10
+
+# Minimum rep duration in seconds (more robust than hardcoding frames)
+# 0.25s is a safe filter against noise but won't kill real reps.
+MIN_REP_SECONDS = 0.25
+
+# YOLO inference resize (no cropping, only downscale).
+# 640 is typically safe with small accuracy loss, big speed gain on high-res videos.
+MAX_INFER_SIDE = 640
 
 
 # ----------------------------
@@ -43,7 +53,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 +66,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
+    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 +103,16 @@ 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):
+    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,17 @@ 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 from target FPS
+    # Example: fps=30, target=10 => stride=3
+    frame_stride = max(1, int(round(float(fps) / float(TARGET_FPS))))
+
+    # Make MIN_REP_FRAMES consistent in real time, not in raw frames
+    # We count "sampled frames", so this should be based on effective fps = fps / stride
+    effective_fps = float(fps) / float(frame_stride)
+    min_rep_frames = max(3, int(math.ceil(MIN_REP_SECONDS * effective_fps)))
+
+    print(f"Video fps={fps:.2f}, TARGET_FPS={TARGET_FPS}, stride={frame_stride}, effective_fps={effective_fps:.2f}, MIN_REP_FRAMES={min_rep_frames}")
+
     # 1) First pass: compute angles + confs per sampled frame
     angles, confs, frame_ids = [], [], []
     frame_i = 0
@@ -117,11 +148,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)
@@ -167,7 +200,7 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
         raise RuntimeError("No valid pose angles detected.")
 
     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 +225,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 +253,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 (keep original resolution)
     annotated_path = os.path.join(out_dir, "pushup_annotated.mp4")
     cap = cv2.VideoCapture(video_path)
     fourcc = cv2.VideoWriter_fourcc(*"mp4v")
@@ -241,11 +274,11 @@ def analyze_pushup_video_yolo(video_path: str, out_dir: str):
         ang_disp = float(angles_smooth[j])
 
         cv2.putText(frame, f"Reps: {count}/{len(reps)}", (20, 40),
-                    cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255,255,255), 2)
+                    cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255, 255, 255), 2)
         cv2.putText(frame, f"Elbow angle: {ang_disp:.1f}", (20, 80),
-                    cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255,255,255), 2)
+                    cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 255, 255), 2)
         cv2.putText(frame, f"Rep score: {active if active is not None else '-'}", (20, 120),
-                    cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255,255,255), 2)
+                    cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 255, 255), 2)
 
         writer.write(frame)
         frame_i += 1
@@ -260,13 +293,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": int(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:
@@ -275,7 +316,6 @@ def api_analyze(uploaded_file):
     workdir = tempfile.mkdtemp()
     in_path = os.path.join(workdir, "input.mp4")
 
-    # Resolve source path robustly
     src_path = None
     if hasattr(uploaded_file, "path") and uploaded_file.path:
         src_path = uploaded_file.path
@@ -286,7 +326,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,9 +346,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
     video_file = gr.File(label="Upload video")
-
     btn = gr.Button("Analyze")
     out_json = gr.JSON(label="Results JSON")
     out_video = gr.Video(label="Annotated Output")