mirror of
https://github.com/blakeblackshear/frigate.git
synced 2024-11-26 19:06:11 +01:00
66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
"""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,
|
|
)
|