blakeblackshear.frigate/frigate/api/classification.py

"""Object classification APIs."""

import logging
import os

from fastapi import APIRouter, Request, UploadFile
from fastapi.responses import JSONResponse

from frigate.api.defs.tags import Tags
from frigate.const import FACE_DIR
from frigate.embeddings import EmbeddingsContext

logger = logging.getLogger(__name__)

router = APIRouter(tags=[Tags.events])


@router.get("/faces")
def get_faces():
    face_dict: dict[str, list[str]] = {}

    for name in os.listdir(FACE_DIR):
        face_dict[name] = []
        for file in os.listdir(os.path.join(FACE_DIR, name)):
            face_dict[name].append(file)

    return JSONResponse(status_code=200, content=face_dict)


@router.post("/faces/{name}")
async def register_face(request: Request, name: str, file: UploadFile):
    # if not file.content_type.startswith("image"):
    #    return JSONResponse(
    #        status_code=400,
    #        content={
    #            "success": False,
    #            "message": "Only an image can be used to register a face.",
    #        },
    #    )

    context: EmbeddingsContext = request.app.embeddings
    context.register_face(name, await file.read())
    return JSONResponse(
        status_code=200,
        content={"success": True, "message": "Successfully registered face."},
    )


@router.delete("/faces")
def deregister_faces(request: Request, body: dict = None):
    json: dict[str, any] = body or {}
    list_of_ids = json.get("ids", "")

    if not list_of_ids or len(list_of_ids) == 0:
        return JSONResponse(
            content=({"success": False, "message": "Not a valid list of ids"}),
            status_code=404,
        )

    context: EmbeddingsContext = request.app.embeddings
    context.delete_face_ids(list_of_ids)
    return JSONResponse(
        content=({"success": True, "message": "Successfully deleted faces."}),
        status_code=200,
    )