Spaces:
Sleeping
Sleeping
File size: 4,490 Bytes
88f108f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 |
# app.py
from fastapi import FastAPI, UploadFile, Form, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pathlib import Path
import subprocess
import shutil
import time
import traceback
import cloudinary
import cloudinary.uploader
from io import BytesIO
from huggingface_hub import hf_hub_download
app = FastAPI(
title="🚦 VRU Detection & Forecasting API",
description="Backend API for VRU Detection, Tracking & Forecasting pipeline (Hugging Face Spaces Edition)",
version="1.0.0"
)
# ==============================
# CONFIG
# ==============================
# Configure Cloudinary (you can put these in .env instead)
cloudinary.config(
cloud_name="YOUR_CLOUD_NAME",
api_key="YOUR_API_KEY",
api_secret="YOUR_API_SECRET"
)
SCRIPTS = [
("Video Creation", "video_creation.py"),
("YOLO + DeepSORT Tracking", "yolo_deepsort_tracker.py"),
("Excel Generation", "excel_generation.py"),
("Feature Engineering", "feature_engineering_forecasting.py"),
("Trajectory Forecasting (Transformer)", "vru_forecasting_transformer.py"),
("Trajectory Visualization", "animated_visualization.py")
]
OUTPUT_GIF = Path("trajectory_comparison.gif")
# ==============================
# CORS
# ==============================
app.add_middleware(
CORSMiddleware,
allow_origins=["https://vru-detection.vercel.app/"], # later replace with your frontend URL
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ==============================
# UTILS
# ==============================
def run_pipeline(input_path: str):
"""Sequentially run all VRU pipeline scripts."""
total = len(SCRIPTS)
progress = []
for i, (label, script) in enumerate(SCRIPTS):
stage_info = {
"stage": i + 1,
"label": label,
"status": "running",
"progress": round((i + 1) / total, 2)
}
progress.append(stage_info)
try:
result = subprocess.run(
["python", script, input_path],
capture_output=True, text=True
)
if result.returncode == 0:
stage_info["status"] = "completed"
stage_info["output"] = result.stdout
else:
stage_info["status"] = "failed"
stage_info["error"] = result.stderr
return {"status": "failed", "progress": progress}
except Exception as e:
stage_info["status"] = "error"
stage_info["error"] = str(e)
return {"status": "failed", "progress": progress}
time.sleep(0.5)
return {"status": "completed", "progress": progress}
# ==============================
# ROUTES
# ==============================
@app.get("/")
def home():
return {"message": "🚦 VRU Detection Backend Running on Hugging Face Spaces!"}
@app.post("/run_pipeline/")
async def run_vru_pipeline(
background_tasks: BackgroundTasks,
dataset_name: str = Form(None),
file: UploadFile = None
):
"""
Run full VRU pipeline.
Accepts either uploaded file or Hugging Face dataset name.
"""
try:
# Handle file upload
if file:
temp_path = Path("uploaded_" + file.filename)
with open(temp_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
input_path = str(temp_path)
# Handle Hugging Face dataset file
elif dataset_name:
input_path = hf_hub_download(
repo_id=dataset_name,
filename="ring_side_left/video.mp4" # Adjust if structure differs
)
else:
return JSONResponse({"error": "Please provide a dataset_name or upload a file."}, status_code=400)
# Run pipeline
result = run_pipeline(input_path)
response = {"input_path": input_path, "result": result}
# Upload visualization if available
if OUTPUT_GIF.exists():
upload_result = cloudinary.uploader.upload(
str(OUTPUT_GIF),
resource_type="image",
folder="vru_results/"
)
response["visualization_url"] = upload_result["secure_url"]
return JSONResponse(response)
except Exception as e:
return JSONResponse(
{"error": str(e), "trace": traceback.format_exc()},
status_code=500
)
|