diff --git a/docker/Dockerfile.base b/docker/Dockerfile.base index a8d9eae4c..f80e3c8f8 100644 --- a/docker/Dockerfile.base +++ b/docker/Dockerfile.base @@ -36,8 +36,8 @@ RUN apt-get -qq update \ RUN pip3 install \ peewee_migrate \ + pydantic \ zeroconf \ - voluptuous\ ws4py COPY --from=nginx /usr/local/nginx/ /usr/local/nginx/ diff --git a/frigate/app.py b/frigate/app.py index a6994d6d9..1d3d7bda5 100644 --- a/frigate/app.py +++ b/frigate/app.py @@ -13,7 +13,7 @@ from peewee_migrate import Router from playhouse.sqlite_ext import SqliteExtDatabase from playhouse.sqliteq import SqliteQueueDatabase -from frigate.config import FrigateConfig +from frigate.config import DetectorTypeEnum, FrigateConfig from frigate.const import CACHE_DIR, CLIPS_DIR, RECORD_DIR from frigate.edgetpu import EdgeTPUProcess from frigate.events import EventCleanup, EventProcessor @@ -35,6 +35,7 @@ logger = logging.getLogger(__name__) class FrigateApp: def __init__(self): self.stop_event = mp.Event() + self.base_config: FrigateConfig = None self.config: FrigateConfig = None self.detection_queue = mp.Queue() self.detectors: Dict[str, EdgeTPUProcess] = {} @@ -65,7 +66,8 @@ class FrigateApp: def init_config(self): config_file = os.environ.get("CONFIG_FILE", "/config/config.yml") - self.config = FrigateConfig(config_file=config_file) + user_config = FrigateConfig.parse_file(config_file) + self.config = user_config.runtime_config for camera_name in self.config.cameras.keys(): # create camera_metrics @@ -116,9 +118,9 @@ class FrigateApp: ) def set_log_levels(self): - logging.getLogger().setLevel(self.config.logger.default) + logging.getLogger().setLevel(self.config.logger.default.value.upper()) for log, level in self.config.logger.logs.items(): - logging.getLogger(log).setLevel(level) + logging.getLogger(log).setLevel(level.value.upper()) if not "werkzeug" in self.config.logger.logs: logging.getLogger("werkzeug").setLevel("ERROR") @@ -183,9 +185,9 @@ class FrigateApp: try: shm_in = mp.shared_memory.SharedMemory( - name=name, - create=True, - size=self.config.model.height*self.config.model.width * 3, + name=name, + create=True, + size=self.config.model.height * self.config.model.width * 3, ) except FileExistsError: shm_in = mp.shared_memory.SharedMemory(name=name) @@ -201,7 +203,7 @@ class FrigateApp: self.detection_shms.append(shm_out) for name, detector in self.config.detectors.items(): - if detector.type == "cpu": + if detector.type == DetectorTypeEnum.cpu: self.detectors[name] = EdgeTPUProcess( name, self.detection_queue, @@ -210,7 +212,7 @@ class FrigateApp: "cpu", detector.num_threads, ) - if detector.type == "edgetpu": + if detector.type == DetectorTypeEnum.edgetpu: self.detectors[name] = EdgeTPUProcess( name, self.detection_queue, diff --git a/frigate/config.py b/frigate/config.py index 0f86243e9..3307fa541 100644 --- a/frigate/config.py +++ b/frigate/config.py @@ -1,20 +1,19 @@ from __future__ import annotations -import base64 -import dataclasses -import datetime +from enum import Enum import json import logging import os -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Dict, List, Optional, Tuple, Union import matplotlib.pyplot as plt import numpy as np -import voluptuous as vol +from pydantic import BaseModel, Field, validator +from pydantic.fields import PrivateAttr import yaml -from frigate.const import BASE_DIR, RECORD_DIR, CLIPS_DIR, CACHE_DIR -from frigate.util import create_mask +from frigate.const import BASE_DIR, RECORD_DIR, CACHE_DIR +from frigate.util import create_mask, deep_merge logger = logging.getLogger(__name__) @@ -23,412 +22,215 @@ DEFAULT_TIME_FORMAT = "%m/%d/%Y %H:%M:%S" # German Style: # DEFAULT_TIME_FORMAT = "%d.%m.%Y %H:%M:%S" +FRIGATE_ENV_VARS = {k: v for k, v in os.environ.items() if k.startswith("FRIGATE_")} + DEFAULT_TRACKED_OBJECTS = ["person"] -DEFAULT_RGB_COLOR = {"red": 255, "green": 255, "blue": 255} DEFAULT_DETECTORS = {"coral": {"type": "edgetpu", "device": "usb"}} -DETECTORS_SCHEMA = vol.Schema( - { - vol.Required(str): { - vol.Required("type", default="edgetpu"): vol.In(["cpu", "edgetpu"]), - vol.Optional("device", default="usb"): str, - vol.Optional("num_threads", default=3): int, - } - } -) -@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) +class DetectorTypeEnum(str, Enum): + edgetpu = "edgetpu" + cpu = "cpu" -MQTT_SCHEMA = vol.Schema( - { - vol.Required("host"): str, - vol.Optional("port", default=1883): int, - vol.Optional("topic_prefix", default="frigate"): str, - vol.Optional("client_id", default="frigate"): str, - vol.Optional("stats_interval", default=60): int, - vol.Inclusive("user", "auth"): str, - vol.Inclusive("password", "auth"): str, - vol.Optional("tls_ca_certs"): str, - vol.Optional("tls_client_cert"): str, - vol.Optional("tls_client_key"): str, - vol.Optional("tls_insecure"): bool, - } -) +class DetectorConfig(BaseModel): + type: DetectorTypeEnum = Field( + default=DetectorTypeEnum.edgetpu, title="Detector Type" + ) + device: str = Field(default="usb", title="Device Type") + num_threads: int = Field(default=3, title="Number of detection threads") -@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] - tls_ca_certs: Optional[str] - tls_client_cert: Optional[str] - tls_client_key: Optional[str] - tls_insecure: Optional[bool] +class MqttConfig(BaseModel): + host: str = Field(title="MQTT Host") + port: int = Field(default=1883, title="MQTT Port") + topic_prefix: str = Field(default="frigate", title="MQTT Topic Prefix") + client_id: str = Field(default="frigate", title="MQTT Client ID") + stats_interval: int = Field(default=60, title="MQTT Camera Stats Interval") + user: Optional[str] = Field(title="MQTT Username") + password: Optional[str] = Field(title="MQTT Password") + tls_ca_certs: Optional[str] = Field(title="MQTT TLS CA Certificates") + tls_client_cert: Optional[str] = Field(title="MQTT TLS Client Certificate") + tls_client_key: Optional[str] = Field(title="MQTT TLS Client Key") + tls_insecure: Optional[bool] = Field(title="MQTT TLS Insecure") - @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"), - config.get("tls_ca_certs"), - config.get("tls_client_cert"), - config.get("tls_client_key"), - config.get("tls_insecure"), - ) - - def to_dict(self) -> Dict[str, Any]: - return dataclasses.asdict(self) + @validator("password", pre=True, always=True) + def validate_password(cls, v, values): + if (v is None) != (values["user"] is None): + raise ValueError("Password must be provided with username.") + return v if v is None else v.format(**FRIGATE_ENV_VARS) -RETAIN_SCHEMA = vol.Schema( - {vol.Required("default", default=10): int, "objects": {str: int}} -) +class RetainConfig(BaseModel): + default: int = Field(default=10, title="Default retention period.") + objects: Dict[str, int] = Field( + default_factory=dict, title="Object retention period." + ) -@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) +class ClipsConfig(BaseModel): + max_seconds: int = Field(default=300, title="Maximum clip duration.") + retain: RetainConfig = Field( + default_factory=RetainConfig, title="Clip retention settings." + ) -CLIPS_SCHEMA = vol.Schema( - { - vol.Optional("max_seconds", default=300): int, - vol.Optional("retain", default={}): RETAIN_SCHEMA, - } -) +class MotionConfig(BaseModel): + threshold: int = Field( + default=25, + title="Motion detection threshold (1-255).", + ge=1, + le=255, + ) + contour_area: int = Field(default=100, title="Contour Area") + delta_alpha: float = Field(default=0.2, title="Delta Alpha") + frame_alpha: float = Field(default=0.2, title="Frame Alpha") + frame_height: Optional[int] = Field(title="Frame Height") + mask: Union[str, List[str]] = Field( + default="", title="Coordinates polygon for the motion mask." + ) -@dataclasses.dataclass(frozen=True) -class ClipsConfig: - max_seconds: int - retain: RetainConfig +class RuntimeMotionConfig(MotionConfig): + raw_mask: Union[str, List[str]] = "" + mask: np.ndarray = None - @classmethod - def build(cls, config) -> ClipsConfig: - return ClipsConfig( - config["max_seconds"], - RetainConfig.build(config["retain"]), - ) + def __init__(self, **config): + frame_shape = config.get("frame_shape", (1, 1)) - def to_dict(self) -> Dict[str, Any]: - return { - "max_seconds": self.max_seconds, - "retain": self.retain.to_dict(), - } + if "frame_height" not in config: + config["frame_height"] = frame_shape[0] // 6 + mask = config.get("mask", "") + config["raw_mask"] = mask -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) + if mask: + config["mask"] = create_mask(frame_shape, mask) else: - mask = np.zeros(frame_shape, np.uint8) - mask[:] = 255 + empty_mask = np.zeros(frame_shape, np.uint8) + empty_mask[:] = 255 + config["mask"] = empty_mask - 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) - ), - ) + super().__init__(**config) - 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, - } + def dict(self, **kwargs): + ret = super().dict(**kwargs) + if "mask" in ret: + ret["mask"] = ret["raw_mask"] + ret.pop("raw_mask") + return ret + + class Config: + arbitrary_types_allowed = True -GLOBAL_DETECT_SCHEMA = vol.Schema({"max_disappeared": int}) -DETECT_SCHEMA = GLOBAL_DETECT_SCHEMA.extend( - {vol.Optional("enabled", default=True): bool} -) +class DetectConfig(BaseModel): + enabled: bool = Field(default=True, title="Detection Enabled.") + max_disappeared: int = Field( + title="Maximum number of frames the object can dissapear before detection ends." + ) -@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, - } +class FilterConfig(BaseModel): + min_area: int = Field( + default=0, title="Minimum area of bounding box for object to be counted." + ) + max_area: int = Field( + default=24000000, title="Maximum area of bounding box for object to be counted." + ) + threshold: float = Field( + default=0.7, + title="Average detection confidence threshold for object to be counted.", + ) + min_score: float = Field( + default=0.5, title="Minimum detection confidence for object to be counted." + ) + mask: Optional[Union[str, List[str]]] = Field( + title="Detection area polygon mask for this filter configuration.", + ) -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 +class RuntimeFilterConfig(FilterConfig): mask: Optional[np.ndarray] - raw_mask: Union[str, List[str]] + raw_mask: Optional[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] + def __init__(self, **config): + mask = config.get("mask") + config["raw_mask"] = 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] + if mask is not None: + config["mask"] = create_mask(config.get("frame_shape", (1, 1)), mask) - mask = create_mask(frame_shape, raw_mask) if raw_mask else None + super().__init__(**config) - 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 dict(self, **kwargs): + ret = super().dict(**kwargs) + if "mask" in ret: + ret["mask"] = ret["raw_mask"] + ret.pop("raw_mask") + return ret - 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, - } + class Config: + arbitrary_types_allowed = True -ZONE_SCHEMA = { - str: { - vol.Required("coordinates"): vol.Any(str, [str]), - vol.Optional("filters", default={}): ZONE_FILTER_SCHEMA, - } -} +class ZoneConfig(BaseModel): + filters: Dict[str, FilterConfig] = Field( + default_factory=dict, title="Zone filters." + ) + coordinates: Union[str, List[str]] = Field( + title="Coordinates polygon for the defined zone." + ) + _color: Optional[Tuple[int, int, int]] = PrivateAttr() + _contour: np.ndarray = PrivateAttr() + @property + def color(self) -> Tuple[int, int, int]: + return self._color -@dataclasses.dataclass(frozen=True) -class ZoneConfig: - filters: Dict[str, FilterConfig] - coordinates: Union[str, List[str]] - contour: np.ndarray - color: Tuple[int, int, int] + @property + def contour(self) -> np.ndarray: + return self._contour - @classmethod - def build(cls, config, color: Tuple[int, int, int]) -> ZoneConfig: + def __init__(self, **config): + super().__init__(**config) + + self._color = config.get("color", (0, 0, 0)) coordinates = config["coordinates"] if isinstance(coordinates, list): - contour = np.array( + self._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( + self._contour = np.array( [[int(points[i]), int(points[i + 1])] for i in range(0, len(points), 2)] ) else: - contour = np.array([]) - - return ZoneConfig( - {name: FilterConfig.build(c) for name, c in config["filters"].items()}, - coordinates, - contour, - color=color, - ) - - def to_dict(self) -> Dict[str, Any]: - return { - "filters": {k: f.to_dict() for k, f in self.filters.items()}, - "coordinates": self.coordinates, - } + self._contour = np.array([]) -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 +class ObjectConfig(BaseModel): + track: List[str] = Field(default=DEFAULT_TRACKED_OBJECTS, title="Objects to track.") + filters: Optional[Dict[str, FilterConfig]] = Field(title="Object filters.") + mask: Union[str, List[str]] = Field(default="", title="Object mask.") -OBJECTS_SCHEMA = vol.Schema( - vol.All( - filters_for_all_tracked_objects, - { - "track": [str], - "mask": vol.Any(str, [str]), - vol.Optional("filters", default={}): FILTER_SCHEMA, - }, +class BirdseyeModeEnum(str, Enum): + objects = "objects" + motion = "motion" + continuous = "continuous" + + +class BirdseyeConfig(BaseModel): + enabled: bool = Field(default=True, title="Enable birdseye view.") + width: int = Field(default=1280, title="Birdseye width.") + height: int = Field(default=720, title="Birdseye height.") + quality: int = Field(default=8, title="Encoding quality.") + mode: BirdseyeModeEnum = Field( + default=BirdseyeModeEnum.objects, title="Tracking mode." ) -) - - -@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()}, - } - - -BIRDSEYE_SCHEMA = vol.Schema( - { - vol.Optional("enabled", default=True): bool, - vol.Optional("width", default=1280): int, - vol.Optional("height", default=720): int, - vol.Optional("quality", default=8): vol.Range(min=1, max=31), - vol.Optional("mode", default="objects"): vol.In( - ["objects", "motion", "continuous"] - ), - } -) - - -@dataclasses.dataclass(frozen=True) -class BirdseyeConfig: - enabled: bool - width: int - height: int - quality: int - mode: str - - @classmethod - def build(cls, config) -> BirdseyeConfig: - return BirdseyeConfig( - config["enabled"], - config["width"], - config["height"], - config["quality"], - config["mode"], - ) - - def to_dict(self) -> Dict[str, Any]: - return dataclasses.asdict(self) FFMPEG_GLOBAL_ARGS_DEFAULT = ["-hide_banner", "-loglevel", "warning"] @@ -477,400 +279,225 @@ RECORD_FFMPEG_OUTPUT_ARGS_DEFAULT = [ "-an", ] -GLOBAL_FFMPEG_SCHEMA = vol.Schema( - { - vol.Optional("global_args", default=FFMPEG_GLOBAL_ARGS_DEFAULT): vol.Any( - str, [str] - ), - vol.Optional("hwaccel_args", default=[]): vol.Any(str, [str]), - vol.Optional("input_args", default=FFMPEG_INPUT_ARGS_DEFAULT): vol.Any( - str, [str] - ), - vol.Optional("output_args", default={}): { - vol.Optional("detect", default=DETECT_FFMPEG_OUTPUT_ARGS_DEFAULT): vol.Any( - str, [str] - ), - vol.Optional("record", default=RECORD_FFMPEG_OUTPUT_ARGS_DEFAULT): vol.Any( - str, [str] - ), - vol.Optional( - "clips", default=SAVE_CLIPS_FFMPEG_OUTPUT_ARGS_DEFAULT - ): vol.Any(str, [str]), - vol.Optional("rtmp", default=RTMP_FFMPEG_OUTPUT_ARGS_DEFAULT): vol.Any( - str, [str] - ), - }, - } -) +class FfmpegOutputArgsConfig(BaseModel): + detect: List[str] = Field( + default=DETECT_FFMPEG_OUTPUT_ARGS_DEFAULT, + title="Detect role FFmpeg output arguments.", + ) + record: List[str] = Field( + default=RECORD_FFMPEG_OUTPUT_ARGS_DEFAULT, + title="Record role FFmpeg output arguments.", + ) + clips: List[str] = Field( + default=SAVE_CLIPS_FFMPEG_OUTPUT_ARGS_DEFAULT, + title="Clips role FFmpeg output arguments.", + ) + rtmp: List[str] = Field( + default=RTMP_FFMPEG_OUTPUT_ARGS_DEFAULT, + title="RTMP role FFmpeg output arguments.", + ) -def each_role_used_once(inputs): - roles = [role for i in inputs for role in i["roles"]] - roles_set = set(roles) - if len(roles) > len(roles_set): - raise ValueError - return inputs - - -def detect_is_required(inputs): - roles = [role for i in inputs for role in i["roles"]] - if not "detect" in roles: - raise ValueError - return inputs - - -CAMERA_FFMPEG_SCHEMA = vol.Schema( - { - vol.Required("inputs"): vol.All( - [ - { - vol.Required("path"): str, - vol.Required("roles"): ["detect", "clips", "record", "rtmp"], - "global_args": vol.Any(str, [str]), - "hwaccel_args": vol.Any(str, [str]), - "input_args": vol.Any(str, [str]), - } - ], - vol.Msg(each_role_used_once, msg="Each input role may only be used once"), - vol.Msg(detect_is_required, msg="The detect role is required"), - ), - "global_args": vol.Any(str, [str]), - "hwaccel_args": vol.Any(str, [str]), - "input_args": vol.Any(str, [str]), - "output_args": { - vol.Optional("detect", default=DETECT_FFMPEG_OUTPUT_ARGS_DEFAULT): vol.Any( - str, [str] - ), - vol.Optional("record", default=RECORD_FFMPEG_OUTPUT_ARGS_DEFAULT): vol.Any( - str, [str] - ), - vol.Optional( - "clips", default=SAVE_CLIPS_FFMPEG_OUTPUT_ARGS_DEFAULT - ): vol.Any(str, [str]), - vol.Optional("rtmp", default=RTMP_FFMPEG_OUTPUT_ARGS_DEFAULT): vol.Any( - str, [str] - ), - }, - } -) - - -@dataclasses.dataclass(frozen=True) -class CameraFfmpegConfig: - inputs: List[CameraInput] - output_args: Dict[str, List[str]] - - @classmethod - def build(cls, 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])) + @validator("detect", "record", "clips", "rtmp", pre=True) + def extract_args(cls, args): 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: - if zone in cameras.keys(): - raise ValueError - return cameras - - -def ensure_timeformat_is_legit(format_string): - datetime.datetime.now().strftime(format_string) - return format_string - - -RGB_COLOR_SCHEMA = vol.Schema( - { - vol.Required("red"): vol.Range(min=0, max=255), - vol.Required("green"): vol.Range(min=0, max=255), - vol.Required("blue"): vol.Range(min=0, max=255), - } -) - -CAMERAS_SCHEMA = vol.Schema( - vol.All( - { - str: { - vol.Required("ffmpeg"): CAMERA_FFMPEG_SCHEMA, - vol.Required("height"): int, - vol.Required("width"): int, - "fps": int, - vol.Optional("best_image_timeout", default=60): int, - vol.Optional("zones", default={}): ZONE_SCHEMA, - vol.Optional("clips", default={}): { - vol.Optional("enabled", default=False): bool, - vol.Optional("pre_capture", default=5): int, - vol.Optional("post_capture", default=5): int, - vol.Optional("required_zones", default=[]): [str], - "objects": [str], - vol.Optional("retain", default={}): RETAIN_SCHEMA, - }, - vol.Optional("record", default={}): { - "enabled": bool, - "retain_days": int, - }, - vol.Optional("rtmp", default={}): { - vol.Required("enabled", default=True): bool, - }, - vol.Optional("live", default={}): { - "height": int, - vol.Optional("quality", default=8): vol.Range(min=1, max=31), - }, - vol.Optional("snapshots", default={}): { - vol.Optional("enabled", default=False): bool, - vol.Optional("clean_copy", default=True): bool, - vol.Optional("timestamp", default=False): bool, - vol.Optional("bounding_box", default=False): bool, - vol.Optional("crop", default=False): bool, - vol.Optional("required_zones", default=[]): [str], - "height": int, - vol.Optional("retain", default={}): RETAIN_SCHEMA, - }, - vol.Optional("mqtt", default={}): { - vol.Optional("enabled", default=True): bool, - vol.Optional("timestamp", default=True): bool, - vol.Optional("bounding_box", default=True): bool, - vol.Optional("crop", default=True): bool, - vol.Optional("height", default=270): int, - vol.Optional("required_zones", default=[]): [str], - }, - vol.Optional("objects", default={}): OBJECTS_SCHEMA, - vol.Optional("motion", default={}): MOTION_SCHEMA, - vol.Optional("detect", default={}): DETECT_SCHEMA, - vol.Optional("timestamp_style", default={}): { - vol.Optional("position", default="tl"): vol.In( - ["tl", "tr", "bl", "br"] - ), - vol.Optional( - "format", default=DEFAULT_TIME_FORMAT - ): ensure_timeformat_is_legit, - vol.Optional("color", default=DEFAULT_RGB_COLOR): RGB_COLOR_SCHEMA, - vol.Optional("scale", default=1.0): float, - vol.Optional("thickness", default=2): int, - vol.Optional("effect", default=None): vol.In( - [None, "solid", "shadow"] - ), - }, - } - }, - vol.Msg( - ensure_zones_and_cameras_have_different_names, - msg="Zones cannot share names with cameras", - ), +class FfmpegConfig(BaseModel): + global_args: Union[str, List[str]] = Field( + default=FFMPEG_GLOBAL_ARGS_DEFAULT, title="Global FFmpeg arguments." + ) + hwaccel_args: Union[str, List[str]] = Field( + default_factory=list, title="FFmpeg hardware acceleration arguments." + ) + input_args: Union[str, List[str]] = Field( + default=FFMPEG_INPUT_ARGS_DEFAULT, title="FFmpeg input arguments." + ) + output_args: FfmpegOutputArgsConfig = Field( + default_factory=FfmpegOutputArgsConfig, + title="FFmpeg output arguments per role.", ) -) -@dataclasses.dataclass -class CameraSnapshotsConfig: - enabled: bool - clean_copy: bool - timestamp: bool - bounding_box: bool - crop: bool - required_zones: List[str] - height: Optional[int] - retain: RetainConfig +class CameraInput(BaseModel): + path: str = Field(title="Camera input path.") + roles: List[str] = Field(title="Roles assigned to this input.") + global_args: List[str] = Field( + default_factory=list, title="FFmpeg global arguments." + ) + hwaccel_args: List[str] = Field( + default_factory=list, title="FFmpeg hardware acceleration arguments." + ) + input_args: List[str] = Field(default_factory=list, title="FFmpeg input arguments.") - @classmethod - def build(cls, config, global_config) -> CameraSnapshotsConfig: - return CameraSnapshotsConfig( - enabled=config["enabled"], - clean_copy=config["clean_copy"], - 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"] - ), - ) + @validator("path") + def sub_env_vars(cls, v): + return v.format(**FRIGATE_ENV_VARS) - def to_dict(self) -> Dict[str, Any]: - return { - "enabled": self.enabled, - "clean_copy": self.clean_copy, - "timestamp": self.timestamp, - "bounding_box": self.bounding_box, - "crop": self.crop, - "height": self.height, - "retain": self.retain.to_dict(), - "required_zones": self.required_zones, - } + @validator("global_args", "hwaccel_args", "input_args") + def extract_args(cls, args): + return args if isinstance(args, list) else args.split(" ") -@dataclasses.dataclass -class CameraMqttConfig: - enabled: bool - timestamp: bool - bounding_box: bool - crop: bool - height: int - required_zones: List[str] +class CameraFfmpegConfig(FfmpegConfig): + inputs: List[CameraInput] = Field(title="Camera inputs.") + global_args: List[str] = Field( + default_factory=list, title="FFmpeg global arguments." + ) + hwaccel_args: List[str] = Field( + default_factory=list, title="FFmpeg hardware acceleration arguments." + ) + input_args: List[str] = Field(default_factory=list, title="FFmpeg input arguments.") + output_args: FfmpegOutputArgsConfig = Field( + default_factory=FfmpegOutputArgsConfig, title="FFmpeg output arguments." + ) - @classmethod - def build(cls, config) -> CameraMqttConfig: - return CameraMqttConfig( - config["enabled"], - config["timestamp"], - config["bounding_box"], - config["crop"], - config.get("height"), - config["required_zones"], - ) + @validator("inputs") + def validate_roles(cls, v): + roles = [role for i in v for role in i.roles] + roles_set = set(roles) - def to_dict(self) -> Dict[str, Any]: - return dataclasses.asdict(self) + if len(roles) > len(roles_set): + raise ValueError("Each input role may only be used once.") + + if not "detect" in roles: + raise ValueError("The detect role is required.") + + return v + + @validator("global_args", "hwaccel_args", "input_args") + def extract_args(cls, args): + return args if isinstance(args, list) else args.split(" ") -@dataclasses.dataclass -class TimestampStyleConfig: - position: str - format: str - color: Tuple[int, int, int] - scale: float - thickness: int - effect: str - - @classmethod - def build(cls, config) -> TimestampStyleConfig: - return TimestampStyleConfig( - config["position"], - config["format"], - (config["color"]["red"], config["color"]["green"], config["color"]["blue"]), - config["scale"], - config["thickness"], - config["effect"], - ) - - def to_dict(self) -> Dict[str, Any]: - return dataclasses.asdict(self) +class CameraSnapshotsConfig(BaseModel): + enabled: bool = Field(default=False, title="Snapshots enabled.") + clean_copy: bool = Field( + default=True, title="Create a clean copy of the snapshot image." + ) + timestamp: bool = Field( + default=False, title="Add a timestamp overlay on the snapshot." + ) + bounding_box: bool = Field( + default=True, title="Add a bounding box overlay on the snapshot." + ) + crop: bool = Field(default=False, title="Crop the snapshot to the detected object.") + required_zones: List[str] = Field( + default_factory=list, + title="List of required zones to be entered in order to save a snapshot.", + ) + height: Optional[int] = Field(title="Snapshot image height.") + retain: RetainConfig = Field( + default_factory=RetainConfig, title="Snapshot retention." + ) -@dataclasses.dataclass -class CameraClipsConfig: - enabled: bool - pre_capture: int - post_capture: int - required_zones: List[str] - objects: Optional[List[str]] - retain: RetainConfig - - @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"], - ), - ) - - def to_dict(self) -> Dict[str, Any]: - return { - "enabled": self.enabled, - "pre_capture": self.pre_capture, - "post_capture": self.post_capture, - "objects": self.objects, - "retain": self.retain.to_dict(), - "required_zones": self.required_zones, - } +class ColorConfig(BaseModel): + red: int = Field(default=255, le=0, ge=255, title="Red") + green: int = Field(default=255, le=0, ge=255, title="Green") + blue: int = Field(default=255, le=0, ge=255, title="Blue") -@dataclasses.dataclass -class CameraRtmpConfig: - enabled: bool - - @classmethod - def build(cls, config) -> CameraRtmpConfig: - return CameraRtmpConfig(config["enabled"]) - - def to_dict(self) -> Dict[str, Any]: - return dataclasses.asdict(self) +class TimestampStyleConfig(BaseModel): + position: str = Field(default="tl", title="Timestamp position.") + format: str = Field(default=DEFAULT_TIME_FORMAT, title="Timestamp format.") + color: ColorConfig = Field(default_factory=ColorConfig, title="Timestamp color.") + scale: float = Field(default=1.0, title="Timestamp scale.") + thickness: int = Field(default=2, title="Timestamp thickness.") + effect: Optional[str] = Field(title="Timestamp effect.") -@dataclasses.dataclass -class CameraLiveConfig: - height: int - width: int - quality: int - - @classmethod - def build(cls, config, camera_height, camera_width) -> CameraRtmpConfig: - if "height" in config and config["height"] <= camera_height: - height = config["height"] - width = int(height * (camera_width / camera_height)) - else: - height = camera_height - width = camera_width - - return CameraLiveConfig(height, width, config["quality"]) - - def to_dict(self) -> Dict[str, Any]: - return dataclasses.asdict(self) +class CameraMqttConfig(BaseModel): + enabled: bool = Field(default=True, title="Send image over MQTT.") + timestamp: bool = Field(default=True, title="Add timestamp to MQTT image.") + bounding_box: bool = Field(default=True, title="Add bounding box to MQTT image.") + crop: bool = Field(default=True, title="Crop MQTT image to detected object.") + height: int = Field(default=270, title="MQTT image height.") + required_zones: List[str] = Field( + default_factory=list, + title="List of required zones to be entered in order to send the image.", + ) -@dataclasses.dataclass(frozen=True) -class CameraConfig: - 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 - live: CameraLiveConfig - snapshots: CameraSnapshotsConfig - mqtt: CameraMqttConfig - objects: ObjectConfig - motion: MotionConfig - detect: DetectConfig - timestamp_style: TimestampStyleConfig +class CameraClipsConfig(BaseModel): + enabled: bool = Field(default=False, title="Save clips.") + pre_capture: int = Field(default=5, title="Seconds to capture before event starts.") + post_capture: int = Field(default=5, title="Seconds to capture after event ends.") + required_zones: List[str] = Field( + default_factory=list, + title="List of required zones to be entered in order to save the clip.", + ) + objects: Optional[List[str]] = Field( + title="List of objects to be detected in order to save the clip.", + ) + retain: RetainConfig = Field(default_factory=RetainConfig, title="Clip retention.") + + +class CameraRtmpConfig(BaseModel): + enabled: bool = Field(default=True, title="RTMP restreaming enabled.") + + +class CameraLiveConfig(BaseModel): + height: Optional[int] = Field(title="Live camera view height") + width: Optional[int] = Field(title="Live camera view width") + quality: int = Field(default=8, ge=1, le=31, title="Live camera view quality") + + +class RecordConfig(BaseModel): + enabled: bool = Field(default=False, title="Enable record on all cameras.") + retain_days: int = Field(default=30, title="Recording retention period in days.") + + +class CameraConfig(BaseModel): + name: Optional[str] = Field(title="Camera name.") + ffmpeg: CameraFfmpegConfig = Field(title="FFmpeg configuration for the camera.") + height: int = Field(title="Height of the stream for the detect role.") + width: int = Field(title="Width of the stream for the detect role.") + fps: Optional[int] = Field( + title="Number of frames per second to process through Frigate." + ) + best_image_timeout: int = Field( + default=60, + title="How long to wait for the image with the highest confidence score.", + ) + zones: Dict[str, ZoneConfig] = Field( + default_factory=dict, title="Zone configuration." + ) + clips: CameraClipsConfig = Field( + default_factory=CameraClipsConfig, title="Clip configuration." + ) + record: RecordConfig = Field( + default_factory=RecordConfig, title="Record configuration." + ) + rtmp: CameraRtmpConfig = Field( + default_factory=CameraRtmpConfig, title="RTMP restreaming configuration." + ) + live: Optional[CameraLiveConfig] = Field(title="Live playback settings.") + snapshots: CameraSnapshotsConfig = Field( + default_factory=CameraSnapshotsConfig, title="Snapshot configuration." + ) + mqtt: CameraMqttConfig = Field( + default_factory=CameraMqttConfig, title="MQTT configuration." + ) + objects: ObjectConfig = Field( + default_factory=ObjectConfig, title="Object configuration." + ) + motion: Optional[MotionConfig] = Field(title="Motion detection configuration.") + detect: Optional[DetectConfig] = Field(title="Object detection configuration.") + timestamp_style: Optional[TimestampStyleConfig] = Field( + title="Timestamp style configuration." + ) + + def __init__(self, **config): + # Set zone colors + if "zones" in config: + colors = plt.cm.get_cmap("tab10", len(config["zones"])) + config["zones"] = { + name: {**z, "color": tuple(round(255 * c) for c in colors(idx)[:3])} + for idx, (name, z) in enumerate(config["zones"].items()) + } + + super().__init__(**config) @property def frame_shape(self) -> Tuple[int, int]: @@ -891,67 +518,29 @@ class CameraConfig: ffmpeg_cmds.append({"roles": ffmpeg_input.roles, "cmd": ffmpeg_cmd}) return ffmpeg_cmds - @classmethod - def build(cls, name, config, global_config) -> CameraConfig: - colors = plt.cm.get_cmap("tab10", len(config["zones"])) - zones = { - name: ZoneConfig.build(z, tuple(round(255 * c) for c in colors(idx)[:3])) - for idx, (name, z) in enumerate(config["zones"].items()) - } - - 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"]), - live=CameraLiveConfig.build( - config["live"], config["height"], config["width"] - ), - 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) - ), - timestamp_style=TimestampStyleConfig.build(config["timestamp_style"]), - ) - - def _get_ffmpeg_cmd(self, ffmpeg_input): + def _get_ffmpeg_cmd(self, ffmpeg_input: CameraInput): ffmpeg_output_args = [] if "detect" in ffmpeg_input.roles: ffmpeg_output_args = ( - self.ffmpeg.output_args["detect"] + ffmpeg_output_args + ["pipe:"] + self.ffmpeg.output_args.detect + ffmpeg_output_args + ["pipe:"] ) if self.fps: ffmpeg_output_args = ["-r", str(self.fps)] + ffmpeg_output_args if "rtmp" in ffmpeg_input.roles and self.rtmp.enabled: ffmpeg_output_args = ( - self.ffmpeg.output_args["rtmp"] + self.ffmpeg.output_args.rtmp + [f"rtmp://127.0.0.1/live/{self.name}"] + ffmpeg_output_args ) if "clips" in ffmpeg_input.roles: ffmpeg_output_args = ( - self.ffmpeg.output_args["clips"] + self.ffmpeg.output_args.clips + [f"{os.path.join(CACHE_DIR, self.name)}-%Y%m%d%H%M%S.mp4"] + ffmpeg_output_args ) if "record" in ffmpeg_input.roles and self.record.enabled: ffmpeg_output_args = ( - self.ffmpeg.output_args["record"] + self.ffmpeg.output_args.record + [f"{os.path.join(RECORD_DIR, self.name)}-%Y%m%d%H%M%S.mp4"] + ffmpeg_output_args ) @@ -962,191 +551,195 @@ class CameraConfig: cmd = ( ["ffmpeg"] - + ffmpeg_input.global_args - + ffmpeg_input.hwaccel_args - + ffmpeg_input.input_args + + (ffmpeg_input.global_args or self.ffmpeg.global_args) + + (ffmpeg_input.hwaccel_args or self.ffmpeg.hwaccel_args) + + (ffmpeg_input.input_args or self.ffmpeg.input_args) + ["-i", ffmpeg_input.path] + ffmpeg_output_args ) return [part for part in cmd if part != ""] - def to_dict(self) -> Dict[str, Any]: - return { - "name": self.name, - "height": self.height, - "width": self.width, - "fps": self.fps, - "best_image_timeout": self.best_image_timeout, - "zones": {k: z.to_dict() for k, z in self.zones.items()}, - "clips": self.clips.to_dict(), - "record": self.record.to_dict(), - "rtmp": self.rtmp.to_dict(), - "snapshots": self.snapshots.to_dict(), - "mqtt": self.mqtt.to_dict(), - "objects": self.objects.to_dict(), - "motion": self.motion.to_dict(), - "detect": self.detect.to_dict(), - "frame_shape": self.frame_shape, - "ffmpeg_cmds": [ - {"roles": c["roles"], "cmd": " ".join(c["cmd"])} - for c in self.ffmpeg_cmds - ], - "timestamp_style": self.timestamp_style.to_dict(), - } + +class DatabaseConfig(BaseModel): + path: str = Field( + default=os.path.join(BASE_DIR, "frigate.db"), title="Database path." + ) -FRIGATE_CONFIG_SCHEMA = vol.Schema( - { - vol.Optional("database", default={}): { - vol.Optional("path", default=os.path.join(BASE_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"]) +class ModelConfig(BaseModel): + width: int = Field(default=320, title="Object detection model input width.") + height: int = Field(default=320, title="Object detection model input height.") + + +class LogLevelEnum(str, Enum): + debug = "debug" + info = "info" + warning = "warning" + error = "error" + critical = "critical" + + +class LoggerConfig(BaseModel): + default: LogLevelEnum = Field( + default=LogLevelEnum.info, title="Default logging level." + ) + logs: Dict[str, LogLevelEnum] = Field( + default_factory=dict, title="Log level for specified processes." + ) + + +class SnapshotsConfig(BaseModel): + retain: RetainConfig = Field( + default_factory=RetainConfig, title="Global snapshot retention configuration." + ) + + +class FrigateConfig(BaseModel): + mqtt: MqttConfig = Field(title="MQTT Configuration.") + database: DatabaseConfig = Field( + default_factory=DatabaseConfig, title="Database configuration." + ) + environment_vars: Dict[str, str] = Field( + default_factory=dict, title="Frigate environment variables." + ) + model: ModelConfig = Field( + default_factory=ModelConfig, title="Detection model configuration." + ) + detectors: Dict[str, DetectorConfig] = Field( + default={name: DetectorConfig(**d) for name, d in DEFAULT_DETECTORS.items()}, + title="Detector hardware configuration.", + ) + logger: LoggerConfig = Field( + default_factory=LoggerConfig, title="Logging configuration." + ) + clips: ClipsConfig = Field( + default_factory=ClipsConfig, title="Global clips configuration." + ) + record: RecordConfig = Field( + default_factory=RecordConfig, title="Global record configuration." + ) + snapshots: SnapshotsConfig = Field( + default_factory=SnapshotsConfig, title="Global snapshots configuration." + ) + birdseye: BirdseyeConfig = Field( + default_factory=BirdseyeConfig, title="Birdseye configuration." + ) + ffmpeg: FfmpegConfig = Field( + default_factory=FfmpegConfig, title="Global FFmpeg configuration." + ) + objects: ObjectConfig = Field( + default_factory=ObjectConfig, title="Global object configuration." + ) + motion: Optional[MotionConfig] = Field( + title="Global motion detection configuration." + ) + detect: Optional[DetectConfig] = Field( + title="Global object tracking configuration." + ) + cameras: Dict[str, CameraConfig] = Field(title="Camera configuration.") + + @property + def runtime_config(self) -> FrigateConfig: + """Merge camera config with globals.""" + config = self.copy(deep=True) + global_config = config.dict( + include={ + "clips": {"retain"}, + "record": ..., + "snapshots": ..., + "objects": ..., + "motion": ..., + "detect": ..., + "ffmpeg": ..., }, - }, - 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.Optional("birdseye", default={}): BIRDSEYE_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()}, + exclude_unset=True, ) - def to_dict(self) -> Dict[str, Any]: - return dataclasses.asdict(self) + for name, camera in config.cameras.items(): + merged_config = deep_merge(camera.dict(exclude_unset=True), global_config) + camera_config = CameraConfig.parse_obj({"name": name, **merged_config}) + # Add default filters + object_keys = camera_config.objects.track + if camera_config.objects.filters is None: + camera_config.objects.filters = {} + object_keys = object_keys - camera_config.objects.filters.keys() + for key in object_keys: + camera_config.objects.filters[key] = FilterConfig() -@dataclasses.dataclass(frozen=True) -class SnapshotsConfig: - retain: RetainConfig + # Apply global object masks and convert masks to numpy array + for object, filter in camera_config.objects.filters.items(): + if camera_config.objects.mask: + filter_mask = [] + if filter.mask is not None: + filter_mask = ( + filter.mask + if isinstance(filter.mask, list) + else [filter.mask] + ) + object_mask = ( + camera_config.objects.mask + if isinstance(camera_config.objects.mask, list) + else [camera_config.objects.mask] + ) + filter.mask = filter_mask + object_mask - @classmethod - def build(cls, config) -> SnapshotsConfig: - return SnapshotsConfig(RetainConfig.build(config["retain"])) + # Set runtime filter to create masks + camera_config.objects.filters[object] = RuntimeFilterConfig( + frame_shape=camera_config.frame_shape, + **filter.dict(exclude_unset=True), + ) - def to_dict(self) -> Dict[str, Any]: - return {"retain": self.retain.to_dict()} + # Convert motion configuration + if camera_config.motion is None: + camera_config.motion = RuntimeMotionConfig( + frame_shape=camera_config.frame_shape + ) + else: + camera_config.motion = RuntimeMotionConfig( + frame_shape=camera_config.frame_shape, + raw_mask=camera_config.motion.mask, + **camera_config.motion.dict(exclude_unset=True), + ) + # Default detect configuration + if camera_config.detect is None: + max_disappeared = (camera_config.fps or 5) * 5 + camera_config.detect = DetectConfig(max_disappeared=max_disappeared) -@dataclasses.dataclass -class RecordConfig: - enabled: bool - retain_days: int + # Default live configuration + if camera_config.live: + if ( + camera_config.live.height + and camera_config.live.height <= camera_config.height + ): + camera_config.live.width = int( + camera_config.live.height + * (camera_config.width / camera_config.height) + ) + else: + camera_config.live.height = camera_config.height + camera_config.live.width = camera_config.width + else: + camera_config.live = CameraLiveConfig( + height=camera_config.height, width=camera_config.width + ) - @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) -> 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: - config = self._load_file(config_file) - - config = FRIGATE_CONFIG_SCHEMA(config) - - config = self._sub_env_vars(config) - - self._database = DatabaseConfig.build(config["database"]) - self._model = ModelConfig.build(config["model"]) - self._detectors = { - name: DetectorConfig.build(d) for name, d in config["detectors"].items() - } - self._mqtt = MqttConfig.build(config["mqtt"]) - self._clips = ClipsConfig.build(config["clips"]) - self._snapshots = SnapshotsConfig.build(config["snapshots"]) - self._birdseye = BirdseyeConfig.build(config["birdseye"]) - self._cameras = { - name: CameraConfig.build(name, c, config) - for name, c in config["cameras"].items() - } - self._logger = LoggerConfig.build(config["logger"]) - self._environment_vars = config["environment_vars"] - - def _sub_env_vars(self, config): - frigate_env_vars = { - k: v for k, v in os.environ.items() if k.startswith("FRIGATE_") - } - - if "password" in config["mqtt"]: - config["mqtt"]["password"] = config["mqtt"]["password"].format( - **frigate_env_vars - ) - - for camera in config["cameras"].values(): - for i in camera["ffmpeg"]["inputs"]: - i["path"] = i["path"].format(**frigate_env_vars) + config.cameras[name] = camera_config return config - def _load_file(self, config_file): + @validator("cameras") + def ensure_zones_and_cameras_have_different_names(cls, v: Dict[str, CameraConfig]): + zones = [zone for camera in v.values() for zone in camera.zones.keys()] + for zone in zones: + if zone in v.keys(): + raise ValueError("Zones cannot share names with cameras") + return v + + @classmethod + def parse_file(cls, config_file): with open(config_file) as f: raw_config = f.read() @@ -1155,58 +748,4 @@ class FrigateConfig: elif config_file.endswith(".json"): config = json.loads(raw_config) - return config - - def to_dict(self) -> Dict[str, Any]: - return { - "database": self.database.to_dict(), - "model": self.model.to_dict(), - "detectors": {k: d.to_dict() for k, d in self.detectors.items()}, - "mqtt": self.mqtt.to_dict(), - "clips": self.clips.to_dict(), - "snapshots": self.snapshots.to_dict(), - "birdseye": self.birdseye.to_dict(), - "cameras": {k: c.to_dict() for k, c in self.cameras.items()}, - "logger": self.logger.to_dict(), - "environment_vars": self._environment_vars, - } - - @property - def database(self) -> DatabaseConfig: - return self._database - - @property - def model(self) -> ModelConfig: - return self._model - - @property - def detectors(self) -> Dict[str, DetectorConfig]: - return self._detectors - - @property - def logger(self) -> LoggerConfig: - return self._logger - - @property - def mqtt(self) -> MqttConfig: - return self._mqtt - - @property - def clips(self) -> ClipsConfig: - return self._clips - - @property - def snapshots(self) -> SnapshotsConfig: - return self._snapshots - - @property - def birdseye(self) -> BirdseyeConfig: - return self._birdseye - - @property - def cameras(self) -> Dict[str, CameraConfig]: - return self._cameras - - @property - def environment_vars(self): - return self._environment_vars + return cls.parse_obj(config) diff --git a/frigate/http.py b/frigate/http.py index 3b962f16f..da64e9ef8 100644 --- a/frigate/http.py +++ b/frigate/http.py @@ -272,7 +272,14 @@ def events(): @bp.route("/config") def config(): - return jsonify(current_app.frigate_config.to_dict()) + return jsonify(current_app.frigate_config.dict()) + + +@bp.route("/config/schema") +def config_schema(): + return current_app.response_class( + current_app.frigate_config.schema_json(), mimetype="application/json" + ) @bp.route("/version") diff --git a/frigate/motion.py b/frigate/motion.py index 3b9eb20bb..a88ea000d 100644 --- a/frigate/motion.py +++ b/frigate/motion.py @@ -2,6 +2,7 @@ import cv2 import imutils import numpy as np from frigate.config import MotionConfig +from frigate.util import create_mask class MotionDetector: @@ -18,7 +19,7 @@ class MotionDetector: self.motion_frame_count = 0 self.frame_counter = 0 resized_mask = cv2.resize( - config.mask, + create_mask(frame_shape, config.mask), dsize=(self.motion_frame_size[1], self.motion_frame_size[0]), interpolation=cv2.INTER_LINEAR, ) diff --git a/frigate/object_processing.py b/frigate/object_processing.py index 386d44ee9..245db1efe 100644 --- a/frigate/object_processing.py +++ b/frigate/object_processing.py @@ -272,6 +272,7 @@ class TrackedObject: best_frame, dsize=(width, height), interpolation=cv2.INTER_AREA ) if timestamp: + color = self.camera_config.timestamp_style.color draw_timestamp( best_frame, self.thumbnail_data["frame_time"], @@ -279,7 +280,7 @@ class TrackedObject: font_effect=self.camera_config.timestamp_style.effect, font_scale=self.camera_config.timestamp_style.scale, font_thickness=self.camera_config.timestamp_style.thickness, - font_color=self.camera_config.timestamp_style.color, + font_color=(color.red, color.green, color.blue), position=self.camera_config.timestamp_style.position, ) @@ -318,7 +319,7 @@ class CameraState: def __init__(self, name, config, frame_manager): self.name = name self.config = config - self.camera_config = config.cameras[name] + self.camera_config: CameraConfig = config.cameras[name] self.frame_manager = frame_manager self.best_objects: Dict[str, TrackedObject] = {} self.object_counts = defaultdict(int) @@ -328,6 +329,7 @@ class CameraState: self._current_frame = np.zeros(self.camera_config.frame_shape_yuv, np.uint8) self.current_frame_lock = threading.Lock() self.current_frame_time = 0.0 + self.motion_mask = self.camera_config.motion.mask self.motion_boxes = [] self.regions = [] self.previous_frame_id = None @@ -389,7 +391,7 @@ class CameraState: cv2.drawContours(frame_copy, [zone.contour], -1, zone.color, thickness) if draw_options.get("mask"): - mask_overlay = np.where(self.camera_config.motion.mask == [0]) + mask_overlay = np.where(self.motion_mask == [0]) frame_copy[mask_overlay] = [0, 0, 0] if draw_options.get("motion_boxes"): diff --git a/frigate/output.py b/frigate/output.py index ce6ba272f..9b265c530 100644 --- a/frigate/output.py +++ b/frigate/output.py @@ -21,7 +21,7 @@ from ws4py.server.wsgirefserver import ( from ws4py.server.wsgiutils import WebSocketWSGIApplication from ws4py.websocket import WebSocket -from frigate.config import FrigateConfig +from frigate.config import BirdseyeModeEnum, FrigateConfig from frigate.util import SharedMemoryFrameManager, copy_yuv_to_position, get_yuv_crop logger = logging.getLogger(__name__) @@ -173,13 +173,16 @@ class BirdsEyeFrameManager: ) def camera_active(self, object_box_count, motion_box_count): - if self.mode == "continuous": + if self.mode == BirdseyeModeEnum.continuous: return True - if self.mode == "motion" and object_box_count + motion_box_count > 0: + if ( + self.mode == BirdseyeModeEnum.motion + and object_box_count + motion_box_count > 0 + ): return True - if self.mode == "objects" and object_box_count > 0: + if self.mode == BirdseyeModeEnum.objects and object_box_count > 0: return True def update_frame(self): diff --git a/frigate/test/test_config.py b/frigate/test/test_config.py index 8f6aba6f3..2530027d0 100644 --- a/frigate/test/test_config.py +++ b/frigate/test/test_config.py @@ -1,10 +1,13 @@ -import json -from unittest import TestCase, main -import voluptuous as vol -from frigate.config import FRIGATE_CONFIG_SCHEMA, FrigateConfig +import unittest +import numpy as np +from pydantic import ValidationError +from frigate.config import ( + FrigateConfig, + DetectorTypeEnum, +) -class TestConfig(TestCase): +class TestConfig(unittest.TestCase): def setUp(self): self.minimal = { "mqtt": {"host": "mqtt"}, @@ -21,14 +24,30 @@ class TestConfig(TestCase): }, } - def test_empty(self): - FRIGATE_CONFIG_SCHEMA({}) - - def test_minimal(self): - FRIGATE_CONFIG_SCHEMA(self.minimal) - def test_config_class(self): - FrigateConfig(config=self.minimal) + frigate_config = FrigateConfig(**self.minimal) + assert self.minimal == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "coral" in runtime_config.detectors.keys() + assert runtime_config.detectors["coral"].type == DetectorTypeEnum.edgetpu + + def test_invalid_mqtt_config(self): + config = { + "mqtt": {"host": "mqtt", "user": "test"}, + "cameras": { + "back": { + "ffmpeg": { + "inputs": [ + {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]} + ] + }, + "height": 1080, + "width": 1920, + } + }, + } + self.assertRaises(ValidationError, lambda: FrigateConfig(**config)) def test_inherit_tracked_objects(self): config = { @@ -46,8 +65,11 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "dog" in frigate_config.cameras["back"].objects.track + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "dog" in runtime_config.cameras["back"].objects.track def test_override_tracked_objects(self): config = { @@ -66,8 +88,11 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "cat" in frigate_config.cameras["back"].objects.track + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "cat" in runtime_config.cameras["back"].objects.track def test_default_object_filters(self): config = { @@ -85,8 +110,11 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "dog" in frigate_config.cameras["back"].objects.filters + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "dog" in runtime_config.cameras["back"].objects.filters def test_inherit_object_filters(self): config = { @@ -107,9 +135,12 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "dog" in frigate_config.cameras["back"].objects.filters - assert frigate_config.cameras["back"].objects.filters["dog"].threshold == 0.7 + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "dog" in runtime_config.cameras["back"].objects.filters + assert runtime_config.cameras["back"].objects.filters["dog"].threshold == 0.7 def test_override_object_filters(self): config = { @@ -130,9 +161,12 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "dog" in frigate_config.cameras["back"].objects.filters - assert frigate_config.cameras["back"].objects.filters["dog"].threshold == 0.7 + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "dog" in runtime_config.cameras["back"].objects.filters + assert runtime_config.cameras["back"].objects.filters["dog"].threshold == 0.7 def test_global_object_mask(self): config = { @@ -154,12 +188,14 @@ 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["person"].raw_mask) == 1 - ) + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + back_camera = runtime_config.cameras["back"] + assert "dog" in back_camera.objects.filters + assert len(back_camera.objects.filters["dog"].raw_mask) == 2 + assert len(back_camera.objects.filters["person"].raw_mask) == 1 def test_ffmpeg_params_global(self): config = { @@ -181,8 +217,11 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"] + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] def test_ffmpeg_params_camera(self): config = { @@ -204,8 +243,11 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"] + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] def test_ffmpeg_params_input(self): config = { @@ -230,8 +272,11 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"] + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] def test_inherit_clips_retention(self): config = { @@ -249,8 +294,11 @@ class TestConfig(TestCase): } }, } - frigate_config = FrigateConfig(config=config) - assert frigate_config.cameras["back"].clips.retain.objects["person"] == 30 + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert runtime_config.cameras["back"].clips.retain.objects["person"] == 30 def test_roles_listed_twice_throws_error(self): config = { @@ -269,7 +317,7 @@ class TestConfig(TestCase): } }, } - self.assertRaises(vol.MultipleInvalid, lambda: FrigateConfig(config=config)) + self.assertRaises(ValidationError, lambda: FrigateConfig(**config)) def test_zone_matching_camera_name_throws_error(self): config = { @@ -288,7 +336,33 @@ class TestConfig(TestCase): } }, } - self.assertRaises(vol.MultipleInvalid, lambda: FrigateConfig(config=config)) + self.assertRaises(ValidationError, lambda: FrigateConfig(**config)) + + def test_zone_assigns_color_and_contour(self): + config = { + "mqtt": {"host": "mqtt"}, + "clips": {"retain": {"default": 20, "objects": {"person": 30}}}, + "cameras": { + "back": { + "ffmpeg": { + "inputs": [ + {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]} + ] + }, + "height": 1080, + "width": 1920, + "zones": {"test": {"coordinates": "1,1,1,1,1,1"}}, + } + }, + } + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + assert isinstance( + runtime_config.cameras["back"].zones["test"].contour, np.ndarray + ) + assert runtime_config.cameras["back"].zones["test"].color != (0, 0, 0) def test_clips_should_default_to_global_objects(self): config = { @@ -308,11 +382,16 @@ class TestConfig(TestCase): } }, } - config = FrigateConfig(config=config) - assert config.cameras["back"].clips.objects is None + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + back_camera = runtime_config.cameras["back"] + assert back_camera.clips.objects is None + assert back_camera.clips.retain.objects["person"] == 30 def test_role_assigned_but_not_enabled(self): - json_config = { + config = { "mqtt": {"host": "mqtt"}, "cameras": { "back": { @@ -331,11 +410,14 @@ class TestConfig(TestCase): }, } - config = FrigateConfig(config=json_config) - ffmpeg_cmds = config.cameras["back"].ffmpeg_cmds + frigate_config = FrigateConfig(**config) + assert config == frigate_config.dict(exclude_unset=True) + + runtime_config = frigate_config.runtime_config + ffmpeg_cmds = runtime_config.cameras["back"].ffmpeg_cmds assert len(ffmpeg_cmds) == 1 assert not "clips" in ffmpeg_cmds[0]["roles"] if __name__ == "__main__": - main(verbosity=2) + unittest.main(verbosity=2) diff --git a/frigate/util.py b/frigate/util.py index 5349ca23b..a09d2721a 100755 --- a/frigate/util.py +++ b/frigate/util.py @@ -1,4 +1,5 @@ import collections +import copy import datetime import hashlib import json @@ -20,6 +21,29 @@ import numpy as np logger = logging.getLogger(__name__) +def deep_merge(dct1: dict, dct2: dict, override=False) -> dict: + """ + :param dct1: First dict to merge + :param dct2: Second dict to merge + :param override: if same key exists in both dictionaries, should override? otherwise ignore. (default=True) + :return: The merge dictionary + """ + merged = copy.deepcopy(dct1) + for k, v2 in dct2.items(): + if k in merged: + v1 = merged[k] + if isinstance(v1, dict) and isinstance(v2, collections.Mapping): + merged[k] = deep_merge(v1, v2, override) + elif isinstance(v1, list) and isinstance(v2, list): + merged[k] = v1 + v2 + else: + if override: + merged[k] = copy.deepcopy(v2) + else: + merged[k] = copy.deepcopy(v2) + return merged + + def draw_timestamp( frame, timestamp,