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: