Update app.py

app.py

@@ -1,11 +1,10 @@
 import os
+import tempfile
+import shutil
 import cv2
 import numpy as np
 import mediapipe as mp
-
 import gradio as gr
-import tempfile
-import shutil
 
 
 # -----------------------
@@ -13,11 +12,12 @@ import shutil
 # -----------------------
 def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated_out_path: str | None = None):
     """
-    Runs MediaPipe Pose on a video, counts pushup reps, and returns:
+    Returns:
     {
       "ok": bool,
       "error": str | None,
       "rep_count": int,
+      "avg_rep_prob": float | None,
       "rep_events": list[dict],
       "annotated_video_path": str | None
     }
@@ -27,16 +27,15 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
             "ok": False,
             "error": f"Could not find input video: {video_path}",
             "rep_count": 0,
+            "avg_rep_prob": None,
             "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)
@@ -48,10 +47,6 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
         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:
@@ -62,7 +57,6 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
     def ema(prev, x, a=0.25):
         return x if prev is None else (a * x + (1 - a) * prev)
 
-    # ---------- Pose setup ----------
     mp_pose = mp.solutions.pose
     pose = mp_pose.Pose(
         static_image_mode=False,
@@ -73,14 +67,14 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
         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.",
+            "error": "OpenCV could not open the video. Try a different MP4 encoding.",
             "rep_count": 0,
+            "avg_rep_prob": None,
             "rep_events": [],
             "annotated_video_path": None,
         }
@@ -89,18 +83,6 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
     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"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:
@@ -110,8 +92,7 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
         fourcc = cv2.VideoWriter_fourcc(*"mp4v")
         writer = cv2.VideoWriter(annotated_path, fourcc, fps, (W, H))
 
-    # ---------- Pushup detection logic ----------
-    state = "UNKNOWN"  # "UP" or "DOWN"
+    state = "UNKNOWN"
     rep_events = []
     current_rep = None
     rep_count = 0
@@ -144,7 +125,6 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
             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)
@@ -209,8 +189,6 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
 
                                     rep_events.append({
                                         "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),
@@ -227,7 +205,6 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
 
                 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),
@@ -245,6 +222,7 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
             "ok": False,
             "error": f"Runtime error: {type(e).__name__}: {e}",
             "rep_count": rep_count,
+            "avg_rep_prob": None,
             "rep_events": rep_events,
             "annotated_video_path": annotated_path,
         }
@@ -254,51 +232,70 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
         writer.release()
     pose.close()
 
+    avg_rep_prob = None
+    if rep_events:
+        avg_rep_prob = float(sum(r["prob"] for r in rep_events) / len(rep_events))
+
     return {
         "ok": True,
         "error": None,
-        "rep_count": rep_count,
+        "rep_count": int(rep_count),
+        "avg_rep_prob": avg_rep_prob,
         "rep_events": rep_events,
         "annotated_video_path": annotated_path,
     }
 
 
 # -----------------------
-# Gradio wrapper
+# Gradio API wrapper
 # -----------------------
-def gradio_run(video_file_path):
-    # video_file_path is a REAL file path on the HF server (because type="filepath")
+def api_run(uploaded_file):
+    """
+    uploaded_file can be:
+      - a Gradio UploadedFile object with .name
+      - a dict with 'path'
+      - a string path
+    We normalize it to a real filepath, copy it, run pipeline, return JSON + annotated video.
+    """
+    # get path
+    path = None
+    if uploaded_file is None:
+        return {"ok": False, "error": "No file uploaded."}, None
+
+    if isinstance(uploaded_file, str):
+        path = uploaded_file
+    elif isinstance(uploaded_file, dict) and "path" in uploaded_file:
+        path = uploaded_file["path"]
+    elif hasattr(uploaded_file, "name"):
+        path = uploaded_file.name
+
+    if not path or not os.path.exists(path):
+        return {"ok": False, "error": f"Upload path missing or not found: {path}"}, None
+
     workdir = tempfile.mkdtemp()
     in_path = os.path.join(workdir, "input.mp4")
-    shutil.copy(video_file_path, in_path)
+    shutil.copy(path, in_path)
 
     out_path = os.path.join(workdir, "annotated.mp4")
     result = analyze_pushup_video(in_path, save_annotated=True, annotated_out_path=out_path)
 
-    if not result["ok"]:
-        return {"ok": False, "error": str(result["error"]), "rep_count": 0, "avg_prob": 0.0, "rep_events": []}
+    # Return JSON + video path (Gradio serves it)
+    return result, result["annotated_video_path"]
 
-    avg_prob = 0.0
-    if result["rep_events"]:
-        avg_prob = sum(r["prob"] for r in result["rep_events"]) / len(result["rep_events"])
-
-    # IMPORTANT: return only JSON-ish data (browser can render it easily)
-    return {
-        "ok": True,
-        "error": None,
-        "rep_count": int(result["rep_count"]),
-        "avg_prob": float(avg_prob),
-        "rep_events": result["rep_events"],
-    }
 
+with gr.Blocks(title="Pushup Prototype API") as demo:
+    gr.Markdown("# Pushup Prototype (API)\nUpload a video -> get JSON result + annotated output.")
+    video_file = gr.File(label="Upload MP4", file_types=["video"])
+    run_btn = gr.Button("Analyze")
+    out_json = gr.JSON(label="Result JSON")
+    out_vid = gr.Video(label="Annotated Output")
 
-demo = gr.Interface(
-    fn=gradio_run,
-    inputs=gr.Video(label="Upload pushup video", type="filepath"),
-    outputs=gr.JSON(label="Result"),
-    api_name="analyze",          # gives you a stable endpoint name
-)
+    run_btn.click(
+        fn=api_run,
+        inputs=[video_file],
+        outputs=[out_json, out_vid],
+        api_name="analyze"   # IMPORTANT for GitHub JS calls
+    )
 
-if __name__ == "__main__":
-    demo.queue()
-    demo.launch()
+demo.queue()
+demo.launch()