Build out generic API for post processing

frigate/data_processors/post/api.py

@@ -0,0 +1,45 @@
+"""Local or remote processors to handle post processing."""
+
+import logging
+from abc import ABC, abstractmethod
+
+import numpy as np
+
+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_processors/real_time/api.py

@@ -1,3 +1,5 @@
+"""Local only processors for handling real time object processing."""
+
 import logging
 from abc import ABC, abstractmethod
 

frigate/data_processors/types.py

@@ -1,6 +1,7 @@
 """Embeddings types."""
 
 import multiprocessing as mp
+from enum import Enum
 from multiprocessing.sharedctypes import Synchronized
 
 
@@ -15,3 +16,9 @@ class DataProcessorMetrics:
         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"