From 84a0827aeec30d64ac7c41d2e8944be67fef2ff3 Mon Sep 17 00:00:00 2001 From: Sean Vig Date: Sat, 22 May 2021 18:28:15 -0400 Subject: [PATCH] Use dataclasses for config handling Use config data classes to eliminate some of the boilerplate associated with setting up the configuration. In particular, using dataclasses removes a lot of the boilerplate around assigning properties to the object and allows these to be easily immutable by freezing them. In the case of simple, non-nested dataclasses, this also provides more convenient `asdict` helpers. To set this up, where previously the objects would be parsed from the config via the `__init__` method, create a `build` classmethod that does this and calls the dataclass initializer. Some of the objects are mutated at runtime, in particular some of the zones are mutated to set the color (this might be able to be refactored out) and some of the camera functionality can be enabled/disabled. Some of the configs with `enabled` properties don't seem to have mqtt hooks to be able to toggle this, in particular, the clips, snapshots, and detect can be toggled but rtmp and record configs do not, but all of these configs are still not frozen in case there is some other functionality I am missing. There are a couple other minor fixes here, one that was introduced by me recently where `max_seconds` was not defined, the other to properly `get()` the message payload when handling publishing mqtt messages sent via websocket. --- .gitignore | 1 + frigate/config.py | 1462 +++++++++++++++-------------------- frigate/events.py | 1 + frigate/mqtt.py | 12 +- frigate/test/test_config.py | 4 +- frigate/video.py | 2 +- 6 files changed, 651 insertions(+), 831 deletions(-) diff --git a/.gitignore b/.gitignore index a09436535..85caded13 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ .DS_Store *.pyc +*.swp debug .vscode config/config.yml diff --git a/frigate/config.py b/frigate/config.py index a12106c29..4ef5b6bc8 100644 --- a/frigate/config.py +++ b/frigate/config.py @@ -1,8 +1,11 @@ +from __future__ import annotations + import base64 +import dataclasses import json import logging import os -from typing import Dict +from typing import Any, Dict, List, Optional, Tuple, Union import cv2 import matplotlib.pyplot as plt @@ -17,6 +20,7 @@ logger = logging.getLogger(__name__) DEFAULT_TRACKED_OBJECTS = ["person"] +DEFAULT_DETECTORS = {"coral": {"type": "edgetpu", "device": "usb"}} DETECTORS_SCHEMA = vol.Schema( { vol.Required(str): { @@ -27,7 +31,20 @@ DETECTORS_SCHEMA = vol.Schema( } ) -DEFAULT_DETECTORS = {"coral": {"type": "edgetpu", "device": "usb"}} + +@dataclasses.dataclass(frozen=True) +class DetectorConfig: + type: str + device: str + num_threads: int + + @classmethod + def build(cls, config) -> DetectorConfig: + return DetectorConfig(config["type"], config["device"], config["num_threads"]) + + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) + MQTT_SCHEMA = vol.Schema( { @@ -36,15 +53,59 @@ MQTT_SCHEMA = vol.Schema( vol.Optional("topic_prefix", default="frigate"): str, vol.Optional("client_id", default="frigate"): str, vol.Optional("stats_interval", default=60): int, - "user": str, - "password": str, + vol.Inclusive("user", "auth"): str, + vol.Inclusive("password", "auth"): str, } ) + +@dataclasses.dataclass(frozen=True) +class MqttConfig: + host: str + port: int + topic_prefix: str + client_id: str + stats_interval: int + user: Optional[str] + password: Optional[str] + + @classmethod + def build(cls, config) -> MqttConfig: + return MqttConfig( + config["host"], + config["port"], + config["topic_prefix"], + config["client_id"], + config["stats_interval"], + config.get("user"), + config.get("password"), + ) + + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) + + RETAIN_SCHEMA = vol.Schema( {vol.Required("default", default=10): int, "objects": {str: int}} ) + +@dataclasses.dataclass(frozen=True) +class RetainConfig: + default: int + objects: Dict[str, int] + + @classmethod + def build(cls, config, global_config={}) -> RetainConfig: + return RetainConfig( + config.get("default", global_config.get("default")), + config.get("objects", global_config.get("objects", {})), + ) + + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) + + CLIPS_SCHEMA = vol.Schema( { vol.Optional("max_seconds", default=300): int, @@ -52,6 +113,273 @@ CLIPS_SCHEMA = vol.Schema( } ) + +@dataclasses.dataclass(frozen=True) +class ClipsConfig: + max_seconds: int + retain: RetainConfig + + @classmethod + def build(cls, config) -> ClipsConfig: + return ClipsConfig( + config["max_seconds"], + RetainConfig.build(config["retain"]), + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "max_seconds": self.max_seconds, + "retain": self.retain.to_dict(), + } + + +MOTION_SCHEMA = vol.Schema( + { + "mask": vol.Any(str, [str]), + "threshold": vol.Range(min=1, max=255), + "contour_area": int, + "delta_alpha": float, + "frame_alpha": float, + "frame_height": int, + } +) + + +@dataclasses.dataclass(frozen=True) +class MotionConfig: + raw_mask: Union[str, List[str]] + mask: np.ndarray + threshold: int + contour_area: int + delta_alpha: float + frame_alpha: float + frame_height: int + + @classmethod + def build(cls, config, global_config, frame_shape) -> MotionConfig: + raw_mask = config.get("mask") + if raw_mask: + mask = create_mask(frame_shape, raw_mask) + else: + mask = np.zeros(frame_shape, np.uint8) + mask[:] = 255 + + return MotionConfig( + raw_mask, + mask, + config.get("threshold", global_config.get("threshold", 25)), + config.get("contour_area", global_config.get("contour_area", 100)), + config.get("delta_alpha", global_config.get("delta_alpha", 0.2)), + config.get("frame_alpha", global_config.get("frame_alpha", 0.2)), + config.get( + "frame_height", global_config.get("frame_height", frame_shape[0] // 6) + ), + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "mask": self.raw_mask, + "threshold": self.threshold, + "contour_area": self.contour_area, + "delta_alpha": self.delta_alpha, + "frame_alpha": self.frame_alpha, + "frame_height": self.frame_height, + } + + +GLOBAL_DETECT_SCHEMA = vol.Schema({"max_disappeared": int}) +DETECT_SCHEMA = GLOBAL_DETECT_SCHEMA.extend( + {vol.Optional("enabled", default=True): bool} +) + + +@dataclasses.dataclass +class DetectConfig: + enabled: bool + max_disappeared: int + + @classmethod + def build(cls, config, global_config, camera_fps) -> DetectConfig: + return DetectConfig( + config["enabled"], + config.get( + "max_disappeared", global_config.get("max_disappeared", camera_fps * 5) + ), + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "enabled": self.enabled, + "max_disappeared": self.max_disappeared, + } + + +ZONE_FILTER_SCHEMA = vol.Schema( + { + str: { + "min_area": int, + "max_area": int, + "threshold": float, + } + } +) +FILTER_SCHEMA = ZONE_FILTER_SCHEMA.extend( + { + str: { + "min_score": float, + "mask": vol.Any(str, [str]), + } + } +) + + +@dataclasses.dataclass(frozen=True) +class FilterConfig: + min_area: int + max_area: int + threshold: float + min_score: float + mask: Optional[np.ndarray] + raw_mask: Union[str, List[str]] + + @classmethod + def build( + cls, config, global_config={}, global_mask=None, frame_shape=None + ) -> FilterConfig: + raw_mask = [] + if global_mask: + if isinstance(global_mask, list): + raw_mask += global_mask + elif isinstance(global_mask, str): + raw_mask += [global_mask] + + config_mask = config.get("mask") + if config_mask: + if isinstance(config_mask, list): + raw_mask += config_mask + elif isinstance(config_mask, str): + raw_mask += [config_mask] + + mask = create_mask(frame_shape, raw_mask) if raw_mask else None + + return FilterConfig( + min_area=config.get("min_area", global_config.get("min_area", 0)), + max_area=config.get("max_area", global_config.get("max_area", 24000000)), + threshold=config.get("threshold", global_config.get("threshold", 0.7)), + min_score=config.get("min_score", global_config.get("min_score", 0.5)), + mask=mask, + raw_mask=raw_mask, + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "min_area": self.min_area, + "max_area": self.max_area, + "threshold": self.threshold, + "min_score": self.min_score, + "mask": self.raw_mask, + } + + +ZONE_SCHEMA = { + str: { + vol.Required("coordinates"): vol.Any(str, [str]), + vol.Optional("filters", default={}): ZONE_FILTER_SCHEMA, + } +} + + +@dataclasses.dataclass +class ZoneConfig: + filters: Dict[str, FilterConfig] + coordinates: Union[str, List[str]] + contour: np.ndarray + color: Tuple[int, int, int] + + @classmethod + def build(cls, name, config) -> ZoneConfig: + coordinates = config["coordinates"] + + if isinstance(coordinates, list): + contour = np.array( + [[int(p.split(",")[0]), int(p.split(",")[1])] for p in coordinates] + ) + elif isinstance(coordinates, str): + points = coordinates.split(",") + contour = np.array( + [[int(points[i]), int(points[i + 1])] for i in range(0, len(points), 2)] + ) + else: + print(f"Unable to parse zone coordinates for {name}") + contour = np.array([]) + + return ZoneConfig( + {name: FilterConfig.build(c) for name, c in config["filters"].items()}, + coordinates, + contour, + color=(0, 0, 0), + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "filters": {k: f.to_dict() for k, f in self.filters.items()}, + "coordinates": self.coordinates, + } + + +def filters_for_all_tracked_objects(object_config): + for tracked_object in object_config.get("track", DEFAULT_TRACKED_OBJECTS): + if not "filters" in object_config: + object_config["filters"] = {} + if not tracked_object in object_config["filters"]: + object_config["filters"][tracked_object] = {} + return object_config + + +OBJECTS_SCHEMA = vol.Schema( + vol.All( + filters_for_all_tracked_objects, + { + "track": [str], + "mask": vol.Any(str, [str]), + vol.Optional("filters", default={}): FILTER_SCHEMA, + }, + ) +) + + +@dataclasses.dataclass(frozen=True) +class ObjectConfig: + track: List[str] + filters: Dict[str, FilterConfig] + raw_mask: Optional[Union[str, List[str]]] + + @classmethod + def build(cls, config, global_config, frame_shape) -> ObjectConfig: + track = config.get("track", global_config.get("track", DEFAULT_TRACKED_OBJECTS)) + raw_mask = config.get("mask") + return ObjectConfig( + track, + { + name: FilterConfig.build( + config["filters"].get(name, {}), + global_config["filters"].get(name, {}), + raw_mask, + frame_shape, + ) + for name in track + }, + raw_mask, + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "track": self.track, + "mask": self.raw_mask, + "filters": {k: f.to_dict() for k, f in self.filters.items()}, + } + + FFMPEG_GLOBAL_ARGS_DEFAULT = ["-hide_banner", "-loglevel", "warning"] FFMPEG_INPUT_ARGS_DEFAULT = [ "-avoid_negative_ts", @@ -124,57 +452,6 @@ GLOBAL_FFMPEG_SCHEMA = vol.Schema( } ) -MOTION_SCHEMA = vol.Schema( - { - "mask": vol.Any(str, [str]), - "threshold": vol.Range(min=1, max=255), - "contour_area": int, - "delta_alpha": float, - "frame_alpha": float, - "frame_height": int, - } -) - -DETECT_SCHEMA = vol.Schema({"max_disappeared": int}) - -FILTER_SCHEMA = vol.Schema( - { - str: { - "min_area": int, - "max_area": int, - "threshold": float, - } - } -) - - -def filters_for_all_tracked_objects(object_config): - for tracked_object in object_config.get("track", DEFAULT_TRACKED_OBJECTS): - if not "filters" in object_config: - object_config["filters"] = {} - if not tracked_object in object_config["filters"]: - object_config["filters"][tracked_object] = {} - return object_config - - -OBJECTS_SCHEMA = vol.Schema( - vol.All( - filters_for_all_tracked_objects, - { - "track": [str], - "mask": vol.Any(str, [str]), - vol.Optional("filters", default={}): FILTER_SCHEMA.extend( - { - str: { - "min_score": float, - "mask": vol.Any(str, [str]), - } - } - ), - }, - ) -) - def each_role_used_once(inputs): roles = [role for i in inputs for role in i["roles"]] @@ -227,6 +504,54 @@ CAMERA_FFMPEG_SCHEMA = vol.Schema( ) +@dataclasses.dataclass(frozen=True) +class CameraFfmpegConfig: + inputs: List[CameraInput] + output_args: Dict[str, List[str]] + + @classmethod + def build(self, config, global_config): + output_args = config.get("output_args", global_config["output_args"]) + output_args = { + k: v if isinstance(v, list) else v.split(" ") + for k, v in output_args.items() + } + return CameraFfmpegConfig( + [CameraInput.build(i, config, global_config) for i in config["inputs"]], + output_args, + ) + + +@dataclasses.dataclass(frozen=True) +class CameraInput: + path: str + roles: List[str] + global_args: List[str] + hwaccel_args: List[str] + input_args: List[str] + + @classmethod + def build(cls, ffmpeg_input, camera_config, global_config) -> CameraInput: + return CameraInput( + ffmpeg_input["path"], + ffmpeg_input["roles"], + CameraInput._extract_args( + "global_args", ffmpeg_input, camera_config, global_config + ), + CameraInput._extract_args( + "hwaccel_args", ffmpeg_input, camera_config, global_config + ), + CameraInput._extract_args( + "input_args", ffmpeg_input, camera_config, global_config + ), + ) + + @staticmethod + def _extract_args(name, ffmpeg_input, camera_config, global_config): + args = ffmpeg_input.get(name, camera_config.get(name, global_config[name])) + return args if isinstance(args, list) else args.split(" ") + + def ensure_zones_and_cameras_have_different_names(cameras): zones = [zone for camera in cameras.values() for zone in camera["zones"].keys()] for zone in zones: @@ -244,12 +569,7 @@ CAMERAS_SCHEMA = vol.Schema( vol.Required("width"): int, "fps": int, vol.Optional("best_image_timeout", default=60): int, - vol.Optional("zones", default={}): { - str: { - vol.Required("coordinates"): vol.Any(str, [str]), - vol.Optional("filters", default={}): FILTER_SCHEMA, - } - }, + vol.Optional("zones", default={}): ZONE_SCHEMA, vol.Optional("clips", default={}): { vol.Optional("enabled", default=False): bool, vol.Optional("pre_capture", default=5): int, @@ -284,9 +604,7 @@ CAMERAS_SCHEMA = vol.Schema( }, vol.Optional("objects", default={}): OBJECTS_SCHEMA, vol.Optional("motion", default={}): MOTION_SCHEMA, - vol.Optional("detect", default={}): DETECT_SCHEMA.extend( - {vol.Optional("enabled", default=True): bool} - ), + vol.Optional("detect", default={}): DETECT_SCHEMA, } }, vol.Msg( @@ -296,426 +614,32 @@ CAMERAS_SCHEMA = vol.Schema( ) ) -FRIGATE_CONFIG_SCHEMA = vol.Schema( - { - vol.Optional("database", default={}): { - vol.Optional("path", default=os.path.join(CLIPS_DIR, "frigate.db")): str - }, - vol.Optional("model", default={"width": 320, "height": 320}): { - vol.Required("width"): int, - vol.Required("height"): int, - }, - vol.Optional("detectors", default=DEFAULT_DETECTORS): DETECTORS_SCHEMA, - "mqtt": MQTT_SCHEMA, - vol.Optional("logger", default={"default": "info", "logs": {}}): { - vol.Optional("default", default="info"): vol.In( - ["info", "debug", "warning", "error", "critical"] - ), - vol.Optional("logs", default={}): { - str: vol.In(["info", "debug", "warning", "error", "critical"]) - }, - }, - vol.Optional("snapshots", default={}): { - vol.Optional("retain", default={}): RETAIN_SCHEMA - }, - vol.Optional("clips", default={}): CLIPS_SCHEMA, - vol.Optional("record", default={}): { - vol.Optional("enabled", default=False): bool, - vol.Optional("retain_days", default=30): int, - }, - vol.Optional("ffmpeg", default={}): GLOBAL_FFMPEG_SCHEMA, - vol.Optional("objects", default={}): OBJECTS_SCHEMA, - vol.Optional("motion", default={}): MOTION_SCHEMA, - vol.Optional("detect", default={}): DETECT_SCHEMA, - vol.Required("cameras", default={}): CAMERAS_SCHEMA, - vol.Optional("environment_vars", default={}): {str: str}, - } -) - - -class DatabaseConfig: - def __init__(self, config): - self._path = config["path"] - - @property - def path(self): - return self._path - - def to_dict(self): - return {"path": self.path} - - -class ModelConfig: - def __init__(self, config): - self._width = config["width"] - self._height = config["height"] - - @property - def width(self): - return self._width - - @property - def height(self): - return self._height - - def to_dict(self): - return {"width": self.width, "height": self.height} - - -class DetectorConfig: - def __init__(self, config): - self._type = config["type"] - self._device = config["device"] - self._num_threads = config["num_threads"] - - @property - def type(self): - return self._type - - @property - def device(self): - return self._device - - @property - def num_threads(self): - return self._num_threads - - def to_dict(self): - return { - "type": self.type, - "device": self.device, - "num_threads": self.num_threads, - } - - -class LoggerConfig: - def __init__(self, config): - self._default = config["default"].upper() - self._logs = {k: v.upper() for k, v in config["logs"].items()} - - @property - def default(self): - return self._default - - @property - def logs(self): - return self._logs - - def to_dict(self): - return {"default": self.default, "logs": self.logs} - - -class MqttConfig: - def __init__(self, config): - self._host = config["host"] - self._port = config["port"] - self._topic_prefix = config["topic_prefix"] - self._client_id = config["client_id"] - self._user = config.get("user") - self._password = config.get("password") - self._stats_interval = config.get("stats_interval") - - @property - def host(self): - return self._host - - @property - def port(self): - return self._port - - @property - def topic_prefix(self): - return self._topic_prefix - - @property - def client_id(self): - return self._client_id - - @property - def user(self): - return self._user - - @property - def password(self): - return self._password - - @property - def stats_interval(self): - return self._stats_interval - - def to_dict(self): - return { - "host": self.host, - "port": self.port, - "topic_prefix": self.topic_prefix, - "client_id": self.client_id, - "user": self.user, - "stats_interval": self.stats_interval, - } - - -class CameraInput: - def __init__(self, camera_config, global_config, ffmpeg_input): - self._path = ffmpeg_input["path"] - self._roles = ffmpeg_input["roles"] - self._global_args = ffmpeg_input.get( - "global_args", - camera_config.get("global_args", global_config["global_args"]), - ) - self._hwaccel_args = ffmpeg_input.get( - "hwaccel_args", - camera_config.get("hwaccel_args", global_config["hwaccel_args"]), - ) - self._input_args = ffmpeg_input.get( - "input_args", camera_config.get("input_args", global_config["input_args"]) - ) - - @property - def path(self): - return self._path - - @property - def roles(self): - return self._roles - - @property - def global_args(self): - return ( - self._global_args - if isinstance(self._global_args, list) - else self._global_args.split(" ") - ) - - @property - def hwaccel_args(self): - return ( - self._hwaccel_args - if isinstance(self._hwaccel_args, list) - else self._hwaccel_args.split(" ") - ) - - @property - def input_args(self): - return ( - self._input_args - if isinstance(self._input_args, list) - else self._input_args.split(" ") - ) - - -class CameraFfmpegConfig: - def __init__(self, global_config, config): - self._inputs = [CameraInput(config, global_config, i) for i in config["inputs"]] - self._output_args = config.get("output_args", global_config["output_args"]) - - @property - def inputs(self): - return self._inputs - - @property - def output_args(self): - return { - k: v if isinstance(v, list) else v.split(" ") - for k, v in self._output_args.items() - } - - -class RetainConfig: - def __init__(self, global_config, config): - self._default = config.get("default", global_config.get("default")) - self._objects = config.get("objects", global_config.get("objects", {})) - - @property - def default(self): - return self._default - - @property - def objects(self): - return self._objects - - def to_dict(self): - return {"default": self.default, "objects": self.objects} - - -class ClipsConfig: - def __init__(self, config): - self._max_seconds = config["max_seconds"] - self._retain = RetainConfig(config["retain"], config["retain"]) - - @property - def max_seconds(self): - return self._max_seconds - - @property - def retain(self): - return self._retain - - def to_dict(self): - return { - "max_seconds": self.max_seconds, - "retain": self.retain.to_dict(), - } - - -class SnapshotsConfig: - def __init__(self, config): - self._retain = RetainConfig(config["retain"], config["retain"]) - - @property - def retain(self): - return self._retain - - def to_dict(self): - return {"retain": self.retain.to_dict()} - - -class RecordConfig: - def __init__(self, global_config, config): - self._enabled = config.get("enabled", global_config["enabled"]) - self._retain_days = config.get("retain_days", global_config["retain_days"]) - - @property - def enabled(self): - return self._enabled - - @property - def retain_days(self): - return self._retain_days - - def to_dict(self): - return { - "enabled": self.enabled, - "retain_days": self.retain_days, - } - - -class FilterConfig: - def __init__(self, global_config, config, global_mask=None, frame_shape=None): - self._min_area = config.get("min_area", global_config.get("min_area", 0)) - self._max_area = config.get("max_area", global_config.get("max_area", 24000000)) - self._threshold = config.get("threshold", global_config.get("threshold", 0.7)) - self._min_score = config.get("min_score", global_config.get("min_score", 0.5)) - - self._raw_mask = [] - if global_mask: - if isinstance(global_mask, list): - self._raw_mask += global_mask - elif isinstance(global_mask, str): - self._raw_mask += [global_mask] - - mask = config.get("mask") - if mask: - if isinstance(mask, list): - self._raw_mask += mask - elif isinstance(mask, str): - self._raw_mask += [mask] - self._mask = ( - create_mask(frame_shape, self._raw_mask) if self._raw_mask else None - ) - - @property - def min_area(self): - return self._min_area - - @property - def max_area(self): - return self._max_area - - @property - def threshold(self): - return self._threshold - - @property - def min_score(self): - return self._min_score - - @property - def mask(self): - return self._mask - - def to_dict(self): - return { - "min_area": self.min_area, - "max_area": self.max_area, - "threshold": self.threshold, - "min_score": self.min_score, - "mask": self._raw_mask, - } - - -class ObjectConfig: - def __init__(self, global_config, config, frame_shape): - self._track = config.get( - "track", global_config.get("track", DEFAULT_TRACKED_OBJECTS) - ) - self._raw_mask = config.get("mask") - self._filters = { - name: FilterConfig( - global_config["filters"].get(name, {}), - config["filters"].get(name, {}), - self._raw_mask, - frame_shape, - ) - for name in self._track - } - - @property - def track(self): - return self._track - - @property - def filters(self) -> Dict[str, FilterConfig]: - return self._filters - - def to_dict(self): - return { - "track": self.track, - "mask": self._raw_mask, - "filters": {k: f.to_dict() for k, f in self.filters.items()}, - } - +@dataclasses.dataclass class CameraSnapshotsConfig: - def __init__(self, global_config, config): - self._enabled = config["enabled"] - self._timestamp = config["timestamp"] - self._bounding_box = config["bounding_box"] - self._crop = config["crop"] - self._height = config.get("height") - self._retain = RetainConfig( - global_config["snapshots"]["retain"], config["retain"] + enabled: bool + timestamp: bool + bounding_box: bool + crop: bool + required_zones: List[str] + height: Optional[int] + retain: RetainConfig + + @classmethod + def build(self, config, global_config) -> CameraSnapshotsConfig: + return CameraSnapshotsConfig( + enabled=config["enabled"], + timestamp=config["timestamp"], + bounding_box=config["bounding_box"], + crop=config["crop"], + required_zones=config["required_zones"], + height=config.get("height"), + retain=RetainConfig.build( + config["retain"], global_config["snapshots"]["retain"] + ), ) - self._required_zones = config["required_zones"] - @property - def enabled(self): - return self._enabled - - @property - def timestamp(self): - return self._timestamp - - @property - def bounding_box(self): - return self._bounding_box - - @property - def crop(self): - return self._crop - - @property - def height(self): - return self._height - - @property - def retain(self): - return self._retain - - @property - def required_zones(self): - return self._required_zones - - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: return { "enabled": self.enabled, "timestamp": self.timestamp, @@ -727,84 +651,54 @@ class CameraSnapshotsConfig: } +@dataclasses.dataclass class CameraMqttConfig: - def __init__(self, config): - self._enabled = config["enabled"] - self._timestamp = config["timestamp"] - self._bounding_box = config["bounding_box"] - self._crop = config["crop"] - self._height = config.get("height") - self._required_zones = config["required_zones"] + enabled: bool + timestamp: bool + bounding_box: bool + crop: bool + height: int + required_zones: List[str] - @property - def enabled(self): - return self._enabled + @classmethod + def build(cls, config) -> CameraMqttConfig: + return CameraMqttConfig( + config["enabled"], + config["timestamp"], + config["bounding_box"], + config["crop"], + config.get("height"), + config["required_zones"], + ) - @property - def timestamp(self): - return self._timestamp - - @property - def bounding_box(self): - return self._bounding_box - - @property - def crop(self): - return self._crop - - @property - def height(self): - return self._height - - @property - def required_zones(self): - return self._required_zones - - def to_dict(self): - return { - "enabled": self.enabled, - "timestamp": self.timestamp, - "bounding_box": self.bounding_box, - "crop": self.crop, - "height": self.height, - "required_zones": self.required_zones, - } + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) +@dataclasses.dataclass class CameraClipsConfig: - def __init__(self, global_config, config): - self._enabled = config["enabled"] - self._pre_capture = config["pre_capture"] - self._post_capture = config["post_capture"] - self._objects = config.get("objects") - self._retain = RetainConfig(global_config["clips"]["retain"], config["retain"]) - self._required_zones = config["required_zones"] + enabled: bool + pre_capture: int + post_capture: int + required_zones: List[str] + objects: Optional[List[str]] + retain: RetainConfig - @property - def enabled(self): - return self._enabled + @classmethod + def build(cls, config, global_config) -> CameraClipsConfig: + return CameraClipsConfig( + enabled=config["enabled"], + pre_capture=config["pre_capture"], + post_capture=config["post_capture"], + required_zones=config["required_zones"], + objects=config.get("objects"), + retain=RetainConfig.build( + config["retain"], + global_config["clips"]["retain"], + ), + ) - @property - def pre_capture(self): - return self._pre_capture - - @property - def post_capture(self): - return self._post_capture - - @property - def objects(self): - return self._objects - - @property - def retain(self): - return self._retain - - @property - def required_zones(self): - return self._required_zones - - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: return { "enabled": self.enabled, "pre_capture": self.pre_capture, @@ -815,191 +709,85 @@ class CameraClipsConfig: } +@dataclasses.dataclass class CameraRtmpConfig: - def __init__(self, global_config, config): - self._enabled = config["enabled"] + enabled: bool - @property - def enabled(self): - return self._enabled + @classmethod + def build(cls, config) -> CameraRtmpConfig: + return CameraRtmpConfig(config["enabled"]) - def to_dict(self): - return { - "enabled": self.enabled, - } - - -class MotionConfig: - def __init__(self, global_config, config, frame_shape): - self._raw_mask = config.get("mask") - if self._raw_mask: - self._mask = create_mask(frame_shape, self._raw_mask) - else: - default_mask = np.zeros(frame_shape, np.uint8) - default_mask[:] = 255 - self._mask = default_mask - self._threshold = config.get("threshold", global_config.get("threshold", 25)) - self._contour_area = config.get( - "contour_area", global_config.get("contour_area", 100) - ) - self._delta_alpha = config.get( - "delta_alpha", global_config.get("delta_alpha", 0.2) - ) - self._frame_alpha = config.get( - "frame_alpha", global_config.get("frame_alpha", 0.2) - ) - self._frame_height = config.get( - "frame_height", global_config.get("frame_height", frame_shape[0] // 6) - ) - - @property - def mask(self): - return self._mask - - @property - def threshold(self): - return self._threshold - - @property - def contour_area(self): - return self._contour_area - - @property - def delta_alpha(self): - return self._delta_alpha - - @property - def frame_alpha(self): - return self._frame_alpha - - @property - def frame_height(self): - return self._frame_height - - def to_dict(self): - return { - "mask": self._raw_mask, - "threshold": self.threshold, - "contour_area": self.contour_area, - "delta_alpha": self.delta_alpha, - "frame_alpha": self.frame_alpha, - "frame_height": self.frame_height, - } - - -class DetectConfig: - def __init__(self, global_config, config, camera_fps): - self._enabled = config["enabled"] - self._max_disappeared = config.get( - "max_disappeared", global_config.get("max_disappeared", camera_fps * 5) - ) - - @property - def enabled(self): - return self._enabled - - @property - def max_disappeared(self): - return self._max_disappeared - - def to_dict(self): - return { - "enabled": self.enabled, - "max_disappeared": self._max_disappeared, - } - - -class ZoneConfig: - def __init__(self, name, config): - self._coordinates = config["coordinates"] - self._filters = { - name: FilterConfig(c, c) for name, c in config["filters"].items() - } - - if isinstance(self._coordinates, list): - self._contour = np.array( - [ - [int(p.split(",")[0]), int(p.split(",")[1])] - for p in self._coordinates - ] - ) - elif isinstance(self._coordinates, str): - points = self._coordinates.split(",") - self._contour = np.array( - [[int(points[i]), int(points[i + 1])] for i in range(0, len(points), 2)] - ) - else: - print(f"Unable to parse zone coordinates for {name}") - self._contour = np.array([]) - - self._color = (0, 0, 0) - - @property - def coordinates(self): - return self._coordinates - - @property - def contour(self): - return self._contour - - @contour.setter - def contour(self, val): - self._contour = val - - @property - def color(self): - return self._color - - @color.setter - def color(self, val): - self._color = val - - @property - def filters(self): - return self._filters - - def to_dict(self): - return { - "filters": {k: f.to_dict() for k, f in self.filters.items()}, - "coordinates": self._coordinates, - } + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) +@dataclasses.dataclass(frozen=True) class CameraConfig: - def __init__(self, name, config, global_config): - self._name = name - self._ffmpeg = CameraFfmpegConfig(global_config["ffmpeg"], config["ffmpeg"]) - self._height = config.get("height") - self._width = config.get("width") - self._frame_shape = (self._height, self._width) - self._frame_shape_yuv = (self._frame_shape[0] * 3 // 2, self._frame_shape[1]) - self._fps = config.get("fps") - self._best_image_timeout = config["best_image_timeout"] - self._zones = {name: ZoneConfig(name, z) for name, z in config["zones"].items()} - self._clips = CameraClipsConfig(global_config, config["clips"]) - self._record = RecordConfig(global_config["record"], config["record"]) - self._rtmp = CameraRtmpConfig(global_config, config["rtmp"]) - self._snapshots = CameraSnapshotsConfig(global_config, config["snapshots"]) - self._mqtt = CameraMqttConfig(config["mqtt"]) - self._objects = ObjectConfig( - global_config["objects"], config.get("objects", {}), self._frame_shape - ) - self._motion = MotionConfig( - global_config["motion"], config["motion"], self._frame_shape - ) - self._detect = DetectConfig( - global_config["detect"], config["detect"], config.get("fps", 5) - ) + name: str + ffmpeg: CameraFfmpegConfig + height: int + width: int + fps: Optional[int] + best_image_timeout: int + zones: Dict[str, ZoneConfig] + clips: CameraClipsConfig + record: RecordConfig + rtmp: CameraRtmpConfig + snapshots: CameraSnapshotsConfig + mqtt: CameraMqttConfig + objects: ObjectConfig + motion: MotionConfig + detect: DetectConfig - self._ffmpeg_cmds = [] - for ffmpeg_input in self._ffmpeg.inputs: + @property + def frame_shape(self) -> Tuple[int, int]: + return self.height, self.width + + @property + def frame_shape_yuv(self) -> Tuple[int, int]: + return self.height * 3 // 2, self.width + + @property + def ffmpeg_cmds(self) -> List[Dict[str, List[str]]]: + ffmpeg_cmds = [] + for ffmpeg_input in self.ffmpeg.inputs: ffmpeg_cmd = self._get_ffmpeg_cmd(ffmpeg_input) if ffmpeg_cmd is None: continue - self._ffmpeg_cmds.append({"roles": ffmpeg_input.roles, "cmd": ffmpeg_cmd}) + ffmpeg_cmds.append({"roles": ffmpeg_input.roles, "cmd": ffmpeg_cmd}) + return ffmpeg_cmds - self._set_zone_colors(self._zones) + @classmethod + def build(cls, name, config, global_config) -> CameraConfig: + zones = {name: ZoneConfig.build(name, z) for name, z in config["zones"].items()} + cls._set_zone_colors(zones) + + frame_shape = config["height"], config["width"] + + return CameraConfig( + name=name, + ffmpeg=CameraFfmpegConfig.build(config["ffmpeg"], global_config["ffmpeg"]), + height=config["height"], + width=config["width"], + fps=config.get("fps"), + best_image_timeout=config["best_image_timeout"], + zones=zones, + clips=CameraClipsConfig.build(config["clips"], global_config), + record=RecordConfig.build(config["record"], global_config["record"]), + rtmp=CameraRtmpConfig.build(config["rtmp"]), + snapshots=CameraSnapshotsConfig.build(config["snapshots"], global_config), + mqtt=CameraMqttConfig.build(config["mqtt"]), + objects=ObjectConfig.build( + config.get("objects", {}), global_config["objects"], frame_shape + ), + motion=MotionConfig.build( + config["motion"], global_config["motion"], frame_shape + ), + detect=DetectConfig.build( + config["detect"], global_config["detect"], config.get("fps", 5) + ), + ) def _get_ffmpeg_cmd(self, ffmpeg_input): ffmpeg_output_args = [] @@ -1043,90 +831,13 @@ class CameraConfig: return [part for part in cmd if part != ""] - def _set_zone_colors(self, zones: Dict[str, ZoneConfig]): - # set colors for zones - all_zone_names = zones.keys() - zone_colors = {} - colors = plt.cm.get_cmap("tab10", len(all_zone_names)) - for i, zone in enumerate(all_zone_names): - zone_colors[zone] = tuple(int(round(255 * c)) for c in colors(i)[:3]) + @classmethod + def _set_zone_colors(cls, zones: Dict[str, ZoneConfig]) -> None: + colors = plt.cm.get_cmap("tab10", len(zones)) + for i, (name, zone) in enumerate(zones.items()): + zone.color = tuple(int(round(255 * c)) for c in colors(i)[:3]) - for name, zone in zones.items(): - zone.color = zone_colors[name] - - @property - def name(self): - return self._name - - @property - def ffmpeg(self): - return self._ffmpeg - - @property - def height(self): - return self._height - - @property - def width(self): - return self._width - - @property - def fps(self): - return self._fps - - @property - def best_image_timeout(self): - return self._best_image_timeout - - @property - def zones(self) -> Dict[str, ZoneConfig]: - return self._zones - - @property - def clips(self): - return self._clips - - @property - def record(self): - return self._record - - @property - def rtmp(self): - return self._rtmp - - @property - def snapshots(self): - return self._snapshots - - @property - def mqtt(self): - return self._mqtt - - @property - def objects(self): - return self._objects - - @property - def motion(self): - return self._motion - - @property - def detect(self): - return self._detect - - @property - def frame_shape(self): - return self._frame_shape - - @property - def frame_shape_yuv(self): - return self._frame_shape_yuv - - @property - def ffmpeg_cmds(self): - return self._ffmpeg_cmds - - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: return { "name": self.name, "height": self.height, @@ -1150,8 +861,114 @@ class CameraConfig: } +FRIGATE_CONFIG_SCHEMA = vol.Schema( + { + vol.Optional("database", default={}): { + vol.Optional("path", default=os.path.join(CLIPS_DIR, "frigate.db")): str + }, + vol.Optional("model", default={"width": 320, "height": 320}): { + vol.Required("width"): int, + vol.Required("height"): int, + }, + vol.Optional("detectors", default=DEFAULT_DETECTORS): DETECTORS_SCHEMA, + "mqtt": MQTT_SCHEMA, + vol.Optional("logger", default={}): { + vol.Optional("default", default="info"): vol.In( + ["info", "debug", "warning", "error", "critical"] + ), + vol.Optional("logs", default={}): { + str: vol.In(["info", "debug", "warning", "error", "critical"]) + }, + }, + vol.Optional("snapshots", default={}): { + vol.Optional("retain", default={}): RETAIN_SCHEMA + }, + vol.Optional("clips", default={}): CLIPS_SCHEMA, + vol.Optional("record", default={}): { + vol.Optional("enabled", default=False): bool, + vol.Optional("retain_days", default=30): int, + }, + vol.Optional("ffmpeg", default={}): GLOBAL_FFMPEG_SCHEMA, + vol.Optional("objects", default={}): OBJECTS_SCHEMA, + vol.Optional("motion", default={}): MOTION_SCHEMA, + vol.Optional("detect", default={}): GLOBAL_DETECT_SCHEMA, + vol.Required("cameras", default={}): CAMERAS_SCHEMA, + vol.Optional("environment_vars", default={}): {str: str}, + } +) + + +@dataclasses.dataclass(frozen=True) +class DatabaseConfig: + path: str + + @classmethod + def build(cls, config) -> DatabaseConfig: + return DatabaseConfig(config["path"]) + + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) + + +@dataclasses.dataclass(frozen=True) +class ModelConfig: + width: int + height: int + + @classmethod + def build(cls, config) -> ModelConfig: + return ModelConfig(config["width"], config["height"]) + + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) + + +@dataclasses.dataclass(frozen=True) +class LoggerConfig: + default: str + logs: Dict[str, str] + + @classmethod + def build(cls, config) -> LoggerConfig: + return LoggerConfig( + config["default"].upper(), + {k: v.upper() for k, v in config["logs"].items()}, + ) + + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) + + +@dataclasses.dataclass(frozen=True) +class SnapshotsConfig: + retain: RetainConfig + + @classmethod + def build(cls, config) -> SnapshotsConfig: + return SnapshotsConfig(RetainConfig.build(config["retain"])) + + def to_dict(self) -> Dict[str, Any]: + return {"retain": self.retain.to_dict()} + + +@dataclasses.dataclass +class RecordConfig: + enabled: bool + retain_days: int + + @classmethod + def build(cls, config, global_config) -> RecordConfig: + return RecordConfig( + config.get("enabled", global_config["enabled"]), + config.get("retain_days", global_config["retain_days"]), + ) + + def to_dict(self) -> Dict[str, Any]: + return dataclasses.asdict(self) + + class FrigateConfig: - def __init__(self, config_file=None, config=None): + def __init__(self, config_file=None, config=None) -> None: if config is None and config_file is None: raise ValueError("config or config_file must be defined") elif not config_file is None: @@ -1161,18 +978,19 @@ class FrigateConfig: config = self._sub_env_vars(config) - self._database = DatabaseConfig(config["database"]) - self._model = ModelConfig(config["model"]) + self._database = DatabaseConfig.build(config["database"]) + self._model = ModelConfig.build(config["model"]) self._detectors = { - name: DetectorConfig(d) for name, d in config["detectors"].items() + name: DetectorConfig.build(d) for name, d in config["detectors"].items() } - self._mqtt = MqttConfig(config["mqtt"]) - self._clips = ClipsConfig(config["clips"]) - self._snapshots = SnapshotsConfig(config["snapshots"]) + self._mqtt = MqttConfig.build(config["mqtt"]) + self._clips = ClipsConfig.build(config["clips"]) + self._snapshots = SnapshotsConfig.build(config["snapshots"]) self._cameras = { - name: CameraConfig(name, c, config) for name, c in config["cameras"].items() + name: CameraConfig.build(name, c, config) + for name, c in config["cameras"].items() } - self._logger = LoggerConfig(config["logger"]) + self._logger = LoggerConfig.build(config["logger"]) self._environment_vars = config["environment_vars"] def _sub_env_vars(self, config): @@ -1202,7 +1020,7 @@ class FrigateConfig: return config - def to_dict(self): + def to_dict(self) -> Dict[str, Any]: return { "database": self.database.to_dict(), "model": self.model.to_dict(), @@ -1216,11 +1034,11 @@ class FrigateConfig: } @property - def database(self): + def database(self) -> DatabaseConfig: return self._database @property - def model(self): + def model(self) -> ModelConfig: return self._model @property @@ -1228,19 +1046,19 @@ class FrigateConfig: return self._detectors @property - def logger(self): + def logger(self) -> LoggerConfig: return self._logger @property - def mqtt(self): + def mqtt(self) -> MqttConfig: return self._mqtt @property - def clips(self): + def clips(self) -> ClipsConfig: return self._clips @property - def snapshots(self): + def snapshots(self) -> SnapshotsConfig: return self._snapshots @property diff --git a/frigate/events.py b/frigate/events.py index 4818e5594..a9b24dbcf 100644 --- a/frigate/events.py +++ b/frigate/events.py @@ -109,6 +109,7 @@ class EventProcessor(threading.Thread): earliest_event = datetime.datetime.now().timestamp() # if the earliest event is more tha max seconds ago, cap it + max_seconds = self.config.clips.max_seconds earliest_event = max( earliest_event, datetime.datetime.now().timestamp() - self.config.clips.max_seconds, diff --git a/frigate/mqtt.py b/frigate/mqtt.py index 352cc4eff..7a0beaabc 100644 --- a/frigate/mqtt.py +++ b/frigate/mqtt.py @@ -22,11 +22,11 @@ def create_mqtt_client(config: FrigateConfig, camera_metrics): if payload == "ON": if not clips_settings.enabled: logger.info(f"Turning on clips for {camera_name} via mqtt") - clips_settings._enabled = True + clips_settings.enabled = True elif payload == "OFF": if clips_settings.enabled: logger.info(f"Turning off clips for {camera_name} via mqtt") - clips_settings._enabled = False + clips_settings.enabled = False else: logger.warning(f"Received unsupported value at {message.topic}: {payload}") @@ -44,11 +44,11 @@ def create_mqtt_client(config: FrigateConfig, camera_metrics): if payload == "ON": if not snapshots_settings.enabled: logger.info(f"Turning on snapshots for {camera_name} via mqtt") - snapshots_settings._enabled = True + snapshots_settings.enabled = True elif payload == "OFF": if snapshots_settings.enabled: logger.info(f"Turning off snapshots for {camera_name} via mqtt") - snapshots_settings._enabled = False + snapshots_settings.enabled = False else: logger.warning(f"Received unsupported value at {message.topic}: {payload}") @@ -67,12 +67,12 @@ def create_mqtt_client(config: FrigateConfig, camera_metrics): if not camera_metrics[camera_name]["detection_enabled"].value: logger.info(f"Turning on detection for {camera_name} via mqtt") camera_metrics[camera_name]["detection_enabled"].value = True - detect_settings._enabled = True + detect_settings.enabled = True elif payload == "OFF": if camera_metrics[camera_name]["detection_enabled"].value: logger.info(f"Turning off detection for {camera_name} via mqtt") camera_metrics[camera_name]["detection_enabled"].value = False - detect_settings._enabled = False + detect_settings.enabled = False else: logger.warning(f"Received unsupported value at {message.topic}: {payload}") diff --git a/frigate/test/test_config.py b/frigate/test/test_config.py index 5a3191b52..8f6aba6f3 100644 --- a/frigate/test/test_config.py +++ b/frigate/test/test_config.py @@ -156,9 +156,9 @@ class TestConfig(TestCase): } frigate_config = FrigateConfig(config=config) assert "dog" in frigate_config.cameras["back"].objects.filters - assert len(frigate_config.cameras["back"].objects.filters["dog"]._raw_mask) == 2 + assert len(frigate_config.cameras["back"].objects.filters["dog"].raw_mask) == 2 assert ( - len(frigate_config.cameras["back"].objects.filters["person"]._raw_mask) == 1 + len(frigate_config.cameras["back"].objects.filters["person"].raw_mask) == 1 ) def test_ffmpeg_params_global(self): diff --git a/frigate/video.py b/frigate/video.py index c2d377a47..e1281874d 100755 --- a/frigate/video.py +++ b/frigate/video.py @@ -100,7 +100,7 @@ def stop_ffmpeg(ffmpeg_process, logger): def start_or_restart_ffmpeg( ffmpeg_cmd, logger, logpipe: LogPipe, frame_size=None, ffmpeg_process=None ): - if not ffmpeg_process is None: + if ffmpeg_process is not None: stop_ffmpeg(ffmpeg_process, logger) if frame_size is None: