Update app.py

app.py

@@ -6,36 +6,37 @@ import mediapipe as mp
 import gradio as gr
 import tempfile
 import shutil
-from typing import Optional, Any, Dict
 
 
 # -----------------------
 # Core pipeline function
 # -----------------------
-def analyze_pushup_video(
-    video_path: str,
-    save_annotated: bool = True,
-    annotated_out_path: Optional[str] = None,
-):
+def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated_out_path: str | None = None):
     """
-    Returns dict:
+    Runs MediaPipe Pose on a video, counts pushup reps, and returns:
     {
       "ok": bool,
-      "error": str|None,
+      "error": str | None,
       "rep_count": int,
       "rep_events": list[dict],
-      "annotated_video_path": str|None
+      "annotated_video_path": str | None
     }
     """
     if not os.path.exists(video_path):
-        return {"ok": False, "error": f"Could not find input video: {video_path}",
-                "rep_count": 0, "rep_events": [], "annotated_video_path": None}
-
-    # ---------- helpers ----------
+        return {
+            "ok": False,
+            "error": f"Could not find input video: {video_path}",
+            "rep_count": 0,
+            "rep_events": [],
+            "annotated_video_path": None,
+        }
+
+    # ---------- Math helpers ----------
     def clamp(x, lo=0.0, hi=1.0):
         return max(lo, min(hi, x))
 
     def angle_deg(a, b, c):
+        """Angle ABC in degrees using points a,b,c as (x,y)."""
         a = np.array(a, dtype=np.float32)
         b = np.array(b, dtype=np.float32)
         c = np.array(c, dtype=np.float32)
@@ -47,16 +48,21 @@ def analyze_pushup_video(
         return float(np.degrees(np.arccos(cosang)))
 
     def score_from_range(val, good_lo, good_hi, ok_lo, ok_hi):
+        """
+        Returns 1 if val in [good_lo, good_hi],
+        fades to 0 by the time it reaches ok_lo/ok_hi.
+        """
         if good_lo <= val <= good_hi:
             return 1.0
         if val < good_lo:
             return clamp((val - ok_lo) / (good_lo - ok_lo))
-        return clamp((ok_hi - val) / (ok_hi - good_hi))
+        else:
+            return clamp((ok_hi - val) / (ok_hi - good_hi))
 
     def ema(prev, x, a=0.25):
         return x if prev is None else (a * x + (1 - a) * prev)
 
-    # ---------- pose ----------
+    # ---------- Pose setup ----------
     mp_pose = mp.solutions.pose
     pose = mp_pose.Pose(
         static_image_mode=False,
@@ -67,26 +73,34 @@ def analyze_pushup_video(
         min_tracking_confidence=0.5,
     )
 
+    # ---------- Video I/O ----------
     cap = cv2.VideoCapture(video_path)
     if not cap.isOpened():
         pose.close()
-        return {"ok": False, "error": "OpenCV could not open the video. Try a different MP4 encoding.",
-                "rep_count": 0, "rep_events": [], "annotated_video_path": None}
-
-    fps = float(cap.get(cv2.CAP_PROP_FPS) or 30.0)
-    W = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
-    H = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
+        return {
+            "ok": False,
+            "error": "OpenCV could not open the video. Try a different mp4 encoding.",
+            "rep_count": 0,
+            "rep_events": [],
+            "annotated_video_path": None,
+        }
+
+    fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
+    W = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) or 0
+    H = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) or 0
 
     if W <= 0 or H <= 0:
         cap.release()
         pose.close()
-        return {"ok": False, "error": f"Video decode failed (W={W}, H={H}). Try re-encoding the video.",
-                "rep_count": 0, "rep_events": [], "annotated_video_path": None}
-
-    # Some encoders hate odd dimensions
-    if W % 2 == 1: W -= 1
-    if H % 2 == 1: H -= 1
-
+        return {
+            "ok": False,
+            "error": f"Bad video dimensions from OpenCV: W={W}, H={H}.",
+            "rep_count": 0,
+            "rep_events": [],
+            "annotated_video_path": None,
+        }
+
+    # Output path handling
     annotated_path = None
     writer = None
     if save_annotated:
@@ -95,13 +109,9 @@ def analyze_pushup_video(
         annotated_path = annotated_out_path
         fourcc = cv2.VideoWriter_fourcc(*"mp4v")
         writer = cv2.VideoWriter(annotated_path, fourcc, fps, (W, H))
-        if not writer.isOpened():
-            # Don’t crash the whole run if writing fails
-            writer = None
-            annotated_path = None
 
-    # ---------- detection ----------
-    state = "UNKNOWN"
+    # ---------- Pushup detection logic ----------
+    state = "UNKNOWN"  # "UP" or "DOWN"
     rep_events = []
     current_rep = None
     rep_count = 0
@@ -125,9 +135,6 @@ def analyze_pushup_video(
                 break
             frame_idx += 1
 
-            # Resize to the exact writer size if we adjusted odd dims
-            frame = frame[:H, :W]
-
             rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
             res = pose.process(rgb)
 
@@ -137,6 +144,7 @@ def analyze_pushup_video(
             if res.pose_landmarks:
                 lms = res.pose_landmarks.landmark
 
+                # Choose side: whichever shoulder has higher visibility
                 Ls = lms[mp_pose.PoseLandmark.LEFT_SHOULDER.value]
                 Rs = lms[mp_pose.PoseLandmark.RIGHT_SHOULDER.value]
                 left_side = (Ls.visibility >= Rs.visibility)
@@ -171,7 +179,7 @@ def analyze_pushup_video(
 
                 s_straight = score_from_range(ema_straight, 165, 185, 145, 195)
                 s_elbow = score_from_range(ema_elbow, 85, 175, 60, 190)
-                s_vis = clamp((ema_vis - MIN_VIS) / (0.85 - MIN_VIS)) if ema_vis is not None else 0.0
+                s_vis = clamp((ema_vis - MIN_VIS) / (0.85 - MIN_VIS))
 
                 frame_prob = clamp(0.15 + 0.45 * s_elbow + 0.30 * s_straight + 0.10 * s_vis)
 
@@ -188,6 +196,7 @@ def analyze_pushup_video(
                                     "min_elbow": float(ema_elbow),
                                     "min_straight": float(ema_straight),
                                 }
+
                     elif state == "DOWN":
                         if ema_elbow >= UP_ELBOW_DEG and frame_prob >= 0.35:
                             end_f = frame_idx
@@ -197,8 +206,11 @@ def analyze_pushup_video(
                                     rep_count += 1
                                     probs = current_rep["frame_probs"] if current_rep["frame_probs"] else [frame_prob]
                                     rep_prob = float(np.mean(probs))
+
                                     rep_events.append({
-                                        "rep": rep_count,
+                                        "rep": int(rep_count),
+                                        "start_f": int(current_rep["start_f"]),
+                                        "end_f": int(end_f),
                                         "start_t": float(current_rep["start_f"] / fps),
                                         "end_t": float(end_f / fps),
                                         "prob": float(rep_prob),
@@ -215,6 +227,7 @@ def analyze_pushup_video(
 
                 debug_txt = f"{'L' if left_side else 'R'} vis={ema_vis:.2f} elbow={ema_elbow:.0f} straight={ema_straight:.0f} p={frame_prob:.2f} state={state}"
 
+            # Overlay text on every frame
             cv2.putText(frame, f"Reps: {rep_count}", (20, 40),
                         cv2.FONT_HERSHEY_SIMPLEX, 1.0, (255, 255, 255), 2, cv2.LINE_AA)
             cv2.putText(frame, debug_txt[:90], (20, 75),
@@ -228,39 +241,35 @@ def analyze_pushup_video(
         if writer is not None:
             writer.release()
         pose.close()
-        return {"ok": False, "error": f"Runtime error: {type(e).__name__}: {e}",
-                "rep_count": rep_count, "rep_events": rep_events, "annotated_video_path": annotated_path}
+        return {
+            "ok": False,
+            "error": f"Runtime error: {type(e).__name__}: {e}",
+            "rep_count": rep_count,
+            "rep_events": rep_events,
+            "annotated_video_path": annotated_path,
+        }
 
     cap.release()
     if writer is not None:
         writer.release()
     pose.close()
 
-    return {"ok": True, "error": None, "rep_count": rep_count,
-            "rep_events": rep_events, "annotated_video_path": annotated_path}
+    return {
+        "ok": True,
+        "error": None,
+        "rep_count": rep_count,
+        "rep_events": rep_events,
+        "annotated_video_path": annotated_path,
+    }
 
 
 # -----------------------
 # Gradio wrapper
 # -----------------------
-def _extract_video_path(video_input: Any) -> str:
-    # Gradio sometimes gives a string path, sometimes a dict with "path"
-    if isinstance(video_input, str):
-        return video_input
-    if isinstance(video_input, dict) and "path" in video_input:
-        return video_input["path"]
-    # Some versions use "name"
-    if isinstance(video_input, dict) and "name" in video_input:
-        return video_input["name"]
-    raise ValueError(f"Unexpected video input type: {type(video_input)} value={video_input}")
-
-
 def gradio_run(video_file):
     workdir = tempfile.mkdtemp()
     in_path = os.path.join(workdir, "input.mp4")
-
-    src_path = _extract_video_path(video_file)
-    shutil.copy(src_path, in_path)
+    shutil.copy(video_file, in_path)
 
     out_path = os.path.join(workdir, "annotated.mp4")
     result = analyze_pushup_video(in_path, save_annotated=True, annotated_out_path=out_path)
@@ -284,7 +293,9 @@ demo = gr.Interface(
         gr.Video(label="Annotated output"),
         gr.JSON(label="Per-rep details"),
     ],
-    allow_flagging="never",
+    title="Pushup Prototype",
+    description="Uploads a video, counts reps, and gives per-rep likelihood.",
+    flagging_mode="never",  # ✅ correct for newer gradio
 )
 
 if __name__ == "__main__":