File size: 2,627 Bytes
fa2fb5f
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
# FastAPI Space: username + image -> decision, score, threshold, jwt
import os, io, json, time
import numpy as np
from fastapi import FastAPI, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware
from PIL import Image
import jwt

# ---- config ----
MODELS_DIR = os.environ.get('MODELS_DIR', 'models')           # put gallery/labels/threshold here
JWT_SECRET = os.environ.get('PORTAL_SECRET', 'change-me')     # set a real secret in Space settings
JWT_EXP_SECS = int(os.environ.get('JWT_EXP_SECS', '600'))     # 10 min

# ---- load artifacts ----
import json as _json
labels = _json.load(open(f"{MODELS_DIR}/labels.json","r",encoding="utf-8"))
gallery = np.load(f"{MODELS_DIR}/gallery_mean.npy")
thr = _json.load(open(f"{MODELS_DIR}/threshold.json","r"))["cosine_threshold"]

# ---- embedder ----
class FaceEmbedder:
    def __init__(self, provider='CPUExecutionProvider'):
        import insightface
        from insightface.app import FaceAnalysis
        self.fa = FaceAnalysis(name='buffalo_l', providers=[provider])
        self.fa.prepare(ctx_id=-1, det_size=(640,640))
    def __call__(self, img_rgb: np.ndarray):
        faces = self.fa.get(img_rgb)
        if not faces: return None
        f = max(faces, key=lambda z: (z.bbox[2]-z.bbox[0])*(z.bbox[3]-z.bbox[1]))
        return f.normed_embedding.astype('float32')

EMB = FaceEmbedder()

# ---- api ----
app = FastAPI(title="Access Gate")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"], allow_credentials=True,
    allow_methods=["*"], allow_headers=["*"]
)

def issue_token(username: str):
    payload = {"sub": username, "iat": int(time.time()), "exp": int(time.time()) + JWT_EXP_SECS}
    return jwt.encode(payload, JWT_SECRET, algorithm="HS256")

@app.get("/labels")
def get_labels():
    return {"labels": labels}

@app.post("/verify")
async def verify(username: str = Form(...), image: UploadFile = File(...)):
    # read image
    arr = np.array(Image.open(io.BytesIO(await image.read())).convert("RGB"))
    # checks
    if username not in labels:
        return {"decision": "Not Found", "score": 0.0, "threshold": thr, "reason": "Unknown user"}

    emb = EMB(arr)
    if emb is None:
        return {"decision": "Invalid", "score": 0.0, "threshold": thr, "reason": "No face detected"}

    idx = labels.index(username)
    score = float(emb @ gallery[idx])
    decision = "Allow" if score >= thr else "Deny"
    token = issue_token(username) if decision == "Allow" else ""

    return {
        "decision": decision,
        "score": score,
        "threshold": thr,
        "reason": "",
        "token": token
    }