Spaces:
Paused
Paused
File size: 3,232 Bytes
fccef52 091117d d5bf444 091117d fba6e1e 2d729d2 d1dd306 fccef52 2d729d2 698a89d 091117d ccd32a3 091117d fba6e1e 091117d 698a89d 1d235a8 091117d fba6e1e 091117d fba6e1e fccef52 825522e 091117d 825522e 091117d 825522e 091117d fccef52 091117d fccef52 d5bf444 d1dd306 d5bf444 5504846 d5bf444 fccef52 d5bf444 5504846 fccef52 d1dd306 |
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 |
from fastapi import (FastAPI,
UploadFile,
File,
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
import logging
import json
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(...)
):
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,
exercise_id,
repetitions)
# Return the file as a response
print(f"returning response")
return JSONResponse(content={"message": "Video uploaded successfully",
"status": 200})
|