Dynamic Management of Cameras (#18671)

* Add base class for global config updates

* Add or remove camera states

* Move camera process management to separate thread

* Move camera management fully to separate class

* Cleanup

* Stop camera processes when stop command is sent

* Start processes dynamically when needed

* Adjust

* Leave extra room in tracked object queue for two cameras

* Dynamically set extra config pieces

* Add some TODOs

* Fix type check

* Simplify config updates

* Improve typing

* Correctly handle indexed entries

* Cleanup

* Create out SHM

* Use ZMQ for signaling object detectoin is completed

* Get camera correctly created

* Cleanup for updating the cameras config

* Cleanup

* Don't enable audio if no cameras have audio transcription

* Use exact string so similar camera names don't interfere

* Add ability to update config via json body to config/set endpoint

Additionally, update the config in a single rather than multiple calls for each updated key

* fix autotracking calibration to support new config updater function

---------

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>

frigate/object_detection/base.py

@@ -13,6 +13,10 @@ import numpy as np
 from setproctitle import setproctitle
 
 import frigate.util as util
+from frigate.comms.object_detector_signaler import (
+    ObjectDetectorPublisher,
+    ObjectDetectorSubscriber,
+)
 from frigate.detectors import create_detector
 from frigate.detectors.detector_config import (
     BaseDetectorConfig,
@@ -89,7 +93,7 @@ class LocalObjectDetector(ObjectDetector):
 def run_detector(
     name: str,
     detection_queue: Queue,
-    out_events: dict[str, MpEvent],
+    cameras: list[str],
     avg_speed: Value,
     start: Value,
     detector_config: BaseDetectorConfig,
@@ -108,15 +112,19 @@ def run_detector(
     signal.signal(signal.SIGTERM, receiveSignal)
     signal.signal(signal.SIGINT, receiveSignal)
 
-    frame_manager = SharedMemoryFrameManager()
-    object_detector = LocalObjectDetector(detector_config=detector_config)
-
-    outputs = {}
-    for name in out_events.keys():
+    def create_output_shm(name: str):
         out_shm = UntrackedSharedMemory(name=f"out-{name}", create=False)
         out_np = np.ndarray((20, 6), dtype=np.float32, buffer=out_shm.buf)
         outputs[name] = {"shm": out_shm, "np": out_np}
 
+    frame_manager = SharedMemoryFrameManager()
+    object_detector = LocalObjectDetector(detector_config=detector_config)
+    detector_publisher = ObjectDetectorPublisher()
+
+    outputs = {}
+    for name in cameras:
+        create_output_shm(name)
+
     while not stop_event.is_set():
         try:
             connection_id = detection_queue.get(timeout=1)
@@ -136,12 +144,18 @@ def run_detector(
         detections = object_detector.detect_raw(input_frame)
         duration = datetime.datetime.now().timestamp() - start.value
         frame_manager.close(connection_id)
+
+        if connection_id not in outputs:
+            create_output_shm(connection_id)
+
         outputs[connection_id]["np"][:] = detections[:]
-        out_events[connection_id].set()
+        signal_id = f"{connection_id}/update"
+        detector_publisher.publish(signal_id, signal_id)
         start.value = 0.0
 
         avg_speed.value = (avg_speed.value * 9 + duration) / 10
 
+    detector_publisher.stop()
     logger.info("Exited detection process...")
 
 
@@ -150,11 +164,11 @@ class ObjectDetectProcess:
         self,
         name: str,
         detection_queue: Queue,
-        out_events: dict[str, MpEvent],
+        cameras: list[str],
         detector_config: BaseDetectorConfig,
     ):
         self.name = name
-        self.out_events = out_events
+        self.cameras = cameras
         self.detection_queue = detection_queue
         self.avg_inference_speed = Value("d", 0.01)
         self.detection_start = Value("d", 0.0)
@@ -185,7 +199,7 @@ class ObjectDetectProcess:
             args=(
                 self.name,
                 self.detection_queue,
-                self.out_events,
+                self.cameras,
                 self.avg_inference_speed,
                 self.detection_start,
                 self.detector_config,
@@ -201,7 +215,6 @@ class RemoteObjectDetector:
         name: str,
         labels: dict[int, str],
         detection_queue: Queue,
-        event: MpEvent,
         model_config: ModelConfig,
         stop_event: MpEvent,
     ):
@@ -209,7 +222,6 @@ class RemoteObjectDetector:
         self.name = name
         self.fps = EventsPerSecond()
         self.detection_queue = detection_queue
-        self.event = event
         self.stop_event = stop_event
         self.shm = UntrackedSharedMemory(name=self.name, create=False)
         self.np_shm = np.ndarray(
@@ -219,6 +231,7 @@ class RemoteObjectDetector:
         )
         self.out_shm = UntrackedSharedMemory(name=f"out-{self.name}", create=False)
         self.out_np_shm = np.ndarray((20, 6), dtype=np.float32, buffer=self.out_shm.buf)
+        self.detector_subscriber = ObjectDetectorSubscriber(f"{name}/update")
 
     def detect(self, tensor_input, threshold=0.4):
         detections = []
@@ -228,9 +241,8 @@ class RemoteObjectDetector:
 
         # copy input to shared memory
         self.np_shm[:] = tensor_input[:]
-        self.event.clear()
         self.detection_queue.put(self.name)
-        result = self.event.wait(timeout=5.0)
+        result = self.detector_subscriber.check_for_update()
 
         # if it timed out
         if result is None:
@@ -246,5 +258,6 @@ class RemoteObjectDetector:
         return detections
 
     def cleanup(self):
+        self.detector_subscriber.stop()
         self.shm.unlink()
         self.out_shm.unlink()