Build out more APIs

frigate/api/classification.py

@@ -20,7 +20,7 @@ def get_faces():
 
 @router.post("/faces/{name}")
 async def register_face(request: Request, name: str, file: UploadFile):
-    #if not file.content_type.startswith("image"):
+    # if not file.content_type.startswith("image"):
     #    return JSONResponse(
     #        status_code=400,
     #        content={
@@ -35,3 +35,22 @@ async def register_face(request: Request, name: str, file: UploadFile):
         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,
+    )

frigate/db/sqlitevecq.py

@@ -29,6 +29,10 @@ class SqliteVecQueueDatabase(SqliteQueueDatabase):
         ids = ",".join(["?" for _ in event_ids])
         self.execute_sql(f"DELETE FROM vec_descriptions WHERE id IN ({ids})", event_ids)
 
+    def delete_embeddings_face(self, face_ids: list[str]) -> None:
+        ids = ",".join(["?" for _ in face_ids])
+        self.execute_sql(f"DELETE FROM vec_faces WHERE id IN ({ids})", face_ids)
+
     def drop_embeddings_tables(self) -> None:
         self.execute_sql("""
             DROP TABLE vec_descriptions;

frigate/embeddings/__init__.py

@@ -193,9 +193,25 @@ class EmbeddingsContext:
     def register_face(self, face_name: str, image_data: bytes) -> None:
         self.requestor.send_data(
             EmbeddingsRequestEnum.register_face.value,
-            {"face_name": face_name, "image": base64.b64encode(image_data).decode("ASCII")},
+            {
+                "face_name": face_name,
+                "image": base64.b64encode(image_data).decode("ASCII"),
+            },
         )
 
+    def get_face_ids(self, name: str) -> list[str]:
+        sql_query = f"""
+            SELECT
+                id
+            FROM vec_descriptions
+            WHERE id LIKE '%{name}%'
+        """
+
+        return self.db.execute_sql(sql_query).fetchall()
+
+    def delete_face_ids(self, ids: list[str]) -> None:
+        self.db.delete_embeddings_face(ids)
+
     def update_description(self, event_id: str, description: str) -> None:
         self.requestor.send_data(
             EmbeddingsRequestEnum.embed_description.value,