diff --git a/frigate/config/camera/updater.py b/frigate/config/camera/updater.py new file mode 100644 index 000000000..e181377f5 --- /dev/null +++ b/frigate/config/camera/updater.py @@ -0,0 +1,123 @@ +"""Convenience classes for updating configurations dynamically.""" + +from dataclasses import dataclass +from enum import Enum +from typing import Any + +from frigate.comms.config_updater import ConfigPublisher, ConfigSubscriber +from frigate.config import CameraConfig + + +class CameraConfigUpdateEnum(str, Enum): + """Supported camera config update types.""" + + audio = "audio" + birdseye = "birdseye" + detect = "detect" + enabled = "enabled" + motion = "motion" # includes motion and motion masks + notifications = "notifications" + record = "record" + review = "review" + snapshots = "snapshots" + zones = "zones" + + +@dataclass +class CameraConfigUpdateTopic: + update_type: CameraConfigUpdateEnum + camera: str + + @property + def topic(self) -> str: + return f"config/cameras/{self.camera}/{self.update_type}" + + +def find_base_topic(topics: list[CameraConfigUpdateTopic]) -> str: + """ + Finds the longest common topic of topic strings. + + Args: + topics: list of update topics. + + Returns: + The longest common topic of the two strings. + """ + strings = [topic.topic for topic in topics] + + for i in range(min([len(s) for s in strings])): + letters = [s[i] for s in strings] + if all([letters[0] == letter for letter in letters]): + continue + + return strings[0][0:i] + + +class CameraConfigUpdatePublisher: + def __init__(self, publisher: ConfigPublisher): + self.publisher = publisher + + def publish_update(self, topic: CameraConfigUpdateTopic, config: Any) -> None: + self.publisher.publish(topic.topic, config) + + def stop(self) -> None: + self.publisher.stop() + + +class CameraConfigUpdateSubscriber: + def __init__( + self, + camera_configs: dict[str, CameraConfig], + topics: list[CameraConfigUpdateEnum], + ): + self.camera_configs = camera_configs + self.exact = len(topics) == 1 + self.subscriber = ConfigSubscriber( + topics[0] if self.exact else find_base_topic(topics), + exact=self.exact, + ) + + def __update_config( + self, camera: str, update_type: CameraConfigUpdateEnum, updated_config: Any + ) -> None: + config = self.camera_configs[camera] + + if not config: + return + + if update_type == CameraConfigUpdateEnum.audio: + config.audio = updated_config + elif update_type == CameraConfigUpdateEnum.birdseye: + config.birdseye = updated_config + elif update_type == CameraConfigUpdateEnum.detect: + config.detect = updated_config + elif update_type == CameraConfigUpdateEnum.enabled: + config.enabled = updated_config + elif update_type == CameraConfigUpdateEnum.motion: + config.motion = updated_config + elif update_type == CameraConfigUpdateEnum.notifications: + config.notifications = updated_config + elif updated_config == CameraConfigUpdateEnum.record: + config.record = updated_config + elif updated_config == CameraConfigUpdateEnum.review: + config.review = updated_config + elif updated_config == CameraConfigUpdateEnum.snapshots: + config.snapshots = updated_config + elif update_type == CameraConfigUpdateEnum.zones: + config.zones = updated_config + + def check_for_update(self) -> bool: + update_topic, update_config = self.subscriber.check_for_update() + + if update_topic and update_config: + _, _, camera, update_type = update_topic.split("/") + + for enum in CameraConfigUpdateEnum.__members__.values(): + if update_type == enum.name: + self.__update_config(camera, enum) + return True + + return False + + def stop(self) -> None: + self.subscriber.stop()