Update app.py

app.py

@@ -1,6 +1,9 @@
 import os
-import tempfile
+import math
 import shutil
+import tempfile
+from typing import Any, Dict, List, Optional, Union
+
 import cv2
 import numpy as np
 import mediapipe as mp
@@ -10,9 +13,13 @@ import gradio as gr
 # -----------------------
 # Core pipeline function
 # -----------------------
-def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated_out_path: str | None = None):
+def analyze_pushup_video(
+    video_path: str,
+    save_annotated: bool = True,
+    annotated_out_path: Optional[str] = None,
+) -> Dict[str, Any]:
     """
-    Returns:
+    Runs MediaPipe Pose on a video, counts pushup reps, and returns:
     {
       "ok": bool,
       "error": str | None,
@@ -22,6 +29,7 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
       "annotated_video_path": str | None
     }
     """
+
     if not os.path.exists(video_path):
         return {
             "ok": False,
@@ -32,6 +40,7 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
             "annotated_video_path": None,
         }
 
+    # ---------- Helpers ----------
     def clamp(x, lo=0.0, hi=1.0):
         return max(lo, min(hi, x))
 
@@ -51,12 +60,12 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
             return 1.0
         if val < good_lo:
             return clamp((val - ok_lo) / (good_lo - ok_lo))
-        else:
-            return clamp((ok_hi - val) / (ok_hi - good_hi))
+        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 ----------
     mp_pose = mp.solutions.pose
     pose = mp_pose.Pose(
         static_image_mode=False,
@@ -67,12 +76,13 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
         min_tracking_confidence=0.5,
     )
 
+    # ---------- Video ----------
     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 re-exporting to a standard H.264 mp4.",
             "rep_count": 0,
             "avg_rep_prob": None,
             "rep_events": [],
@@ -83,8 +93,8 @@ 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
 
-    annotated_path = None
     writer = None
+    annotated_path = None
     if save_annotated:
         if annotated_out_path is None:
             annotated_out_path = os.path.join(tempfile.mkdtemp(), "annotated.mp4")
@@ -92,9 +102,10 @@ 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))
 
-    state = "UNKNOWN"
-    rep_events = []
-    current_rep = None
+    # ---------- Detection state ----------
+    state = "UNKNOWN"  # UP/DOWN
+    rep_events: List[Dict[str, Any]] = []
+    current_rep: Optional[Dict[str, Any]] = None
     rep_count = 0
 
     ema_elbow = None
@@ -188,7 +199,7 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
                                     rep_prob = float(np.mean(probs))
 
                                     rep_events.append({
-                                        "rep": int(rep_count),
+                                        "rep": rep_count,
                                         "start_t": float(current_rep["start_f"] / fps),
                                         "end_t": float(end_f / fps),
                                         "prob": float(rep_prob),
@@ -205,6 +216,7 @@ 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}"
 
+            # annotate 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),
@@ -247,55 +259,57 @@ def analyze_pushup_video(video_path: str, save_annotated: bool = True, annotated
 
 
 # -----------------------
-# Gradio API wrapper
+# Gradio API function
 # -----------------------
-def api_run(uploaded_file):
+def api_analyze(file_obj: Union[str, Dict[str, Any], None]):
     """
-    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.
+    Accepts Gradio File input. Depending on Gradio version, it may come in as:
+    - a string path
+    - a dict with {"name": "..."}
     """
-    # get path
-    path = None
-    if uploaded_file is None:
-        return {"ok": False, "error": "No file uploaded."}, None
+    if file_obj is None:
+        return {"ok": False, "error": "No file received."}
 
-    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
+    # Extract path robustly
+    if isinstance(file_obj, str):
+        src_path = file_obj
+    elif isinstance(file_obj, dict) and "name" in file_obj:
+        src_path = file_obj["name"]
+    else:
+        return {"ok": False, "error": f"Unsupported file object: {type(file_obj)}"}
 
-    if not path or not os.path.exists(path):
-        return {"ok": False, "error": f"Upload path missing or not found: {path}"}, None
+    if not os.path.exists(src_path):
+        return {"ok": False, "error": f"Upload path not found on server: {src_path}"}
 
     workdir = tempfile.mkdtemp()
     in_path = os.path.join(workdir, "input.mp4")
-    shutil.copy(path, in_path)
+    shutil.copy(src_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)
 
-    # Return JSON + video path (Gradio serves it)
-    return result, result["annotated_video_path"]
+    # Return only what frontend needs + a file path for annotated video
+    return {
+        "ok": result["ok"],
+        "error": result["error"],
+        "rep_count": result["rep_count"],
+        "avg_rep_prob": result["avg_rep_prob"],
+        "rep_events": result["rep_events"],
+        "annotated_video_path": result["annotated_video_path"],
+    }
+
 
+# -----------------------
+# Gradio app
+# -----------------------
+with gr.Blocks(title="Pushup Analyzer API") as demo:
+    gr.Markdown("# Pushup Analyzer (Backend)\nThis Space provides an API + optional Gradio UI.")
 
-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")
+    file_in = gr.File(label="Upload video", file_types=["video"])
+    btn = gr.Button("Analyze")
+    json_out = gr.JSON(label="Result JSON")
 
-    run_btn.click(
-        fn=api_run,
-        inputs=[video_file],
-        outputs=[out_json, out_vid],
-        api_name="analyze"   # IMPORTANT for GitHub JS calls
-    )
+    # api_name is IMPORTANT: frontend will call this endpoint
+    btn.click(fn=api_analyze, inputs=file_in, outputs=json_out, api_name="analyze")
 
-demo.queue()
 demo.launch()