File size: 3,782 Bytes
36cbda0 |
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 |
# backend/app.py
from fastapi import FastAPI, File, UploadFile, HTTPException, Request
from fastapi.responses import JSONResponse, FileResponse, HTMLResponse
from fastapi.templating import Jinja2Templates
import shutil
import os
from utils.video_processing import track_ball, generate_replay
from utils.drs_logic import predict_impact, is_drs_out
from utils.db import update_leaderboard, get_leaderboard, store_drs_result
from utils.huggingface_api import submit_to_huggingface
app = FastAPI(title="Gully DRS API")
templates = Jinja2Templates(directory="templates")
# Ensure upload and replay directories exist
os.makedirs("uploads", exist_ok=True)
os.makedirs("replays", exist_ok=True)
@app.post("/upload")
async def upload_video(request: Request, file: UploadFile = File(...), player_name: str = "Player1"):
"""
Upload a video for DRS analysis, process it, and store results in Salesforce.
"""
try:
# Save uploaded video
video_path = f"uploads/{file.filename}"
with open(video_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Process video to extract ball trajectory
ball_positions = track_ball(video_path)
# Predict impact and determine DRS decision
impact_point = predict_impact(ball_positions)
decision = "Out" if is_drs_out(impact_point) else "Not Out"
# Generate replay video
replay_path = generate_replay(video_path, ball_positions, decision)
# Submit to HuggingFace API for additional analysis (e.g., speed)
huggingface_response = submit_to_huggingface(video_path)
speed = huggingface_response.get("speed", 0.0)
replay_url = f"/replay/{os.path.basename(replay_path)}"
# Store DRS result in Salesforce (DRS_Result__c)
match_id = store_drs_result(
video_url=video_path,
verdict=decision,
speed=speed,
replay_link=replay_url,
player_name=player_name
)
# Update leaderboard in Salesforce (User_Profile__c)
update_leaderboard(player_name, drs_out=(decision == "Out"))
# Render result page
return templates.TemplateResponse("result.html", {
"request": request,
"decision": decision,
"replay_url": replay_url,
"speed": speed,
"match_id": match_id,
"player_name": player_name
})
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/replay/{replay_name}")
async def get_replay(replay_name: str):
"""
Serve the replay video file.
"""
replay_path = f"replays/{replay_name}"
if not os.path.exists(replay_path):
raise HTTPException(status_code=404, detail="Replay not found")
return FileResponse(replay_path)
@app.get("/leaderboard", response_class=HTMLResponse)
async def get_leaderboard_data(request: Request):
"""
Render the leaderboard page with data from Salesforce.
"""
try:
leaderboard = get_leaderboard()
return templates.TemplateResponse("leaderboard.html", {
"request": request,
"leaderboard": [{
"id": player["Id"],
"name": player["Name__c"],
"matches": player["Matches__c"],
"drs_dismissals": player["LBW_Dismissals__c"],
"score": player["Score__c"]
} for player in leaderboard]
})
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
"""
Render the home page with upload form.
"""
return templates.TemplateResponse("index.html", {"request": request}) |