Spaces:
Sleeping
Sleeping
File size: 3,097 Bytes
7fd360d b0d9851 7fd360d | 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 | import base64
import logging
from pathlib import Path
from typing import Any
from fastapi import FastAPI, File, HTTPException, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from api_server import (
MODEL_STATUS,
analyze_emotion_base64,
build_error_response,
get_cnn_model_repo,
get_model_endpoint,
get_wav2vec_model_id,
)
logger = logging.getLogger("voice-emotion-fastapi")
BASE_DIR = Path(__file__).resolve().parent
FRONTEND_DIR = BASE_DIR / "frontend"
app = FastAPI(title="Voice Emotion Analysis")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")
@app.get("/")
async def index() -> FileResponse:
return FileResponse(FRONTEND_DIR / "index.html")
@app.api_route("/ping", methods=["GET", "HEAD"])
async def ping() -> dict[str, str]:
"""Lightweight endpoint for uptime bots (e.g. UptimeRobot)."""
return {"status": "alive"}
@app.get("/health")
async def health() -> dict[str, Any]:
return {
"status": "healthy",
"cnnEndpoint": get_model_endpoint("cnn") or "",
"wav2vecEndpoint": get_model_endpoint("wav2vec") or "",
"cnnModel": get_cnn_model_repo(),
"wav2vecModel": get_wav2vec_model_id(),
"modelStatus": MODEL_STATUS,
}
def get_audio_from_payload(payload: dict[str, Any]) -> str:
if "audio" in payload and isinstance(payload["audio"], str):
return payload["audio"]
data = payload.get("data")
if isinstance(data, list) and data and isinstance(data[0], str):
return data[0]
raise ValueError("Request must include audio as base64 in `audio` or `data[0]`.")
@app.post("/api/predict")
async def predict(request: Request) -> JSONResponse:
try:
payload = await request.json()
audio_base64 = get_audio_from_payload(payload)
return JSONResponse(analyze_emotion_base64(audio_base64))
except ValueError as error:
return JSONResponse(
build_error_response(str(error), warnings=[str(error)]),
status_code=400,
)
except Exception as error:
logger.exception("FastAPI prediction failed")
return JSONResponse(
build_error_response(
f"Server error: {error}",
warnings=[f"Server error: {error}"],
),
status_code=500,
)
@app.post("/api/analyze")
async def analyze_upload(audio: UploadFile = File(...)) -> JSONResponse:
content = await audio.read()
if not content:
raise HTTPException(status_code=400, detail="Uploaded audio file is empty.")
audio_base64 = base64.b64encode(content).decode("utf-8")
return JSONResponse(analyze_emotion_base64(audio_base64))
if __name__ == "__main__":
import uvicorn
uvicorn.run("fastapi_server:app", host="0.0.0.0", port=7860, reload=False)
|