Update app.py

app.py

@@ -1,46 +1,37 @@
 import os
-import math
-import shutil
-import tempfile
-from typing import Any, Dict, List, Optional, Union
-
 import cv2
 import numpy as np
 import mediapipe as mp
+import tempfile
+import shutil
 import gradio as gr
 
 
 # -----------------------
 # Core pipeline function
 # -----------------------
-def analyze_pushup_video(
-    video_path: str,
-    save_annotated: bool = True,
-    annotated_out_path: Optional[str] = None,
-) -> Dict[str, Any]:
+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,
+      "avg_rep_prob": float,
       "rep_events": list[dict],
       "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,
-            "avg_rep_prob": None,
+            "avg_rep_prob": 0.0,
             "rep_events": [],
             "annotated_video_path": None,
         }
 
-    # ---------- Helpers ----------
     def clamp(x, lo=0.0, hi=1.0):
         return max(lo, min(hi, x))
 
@@ -65,7 +56,6 @@ def analyze_pushup_video(
     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,
@@ -76,15 +66,14 @@ def analyze_pushup_video(
         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 re-exporting to a standard H.264 mp4.",
+            "error": "OpenCV could not open the video. Try a different mp4 encoding.",
             "rep_count": 0,
-            "avg_rep_prob": None,
+            "avg_rep_prob": 0.0,
             "rep_events": [],
             "annotated_video_path": None,
         }
@@ -93,8 +82,8 @@ def analyze_pushup_video(
     W = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) or 0
     H = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) or 0
 
-    writer = None
     annotated_path = None
+    writer = None
     if save_annotated:
         if annotated_out_path is None:
             annotated_out_path = os.path.join(tempfile.mkdtemp(), "annotated.mp4")
@@ -102,10 +91,9 @@ def analyze_pushup_video(
         fourcc = cv2.VideoWriter_fourcc(*"mp4v")
         writer = cv2.VideoWriter(annotated_path, fourcc, fps, (W, H))
 
-    # ---------- Detection state ----------
-    state = "UNKNOWN"  # UP/DOWN
-    rep_events: List[Dict[str, Any]] = []
-    current_rep: Optional[Dict[str, Any]] = None
+    state = "UNKNOWN"
+    rep_events = []
+    current_rep = None
     rep_count = 0
 
     ema_elbow = None
@@ -197,7 +185,6 @@ 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,
                                         "start_t": float(current_rep["start_f"] / fps),
@@ -216,7 +203,6 @@ 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}"
 
-            # 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),
@@ -234,7 +220,7 @@ def analyze_pushup_video(
             "ok": False,
             "error": f"Runtime error: {type(e).__name__}: {e}",
             "rep_count": rep_count,
-            "avg_rep_prob": None,
+            "avg_rep_prob": float(np.mean([r["prob"] for r in rep_events])) if rep_events else 0.0,
             "rep_events": rep_events,
             "annotated_video_path": annotated_path,
         }
@@ -244,72 +230,72 @@ def analyze_pushup_video(
         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))
+    avg_prob = float(np.mean([r["prob"] for r in rep_events])) if rep_events else 0.0
 
     return {
         "ok": True,
         "error": None,
-        "rep_count": int(rep_count),
-        "avg_rep_prob": avg_rep_prob,
+        "rep_count": rep_count,
+        "avg_rep_prob": avg_prob,
         "rep_events": rep_events,
         "annotated_video_path": annotated_path,
     }
 
 
 # -----------------------
-# Gradio API function
+# API function (Space endpoint)
 # -----------------------
-def api_analyze(file_obj: Union[str, Dict[str, Any], None]):
+def api_analyze(uploaded_file):
     """
-    Accepts Gradio File input. Depending on Gradio version, it may come in as:
-    - a string path
-    - a dict with {"name": "..."}
+    uploaded_file is a dict-like object from gr.File in many gradio versions:
+    - sometimes it is {"path": "..."}
+    - sometimes it is an object with .name
+    We'll support both.
     """
-    if file_obj is None:
-        return {"ok": False, "error": "No file received."}
-
-    # 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 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")
+
+    # Resolve input path
+    if isinstance(uploaded_file, dict) and "path" in uploaded_file:
+        src_path = uploaded_file["path"]
+    else:
+        # gradio can pass a file object with .name
+        src_path = getattr(uploaded_file, "name", None) or str(uploaded_file)
+
     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 only what frontend needs + a file path for annotated video
-    return {
-        "ok": result["ok"],
-        "error": result["error"],
+    if not result["ok"]:
+        # Return placeholders so frontend won't crash
+        return {"ok": False, "error": result["error"], "rep_count": 0, "avg_rep_prob": 0.0, "rep_events": []}, None
+
+    summary = {
+        "ok": True,
+        "error": None,
         "rep_count": result["rep_count"],
         "avg_rep_prob": result["avg_rep_prob"],
         "rep_events": result["rep_events"],
-        "annotated_video_path": result["annotated_video_path"],
     }
+    return summary, 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 API") as demo:
+    gr.Markdown("# Pushup Analyzer API\nUpload a video, get rep count + confidence.\n")
 
-    file_in = gr.File(label="Upload video", file_types=["video"])
+    video_file = gr.File(label="Upload .mp4", file_types=["video"])
     btn = gr.Button("Analyze")
-    json_out = gr.JSON(label="Result JSON")
 
-    # api_name is IMPORTANT: frontend will call this endpoint
-    btn.click(fn=api_analyze, inputs=file_in, outputs=json_out, api_name="analyze")
+    out_json = gr.JSON(label="Results JSON")
+    out_video = gr.Video(label="Annotated Output")
+
+    btn.click(
+        fn=api_analyze,
+        inputs=[video_file],
+        outputs=[out_json, out_video],
+        api_name="analyze",  # IMPORTANT: gives you a stable endpoint name
+    )
 
-demo.launch()
+if __name__ == "__main__":
+    demo.launch()