Dynamic embeddings reindexing (#17418)

* reindex with api endpoint and zmq

* threading

* frontend

* require admin role

frigate/api/classification.py

@@ -298,3 +298,49 @@ def reprocess_license_plate(request: Request, event_id: str):
         content=response,
         status_code=200,
     )
+
+
+@router.put("/reindex", dependencies=[Depends(require_role(["admin"]))])
+def reindex_embeddings(request: Request):
+    if not request.app.frigate_config.semantic_search.enabled:
+        message = (
+            "Cannot reindex tracked object embeddings, Semantic Search is not enabled."
+        )
+        logger.error(message)
+        return JSONResponse(
+            content=(
+                {
+                    "success": False,
+                    "message": message,
+                }
+            ),
+            status_code=400,
+        )
+
+    context: EmbeddingsContext = request.app.embeddings
+    response = context.reindex_embeddings()
+
+    if response == "started":
+        return JSONResponse(
+            content={
+                "success": True,
+                "message": "Embeddings reindexing has started.",
+            },
+            status_code=202,  # 202 Accepted
+        )
+    elif response == "in_progress":
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "Embeddings reindexing is already in progress.",
+            },
+            status_code=409,  # 409 Conflict
+        )
+    else:
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "Failed to start reindexing.",
+            },
+            status_code=500,
+        )

frigate/comms/embeddings_updater.py

@@ -17,6 +17,7 @@ class EmbeddingsRequestEnum(Enum):
     register_face = "register_face"
     reprocess_face = "reprocess_face"
     reprocess_plate = "reprocess_plate"
+    reindex = "reindex"
 
 
 class EmbeddingsResponder:

frigate/embeddings/__init__.py

@@ -250,3 +250,6 @@ class EmbeddingsContext:
         return self.requestor.send_data(
             EmbeddingsRequestEnum.reprocess_plate.value, {"event": event}
         )
+
+    def reindex_embeddings(self) -> dict[str, any]:
+        return self.requestor.send_data(EmbeddingsRequestEnum.reindex.value, {})

frigate/embeddings/embeddings.py

@@ -3,6 +3,7 @@
 import datetime
 import logging
 import os
+import threading
 import time
 
 from numpy import ndarray
@@ -74,6 +75,10 @@ class Embeddings:
         self.metrics = metrics
         self.requestor = InterProcessRequestor()
 
+        self.reindex_lock = threading.Lock()
+        self.reindex_thread = None
+        self.reindex_running = False
+
         # Create tables if they don't exist
         self.db.create_embeddings_tables()
 
@@ -368,3 +373,27 @@ class Embeddings:
         totals["status"] = "completed"
 
         self.requestor.send_data(UPDATE_EMBEDDINGS_REINDEX_PROGRESS, totals)
+
+    def start_reindex(self) -> bool:
+        """Start reindexing in a separate thread if not already running."""
+        with self.reindex_lock:
+            if self.reindex_running:
+                logger.warning("Reindex embeddings is already running.")
+                return False
+
+            # Mark as running and start the thread
+            self.reindex_running = True
+            self.reindex_thread = threading.Thread(
+                target=self._reindex_wrapper, daemon=True
+            )
+            self.reindex_thread.start()
+            return True
+
+    def _reindex_wrapper(self) -> None:
+        """Wrapper to run reindex and reset running flag when done."""
+        try:
+            self.reindex()
+        finally:
+            with self.reindex_lock:
+                self.reindex_running = False
+                self.reindex_thread = None

frigate/embeddings/maintainer.py

@@ -206,6 +206,9 @@ class EmbeddingMaintainer(threading.Thread):
                             self.embeddings.embed_description("", data, upsert=False),
                             pack=False,
                         )
+                    elif topic == EmbeddingsRequestEnum.reindex.value:
+                        response = self.embeddings.start_reindex()
+                        return "started" if response else "in_progress"
 
                 processors = [self.realtime_processors, self.post_processors]
                 for processor_list in processors: