Spaces:

Simon9
/

afrigoals

Running

App Files Files Community

Simon9 commited on 18 days ago

Commit

655c5fa

verified ·

1 Parent(s): ca40031

Update app.py

Browse files

Files changed (1) hide show

app.py +451 -291

app.py CHANGED Viewed

@@ -1,303 +1,463 @@
-# app.py
 import os
 import uuid
-import json
 import shutil
-import threading
-from fastapi import FastAPI, UploadFile, File
-from fastapi.responses import (
-    JSONResponse,
-    FileResponse,
-    HTMLResponse,
-    PlainTextResponse,
-    RedirectResponse,
-)
-from fastapi.middleware.cors import CORSMiddleware
 from pipeline_full import run_full_pipeline
 BASE_RESULTS_DIR = "jobs"
-app = FastAPI(title="Football Analytics Space")
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],  # lock down to your backend/nextjs domains later
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-# -------------------- UI entrypoint --------------------
-@app.get("/")
-def root_ui():
-    html = """
-    <html>
-      <head>
-        <title>Football Analytics</title>
-        <style>
-          body { font-family: sans-serif; padding: 20px; }
-          input[type=file] { margin-bottom: 10px; }
-          pre { background: #f3f3f3; padding: 10px; border-radius: 4px; }
-        </style>
-      </head>
-      <body>
-        <h1>⚽ Football Analytics</h1>
-        <p>Upload a match video to run the full pipeline (detections, stats, radar, heatmaps, Voronoi, ball path, clustering).</p>
-        <form action="/analyze-video-ui" method="post" enctype="multipart/form-data">
-          <input type="file" name="file" accept="video/*" required />
-          <button type="submit">Analyze</button>
-        </form>
-      </body>
-    </html>
     """
-    return HTMLResponse(html)
-# -------------------- Start a job (UI) --------------------
-@app.post("/analyze-video-ui")
-async def analyze_video_ui(file: UploadFile = File(...)):
-    job_id = str(uuid.uuid4())
-    job_dir = os.path.join(BASE_RESULTS_DIR, job_id)
-    os.makedirs(job_dir, exist_ok=True)
-    video_path = os.path.join(job_dir, file.filename)
-    with open(video_path, "wb") as f:
-        shutil.copyfileobj(file.file, f)
-    # Run pipeline in background so we can show progress page
-    def worker():
-        from pipeline_full import run_full_pipeline  # import inside to avoid circular imports
-        try:
-            run_full_pipeline(video_path, job_dir)
-        except Exception as e:
-            # write error to status
-            status_path = os.path.join(job_dir, "status.json")
-            with open(status_path, "w", encoding="utf-8") as f:
-                json.dump(
-                    {"stage": "error", "progress": 1.0, "message": str(e)},
-                    f,
-                )
-    threading.Thread(target=worker, daemon=True).start()
-    # Redirect immediately to job progress page
-    return RedirectResponse(f"/job/{job_id}", status_code=303)
-# -------------------- Job progress page --------------------
-@app.get("/job/{job_id}")
-def job_page(job_id: str):
-    html = f"""
-    <html>
-      <head>
-        <title>Job {job_id} - Progress</title>
-        <style>
-          body {{ font-family: sans-serif; padding: 20px; }}
-          #message {{ margin-top: 10px; }}
-          #progress-container {{ margin-top: 20px; width: 100%; max-width: 400px; }}
-          progress {{ width: 100%; height: 20px; }}
-        </style>
-      </head>
-      <body>
-        <h1>Job {job_id}</h1>
-        <p>Processing video... This may take a while for long clips.</p>
-        <div id="progress-container">
-          <progress id="progress-bar" value="0" max="1"></progress>
-          <div id="message">Starting...</div>
-        </div>
-        <script>
-          async function pollStatus() {{
-            try {{
-              const res = await fetch("/status/{job_id}");
-              if (!res.ok) throw new Error("status error");
-              const data = await res.json();
-              document.getElementById("progress-bar").value = data.progress || 0;
-              document.getElementById("message").innerText = data.stage + ": " + (data.message || "");
-              if (data.stage === "done") {{
-                window.location.href = "/job-result/{job_id}";
-              }} else if (data.stage === "error") {{
-                document.getElementById("message").innerText = "Error: " + data.message;
-              }} else {{
-                setTimeout(pollStatus, 2000);
-              }}
-            }} catch (e) {{
-              document.getElementById("message").innerText = "Error fetching status. Retrying...";
-              setTimeout(pollStatus, 5000);
-            }}
-          }}
-          pollStatus();
-        </script>
-      </body>
-    </html>
     """
