Convert config updating and video/audio queues to use zmq (#9893)

* Add config pub / sub pattern

* remove recording from feature metrics

* remove audio and feature metrics

* Check for updates from all cameras

* remove birdseye from camera metrics

* remove motion and detection camera metrics

* Ensure that all processes are stopped

* Stop communicators

* Detections

* Cleanup video output queue

* Use select for time sensitive polls

* Use ipc instead of tcp

frigate/comms/config_updater.py

@@ -0,0 +1,51 @@
+"""Facilitates communication between processes."""
+
+import multiprocessing as mp
+from multiprocessing.synchronize import Event as MpEvent
+from typing import Optional
+
+import zmq
+
+SOCKET_PUB_SUB = "ipc:///tmp/cache/config"
+
+
+class ConfigPublisher:
+    """Publishes config changes to different processes."""
+
+    def __init__(self) -> None:
+        self.context = zmq.Context()
+        self.socket = self.context.socket(zmq.PUB)
+        self.socket.bind(SOCKET_PUB_SUB)
+        self.stop_event: MpEvent = mp.Event()
+
+    def publish(self, topic: str, payload: any) -> None:
+        """There is no communication back to the processes."""
+        self.socket.send_string(topic, flags=zmq.SNDMORE)
+        self.socket.send_pyobj(payload)
+
+    def stop(self) -> None:
+        self.stop_event.set()
+        self.socket.close()
+        self.context.destroy()
+
+
+class ConfigSubscriber:
+    """Simplifies receiving an updated config."""
+
+    def __init__(self, topic: str) -> None:
+        self.context = zmq.Context()
+        self.socket = self.context.socket(zmq.SUB)
+        self.socket.setsockopt_string(zmq.SUBSCRIBE, topic)
+        self.socket.connect(SOCKET_PUB_SUB)
+
+    def check_for_update(self) -> Optional[tuple[str, any]]:
+        """Returns updated config or None if no update."""
+        try:
+            topic = self.socket.recv_string(flags=zmq.NOBLOCK)
+            return (topic, self.socket.recv_pyobj())
+        except zmq.ZMQError:
+            return (None, None)
+
+    def stop(self) -> None:
+        self.socket.close()
+        self.context.destroy()

frigate/comms/detections_updater.py

@@ -0,0 +1,102 @@
+"""Facilitates communication between processes."""
+
+import threading
+from enum import Enum
+from typing import Optional
+
+import zmq
+
+SOCKET_CONTROL = "inproc://control.detections_updater"
+SOCKET_PUB = "ipc:///tmp/cache/detect_pub"
+SOCKET_SUB = "ipc:///tmp/cache/detect_sun"
+
+
+class DetectionTypeEnum(str, Enum):
+    all = ""
+    video = "video"
+    audio = "audio"
+
+
+class DetectionProxyRunner(threading.Thread):
+    def __init__(self, context: zmq.Context[zmq.Socket]) -> None:
+        threading.Thread.__init__(self)
+        self.name = "detection_proxy"
+        self.context = context
+
+    def run(self) -> None:
+        """Run the proxy."""
+        control = self.context.socket(zmq.SUB)
+        control.connect(SOCKET_CONTROL)
+        control.setsockopt_string(zmq.SUBSCRIBE, "")
+        incoming = self.context.socket(zmq.XSUB)
+        incoming.bind(SOCKET_PUB)
+        outgoing = self.context.socket(zmq.XPUB)
+        outgoing.bind(SOCKET_SUB)
+
+        zmq.proxy_steerable(
+            incoming, outgoing, None, control
+        )  # blocking, will unblock terminate message is received
+        incoming.close()
+        outgoing.close()
+
+
+class DetectionProxy:
+    """Proxies video and audio detections."""
+
+    def __init__(self) -> None:
+        self.context = zmq.Context()
+        self.control = self.context.socket(zmq.PUB)
+        self.control.bind(SOCKET_CONTROL)
+        self.runner = DetectionProxyRunner(self.context)
+        self.runner.start()
+
+    def stop(self) -> None:
+        self.control.send_string("TERMINATE")  # tell the proxy to stop
+        self.runner.join()
+        self.context.destroy()
+
+
+class DetectionPublisher:
+    """Simplifies receiving video and audio detections."""
+
+    def __init__(self, topic: DetectionTypeEnum) -> None:
+        self.topic = topic
+        self.context = zmq.Context()
+        self.socket = self.context.socket(zmq.PUB)
+        self.socket.connect(SOCKET_PUB)
+
+    def send_data(self, payload: any) -> None:
+        """Publish detection."""
+        self.socket.send_string(self.topic.value, flags=zmq.SNDMORE)
+        self.socket.send_pyobj(payload)
+
+    def stop(self) -> None:
+        self.socket.close()
+        self.context.destroy()
+
+
+class DetectionSubscriber:
+    """Simplifies receiving video and audio detections."""
+
+    def __init__(self, topic: DetectionTypeEnum) -> None:
+        self.context = zmq.Context()
+        self.socket = self.context.socket(zmq.SUB)
+        self.socket.setsockopt_string(zmq.SUBSCRIBE, topic.value)
+        self.socket.connect(SOCKET_SUB)
+
+    def get_data(self, timeout: float = None) -> Optional[tuple[str, any]]:
+        """Returns detections or None if no update."""
+        try:
+            has_update, _, _ = zmq.select([self.socket], [], [], timeout)
+
+            if has_update:
+                topic = DetectionTypeEnum[self.socket.recv_string(flags=zmq.NOBLOCK)]
+                return (topic, self.socket.recv_pyobj())
+        except zmq.ZMQError:
+            pass
+
+        return (None, None)
+
+    def stop(self) -> None:
+        self.socket.close()
+        self.context.destroy()

frigate/comms/dispatcher.py

@@ -4,11 +4,12 @@ import logging
 from abc import ABC, abstractmethod
 from typing import Any, Callable, Optional
 
+from frigate.comms.config_updater import ConfigPublisher
 from frigate.config import BirdseyeModeEnum, FrigateConfig
 from frigate.const import INSERT_MANY_RECORDINGS, INSERT_PREVIEW, REQUEST_REGION_GRID
 from frigate.models import Previews, Recordings
 from frigate.ptz.onvif import OnvifCommandEnum, OnvifController
-from frigate.types import CameraMetricsTypes, FeatureMetricsTypes, PTZMetricsTypes
+from frigate.types import PTZMetricsTypes
 from frigate.util.object import get_camera_regions_grid
 from frigate.util.services import restart_frigate
 
@@ -40,16 +41,14 @@ class Dispatcher:
     def __init__(
         self,
         config: FrigateConfig,
+        config_updater: ConfigPublisher,
         onvif: OnvifController,
-        camera_metrics: dict[str, CameraMetricsTypes],
-        feature_metrics: dict[str, FeatureMetricsTypes],
         ptz_metrics: dict[str, PTZMetricsTypes],
         communicators: list[Communicator],
     ) -> None:
         self.config = config
+        self.config_updater = config_updater
         self.onvif = onvif
-        self.camera_metrics = camera_metrics
-        self.feature_metrics = feature_metrics
         self.ptz_metrics = ptz_metrics
         self.comms = communicators
 
@@ -118,44 +117,51 @@ class Dispatcher:
     def _on_detect_command(self, camera_name: str, payload: str) -> None:
         """Callback for detect topic."""
         detect_settings = self.config.cameras[camera_name].detect
+        motion_settings = self.config.cameras[camera_name].motion
 
         if payload == "ON":
-            if not self.camera_metrics[camera_name]["detection_enabled"].value:
+            if not detect_settings.enabled:
                 logger.info(f"Turning on detection for {camera_name}")
-                self.camera_metrics[camera_name]["detection_enabled"].value = True
                 detect_settings.enabled = True
 
-                if not self.camera_metrics[camera_name]["motion_enabled"].value:
+                if not motion_settings.enabled:
                     logger.info(
                         f"Turning on motion for {camera_name} due to detection being enabled."
                     )
-                    self.camera_metrics[camera_name]["motion_enabled"].value = True
+                    motion_settings.enabled = True
+                    self.config_updater.publish(
+                        f"config/motion/{camera_name}", motion_settings
+                    )
                     self.publish(f"{camera_name}/motion/state", payload, retain=True)
         elif payload == "OFF":
-            if self.camera_metrics[camera_name]["detection_enabled"].value:
+            if detect_settings.enabled:
                 logger.info(f"Turning off detection for {camera_name}")
-                self.camera_metrics[camera_name]["detection_enabled"].value = False
                 detect_settings.enabled = False
 
+        self.config_updater.publish(f"config/detect/{camera_name}", detect_settings)
         self.publish(f"{camera_name}/detect/state", payload, retain=True)
 
     def _on_motion_command(self, camera_name: str, payload: str) -> None:
         """Callback for motion topic."""
+        detect_settings = self.config.cameras[camera_name].detect
+        motion_settings = self.config.cameras[camera_name].motion
+
         if payload == "ON":
-            if not self.camera_metrics[camera_name]["motion_enabled"].value:
+            if not motion_settings.enabled:
                 logger.info(f"Turning on motion for {camera_name}")
-                self.camera_metrics[camera_name]["motion_enabled"].value = True
+                motion_settings.enabled = True
         elif payload == "OFF":
-            if self.camera_metrics[camera_name]["detection_enabled"].value:
+            if detect_settings.enabled:
                 logger.error(
                     "Turning off motion is not allowed when detection is enabled."
                 )
                 return
 
-            if self.camera_metrics[camera_name]["motion_enabled"].value:
+            if motion_settings.enabled:
                 logger.info(f"Turning off motion for {camera_name}")
-                self.camera_metrics[camera_name]["motion_enabled"].value = False
+                motion_settings.enabled = False
 
+        self.config_updater.publish(f"config/motion/{camera_name}", motion_settings)
         self.publish(f"{camera_name}/motion/state", payload, retain=True)
 
     def _on_motion_improve_contrast_command(
@@ -165,20 +171,15 @@ class Dispatcher:
         motion_settings = self.config.cameras[camera_name].motion
 
         if payload == "ON":
-            if not self.camera_metrics[camera_name]["improve_contrast_enabled"].value:
+            if not motion_settings.improve_contrast:
                 logger.info(f"Turning on improve contrast for {camera_name}")
-                self.camera_metrics[camera_name][
-                    "improve_contrast_enabled"
-                ].value = True
                 motion_settings.improve_contrast = True  # type: ignore[union-attr]
         elif payload == "OFF":
-            if self.camera_metrics[camera_name]["improve_contrast_enabled"].value:
+            if motion_settings.improve_contrast:
                 logger.info(f"Turning off improve contrast for {camera_name}")
-                self.camera_metrics[camera_name][
-                    "improve_contrast_enabled"
-                ].value = False
                 motion_settings.improve_contrast = False  # type: ignore[union-attr]
 
+        self.config_updater.publish(f"config/motion/{camera_name}", motion_settings)
         self.publish(f"{camera_name}/improve_contrast/state", payload, retain=True)
 
     def _on_ptz_autotracker_command(self, camera_name: str, payload: str) -> None:
@@ -217,8 +218,8 @@ class Dispatcher:
 
         motion_settings = self.config.cameras[camera_name].motion
         logger.info(f"Setting motion contour area for {camera_name}: {payload}")
-        self.camera_metrics[camera_name]["motion_contour_area"].value = payload
         motion_settings.contour_area = payload  # type: ignore[union-attr]
+        self.config_updater.publish(f"config/motion/{camera_name}", motion_settings)
         self.publish(f"{camera_name}/motion_contour_area/state", payload, retain=True)
 
     def _on_motion_threshold_command(self, camera_name: str, payload: int) -> None:
@@ -231,8 +232,8 @@ class Dispatcher:
 
         motion_settings = self.config.cameras[camera_name].motion
         logger.info(f"Setting motion threshold for {camera_name}: {payload}")
-        self.camera_metrics[camera_name]["motion_threshold"].value = payload
         motion_settings.threshold = payload  # type: ignore[union-attr]
+        self.config_updater.publish(f"config/motion/{camera_name}", motion_settings)
         self.publish(f"{camera_name}/motion_threshold/state", payload, retain=True)
 
     def _on_audio_command(self, camera_name: str, payload: str) -> None:
@@ -249,13 +250,12 @@ class Dispatcher:
             if not audio_settings.enabled:
                 logger.info(f"Turning on audio detection for {camera_name}")
                 audio_settings.enabled = True
-                self.feature_metrics[camera_name]["audio_enabled"].value = True
         elif payload == "OFF":
-            if self.feature_metrics[camera_name]["audio_enabled"].value:
+            if audio_settings.enabled:
                 logger.info(f"Turning off audio detection for {camera_name}")
                 audio_settings.enabled = False
-                self.feature_metrics[camera_name]["audio_enabled"].value = False
 
+        self.config_updater.publish(f"config/audio/{camera_name}", audio_settings)
         self.publish(f"{camera_name}/audio/state", payload, retain=True)
 
     def _on_recordings_command(self, camera_name: str, payload: str) -> None:
@@ -272,13 +272,12 @@ class Dispatcher:
             if not record_settings.enabled:
                 logger.info(f"Turning on recordings for {camera_name}")
                 record_settings.enabled = True
-                self.feature_metrics[camera_name]["record_enabled"].value = True
         elif payload == "OFF":
-            if self.feature_metrics[camera_name]["record_enabled"].value:
+            if record_settings.enabled:
                 logger.info(f"Turning off recordings for {camera_name}")
                 record_settings.enabled = False
-                self.feature_metrics[camera_name]["record_enabled"].value = False
 
+        self.config_updater.publish(f"config/record/{camera_name}", record_settings)
         self.publish(f"{camera_name}/recordings/state", payload, retain=True)
 
     def _on_snapshots_command(self, camera_name: str, payload: str) -> None:
@@ -316,17 +315,16 @@ class Dispatcher:
         birdseye_settings = self.config.cameras[camera_name].birdseye
 
         if payload == "ON":
-            if not self.camera_metrics[camera_name]["birdseye_enabled"].value:
+            if not birdseye_settings.enabled:
                 logger.info(f"Turning on birdseye for {camera_name}")
-                self.camera_metrics[camera_name]["birdseye_enabled"].value = True
                 birdseye_settings.enabled = True
 
         elif payload == "OFF":
-            if self.camera_metrics[camera_name]["birdseye_enabled"].value:
+            if birdseye_settings.enabled:
                 logger.info(f"Turning off birdseye for {camera_name}")
-                self.camera_metrics[camera_name]["birdseye_enabled"].value = False
                 birdseye_settings.enabled = False
 
+        self.config_updater.publish(f"config/birdseye/{camera_name}", birdseye_settings)
         self.publish(f"{camera_name}/birdseye/state", payload, retain=True)
 
     def _on_birdseye_mode_command(self, camera_name: str, payload: str) -> None:
@@ -336,17 +334,16 @@ class Dispatcher:
             logger.info(f"Invalid birdseye_mode command: {payload}")
             return
 
-        birdseye_config = self.config.cameras[camera_name].birdseye
-        if not birdseye_config.enabled:
+        birdseye_settings = self.config.cameras[camera_name].birdseye
+
+        if not birdseye_settings.enabled:
             logger.info(f"Birdseye mode not enabled for {camera_name}")
             return
 
-        new_birdseye_mode = BirdseyeModeEnum(payload.lower())
-        logger.info(f"Setting birdseye mode for {camera_name} to {new_birdseye_mode}")
-
-        # update the metric (need the mode converted to an int)
-        self.camera_metrics[camera_name][
-            "birdseye_mode"
-        ].value = BirdseyeModeEnum.get_index(new_birdseye_mode)
+        birdseye_settings.mode = BirdseyeModeEnum(payload.lower())
+        logger.info(
+            f"Setting birdseye mode for {camera_name} to {birdseye_settings.mode}"
+        )
 
+        self.config_updater.publish(f"config/birdseye/{camera_name}", birdseye_settings)
         self.publish(f"{camera_name}/birdseye_mode/state", payload, retain=True)

frigate/comms/inter_process.py

@@ -1,7 +1,6 @@
 """Facilitates communication between processes."""
 
 import multiprocessing as mp
-import os
 import threading
 from multiprocessing.synchronize import Event as MpEvent
 from typing import Callable
@@ -9,17 +8,15 @@ from typing import Callable
 import zmq
 
 from frigate.comms.dispatcher import Communicator
-from frigate.const import PORT_INTER_PROCESS_COMM
+
+SOCKET_REP_REQ = "ipc:///tmp/cache/comms"
 
 
 class InterProcessCommunicator(Communicator):
     def __init__(self) -> None:
-        INTER_PROCESS_COMM_PORT = (
-            os.environ.get("INTER_PROCESS_COMM_PORT") or PORT_INTER_PROCESS_COMM
-        )
         self.context = zmq.Context()
         self.socket = self.context.socket(zmq.REP)
-        self.socket.bind(f"tcp://127.0.0.1:{INTER_PROCESS_COMM_PORT}")
+        self.socket.bind(SOCKET_REP_REQ)
         self.stop_event: MpEvent = mp.Event()
 
     def publish(self, topic: str, payload: str, retain: bool) -> None:
@@ -32,8 +29,13 @@ class InterProcessCommunicator(Communicator):
         self.reader_thread.start()
 
     def read(self) -> None:
-        while not self.stop_event.wait(0.5):
+        while not self.stop_event.is_set():
             while True:  # load all messages that are queued
+                has_message, _, _ = zmq.select([self.socket], [], [], 1)
+
+                if not has_message:
+                    break
+
                 try:
                     (topic, value) = self.socket.recv_pyobj(flags=zmq.NOBLOCK)
 
@@ -57,10 +59,9 @@ class InterProcessRequestor:
     """Simplifies sending data to InterProcessCommunicator and getting a reply."""
 
     def __init__(self) -> None:
-        port = os.environ.get("INTER_PROCESS_COMM_PORT") or PORT_INTER_PROCESS_COMM
         self.context = zmq.Context()
         self.socket = self.context.socket(zmq.REQ)
-        self.socket.connect(f"tcp://127.0.0.1:{port}")
+        self.socket.connect(SOCKET_REP_REQ)
 
     def send_data(self, topic: str, data: any) -> any:
         """Sends data and then waits for reply."""