mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-03-16 02:17:46 +01:00
faadea8e1f
* 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>
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
"""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, FrigateConfig
|
|
|
|
|
|
class CameraConfigUpdateEnum(str, Enum):
|
|
"""Supported camera config update types."""
|
|
|
|
add = "add" # for adding a camera
|
|
audio = "audio"
|
|
audio_transcription = "audio_transcription"
|
|
birdseye = "birdseye"
|
|
detect = "detect"
|
|
enabled = "enabled"
|
|
motion = "motion" # includes motion and motion masks
|
|
notifications = "notifications"
|
|
objects = "objects"
|
|
record = "record"
|
|
remove = "remove" # for removing a camera
|
|
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.name}"
|
|
|
|
|
|
class CameraConfigUpdatePublisher:
|
|
def __init__(self):
|
|
self.publisher = ConfigPublisher()
|
|
|
|
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,
|
|
config: FrigateConfig | None,
|
|
camera_configs: dict[str, CameraConfig],
|
|
topics: list[CameraConfigUpdateEnum],
|
|
):
|
|
self.config = config
|
|
self.camera_configs = camera_configs
|
|
self.topics = topics
|
|
|
|
base_topic = "config/cameras"
|
|
|
|
if len(self.camera_configs) == 1:
|
|
base_topic += f"/{list(self.camera_configs.keys())[0]}"
|
|
|
|
self.subscriber = ConfigSubscriber(
|
|
base_topic,
|
|
exact=False,
|
|
)
|
|
|
|
def __update_config(
|
|
self, camera: str, update_type: CameraConfigUpdateEnum, updated_config: Any
|
|
) -> None:
|
|
if update_type == CameraConfigUpdateEnum.add:
|
|
self.config.cameras[camera] = updated_config
|
|
self.camera_configs[camera] = updated_config
|
|
return
|
|
elif update_type == CameraConfigUpdateEnum.remove:
|
|
self.config.cameras.pop(camera)
|
|
self.camera_configs.pop(camera)
|
|
return
|
|
|
|
config = self.camera_configs.get(camera)
|
|
|
|
if not config:
|
|
return
|
|
|
|
if update_type == CameraConfigUpdateEnum.audio:
|
|
config.audio = updated_config
|
|
elif update_type == CameraConfigUpdateEnum.audio_transcription:
|
|
config.audio_transcription = 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 update_type == CameraConfigUpdateEnum.objects:
|
|
config.objects = updated_config
|
|
elif update_type == CameraConfigUpdateEnum.record:
|
|
config.record = updated_config
|
|
elif update_type == CameraConfigUpdateEnum.review:
|
|
config.review = updated_config
|
|
elif update_type == CameraConfigUpdateEnum.snapshots:
|
|
config.snapshots = updated_config
|
|
elif update_type == CameraConfigUpdateEnum.zones:
|
|
config.zones = updated_config
|
|
|
|
def check_for_updates(self) -> dict[str, list[str]]:
|
|
updated_topics: dict[str, list[str]] = {}
|
|
|
|
# get all updates available
|
|
while True:
|
|
update_topic, update_config = self.subscriber.check_for_update()
|
|
|
|
if update_topic is None or update_config is None:
|
|
break
|
|
|
|
_, _, camera, raw_type = update_topic.split("/")
|
|
update_type = CameraConfigUpdateEnum[raw_type]
|
|
|
|
if update_type in self.topics:
|
|
if update_type.name in updated_topics:
|
|
updated_topics[update_type.name].append(camera)
|
|
else:
|
|
updated_topics[update_type.name] = [camera]
|
|
|
|
self.__update_config(camera, update_type, update_config)
|
|
|
|
return updated_topics
|
|
|
|
def stop(self) -> None:
|
|
self.subscriber.stop()
|