api/api.py

from unittest import result
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi.responses import JSONResponse
from starlette.requests import Request
from fastapi import FastAPI, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from PIL import Image
import io
import time
from model.inference import predict

LAST_CONFIDENT_TS = None
DECAY_SECONDS = 2.0


app = FastAPI()

CURRENT_STATE = {
    "emotion": None,
    "confidence": 0.0
}


app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
def root():
    return {"status": "API running"}

@app.post("/api/predict")
async def predict_emotion(file: UploadFile = File(...)):
    contents = await file.read()
    image = Image.open(io.BytesIO(contents)).convert("RGB")
    emotion = predict(image)
    return {"emotion": emotion}

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.exception_handler(RateLimitExceeded)
def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={"detail": "Too many requests, slow down 😅"}
    )

@app.post("/api/predict")
@limiter.limit("5/minute")
async def predict_emotion(request: Request, file: UploadFile = File(...)):
    contents = await file.read()

    try:
        image = Image.open(io.BytesIO(contents)).convert("RGB")
    except Exception:
        return {
            "state": "error",
            "reason": "invalid_image"
        }

    result = predict(image)

    if result["confidence"] >= 0.6:
        RECENT_PREDICTIONS.append(result["emotion"])
        if len(RECENT_PREDICTIONS) > WINDOW_SIZE:
            RECENT_PREDICTIONS.pop(0)


    if result["confidence"] < 0.6:
        return {
            "state": "uncertain",
            "emotion": CURRENT_STATE["emotion"],
            "confidence": result["confidence"],
            "is_confident": False
        }

    # update memory
    if RECENT_PREDICTIONS:
        dominant_emotion = max(
            set(RECENT_PREDICTIONS),
            key=RECENT_PREDICTIONS.count
        )

        # update memory
        CURRENT_STATE["emotion"] = dominant_emotion
        CURRENT_STATE["confidence"] = result["confidence"]
        LAST_CONFIDENT_TS = time.time()

    RECENT_PREDICTIONS = []
    WINDOW_SIZE = 5

    if LAST_CONFIDENT_TS is not None:
        if time.time() - LAST_CONFIDENT_TS > DECAY_SECONDS:
            CURRENT_STATE["emotion"] = None
            CURRENT_STATE["confidence"] = 0.0


    return {
        "state": "stable" if CURRENT_STATE["emotion"] else "unknown",
        "emotion": CURRENT_STATE["emotion"],
        "confidence": CURRENT_STATE["confidence"],
        "is_confident": CURRENT_STATE["emotion"] is not None
    }