-    return HTMLResponse(html)
-# -------------------- Status endpoint (for progress polling) --------------------
-@app.get("/status/{job_id}")
-def job_status(job_id: str):
-    job_dir = os.path.join(BASE_RESULTS_DIR, job_id)
-    status_path = os.path.join(job_dir, "status.json")
-    if not os.path.exists(job_dir):
-        return JSONResponse(
-            {"stage": "unknown", "progress": 0.0, "message": "Job not found"},
-            status_code=404,
-        )
-    if not os.path.exists(status_path):
-        # job queued / just started
-        return JSONResponse(
-            {"stage": "pending", "progress": 0.0, "message": "Queued"},
-            status_code=200,
-        )
-    with open(status_path, "r", encoding="utf-8") as f:
-        status = json.load(f)
-    return JSONResponse(status)
-# -------------------- Final result page (UI) --------------------
-@app.get("/job-result/{job_id}")
-def job_result(job_id: str):
-    job_dir = os.path.join(BASE_RESULTS_DIR, job_id)
-    result_path = os.path.join(job_dir, "result.json")
-    if not os.path.exists(result_path):
-        return PlainTextResponse("Result not ready yet or job not found.", status_code=404)
-    with open(result_path, "r", encoding="utf-8") as f:
-        result = json.load(f)
-    basic = result["basic"]
-    adv = result["advanced"]
-    ball = result["ball"]
-    stats = result["stats"]
-    siglip_html = result["siglip_html"]
-    def make_link(path, label=None):
-        if not path or not os.path.exists(path):
-            return ""
-        rel = os.path.relpath(path, BASE_RESULTS_DIR)
-        url = f"/results/{rel}"
-        label = label or os.path.basename(path)
-        return f'<li><a href="{url}" target="_blank">{label}</a></li>'
-    html = f"""
-    <html>
-      <head>
-        <title>Football Analytics - Result {job_id}</title>
-        <style>
-          body {{ font-family: sans-serif; padding: 20px; }}
-          pre {{ background: #f3f3f3; padding: 10px; border-radius: 4px; max-height: 400px; overflow: auto; }}
-          h2 {{ margin-top: 30px; }}
-        </style>
-      </head>
-      <body>
-        <h1>Result for job {job_id}</h1>
-        <h2>Basic Frames</h2>
-        <ul>
-          {make_link(basic["raw_frame"], "Raw Frame")}
-          {make_link(basic["boxes_labels"], "Detections with Boxes + Labels")}
-          {make_link(basic["ball_players"], "Ball vs Players (ellipse/triangle)")}
-        </ul>
-        <h2>Advanced Views</h2>
-        <ul>
-          {make_link(adv["frame_advanced"], "Annotated Frame with Teams + IDs")}
-          {make_link(adv["radar"], "Radar View")}
-          {make_link(adv["voronoi"], "Voronoi Diagram")}
-          {make_link(adv["voronoi_blended"], "Voronoi Diagram (Blended)")}
-        </ul>
-        <h2>Ball Path</h2>
-        <ul>
-          {make_link(ball["ball_path_raw_img"], "Ball Path (Raw)")}
-          {make_link(ball["ball_path_cleaned_img"], "Ball Path (Cleaned)")}
-        </ul>
-        <h2>SigLIP Clustering</h2>
-        <ul>
-          {make_link(siglip_html, "Open 3D Clustering Visualization")}
-        </ul>
-        <h2>Stats JSON</h2>
-        <pre>{json.dumps(stats, indent=2)}</pre>
-        <h2>API Usage</h2>
-        <p>Your backend can call <code>POST /analyze-video</code> with a file to get the JSON response (same structure) with URLs to all artifacts:</p>
-        <pre>
-POST /analyze-video
-  - form-data: file=@your_video.mp4
-Artifacts can be accessed via:
-GET /results/&lt;path under jobs/&gt;
-Example: /results/{os.path.relpath(basic["raw_frame"], BASE_RESULTS_DIR)}
-        </pre>
-      </body>
-    </html>
     """
