Processing refactor (#15935)

* Refactor post processor to be real time processor

* Build out generic API for post processing

* Cleanup

* Fix

frigate/app.py

@@ -39,6 +39,7 @@ from frigate.const import (
     RECORD_DIR,
     SHM_FRAMES_VAR,
 )
+from frigate.data_processing.types import DataProcessorMetrics
 from frigate.db.sqlitevecq import SqliteVecQueueDatabase
 from frigate.embeddings import EmbeddingsContext, manage_embeddings
 from frigate.events.audio import AudioProcessor
@@ -59,7 +60,6 @@ from frigate.models import (
 from frigate.object_detection import ObjectDetectProcess
 from frigate.object_processing import TrackedObjectProcessor
 from frigate.output.output import output_frames
-from frigate.postprocessing.types import PostProcessingMetrics
 from frigate.ptz.autotrack import PtzAutoTrackerThread
 from frigate.ptz.onvif import OnvifController
 from frigate.record.cleanup import RecordingCleanup
@@ -90,8 +90,8 @@ class FrigateApp:
         self.detection_shms: list[mp.shared_memory.SharedMemory] = []
         self.log_queue: Queue = mp.Queue()
         self.camera_metrics: dict[str, CameraMetrics] = {}
-        self.embeddings_metrics: PostProcessingMetrics | None = (
+        self.embeddings_metrics: DataProcessorMetrics | None = (
-            PostProcessingMetrics() if config.semantic_search.enabled else None
+            DataProcessorMetrics() if config.semantic_search.enabled else None
         )
         self.ptz_metrics: dict[str, PTZMetrics] = {}
         self.processes: dict[str, int] = {}

frigate/data_processing/post/api.py

@@ -0,0 +1,43 @@
+"""Local or remote processors to handle post processing."""
+
+import logging
+from abc import ABC, abstractmethod
+
+from frigate.config import FrigateConfig
+
+from ..types import DataProcessorMetrics, PostProcessDataEnum
+
+logger = logging.getLogger(__name__)
+
+
+class PostProcessorApi(ABC):
+    @abstractmethod
+    def __init__(self, config: FrigateConfig, metrics: DataProcessorMetrics) -> None:
+        self.config = config
+        self.metrics = metrics
+        pass
+
+    @abstractmethod
+    def process_data(
+        self, data: dict[str, any], data_type: PostProcessDataEnum
+    ) -> None:
+        """Processes the data of data type.
+        Args:
+            data (dict): containing data about the input.
+            data_type (enum): Describing the data that is being processed.
+
+        Returns:
+            None.
+        """
+        pass
+
+    @abstractmethod
+    def handle_request(self, request_data: dict[str, any]) -> dict[str, any] | None:
+        """Handle metadata requests.
+        Args:
+            request_data (dict): containing data about requested change to process.
+
+        Returns:
+            None if request was not handled, otherwise return response.
+        """
+        pass

frigate/data_processing/real_time/api.py

@@ -1,3 +1,5 @@
+"""Local only processors for handling real time object processing."""
+
 import logging
 from abc import ABC, abstractmethod
 
@@ -5,14 +7,14 @@ import numpy as np
 
 from frigate.config import FrigateConfig
 
-from .types import PostProcessingMetrics
+from ..types import DataProcessorMetrics
 
 logger = logging.getLogger(__name__)
 
 
-class ProcessorApi(ABC):
+class RealTimeProcessorApi(ABC):
     @abstractmethod
-    def __init__(self, config: FrigateConfig, metrics: PostProcessingMetrics) -> None:
+    def __init__(self, config: FrigateConfig, metrics: DataProcessorMetrics) -> None:
         self.config = config
         self.metrics = metrics
         pass

frigate/data_processing/real_time/face_processor.py

@@ -16,8 +16,8 @@ from frigate.config import FrigateConfig
 from frigate.const import FACE_DIR, FRIGATE_LOCALHOST, MODEL_CACHE_DIR
 from frigate.util.image import area
 
-from .processor_api import ProcessorApi
+from ..types import DataProcessorMetrics
-from .types import PostProcessingMetrics
+from .api import RealTimeProcessorApi
 
 logger = logging.getLogger(__name__)
 
@@ -25,8 +25,8 @@ logger = logging.getLogger(__name__)
 MIN_MATCHING_FACES = 2
 
 
-class FaceProcessor(ProcessorApi):
+class FaceProcessor(RealTimeProcessorApi):
-    def __init__(self, config: FrigateConfig, metrics: PostProcessingMetrics):
+    def __init__(self, config: FrigateConfig, metrics: DataProcessorMetrics):
         super().__init__(config, metrics)
         self.face_config = config.face_recognition
         self.face_detector: cv2.FaceDetectorYN = None

frigate/data_processing/types.py

@@ -1,10 +1,11 @@
 """Embeddings types."""
 
 import multiprocessing as mp
+from enum import Enum
 from multiprocessing.sharedctypes import Synchronized
 
 
-class PostProcessingMetrics:
+class DataProcessorMetrics:
     image_embeddings_fps: Synchronized
     text_embeddings_sps: Synchronized
     face_rec_fps: Synchronized
@@ -15,3 +16,9 @@ class PostProcessingMetrics:
         self.text_embeddings_sps = mp.Value("d", 0.01)
         self.face_rec_fps = mp.Value("d", 0.01)
         self.alpr_pps = mp.Value("d", 0.01)
+
+
+class PostProcessDataEnum(str, Enum):
+    recording = "recording"
+    review = "review"
+    tracked_object = "tracked_object"

frigate/embeddings/__init__.py

@@ -15,19 +15,19 @@ from setproctitle import setproctitle
 from frigate.comms.embeddings_updater import EmbeddingsRequestEnum, EmbeddingsRequestor
 from frigate.config import FrigateConfig
 from frigate.const import CONFIG_DIR, FACE_DIR
+from frigate.data_processing.types import DataProcessorMetrics
 from frigate.db.sqlitevecq import SqliteVecQueueDatabase
 from frigate.models import Event
 from frigate.util.builtin import serialize
 from frigate.util.services import listen
 
-from ..postprocessing.types import PostProcessingMetrics
 from .maintainer import EmbeddingMaintainer
 from .util import ZScoreNormalization
 
 logger = logging.getLogger(__name__)
 
 
-def manage_embeddings(config: FrigateConfig, metrics: PostProcessingMetrics) -> None:
+def manage_embeddings(config: FrigateConfig, metrics: DataProcessorMetrics) -> None:
     # Only initialize embeddings if semantic search is enabled
     if not config.semantic_search.enabled:
         return

frigate/embeddings/embeddings.py

@@ -16,12 +16,12 @@ from frigate.const import (
     UPDATE_EMBEDDINGS_REINDEX_PROGRESS,
     UPDATE_MODEL_STATE,
 )
+from frigate.data_processing.types import DataProcessorMetrics
 from frigate.db.sqlitevecq import SqliteVecQueueDatabase
 from frigate.models import Event
 from frigate.types import ModelStatusTypesEnum
 from frigate.util.builtin import serialize
 
-from ..postprocessing.types import PostProcessingMetrics
 from .functions.onnx import GenericONNXEmbedding, ModelTypeEnum
 
 logger = logging.getLogger(__name__)
@@ -65,7 +65,7 @@ class Embeddings:
         self,
         config: FrigateConfig,
         db: SqliteVecQueueDatabase,
-        metrics: PostProcessingMetrics,
+        metrics: DataProcessorMetrics,
     ) -> None:
         self.config = config
         self.db = db

frigate/embeddings/maintainer.py

@@ -29,17 +29,17 @@ from frigate.const import (
     FRIGATE_LOCALHOST,
     UPDATE_EVENT_DESCRIPTION,
 )
+from frigate.data_processing.real_time.api import RealTimeProcessorApi
+from frigate.data_processing.real_time.face_processor import FaceProcessor
+from frigate.data_processing.types import DataProcessorMetrics
 from frigate.embeddings.lpr.lpr import LicensePlateRecognition
 from frigate.events.types import EventTypeEnum
 from frigate.genai import get_genai_client
 from frigate.models import Event
-from frigate.postprocessing.face_processor import FaceProcessor
-from frigate.postprocessing.processor_api import ProcessorApi
 from frigate.types import TrackedObjectUpdateTypesEnum
 from frigate.util.builtin import serialize
 from frigate.util.image import SharedMemoryFrameManager, area, calculate_region
 
-from ..postprocessing.types import PostProcessingMetrics
 from .embeddings import Embeddings
 
 logger = logging.getLogger(__name__)
@@ -54,7 +54,7 @@ class EmbeddingMaintainer(threading.Thread):
         self,
         db: SqliteQueueDatabase,
         config: FrigateConfig,
-        metrics: PostProcessingMetrics,
+        metrics: DataProcessorMetrics,
         stop_event: MpEvent,
     ) -> None:
         super().__init__(name="embeddings_maintainer")
@@ -73,7 +73,7 @@ class EmbeddingMaintainer(threading.Thread):
         )
         self.embeddings_responder = EmbeddingsResponder()
         self.frame_manager = SharedMemoryFrameManager()
