flask/app.py

import sys
sys.path.append('../')

import os
import base64
import json
import cv2
import numpy as np
from time import gmtime, strftime
from pydantic import BaseModel
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import JSONResponse
from typing import Dict

from engine.header import *

file_path = os.path.abspath(__file__)
dir_path = os.path.dirname(file_path)
root_path = os.path.dirname(dir_path)

SPOOF_THRESHOLD = 0.5

version = get_version().decode('utf-8')
print_info('\t <Recognito Liveness> \t version {}'.format(version))

device_id = get_deviceid().decode('utf-8')
print_info('\t <Hardware ID> \t\t {}'.format(device_id))

def activate_sdk():
    online_key = os.environ.get("FL_LICENSE_KEY")
    offline_key_path = os.path.join(root_path, "license.txt")
    dict_path = os.path.join(root_path, "engine/bin")

    ret = -1
    if online_key is None:
        print_warning("Liveness online license key not found!")
    else:
        ret = init_sdk(dict_path.encode('utf-8'), online_key.encode('utf-8'))

    if ret == 0:
        print_log("Successfully online init SDK!")
    else:
        print_error(f"Failed to online init SDK, Error code {ret}\n Trying offline init SDK...");
        if os.path.exists(offline_key_path) is False:
            print_warning("Liveness offline license key file not found!")
            print_error(f"Falied to offline init SDK, Error code {ret}")
            return ret
        else:
            ret = init_sdk_offline(dict_path.encode('utf-8'), offline_key_path.encode('utf-8'))
            if ret == 0:
                print_log("Successfully offline init SDK!")
            else:
                print_error(f"Falied to offline init SDK, Error code {ret}")
                return ret
    return ret

def generate_response(result, face_rect, score, angles):
    status = "ok"
    data = {
        "status": status, 
        "data": {}
    }

    data["data"]["result"] = result    

    if score is not None:
        data["data"]["liveness_score"] = score

    if face_rect is not None:
        data["data"]["face_rect"] = {
            "x": int(face_rect[0]), 
            "y": int(face_rect[1]), 
            "w": int(face_rect[2] - face_rect[0] + 1), 
            "h": int(face_rect[3] - face_rect[1] + 1)
        }

    if angles is not None:
        data["data"]["angles"] = {
            "yaw": angles[0], 
            "roll": angles[1], 
            "pitch": angles[2]
        }

    return JSONResponse(content=data, status_code=200)

app = FastAPI()

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

def read_image(file: UploadFile) -> np.ndarray:
    # Read the image file and convert it to OpenCV format
    image_bytes = file.file.read()
    image_np = np.frombuffer(image_bytes, np.uint8)
    image = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
    return image

@app.post("/api/check_liveness")
async def check_liveness_api(
    image: UploadFile = File(...)
) -> JSONResponse:
    try:
        image_mat = read_image(image)
    except:
        response = generate_response("Failed to open file!", None, None, None)
        return response
    
    result, face_rect, score, angles = check_liveness(image_mat, SPOOF_THRESHOLD)
    response = generate_response(result, face_rect, score, angles)
    return response

def decode_base64_image(base64_string: str) -> np.ndarray:
    try:
        image_data = base64.b64decode(base64_string)
        image_np = np.frombuffer(image_data, np.uint8)
        image = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
        if image is None:
            raise ValueError("Decoded image is None")
        return image
    except Exception as e:
        raise ValueError(f"Failed to decode base64 image: {str(e)}")

class CheckLivenessRequest(BaseModel):
    image: str

@app.post("/api/check_liveness_base64")
async def check_liveness_base64_api(request: CheckLivenessRequest) -> JSONResponse:
    try: 
        image_mat = decode_base64_image(request.image)
    except:
        response = generate_response("Failed to open file!", None, None, None)
        return response
    
    result, face_rect, score, angles = check_liveness(image_mat, SPOOF_THRESHOLD)
    response = generate_response(result, face_rect, score, angles)
    return response

if __name__ == '__main__':
    ret = activate_sdk()
    if ret != 0:
        exit(-1)
        
    dummy_interface = gr.Interface(
        fn=lambda x: "API ready.",
        inputs=gr.Textbox(label="Info"),
        outputs=gr.Textbox(label="Response"),
        allow_flagging="never"  # 🚫 disables writing to `flagged/`
    )
    
    gr_app = gr.mount_gradio_app(app, dummy_interface, path="/gradio")
    
    import uvicorn
    uvicorn.run(gr_app, host="0.0.0.0", port=7860)