Semantic Search API (#12105)

* initial event search api implementation

* fix lint

* fix tests

* move chromadb imports and pysqlite hotswap to fix tests

* remove unused import

* switch default limit to 50

* fix events accidently pulling inside chroma results loop

frigate/embeddings/__init__.py

@@ -1,9 +1,9 @@
 """ChromaDB embeddings database."""
 
+import json
 import logging
 import multiprocessing as mp
 import signal
-import sys
 import threading
 from types import FrameType
 from typing import Optional
@@ -12,9 +12,14 @@ from playhouse.sqliteq import SqliteQueueDatabase
 from setproctitle import setproctitle
 
 from frigate.config import FrigateConfig
+from frigate.const import CONFIG_DIR
 from frigate.models import Event
 from frigate.util.services import listen
 
+from .embeddings import Embeddings
+from .maintainer import EmbeddingMaintainer
+from .util import ZScoreNormalization
+
 logger = logging.getLogger(__name__)
 
 
@@ -48,12 +53,6 @@ def manage_embeddings(config: FrigateConfig) -> None:
     models = [Event]
     db.bind(models)
 
-    # Hotsawp the sqlite3 module for Chroma compatibility
-    __import__("pysqlite3")
-    sys.modules["sqlite3"] = sys.modules.pop("pysqlite3")
-    from .embeddings import Embeddings
-    from .maintainer import EmbeddingMaintainer
-
     embeddings = Embeddings()
 
     # Check if we need to re-index events
@@ -65,3 +64,28 @@ def manage_embeddings(config: FrigateConfig) -> None:
         stop_event,
     )
     maintainer.start()
+
+
+class EmbeddingsContext:
+    def __init__(self):
+        self.embeddings = Embeddings()
+        self.thumb_stats = ZScoreNormalization()
+        self.desc_stats = ZScoreNormalization()
+
+        # load stats from disk
+        try:
+            with open(f"{CONFIG_DIR}/.search_stats.json", "r") as f:
+                data = json.loads(f.read())
+                self.thumb_stats.from_dict(data["thumb_stats"])
+                self.desc_stats.from_dict(data["desc_stats"])
+        except FileNotFoundError:
+            pass
+
+    def save_stats(self):
+        """Write the stats to disk as JSON on exit."""
+        contents = {
+            "thumb_stats": self.thumb_stats.to_dict(),
+            "desc_stats": self.desc_stats.to_dict(),
+        }
+        with open(f"{CONFIG_DIR}/.search_stats.json", "w") as f:
+            f.write(json.dumps(contents))