File size: 2,103 Bytes
b87deef
 
cd7b3bb
0edd049
b87deef
0edd049
 
 
790a68a
b87deef
067d3bb
 
 
b87deef
0edd049
515d778
b87deef
 
 
 
 
 
 
515d778
 
 
 
790a68a
515d778
b87deef
515d778
b87deef
 
 
2f18963
 
b87deef
cd7b3bb
b87deef
 
 
 
 
 
515d778
 
 
 
790a68a
515d778
b87deef
 
 
 
 
 
2f18963
 
b87deef
2f18963
 
cd7b3bb
 
 
 
 
575359b
 
 
b87deef
 
 
 
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
import os

import numpy as np
from fastapi import Depends, FastAPI, File, Response, UploadFile, status

from auth import validate_token
from model.model import Model
from routers.fecnet_router import router as fecnet_router
from schema import EmbeddingResponse, SimilarityResponse

app = FastAPI(
    title="Facial Expression Embedding Service",
)

app.include_router(fecnet_router)

model = Model(
    os.getenv("MODEL_REPO_ID", ""),
    os.getenv("MODEL_FILENAME", ""),
    os.getenv("HF_TOKEN", ""),
)


@app.post(
    "/embed",
    status_code=status.HTTP_200_OK,
    dependencies=[Depends(validate_token)],
    response_model=EmbeddingResponse,
)
async def calculate_embedding(
    image: UploadFile = File(...),
):
    try:
        image_content = await image.read()
        if isinstance(image_content, str):
            image_content = image_content.encode()
        pred = model.predict(model.preprocess(image_content))
        return EmbeddingResponse(embedding=pred[0].tolist())
    except Exception as e:
        return Response(
            content=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
        )


@app.post(
    "/similarity",
    status_code=status.HTTP_200_OK,
    dependencies=[Depends(validate_token)],
    response_model=SimilarityResponse,
)
async def calculate_similarity_score(
    image1: UploadFile = File(...),
    image2: UploadFile = File(...),
):
    try:
        image1_content = await image1.read()
        if isinstance(image1_content, str):
            image1_content = image1_content.encode()
        image2_content = await image2.read()
        if isinstance(image2_content, str):
            image2_content = image2_content.encode()
        pred = model.predict(
            np.vstack(
                [model.preprocess(image1_content), model.preprocess(image2_content)]
            )
        )
        return SimilarityResponse(
            score=float(model.cosine_similarity(pred[0], pred[1]))
        )
    except Exception as e:
        return Response(
            content=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
        )