Spaces:
Paused
Paused
File size: 2,902 Bytes
5504846 091117d d5bf444 091117d fba6e1e d5bf444 2d729d2 698a89d 091117d ccd32a3 091117d fba6e1e 091117d 698a89d 1d235a8 091117d fba6e1e 091117d fba6e1e 9085b9a 825522e 091117d 825522e 091117d 825522e 091117d fba6e1e 091117d d5bf444 5504846 d5bf444 5504846 d5bf444 |
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 |
from fastapi import FastAPI, UploadFile, File, Response,Header, BackgroundTasks,Body,HTTPException
from fastapi.staticfiles import StaticFiles
from vitpose import VitPose
from dotenv import load_dotenv
from tasks import process_video,process_salto_alto
from fastapi.responses import JSONResponse
from config import AI_API_TOKEN
from typing import Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI()
vitpose = VitPose()
# vitpose.pipeline.warmup()
load_dotenv()
app.mount("/static", StaticFiles())
@app.get("/")
def read_root():
return {"message": "Hello World"}
@app.get("/test")
def test():
return {"message": "from test"}
@app.post("/upload")
async def upload(background_tasks: BackgroundTasks,
file: UploadFile = File(...),
token: str = Header(...),
user_id: str = Body(...),
player_id: str = Body(...)):
if token != AI_API_TOKEN:
return JSONResponse(content={"message": "Unauthorized", "status": 401})
logger.info("reading contents")
contents = await file.read()
# Save the file to the local directory
logger.info("saving file")
with open(file.filename, "wb") as f:
f.write(contents)
logger.info(f"file saved {file.filename}")
# Create a clone of the file with content already read
background_tasks.add_task(process_video, file.filename, vitpose, user_id,player_id)
# Return the file as a response
return JSONResponse(content={"message": "Video uploaded successfully", "status": 200})
@app.post("/exercise/salto_alto")
async def upload(background_tasks: BackgroundTasks,
file: UploadFile = File(...),
token: str = Header(...),
player_data: str = Body(...),
repetitions: int|str = Body(...),
exercise_id: str = Body(...)
):
import json
player_data = json.loads(player_data)
if token != AI_API_TOKEN:
raise HTTPException(status_code=401, detail="Unauthorized")
logger.info("reading contents")
contents = await file.read()
# Save the file to the local directory
logger.info("saving file")
with open(file.filename, "wb") as f:
f.write(contents)
logger.info(f"file saved {file.filename}")
# Create a clone of the file with content already read
background_tasks.add_task(process_salto_alto,
file.filename,
vitpose,
player_data,
repetitions,
exercise_id)
# Return the file as a response
print(f"returning response")
return JSONResponse(content={"message": "Video uploaded successfully", "status": 200})
|