-    return HTMLResponse(html)
-# -------------------- Public API for backend / Next.js --------------------
-@app.post("/analyze-video")
-async def analyze_video_api(file: UploadFile = File(...)):
-    job_id = str(uuid.uuid4())
-    job_dir = os.path.join(BASE_RESULTS_DIR, job_id)
-    os.makedirs(job_dir, exist_ok=True)
-    video_path = os.path.join(job_dir, file.filename)
-    with open(video_path, "wb") as f:
-        shutil.copyfileobj(file.file, f)
-    # This one runs synchronously (for backend usage)
-    result = run_full_pipeline(video_path, job_dir)
-    def to_url(path: str) -> str:
-        if not path:
-            return ""
-        rel = os.path.relpath(path, BASE_RESULTS_DIR)
-        return f"/results/{rel}"
-    response = {
-        "job_id": job_id,
-        "basic": {k: to_url(v) for k, v in result["basic"].items()},
-        "advanced": {k: to_url(v) for k, v in result["advanced"].items()},
-        "ball": {
-            "ball_path_raw_img": to_url(result["ball"]["ball_path_raw_img"]),
-            "ball_path_cleaned_img": to_url(result["ball"]["ball_path_cleaned_img"]),
-            "ball_path_cleaned_coords": result["ball"]["ball_path_cleaned_coords"],
-        },
-        "stats": result["stats"],
-        "siglip_html": to_url(result["siglip_html"]),
-    }
-    return JSONResponse(response)
-# -------------------- Serve artifacts under /results/... --------------------
-@app.get("/results/{path:path}")
-def get_result_file(path: str):
-    full_path = os.path.join(BASE_RESULTS_DIR, path)
-    if not os.path.exists(full_path):
-        return PlainTextResponse("Not found", status_code=404)
-    return FileResponse(full_path)

 import os
+# Suppress inference model warnings (must be at the very top)
+os.environ["CORE_MODEL_SAM_ENABLED"] = "False"
+os.environ["CORE_MODEL_SAM2_ENABLED"] = "False"
+os.environ["CORE_MODEL_SAM3_ENABLED"] = "False"
+os.environ["CORE_MODEL_GAZE_ENABLED"] = "False"
+os.environ["CORE_MODEL_GROUNDINGDINO_ENABLED"] = "False"
+os.environ["CORE_MODEL_YOLO_WORLD_ENABLED"] = "False"
+import gradio as gr
 import uuid
 import shutil
+import json
+import time
+from pathlib import Path
+from PIL import Image
+# Import your existing pipeline
 from pipeline_full import run_full_pipeline
 BASE_RESULTS_DIR = "jobs"
+Path(BASE_RESULTS_DIR).mkdir(exist_ok=True)
+def monitor_progress(job_dir):
     """
+    Monitor the status.json file and yield progress updates
     """
+    status_file = os.path.join(job_dir, "status.json")
+    last_progress = 0
+    last_stage = ""
+    # Keep checking until done or error
+    max_wait = 300  # 5 minutes max
+    start_time = time.time()
+    while (time.time() - start_time) < max_wait:
+        if os.path.exists(status_file):
+            try:
+                with open(status_file, 'r') as f:
+                    status = json.load(f)
+                stage = status.get('stage', '')
+                progress = status.get('progress', 0)
+                message = status.get('message', '')
+                # Only yield if something changed
+                if stage != last_stage or progress != last_progress:
+                    last_stage = stage
+                    last_progress = progress
+                    # Map stages to friendly names with emojis
+                    stage_map = {
+                        'initializing': '🚀 Initializing',
+                        'siglip': '🤖 SigLIP Clustering',
+                        'team_classifier': '👕 Team Classification',
+                        'basic_frames': '🎯 Basic Detection',
+                        'advanced_views': '🎨 Tactical Views',
+                        'ball_path': '⚽ Ball Tracking',
+                        'stats': '📊 Statistics',
+                        'done': '✅ Complete',
+                        'error': '❌ Error'
+                    }
+                    friendly_stage = stage_map.get(stage, stage)
+                    percentage = int(progress * 100)
+                    status_msg = f"**[{percentage}%]** {friendly_stage}"
+                    if message:
+                        status_msg += f"\n\n_{message}_"
+                    yield status_msg
+                    # Stop if done or error
+                    if stage in ['done', 'error']:
+                        break
+            except Exception as e:
+                print(f"Error reading status: {e}")
+        time.sleep(1)  # Poll every second
+    # Final status
+    yield "✅ **Processing complete!** Loading results..."
+def analyze_video(video_file):
     """
+    Main analysis function with real-time progress monitoring
+    """
+    # Initial empty outputs
+    empty_outputs = (None, None, None, None, None, None, None, None)
+    if video_file is None:
+        yield (*empty_outputs, "❌ **Please upload a video file**")
+        return
+    try:
+        # Create job directory
+        job_id = str(uuid.uuid4())
+        job_dir = os.path.join(BASE_RESULTS_DIR, job_id)
+        os.makedirs(job_dir, exist_ok=True)
+        yield (*empty_outputs, f"🔧 **[0%]** Setting up job `{job_id[:8]}...`")
+        # Copy video file
+        video_input_path = video_file
+        if not os.path.exists(video_input_path):
+            yield (*empty_outputs, f"❌ **Video file not found:** `{video_input_path}`")
+            return
+        video_filename = f"input_{uuid.uuid4().hex[:8]}.mp4"
+        video_path = os.path.join(job_dir, video_filename)
+        shutil.copy2(video_input_path, video_path)
+        yield (*empty_outputs, f"📁 **[5%]** Video copied successfully")
+        # Start pipeline in separate thread
+        import threading
+        pipeline_error = []
+        pipeline_complete = []
+        def run_pipeline_thread():
+            try:
+                result = run_full_pipeline(video_path, job_dir)
+                pipeline_complete.append(result)
+            except Exception as e:
+                pipeline_error.append(str(e))
+        thread = threading.Thread(target=run_pipeline_thread, daemon=True)
+        thread.start()
+        # Monitor progress while pipeline runs
+        for progress_msg in monitor_progress(job_dir):
+            yield (*empty_outputs, progress_msg)
+        # Wait for thread to complete
+        thread.join(timeout=10)
+        # Check for errors
+        if pipeline_error:
+            error_msg = f"""❌ **Error during analysis:**
+```
+{pipeline_error[0]}
+```
+**Troubleshooting:**
+- Ensure ROBOFLOW_API_KEY is set in Space secrets
+- Try a shorter video clip
+- Check container logs for details
+"""
+            yield (*empty_outputs, error_msg)
+            return
+        # Load results
+        result_path = os.path.join(job_dir, "result.json")
+        if not os.path.exists(result_path):
+            yield (*empty_outputs, "⚠️ **Result file not found.** Pipeline may still be processing.")
+            return
+        with open(result_path, 'r') as f:
+            result = json.load(f)
+        # Extract results
+        basic = result["basic"]
+        adv = result["advanced"]
+        ball = result["ball"]
+        stats = result["stats"]
+        siglip_html = result["siglip_html"]
+        # Load images
+        def load_image(path, name="image"):
+            if not path or not os.path.exists(path):
+                print(f"⚠️ {name} not found at {path}")
+                return None
+            try:
+                img = Image.open(path)
+                if img.mode != 'RGB':
+                    img = img.convert('RGB')
+                print(f"✅ Loaded {name}: {img.size}")
+                return img
+            except Exception as e:
+                print(f"❌ Error loading {name}: {e}")
+                return None
+        print("=" * 60)
+        print("LOADING RESULT IMAGES...")
+        raw_frame = load_image(basic["raw_frame"], "Raw frame")
+        boxes_labels = load_image(basic["boxes_labels"], "Boxes/labels")
+        ball_players = load_image(basic["ball_players"], "Ball/players")
+        frame_advanced = load_image(adv["frame_advanced"], "Advanced frame")
+        radar = load_image(adv["radar"], "Radar")
+        voronoi = load_image(adv["voronoi"], "Voronoi")
+        voronoi_blended = load_image(adv["voronoi_blended"], "Voronoi blended")
+        ball_path_cleaned = load_image(ball["ball_path_cleaned_img"], "Ball path")
+        # Format stats
+        stats_text = json.dumps(stats, indent=2)
+        # Count loaded images
+        images_loaded = sum([
+            raw_frame is not None,
+            boxes_labels is not None,
+            ball_players is not None,
+            frame_advanced is not None,
+            radar is not None,
+            voronoi_blended is not None,
+            ball_path_cleaned is not None
+        ])
+        print(f"Images loaded: {images_loaded}/7")
+        print("=" * 60)
+        # Create success message
+        clustering_link = ""
+        if siglip_html and os.path.exists(siglip_html):
+            rel_path = os.path.relpath(siglip_html, ".")
+            clustering_link = f'\n\n📊 <a href="file/{rel_path}" target="_blank">**View 3D Clustering Visualization**</a>'
+        success_msg = f"""✅ **[100%] Analysis Complete!**
+**Job ID:** `{job_id}`
+**Results Generated:**
+- ✅ Player detections ({images_loaded}/7 images loaded)
+- ✅ Team classifications
+- ✅ Tactical visualizations
+- ✅ Ball trajectory analysis
+- ✅ Match statistics
+{clustering_link}
+---
+*Scroll through the tabs above to see all visualizations*
+**Note:** If images appear dark, try uploading a brighter video with good lighting.
+"""
+        # Final yield with all results
+        yield (
+            raw_frame,
+            boxes_labels,
+            ball_players,
+            frame_advanced,
+            radar,
+            voronoi_blended,
+            ball_path_cleaned,
+            stats_text,
+            success_msg
+        )
+    except Exception as e:
+        import traceback
+        error_detail = traceback.format_exc()
+        error_msg = f"""❌ **Unexpected Error:**
+```python
+{str(e)}
+```
+**Debug Info:**
+- Video: `{video_file if video_file else 'None'}`
+- Job: `{job_dir if 'job_dir' in locals() else 'Not created'}`
+**Full Traceback:**
+```
+{error_detail}
+```
+"""
+        print("=" * 60)
+        print("UNEXPECTED ERROR:")
+        print(error_detail)
+        print("=" * 60)
+        yield (*empty_outputs, error_msg)
+# Create Gradio interface
+with gr.Blocks(
+    title="⚽ Afrigoals - Football Analytics",
+    theme=gr.themes.Soft()
+) as demo:
+    gr.Markdown("""
+    # ⚽ Afrigoals - Football Analytics Platform
+    AI-powered match analysis using computer vision and machine learning.
+    **Features:**
+    - 🎯 Player detection and tracking
+    - 👕 Automatic team classification
+    - ⚽ Ball path tracking
+    - 🎨 Tactical visualizations (radar, Voronoi)
+    - 📊 3D clustering analysis
+    - 📈 Match statistics
+    ---
+    """)
+    with gr.Row():
+        with gr.Column(scale=1):
+            video_input = gr.Video(
+                label="📹 Upload Match Video",
+                sources=["upload"],
+            )
+            gr.Markdown("""
+            **Supported formats:** MP4, AVI, MOV, MKV
+            **Best results with:**
+            - 30-60 second clips
+            - Good lighting (daytime matches)
+            - Clear view of the pitch
+            - 720p or higher resolution
+            **Progress updates appear below** 👇
+            """)
+            analyze_btn = gr.Button(
+                "🔍 Analyze Video",
+                variant="primary",
+                size="lg"
+            )
+    with gr.Row():
+        with gr.Column():
+            gr.Markdown("### 📊 Progress & Status")
+            status_output = gr.Markdown(
+                value="⏳ **Ready** - Upload a video and click Analyze to start"
+            )
+    gr.Markdown("---")
+    gr.Markdown("## 📊 Analysis Results")
+    gr.Markdown("*Results will appear here after analysis (typically 30-90 seconds)*")
+    with gr.Tabs():
+        with gr.Tab("🎯 Basic Detections"):
+            gr.Markdown("### Player and Ball Detection Results")
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("#### Raw Frame")
+                    raw_frame_output = gr.Image(
+                        label="Original Frame",
+                        type="pil"
+                    )
+                with gr.Column():
+                    gr.Markdown("#### Detections with Boxes")
+                    boxes_output = gr.Image(
+                        label="Bounding Boxes + Labels",
+                        type="pil"
+                    )
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("#### Ball vs Players")
+                    ball_players_output = gr.Image(
+                        label="Ball & Players Markers",
+                        type="pil"
+                    )
+        with gr.Tab("🎨 Advanced Views"):
+            gr.Markdown("### Tactical Analysis")
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("#### Annotated Frame with Teams")
+                    advanced_output = gr.Image(
+                        label="Teams + Player IDs",
+                        type="pil"
+                    )
+                with gr.Column():
+                    gr.Markdown("#### Radar View")
+                    radar_output = gr.Image(
+                        label="Top-Down Tactical View",
+                        type="pil"
+                    )
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("#### Voronoi Diagram - Space Control")
+                    voronoi_output = gr.Image(
+                        label="Territory Control",
+                        type="pil"
+                    )
+        with gr.Tab("⚽ Ball Analysis"):
+            gr.Markdown("### Ball Movement Tracking")
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("#### Ball Path (Cleaned)")
+                    ball_path_output = gr.Image(
+                        label="Ball Trajectory",
+                        type="pil"
+                    )
+        with gr.Tab("📈 Statistics"):
+            gr.Markdown("### Match Statistics")
+            stats_output = gr.Code(
+                label="JSON Data",
+                language="json",
+                lines=25
+            )
+    # Wire up the button
+    analyze_btn.click(
+        fn=analyze_video,
+        inputs=[video_input],
+        outputs=[
+            raw_frame_output,
+            boxes_output,
+            ball_players_output,
+            advanced_output,
+            radar_output,
+            voronoi_output,
+            ball_path_output,
+            stats_output,
+            status_output
+        ]
+    )
+    gr.Markdown("""
+    ---
+    ## 📝 Information
+    ### ⏱️ Processing Time
+    | Video Length | Est. Time |
+    |--------------|-----------|
+    | 30 seconds   | 30-60s    |
+    | 1 minute     | 60-120s   |
+    | 2 minutes    | 2-4 min   |
+    ### 🔧 Technical Stack
+    - **Detection**: Roboflow YOLO models
+    - **Tracking**: ByteTrack algorithm
+    - **Embeddings**: SigLIP
+    - **Clustering**: UMAP + K-means
+    - **Visualization**: Supervision, Plotly, OpenCV
+    ### ⚠️ Requirements
+    - ROBOFLOW_API_KEY must be set in Space secrets
+    - GPU recommended for faster processing
+    - Good lighting improves detection quality
+    ---
+    **Built with ❤️ for African Football Analytics**
+    """)
+# Launch
+if __name__ == "__main__":
+    demo.queue()  # Enable queuing for progress
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        show_error=True
+    )