-        self.processors: list[ProcessorApi] = []
+        self.processors: list[RealTimeProcessorApi] = []
 
         if self.config.face_recognition.enabled:
             self.processors.append(FaceProcessor(self.config, metrics))

frigate/stats/util.py

@@ -14,8 +14,8 @@ from requests.exceptions import RequestException
 from frigate.camera import CameraMetrics
 from frigate.config import FrigateConfig
 from frigate.const import CACHE_DIR, CLIPS_DIR, RECORD_DIR
+from frigate.data_processing.types import DataProcessorMetrics
 from frigate.object_detection import ObjectDetectProcess
-from frigate.postprocessing.types import PostProcessingMetrics
 from frigate.types import StatsTrackingTypes
 from frigate.util.services import (
     get_amd_gpu_stats,
@@ -52,7 +52,7 @@ def get_latest_version(config: FrigateConfig) -> str:
 def stats_init(
     config: FrigateConfig,
     camera_metrics: dict[str, CameraMetrics],
-    embeddings_metrics: PostProcessingMetrics | None,
+    embeddings_metrics: DataProcessorMetrics | None,
     detectors: dict[str, ObjectDetectProcess],
     processes: dict[str, int],
 ) -> StatsTrackingTypes:

frigate/types.py

@@ -2,13 +2,13 @@ from enum import Enum
 from typing import TypedDict
 
 from frigate.camera import CameraMetrics
+from frigate.data_processing.types import DataProcessorMetrics
 from frigate.object_detection import ObjectDetectProcess
-from frigate.postprocessing.types import PostProcessingMetrics
 
 
 class StatsTrackingTypes(TypedDict):
     camera_metrics: dict[str, CameraMetrics]
-    embeddings_metrics: PostProcessingMetrics | None
+    embeddings_metrics: DataProcessorMetrics | None
     detectors: dict[str, ObjectDetectProcess]
     started: int
     latest_frigate_version: str