administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-02-15 07:35:27 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'dev' into frigate-http-api-with-fastapi

Browse Source 
# Conflicts:
#	frigate/api/app.py
#	frigate/test/test_http.py
This commit is contained in:
Rui Alves 2024-09-22 19:34:32 +01:00
commit 59dfe2ddc1
13 changed files with 233 additions and 363 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.devcontainer/devcontainer.json
View File

@ -52,7 +52,8 @@
"csstools.postcss", "csstools.postcss",
"blanu.vscode-styled-jsx", "blanu.vscode-styled-jsx",
"bradlc.vscode-tailwindcss", "bradlc.vscode-tailwindcss",
"charliermarsh.ruff" "charliermarsh.ruff",
"eamodio.gitlens"
], ],
"settings": { "settings": {
"remote.autoForwardPorts": false, "remote.autoForwardPorts": false,

3
.gitignore vendored
View File

@ -1,5 +1,6 @@
.DS_Store .DS_Store
*.pyc __pycache__
.mypy_cache
*.swp *.swp
debug debug
.vscode/* .vscode/*

2
docker/main/install_deps.sh
View File

@ -91,7 +91,7 @@ fi
if [[ "${TARGETARCH}" == "arm64" ]]; then if [[ "${TARGETARCH}" == "arm64" ]]; then
apt-get -qq install --no-install-recommends --no-install-suggests -y \ apt-get -qq install --no-install-recommends --no-install-suggests -y \
libva-drm2 mesa-va-drivers libva-drm2 mesa-va-drivers radeontop
fi fi
# install vulkan # install vulkan

6
frigate/api/app.py
View File

@ -177,7 +177,7 @@ def config_save(save_option: str, body: dict):
# Validate the config schema # Validate the config schema
try: try:
FrigateConfig.parse_raw(new_config) FrigateConfig.parse_yaml(new_config)
except Exception: except Exception:
return JSONResponse( return JSONResponse(
content=( content=(
@ -265,7 +265,7 @@ def config_set(request: Request, body: AppConfigSetBody):
f.close() f.close()
# Validate the config schema # Validate the config schema
try: try:
config_obj = FrigateConfig.parse_raw(new_raw_config) config_obj = FrigateConfig.parse_yaml(new_raw_config)
except Exception: except Exception:
with open(config_file, "w") as f: with open(config_file, "w") as f:
f.write(old_raw_config) f.write(old_raw_config)
@ -288,7 +288,7 @@ def config_set(request: Request, body: AppConfigSetBody):
) )
if body.requires_restart == 0: if body.requires_restart == 0:
request.app.frigate_config = FrigateConfig.runtime_config( request.app.frigate_config = FrigateConfig.parse_object(
config_obj, request.app.plus_api config_obj, request.app.plus_api
) )

3
frigate/app.py
View File

@ -130,8 +130,7 @@ class FrigateApp:
# check if the config file needs to be migrated # check if the config file needs to be migrated
migrate_frigate_config(config_file) migrate_frigate_config(config_file)
user_config = FrigateConfig.parse_file(config_file) self.config = FrigateConfig.parse_file(config_file, plus_api=self.plus_api)
self.config = user_config.runtime_config(self.plus_api)
for camera_name in self.config.cameras.keys(): for camera_name in self.config.cameras.keys():
# create camera_metrics # create camera_metrics

186
frigate/config.py
View File

@ -6,10 +6,11 @@ import os
import shutil import shutil
from enum import Enum from enum import Enum
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union from typing import Annotated, Any, Dict, List, Optional, Tuple, Union
import numpy as np import numpy as np
from pydantic import ( from pydantic import (
AfterValidator,
BaseModel, BaseModel,
ConfigDict, ConfigDict,
Field, Field,
@ -17,8 +18,11 @@ from pydantic import (
ValidationInfo, ValidationInfo,
field_serializer, field_serializer,
field_validator, field_validator,
model_validator,
) )
from pydantic.fields import PrivateAttr from pydantic.fields import PrivateAttr
from ruamel.yaml import YAML
from typing_extensions import Self
from frigate.const import ( from frigate.const import (
ALL_ATTRIBUTE_LABELS, ALL_ATTRIBUTE_LABELS,
@ -31,7 +35,7 @@ from frigate.const import (
INCLUDED_FFMPEG_VERSIONS, INCLUDED_FFMPEG_VERSIONS,
MAX_PRE_CAPTURE, MAX_PRE_CAPTURE,
REGEX_CAMERA_NAME, REGEX_CAMERA_NAME,
YAML_EXT, REGEX_JSON,
) )
from frigate.detectors import DetectorConfig, ModelConfig from frigate.detectors import DetectorConfig, ModelConfig
from frigate.detectors.detector_config import BaseDetectorConfig from frigate.detectors.detector_config import BaseDetectorConfig
@ -41,13 +45,11 @@ from frigate.ffmpeg_presets import (
parse_preset_input, parse_preset_input,
parse_preset_output_record, parse_preset_output_record,
) )
from frigate.plus import PlusApi
from frigate.util.builtin import ( from frigate.util.builtin import (
deep_merge, deep_merge,
escape_special_characters, escape_special_characters,
generate_color_palette, generate_color_palette,
get_ffmpeg_arg_list, get_ffmpeg_arg_list,
load_config_with_no_duplicates,
) )
from frigate.util.config import StreamInfoRetriever, get_relative_coordinates from frigate.util.config import StreamInfoRetriever, get_relative_coordinates
from frigate.util.image import create_mask from frigate.util.image import create_mask
@ -55,6 +57,8 @@ from frigate.util.services import auto_detect_hwaccel
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
yaml = YAML()
# TODO: Identify what the default format to display timestamps is # TODO: Identify what the default format to display timestamps is
DEFAULT_TIME_FORMAT = "%m/%d/%Y %H:%M:%S" DEFAULT_TIME_FORMAT = "%m/%d/%Y %H:%M:%S"
# German Style: # German Style:
@ -103,6 +107,13 @@ class DateTimeStyleEnum(str, Enum):
short = "short" short = "short"
def validate_env_string(v: str) -> str:
return v.format(**FRIGATE_ENV_VARS)
EnvString = Annotated[str, AfterValidator(validate_env_string)]
class UIConfig(FrigateBaseModel): class UIConfig(FrigateBaseModel):
timezone: Optional[str] = Field(default=None, title="Override UI timezone.") timezone: Optional[str] = Field(default=None, title="Override UI timezone.")
time_format: TimeFormatEnum = Field( time_format: TimeFormatEnum = Field(
@ -137,7 +148,7 @@ class ProxyConfig(FrigateBaseModel):
logout_url: Optional[str] = Field( logout_url: Optional[str] = Field(
default=None, title="Redirect url for logging out with proxy." default=None, title="Redirect url for logging out with proxy."
) )
auth_secret: Optional[str] = Field( auth_secret: Optional[EnvString] = Field(
default=None, default=None,
title="Secret value for proxy authentication.", title="Secret value for proxy authentication.",
) )
@ -208,8 +219,10 @@ class MqttConfig(FrigateBaseModel):
stats_interval: int = Field( stats_interval: int = Field(
default=60, ge=FREQUENCY_STATS_POINTS, title="MQTT Camera Stats Interval" default=60, ge=FREQUENCY_STATS_POINTS, title="MQTT Camera Stats Interval"
) )
user: Optional[str] = Field(None, title="MQTT Username") user: Optional[EnvString] = Field(None, title="MQTT Username")
password: Optional[str] = Field(None, title="MQTT Password", validate_default=True) password: Optional[EnvString] = Field(
None, title="MQTT Password", validate_default=True
)
tls_ca_certs: Optional[str] = Field(None, title="MQTT TLS CA Certificates") tls_ca_certs: Optional[str] = Field(None, title="MQTT TLS CA Certificates")
tls_client_cert: Optional[str] = Field(None, title="MQTT TLS Client Certificate") tls_client_cert: Optional[str] = Field(None, title="MQTT TLS Client Certificate")
tls_client_key: Optional[str] = Field(None, title="MQTT TLS Client Key") tls_client_key: Optional[str] = Field(None, title="MQTT TLS Client Key")
@ -284,8 +297,8 @@ class PtzAutotrackConfig(FrigateBaseModel):
class OnvifConfig(FrigateBaseModel): class OnvifConfig(FrigateBaseModel):
host: str = Field(default="", title="Onvif Host") host: str = Field(default="", title="Onvif Host")
port: int = Field(default=8000, title="Onvif Port") port: int = Field(default=8000, title="Onvif Port")
user: Optional[str] = Field(None, title="Onvif Username") user: Optional[EnvString] = Field(None, title="Onvif Username")
password: Optional[str] = Field(None, title="Onvif Password") password: Optional[EnvString] = Field(None, title="Onvif Password")
autotracking: PtzAutotrackConfig = Field( autotracking: PtzAutotrackConfig = Field(
default_factory=PtzAutotrackConfig, default_factory=PtzAutotrackConfig,
title="PTZ auto tracking config.", title="PTZ auto tracking config.",
@ -756,7 +769,7 @@ class GenAIConfig(FrigateBaseModel):
default=GenAIProviderEnum.openai, title="GenAI provider." default=GenAIProviderEnum.openai, title="GenAI provider."
) )
base_url: Optional[str] = Field(None, title="Provider base url.") base_url: Optional[str] = Field(None, title="Provider base url.")
api_key: Optional[str] = Field(None, title="Provider API key.") api_key: Optional[EnvString] = Field(None, title="Provider API key.")
model: str = Field(default="gpt-4o", title="GenAI model.") model: str = Field(default="gpt-4o", title="GenAI model.")
prompt: str = Field( prompt: str = Field(
default="Describe the {label} in the sequence of images with as much detail as possible. Do not describe the background.", default="Describe the {label} in the sequence of images with as much detail as possible. Do not describe the background.",
@ -926,7 +939,7 @@ class CameraRoleEnum(str, Enum):
class CameraInput(FrigateBaseModel): class CameraInput(FrigateBaseModel):
path: str = Field(title="Camera input path.") path: EnvString = Field(title="Camera input path.")
roles: List[CameraRoleEnum] = Field(title="Roles assigned to this input.") roles: List[CameraRoleEnum] = Field(title="Roles assigned to this input.")
global_args: Union[str, List[str]] = Field( global_args: Union[str, List[str]] = Field(
default_factory=list, title="FFmpeg global arguments." default_factory=list, title="FFmpeg global arguments."
@ -1346,17 +1359,15 @@ def verify_recording_segments_setup_with_reasonable_time(
if record_args[0].startswith("preset"): if record_args[0].startswith("preset"):
return return
try:
seg_arg_index = record_args.index("-segment_time") seg_arg_index = record_args.index("-segment_time")
except ValueError:
if seg_arg_index < 0: raise ValueError(f"Camera {camera_config.name} has no segment_time in \
raise ValueError( recording output args, segment args are required for record.")
f"Camera {camera_config.name} has no segment_time in recording output args, segment args are required for record."
)
if int(record_args[seg_arg_index + 1]) > 60: if int(record_args[seg_arg_index + 1]) > 60:
raise ValueError( raise ValueError(f"Camera {camera_config.name} has invalid segment_time output arg, \
f"Camera {camera_config.name} has invalid segment_time output arg, segment_time must be 60 or less." segment_time must be 60 or less.")
)
def verify_zone_objects_are_tracked(camera_config: CameraConfig) -> None: def verify_zone_objects_are_tracked(camera_config: CameraConfig) -> None:
@ -1481,41 +1492,28 @@ class FrigateConfig(FrigateBaseModel):
) )
version: Optional[str] = Field(default=None, title="Current config version.") version: Optional[str] = Field(default=None, title="Current config version.")
def runtime_config(self, plus_api: PlusApi = None) -> FrigateConfig: @model_validator(mode="after")
"""Merge camera config with globals.""" def post_validation(self, info: ValidationInfo) -> Self:
config = self.model_copy(deep=True) plus_api = None
if isinstance(info.context, dict):
# Proxy secret substitution plus_api = info.context.get("plus_api")
if config.proxy.auth_secret:
config.proxy.auth_secret = config.proxy.auth_secret.format(
**FRIGATE_ENV_VARS
)
# MQTT user/password substitutions
if config.mqtt.user or config.mqtt.password:
config.mqtt.user = config.mqtt.user.format(**FRIGATE_ENV_VARS)
config.mqtt.password = config.mqtt.password.format(**FRIGATE_ENV_VARS)
# set notifications state # set notifications state
config.notifications.enabled_in_config = config.notifications.enabled self.notifications.enabled_in_config = self.notifications.enabled
# GenAI substitution
if config.genai.api_key:
config.genai.api_key = config.genai.api_key.format(**FRIGATE_ENV_VARS)
# set default min_score for object attributes # set default min_score for object attributes
for attribute in ALL_ATTRIBUTE_LABELS: for attribute in ALL_ATTRIBUTE_LABELS:
if not config.objects.filters.get(attribute): if not self.objects.filters.get(attribute):
config.objects.filters[attribute] = FilterConfig(min_score=0.7) self.objects.filters[attribute] = FilterConfig(min_score=0.7)
elif config.objects.filters[attribute].min_score == 0.5: elif self.objects.filters[attribute].min_score == 0.5:
config.objects.filters[attribute].min_score = 0.7 self.objects.filters[attribute].min_score = 0.7
# auto detect hwaccel args # auto detect hwaccel args
if config.ffmpeg.hwaccel_args == "auto": if self.ffmpeg.hwaccel_args == "auto":
config.ffmpeg.hwaccel_args = auto_detect_hwaccel() self.ffmpeg.hwaccel_args = auto_detect_hwaccel()
# Global config to propagate down to camera level # Global config to propagate down to camera level
global_config = config.model_dump( global_config = self.model_dump(
include={ include={
"audio": ..., "audio": ...,
"birdseye": ..., "birdseye": ...,
@ -1533,7 +1531,7 @@ class FrigateConfig(FrigateBaseModel):
exclude_unset=True, exclude_unset=True,
) )
for name, camera in config.cameras.items(): for name, camera in self.cameras.items():
merged_config = deep_merge( merged_config = deep_merge(
camera.model_dump(exclude_unset=True), global_config camera.model_dump(exclude_unset=True), global_config
) )
@ -1542,7 +1540,7 @@ class FrigateConfig(FrigateBaseModel):
) )
if camera_config.ffmpeg.hwaccel_args == "auto": if camera_config.ffmpeg.hwaccel_args == "auto":
camera_config.ffmpeg.hwaccel_args = config.ffmpeg.hwaccel_args camera_config.ffmpeg.hwaccel_args = self.ffmpeg.hwaccel_args
for input in camera_config.ffmpeg.inputs: for input in camera_config.ffmpeg.inputs:
need_record_fourcc = False and "record" in input.roles need_record_fourcc = False and "record" in input.roles
@ -1555,7 +1553,7 @@ class FrigateConfig(FrigateBaseModel):
stream_info = {"width": 0, "height": 0, "fourcc": None} stream_info = {"width": 0, "height": 0, "fourcc": None}
try: try:
stream_info = stream_info_retriever.get_stream_info( stream_info = stream_info_retriever.get_stream_info(
config.ffmpeg, input.path self.ffmpeg, input.path
) )
except Exception: except Exception:
logger.warn( logger.warn(
@ -1607,18 +1605,6 @@ class FrigateConfig(FrigateBaseModel):
if camera_config.detect.stationary.interval is None: if camera_config.detect.stationary.interval is None:
camera_config.detect.stationary.interval = stationary_threshold camera_config.detect.stationary.interval = stationary_threshold
# FFMPEG input substitution
for input in camera_config.ffmpeg.inputs:
input.path = input.path.format(**FRIGATE_ENV_VARS)
# ONVIF substitution
if camera_config.onvif.user or camera_config.onvif.password:
camera_config.onvif.user = camera_config.onvif.user.format(
**FRIGATE_ENV_VARS
)
camera_config.onvif.password = camera_config.onvif.password.format(
**FRIGATE_ENV_VARS
)
# set config pre-value # set config pre-value
camera_config.audio.enabled_in_config = camera_config.audio.enabled camera_config.audio.enabled_in_config = camera_config.audio.enabled
camera_config.record.enabled_in_config = camera_config.record.enabled camera_config.record.enabled_in_config = camera_config.record.enabled
@ -1685,8 +1671,12 @@ class FrigateConfig(FrigateBaseModel):
if not camera_config.live.stream_name: if not camera_config.live.stream_name:
camera_config.live.stream_name = name camera_config.live.stream_name = name
# generate the ffmpeg commands
camera_config.create_ffmpeg_cmds()
self.cameras[name] = camera_config
verify_config_roles(camera_config) verify_config_roles(camera_config)
verify_valid_live_stream_name(config, camera_config) verify_valid_live_stream_name(self, camera_config)
verify_recording_retention(camera_config) verify_recording_retention(camera_config)
verify_recording_segments_setup_with_reasonable_time(camera_config) verify_recording_segments_setup_with_reasonable_time(camera_config)
verify_zone_objects_are_tracked(camera_config) verify_zone_objects_are_tracked(camera_config)
@ -1694,20 +1684,16 @@ class FrigateConfig(FrigateBaseModel):
verify_autotrack_zones(camera_config) verify_autotrack_zones(camera_config)
verify_motion_and_detect(camera_config) verify_motion_and_detect(camera_config)
# generate the ffmpeg commands
camera_config.create_ffmpeg_cmds()
config.cameras[name] = camera_config
# get list of unique enabled labels for tracking # get list of unique enabled labels for tracking
enabled_labels = set(config.objects.track) enabled_labels = set(self.objects.track)
for _, camera in config.cameras.items(): for camera in self.cameras.values():
enabled_labels.update(camera.objects.track) enabled_labels.update(camera.objects.track)
config.model.create_colormap(sorted(enabled_labels)) self.model.create_colormap(sorted(enabled_labels))
config.model.check_and_load_plus_model(plus_api) self.model.check_and_load_plus_model(plus_api)
for key, detector in config.detectors.items(): for key, detector in self.detectors.items():
adapter = TypeAdapter(DetectorConfig) adapter = TypeAdapter(DetectorConfig)
model_dict = ( model_dict = (
detector detector
@ -1716,10 +1702,10 @@ class FrigateConfig(FrigateBaseModel):
) )
detector_config: DetectorConfig = adapter.validate_python(model_dict) detector_config: DetectorConfig = adapter.validate_python(model_dict)
if detector_config.model is None: if detector_config.model is None:
detector_config.model = config.model.model_copy() detector_config.model = self.model.model_copy()
else: else:
path = detector_config.model.path path = detector_config.model.path
detector_config.model = config.model.model_copy() detector_config.model = self.model.model_copy()
detector_config.model.path = path detector_config.model.path = path
if "path" not in model_dict or len(model_dict.keys()) > 1: if "path" not in model_dict or len(model_dict.keys()) > 1:
@ -1729,7 +1715,7 @@ class FrigateConfig(FrigateBaseModel):
merged_model = deep_merge( merged_model = deep_merge(
detector_config.model.model_dump(exclude_unset=True, warnings="none"), detector_config.model.model_dump(exclude_unset=True, warnings="none"),
config.model.model_dump(exclude_unset=True, warnings="none"), self.model.model_dump(exclude_unset=True, warnings="none"),
) )
if "path" not in merged_model: if "path" not in merged_model:
@ -1743,9 +1729,9 @@ class FrigateConfig(FrigateBaseModel):
plus_api, detector_config.type plus_api, detector_config.type
) )
detector_config.model.compute_model_hash() detector_config.model.compute_model_hash()
config.detectors[key] = detector_config self.detectors[key] = detector_config
return config return self
@field_validator("cameras") @field_validator("cameras")
@classmethod @classmethod
@ -1757,18 +1743,42 @@ class FrigateConfig(FrigateBaseModel):
return v return v
@classmethod @classmethod
def parse_file(cls, config_file): def parse_file(cls, config_path, **kwargs):
with open(config_file) as f: with open(config_path) as f:
raw_config = f.read() return FrigateConfig.parse(f, **kwargs)
if config_file.endswith(YAML_EXT):
config = load_config_with_no_duplicates(raw_config)
elif config_file.endswith(".json"):
config = json.loads(raw_config)
return cls.model_validate(config)
@classmethod @classmethod
def parse_raw(cls, raw_config): def parse(cls, config, *, is_json=None, **context):
config = load_config_with_no_duplicates(raw_config) # If config is a file, read its contents.
return cls.model_validate(config) if hasattr(config, "read"):
fname = getattr(config, "name", None)
config = config.read()
# Try to guess the value of is_json from the file extension.
if is_json is None and fname:
_, ext = os.path.splitext(fname)
if ext in (".yaml", ".yml"):
is_json = False
elif ext == ".json":
is_json = True
# At this point, ry to sniff the config string, to guess if it is json or not.
if is_json is None:
is_json = REGEX_JSON.match(config) is not None
# Parse the config into a dictionary.
if is_json:
config = json.load(config)
else:
config = yaml.load(config)
# Validate and return the config dict.
return cls.parse_object(config, **context)
@classmethod
def parse_object(cls, obj: Any, **context):
return cls.model_validate(obj, context=context)
@classmethod
def parse_yaml(cls, config_yaml, **context):
return cls.parse(config_yaml, is_json=False, **context)

4
frigate/const.py
View File

@ -1,3 +1,5 @@
import re
CONFIG_DIR = "/config" CONFIG_DIR = "/config"
DEFAULT_DB_PATH = f"{CONFIG_DIR}/frigate.db" DEFAULT_DB_PATH = f"{CONFIG_DIR}/frigate.db"
MODEL_CACHE_DIR = f"{CONFIG_DIR}/model_cache" MODEL_CACHE_DIR = f"{CONFIG_DIR}/model_cache"
@ -7,7 +9,6 @@ RECORD_DIR = f"{BASE_DIR}/recordings"
EXPORT_DIR = f"{BASE_DIR}/exports" EXPORT_DIR = f"{BASE_DIR}/exports"
BIRDSEYE_PIPE = "/tmp/cache/birdseye" BIRDSEYE_PIPE = "/tmp/cache/birdseye"
CACHE_DIR = "/tmp/cache" CACHE_DIR = "/tmp/cache"
YAML_EXT = (".yaml", ".yml")
FRIGATE_LOCALHOST = "http://127.0.0.1:5000" FRIGATE_LOCALHOST = "http://127.0.0.1:5000"
PLUS_ENV_VAR = "PLUS_API_KEY" PLUS_ENV_VAR = "PLUS_API_KEY"
PLUS_API_HOST = "https://api.frigate.video" PLUS_API_HOST = "https://api.frigate.video"
@ -56,6 +57,7 @@ FFMPEG_HWACCEL_VULKAN = "preset-vulkan"
REGEX_CAMERA_NAME = r"^[a-zA-Z0-9_-]+$" REGEX_CAMERA_NAME = r"^[a-zA-Z0-9_-]+$"
REGEX_RTSP_CAMERA_USER_PASS = r":\/\/[a-zA-Z0-9_-]+:[\S]+@" REGEX_RTSP_CAMERA_USER_PASS = r":\/\/[a-zA-Z0-9_-]+:[\S]+@"
REGEX_HTTP_CAMERA_USER_PASS = r"user=[a-zA-Z0-9_-]+&password=[\S]+" REGEX_HTTP_CAMERA_USER_PASS = r"user=[a-zA-Z0-9_-]+&password=[\S]+"
REGEX_JSON = re.compile(r"^\s*\{")
# Known Driver Names # Known Driver Names

4
frigate/detectors/plugins/onnx.py
View File

@ -53,11 +53,9 @@ class ONNXDetector(DetectionApi):
options.append( options.append(
{ {
"trt_timing_cache_enable": True, "trt_timing_cache_enable": True,
"trt_timing_cache_path": "/config/model_cache/tensorrt/ort",
"trt_engine_cache_enable": True, "trt_engine_cache_enable": True,
"trt_dump_ep_context_model": True, "trt_timing_cache_path": "/config/model_cache/tensorrt/ort",
"trt_engine_cache_path": "/config/model_cache/tensorrt/ort/trt-engines", "trt_engine_cache_path": "/config/model_cache/tensorrt/ort/trt-engines",
"trt_ep_context_file_path": "/config/model_cache/tensorrt/ort",
} }
) )
elif provider == "OpenVINOExecutionProvider": elif provider == "OpenVINOExecutionProvider":

328
frigate/test/test_config.py
View File

@ -5,12 +5,12 @@ from unittest.mock import patch
import numpy as np import numpy as np
from pydantic import ValidationError from pydantic import ValidationError
from ruamel.yaml.constructor import DuplicateKeyError
from frigate.config import BirdseyeModeEnum, FrigateConfig from frigate.config import BirdseyeModeEnum, FrigateConfig
from frigate.const import MODEL_CACHE_DIR from frigate.const import MODEL_CACHE_DIR
from frigate.detectors import DetectorTypeEnum from frigate.detectors import DetectorTypeEnum
from frigate.plus import PlusApi from frigate.util.builtin import deep_merge
from frigate.util.builtin import deep_merge, load_config_with_no_duplicates
class TestConfig(unittest.TestCase): class TestConfig(unittest.TestCase):
@ -64,12 +64,9 @@ class TestConfig(unittest.TestCase):
def test_config_class(self): def test_config_class(self):
frigate_config = FrigateConfig(**self.minimal) frigate_config = FrigateConfig(**self.minimal)
assert self.minimal == frigate_config.model_dump(exclude_unset=True) assert "cpu" in frigate_config.detectors.keys()
assert frigate_config.detectors["cpu"].type == DetectorTypeEnum.cpu
runtime_config = frigate_config.runtime_config() assert frigate_config.detectors["cpu"].model.width == 320
assert "cpu" in runtime_config.detectors.keys()
assert runtime_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert runtime_config.detectors["cpu"].model.width == 320
@patch("frigate.detectors.detector_config.load_labels") @patch("frigate.detectors.detector_config.load_labels")
def test_detector_custom_model_path(self, mock_labels): def test_detector_custom_model_path(self, mock_labels):
@ -93,24 +90,23 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal))) frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
runtime_config = frigate_config.runtime_config()
assert "cpu" in runtime_config.detectors.keys() assert "cpu" in frigate_config.detectors.keys()
assert "edgetpu" in runtime_config.detectors.keys() assert "edgetpu" in frigate_config.detectors.keys()
assert "openvino" in runtime_config.detectors.keys() assert "openvino" in frigate_config.detectors.keys()
assert runtime_config.detectors["cpu"].type == DetectorTypeEnum.cpu assert frigate_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert runtime_config.detectors["edgetpu"].type == DetectorTypeEnum.edgetpu assert frigate_config.detectors["edgetpu"].type == DetectorTypeEnum.edgetpu
assert runtime_config.detectors["openvino"].type == DetectorTypeEnum.openvino assert frigate_config.detectors["openvino"].type == DetectorTypeEnum.openvino
assert runtime_config.detectors["cpu"].num_threads == 3 assert frigate_config.detectors["cpu"].num_threads == 3
assert runtime_config.detectors["edgetpu"].device is None assert frigate_config.detectors["edgetpu"].device is None
assert runtime_config.detectors["openvino"].device is None assert frigate_config.detectors["openvino"].device is None
assert runtime_config.model.path == "/etc/hosts" assert frigate_config.model.path == "/etc/hosts"
assert runtime_config.detectors["cpu"].model.path == "/cpu_model.tflite" assert frigate_config.detectors["cpu"].model.path == "/cpu_model.tflite"
assert runtime_config.detectors["edgetpu"].model.path == "/edgetpu_model.tflite" assert frigate_config.detectors["edgetpu"].model.path == "/edgetpu_model.tflite"
assert runtime_config.detectors["openvino"].model.path == "/etc/hosts" assert frigate_config.detectors["openvino"].model.path == "/etc/hosts"
def test_invalid_mqtt_config(self): def test_invalid_mqtt_config(self):
config = { config = {
@ -151,11 +147,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "dog" in runtime_config.cameras["back"].objects.track assert "dog" in frigate_config.cameras["back"].objects.track
def test_override_birdseye(self): def test_override_birdseye(self):
config = { config = {
@ -177,12 +171,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert not runtime_config.cameras["back"].birdseye.enabled assert not frigate_config.cameras["back"].birdseye.enabled
assert runtime_config.cameras["back"].birdseye.mode is BirdseyeModeEnum.motion assert frigate_config.cameras["back"].birdseye.mode is BirdseyeModeEnum.motion
def test_override_birdseye_non_inheritable(self): def test_override_birdseye_non_inheritable(self):
config = { config = {
@ -203,11 +195,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].birdseye.enabled assert frigate_config.cameras["back"].birdseye.enabled
def test_inherit_birdseye(self): def test_inherit_birdseye(self):
config = { config = {
@ -228,13 +218,11 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].birdseye.enabled assert frigate_config.cameras["back"].birdseye.enabled
assert ( assert (
runtime_config.cameras["back"].birdseye.mode is BirdseyeModeEnum.continuous frigate_config.cameras["back"].birdseye.mode is BirdseyeModeEnum.continuous
) )
def test_override_tracked_objects(self): def test_override_tracked_objects(self):
@ -257,11 +245,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "cat" in runtime_config.cameras["back"].objects.track assert "cat" in frigate_config.cameras["back"].objects.track
def test_default_object_filters(self): def test_default_object_filters(self):
config = { config = {
@ -282,11 +268,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "dog" in runtime_config.cameras["back"].objects.filters assert "dog" in frigate_config.cameras["back"].objects.filters
def test_inherit_object_filters(self): def test_inherit_object_filters(self):
config = { config = {
@ -310,12 +294,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "dog" in runtime_config.cameras["back"].objects.filters assert "dog" in frigate_config.cameras["back"].objects.filters
assert runtime_config.cameras["back"].objects.filters["dog"].threshold == 0.7 assert frigate_config.cameras["back"].objects.filters["dog"].threshold == 0.7
def test_override_object_filters(self): def test_override_object_filters(self):
config = { config = {
@ -339,12 +321,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "dog" in runtime_config.cameras["back"].objects.filters assert "dog" in frigate_config.cameras["back"].objects.filters
assert runtime_config.cameras["back"].objects.filters["dog"].threshold == 0.7 assert frigate_config.cameras["back"].objects.filters["dog"].threshold == 0.7
def test_global_object_mask(self): def test_global_object_mask(self):
config = { config = {
@ -369,11 +349,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
back_camera = runtime_config.cameras["back"] back_camera = frigate_config.cameras["back"]
assert "dog" in back_camera.objects.filters assert "dog" in back_camera.objects.filters
assert len(back_camera.objects.filters["dog"].raw_mask) == 2 assert len(back_camera.objects.filters["dog"].raw_mask) == 2
assert len(back_camera.objects.filters["person"].raw_mask) == 1 assert len(back_camera.objects.filters["person"].raw_mask) == 1
@ -419,7 +397,8 @@ class TestConfig(unittest.TestCase):
}, },
}, },
} }
frigate_config = FrigateConfig(**config).runtime_config()
frigate_config = FrigateConfig(**config)
assert np.array_equal( assert np.array_equal(
frigate_config.cameras["explicit"].motion.mask, frigate_config.cameras["explicit"].motion.mask,
frigate_config.cameras["relative"].motion.mask, frigate_config.cameras["relative"].motion.mask,
@ -448,10 +427,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert "-rtsp_transport" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
runtime_config = frigate_config.runtime_config()
assert "-rtsp_transport" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
def test_ffmpeg_params_global(self): def test_ffmpeg_params_global(self):
config = { config = {
@ -476,11 +452,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
def test_ffmpeg_params_camera(self): def test_ffmpeg_params_camera(self):
config = { config = {
@ -506,12 +480,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test" not in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] assert "test" not in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
def test_ffmpeg_params_input(self): def test_ffmpeg_params_input(self):
config = { config = {
@ -541,14 +513,12 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] assert "test" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test2" not in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] assert "test2" not in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test3" not in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"] assert "test3" not in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
def test_inherit_clips_retention(self): def test_inherit_clips_retention(self):
config = { config = {
@ -569,11 +539,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].record.alerts.retain.days == 20 assert frigate_config.cameras["back"].record.alerts.retain.days == 20
def test_roles_listed_twice_throws_error(self): def test_roles_listed_twice_throws_error(self):
config = { config = {
@ -657,14 +625,12 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert isinstance( assert isinstance(
runtime_config.cameras["back"].zones["test"].contour, np.ndarray frigate_config.cameras["back"].zones["test"].contour, np.ndarray
) )
assert runtime_config.cameras["back"].zones["test"].color != (0, 0, 0) assert frigate_config.cameras["back"].zones["test"].color != (0, 0, 0)
def test_zone_relative_matches_explicit(self): def test_zone_relative_matches_explicit(self):
config = { config = {
@ -699,7 +665,8 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config).runtime_config()
frigate_config = FrigateConfig(**config)
assert np.array_equal( assert np.array_equal(
frigate_config.cameras["back"].zones["explicit"].contour, frigate_config.cameras["back"].zones["explicit"].contour,
frigate_config.cameras["back"].zones["relative"].contour, frigate_config.cameras["back"].zones["relative"].contour,
@ -729,10 +696,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) ffmpeg_cmds = frigate_config.cameras["back"].ffmpeg_cmds
runtime_config = frigate_config.runtime_config()
ffmpeg_cmds = runtime_config.cameras["back"].ffmpeg_cmds
assert len(ffmpeg_cmds) == 1 assert len(ffmpeg_cmds) == 1
assert "clips" not in ffmpeg_cmds[0]["roles"] assert "clips" not in ffmpeg_cmds[0]["roles"]
@ -760,10 +724,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert frigate_config.cameras["back"].detect.max_disappeared == 5 * 5
runtime_config = frigate_config.runtime_config()
assert runtime_config.cameras["back"].detect.max_disappeared == 5 * 5
def test_motion_frame_height_wont_go_below_120(self): def test_motion_frame_height_wont_go_below_120(self):
config = { config = {
@ -788,10 +749,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert frigate_config.cameras["back"].motion.frame_height == 100
runtime_config = frigate_config.runtime_config()
assert runtime_config.cameras["back"].motion.frame_height == 100
def test_motion_contour_area_dynamic(self): def test_motion_contour_area_dynamic(self):
config = { config = {
@ -816,10 +774,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert round(frigate_config.cameras["back"].motion.contour_area) == 10
runtime_config = frigate_config.runtime_config()
assert round(runtime_config.cameras["back"].motion.contour_area) == 10
def test_merge_labelmap(self): def test_merge_labelmap(self):
config = { config = {
@ -845,10 +800,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert frigate_config.model.merged_labelmap[7] == "truck"
runtime_config = frigate_config.runtime_config()
assert runtime_config.model.merged_labelmap[7] == "truck"
def test_default_labelmap_empty(self): def test_default_labelmap_empty(self):
config = { config = {
@ -873,10 +825,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert frigate_config.model.merged_labelmap[0] == "person"
runtime_config = frigate_config.runtime_config()
assert runtime_config.model.merged_labelmap[0] == "person"
def test_default_labelmap(self): def test_default_labelmap(self):
config = { config = {
@ -902,10 +851,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert frigate_config.model.merged_labelmap[0] == "person"
runtime_config = frigate_config.runtime_config()
assert runtime_config.model.merged_labelmap[0] == "person"
def test_plus_labelmap(self): def test_plus_labelmap(self):
with open("/config/model_cache/test", "w") as f: with open("/config/model_cache/test", "w") as f:
@ -936,10 +882,7 @@ class TestConfig(unittest.TestCase):
} }
frigate_config = FrigateConfig(**config) frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True) assert frigate_config.model.merged_labelmap[0] == "amazon"
runtime_config = frigate_config.runtime_config(PlusApi())
assert runtime_config.model.merged_labelmap[0] == "amazon"
def test_fails_on_invalid_role(self): def test_fails_on_invalid_role(self):
config = { config = {
@ -996,8 +939,7 @@ class TestConfig(unittest.TestCase):
}, },
} }
frigate_config = FrigateConfig(**config) self.assertRaises(ValueError, lambda: FrigateConfig(**config))
self.assertRaises(ValueError, lambda: frigate_config.runtime_config())
def test_works_on_missing_role_multiple_cams(self): def test_works_on_missing_role_multiple_cams(self):
config = { config = {
@ -1044,8 +986,7 @@ class TestConfig(unittest.TestCase):
}, },
} }
frigate_config = FrigateConfig(**config) FrigateConfig(**config)
frigate_config.runtime_config()
def test_global_detect(self): def test_global_detect(self):
config = { config = {
@ -1069,12 +1010,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].detect.max_disappeared == 1 assert frigate_config.cameras["back"].detect.max_disappeared == 1
assert runtime_config.cameras["back"].detect.height == 1080 assert frigate_config.cameras["back"].detect.height == 1080
def test_default_detect(self): def test_default_detect(self):
config = { config = {
@ -1097,12 +1036,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].detect.max_disappeared == 25 assert frigate_config.cameras["back"].detect.max_disappeared == 25
assert runtime_config.cameras["back"].detect.height == 720 assert frigate_config.cameras["back"].detect.height == 720
def test_global_detect_merge(self): def test_global_detect_merge(self):
config = { config = {
@ -1126,13 +1063,11 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].detect.max_disappeared == 1 assert frigate_config.cameras["back"].detect.max_disappeared == 1
assert runtime_config.cameras["back"].detect.height == 1080 assert frigate_config.cameras["back"].detect.height == 1080
assert runtime_config.cameras["back"].detect.width == 1920 assert frigate_config.cameras["back"].detect.width == 1920
def test_global_snapshots(self): def test_global_snapshots(self):
config = { config = {
@ -1159,12 +1094,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].snapshots.enabled assert frigate_config.cameras["back"].snapshots.enabled
assert runtime_config.cameras["back"].snapshots.height == 100 assert frigate_config.cameras["back"].snapshots.height == 100
def test_default_snapshots(self): def test_default_snapshots(self):
config = { config = {
@ -1187,12 +1120,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].snapshots.bounding_box assert frigate_config.cameras["back"].snapshots.bounding_box
assert runtime_config.cameras["back"].snapshots.quality == 70 assert frigate_config.cameras["back"].snapshots.quality == 70
def test_global_snapshots_merge(self): def test_global_snapshots_merge(self):
config = { config = {
@ -1220,13 +1151,11 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].snapshots.bounding_box is False assert frigate_config.cameras["back"].snapshots.bounding_box is False
assert runtime_config.cameras["back"].snapshots.height == 150 assert frigate_config.cameras["back"].snapshots.height == 150
assert runtime_config.cameras["back"].snapshots.enabled assert frigate_config.cameras["back"].snapshots.enabled
def test_global_jsmpeg(self): def test_global_jsmpeg(self):
config = { config = {
@ -1250,11 +1179,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].live.quality == 4 assert frigate_config.cameras["back"].live.quality == 4
def test_default_live(self): def test_default_live(self):
config = { config = {
@ -1277,11 +1204,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].live.quality == 8 assert frigate_config.cameras["back"].live.quality == 8
def test_global_live_merge(self): def test_global_live_merge(self):
config = { config = {
@ -1308,12 +1233,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].live.quality == 7 assert frigate_config.cameras["back"].live.quality == 7
assert runtime_config.cameras["back"].live.height == 480 assert frigate_config.cameras["back"].live.height == 480
def test_global_timestamp_style(self): def test_global_timestamp_style(self):
config = { config = {
@ -1337,11 +1260,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].timestamp_style.position == "bl" assert frigate_config.cameras["back"].timestamp_style.position == "bl"
def test_default_timestamp_style(self): def test_default_timestamp_style(self):
config = { config = {
@ -1364,11 +1285,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].timestamp_style.position == "tl" assert frigate_config.cameras["back"].timestamp_style.position == "tl"
def test_global_timestamp_style_merge(self): def test_global_timestamp_style_merge(self):
config = { config = {
@ -1393,12 +1312,10 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].timestamp_style.position == "bl" assert frigate_config.cameras["back"].timestamp_style.position == "bl"
assert runtime_config.cameras["back"].timestamp_style.thickness == 4 assert frigate_config.cameras["back"].timestamp_style.thickness == 4
def test_allow_retain_to_be_a_decimal(self): def test_allow_retain_to_be_a_decimal(self):
config = { config = {
@ -1422,11 +1339,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].snapshots.retain.default == 1.5 assert frigate_config.cameras["back"].snapshots.retain.default == 1.5
def test_fails_on_bad_camera_name(self): def test_fails_on_bad_camera_name(self):
config = { config = {
@ -1451,11 +1366,7 @@ class TestConfig(unittest.TestCase):
}, },
} }
frigate_config = FrigateConfig(**config) self.assertRaises(ValidationError, lambda: FrigateConfig(**config).cameras)
self.assertRaises(
ValidationError, lambda: frigate_config.runtime_config().cameras
)
def test_fails_on_bad_segment_time(self): def test_fails_on_bad_segment_time(self):
config = { config = {
@ -1483,11 +1394,9 @@ class TestConfig(unittest.TestCase):
}, },
} }
frigate_config = FrigateConfig(**config)
self.assertRaises( self.assertRaises(
ValueError, ValueError,
lambda: frigate_config.runtime_config().ffmpeg.output_args.record, lambda: FrigateConfig(**config).ffmpeg.output_args.record,
) )
def test_fails_zone_defines_untracked_object(self): def test_fails_zone_defines_untracked_object(self):
@ -1519,9 +1428,7 @@ class TestConfig(unittest.TestCase):
}, },
} }
frigate_config = FrigateConfig(**config) self.assertRaises(ValueError, lambda: FrigateConfig(**config).cameras)
self.assertRaises(ValueError, lambda: frigate_config.runtime_config().cameras)
def test_fails_duplicate_keys(self): def test_fails_duplicate_keys(self):
raw_config = """ raw_config = """
@ -1537,7 +1444,7 @@ class TestConfig(unittest.TestCase):
""" """
self.assertRaises( self.assertRaises(
ValueError, lambda: load_config_with_no_duplicates(raw_config) DuplicateKeyError, lambda: FrigateConfig.parse_yaml(raw_config)
) )
def test_object_filter_ratios_work(self): def test_object_filter_ratios_work(self):
@ -1562,13 +1469,11 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert "dog" in runtime_config.cameras["back"].objects.filters assert "dog" in frigate_config.cameras["back"].objects.filters
assert runtime_config.cameras["back"].objects.filters["dog"].min_ratio == 0.2 assert frigate_config.cameras["back"].objects.filters["dog"].min_ratio == 0.2
assert runtime_config.cameras["back"].objects.filters["dog"].max_ratio == 10.1 assert frigate_config.cameras["back"].objects.filters["dog"].max_ratio == 10.1
def test_valid_movement_weights(self): def test_valid_movement_weights(self):
config = { config = {
@ -1591,10 +1496,9 @@ class TestConfig(unittest.TestCase):
} }
}, },
} }
frigate_config = FrigateConfig(**config)
runtime_config = frigate_config.runtime_config() frigate_config = FrigateConfig(**config)
assert runtime_config.cameras["back"].onvif.autotracking.movement_weights == [ assert frigate_config.cameras["back"].onvif.autotracking.movement_weights == [
"0.0", "0.0",
"1.0", "1.0",
"1.23", "1.23",

16
frigate/test/test_ffmpeg_presets.py
View File

@ -36,16 +36,13 @@ class TestFfmpegPresets(unittest.TestCase):
} }
def test_default_ffmpeg(self): def test_default_ffmpeg(self):
frigate_config = FrigateConfig(**self.default_ffmpeg) FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert self.default_ffmpeg == frigate_config.dict(exclude_unset=True)
def test_ffmpeg_hwaccel_preset(self): def test_ffmpeg_hwaccel_preset(self):
self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["hwaccel_args"] = ( self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["hwaccel_args"] = (
"preset-rpi-64-h264" "preset-rpi-64-h264"
) )
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert "preset-rpi-64-h264" not in ( assert "preset-rpi-64-h264" not in (
" ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]) " ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"])
) )
@ -58,7 +55,6 @@ class TestFfmpegPresets(unittest.TestCase):
"-other-hwaccel args" "-other-hwaccel args"
) )
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert "-other-hwaccel args" in ( assert "-other-hwaccel args" in (
" ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]) " ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"])
) )
@ -73,7 +69,6 @@ class TestFfmpegPresets(unittest.TestCase):
"fps": 10, "fps": 10,
} }
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert "preset-nvidia-h264" not in ( assert "preset-nvidia-h264" not in (
" ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]) " ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"])
) )
@ -89,8 +84,6 @@ class TestFfmpegPresets(unittest.TestCase):
"preset-rtsp-generic" "preset-rtsp-generic"
) )
frigate_preset_config = FrigateConfig(**self.default_ffmpeg) frigate_preset_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
frigate_preset_config.cameras["back"].create_ffmpeg_cmds()
assert ( assert (
# Ignore global and user_agent args in comparison # Ignore global and user_agent args in comparison
frigate_preset_config.cameras["back"].ffmpeg_cmds[0]["cmd"] frigate_preset_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
@ -102,7 +95,6 @@ class TestFfmpegPresets(unittest.TestCase):
"preset-rtmp-generic" "preset-rtmp-generic"
) )
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert "preset-rtmp-generic" not in ( assert "preset-rtmp-generic" not in (
" ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]) " ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"])
) )
@ -117,7 +109,6 @@ class TestFfmpegPresets(unittest.TestCase):
argsList = defaultArgsList + ["-some", "arg with space"] argsList = defaultArgsList + ["-some", "arg with space"]
self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["input_args"] = argsString self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["input_args"] = argsString
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert set(argsList).issubset( assert set(argsList).issubset(
frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"] frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
) )
@ -125,7 +116,6 @@ class TestFfmpegPresets(unittest.TestCase):
def test_ffmpeg_input_not_preset(self): def test_ffmpeg_input_not_preset(self):
self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["input_args"] = "-some inputs" self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["input_args"] = "-some inputs"
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert "-some inputs" in ( assert "-some inputs" in (
" ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]) " ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"])
) )
@ -135,7 +125,6 @@ class TestFfmpegPresets(unittest.TestCase):
"preset-record-generic-audio-aac" "preset-record-generic-audio-aac"
) )
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert "preset-record-generic-audio-aac" not in ( assert "preset-record-generic-audio-aac" not in (
" ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]) " ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"])
) )
@ -145,10 +134,9 @@ class TestFfmpegPresets(unittest.TestCase):
def test_ffmpeg_output_record_not_preset(self): def test_ffmpeg_output_record_not_preset(self):
self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["output_args"]["record"] = ( self.default_ffmpeg["cameras"]["back"]["ffmpeg"]["output_args"]["record"] = (
"-some output" "-some output -segment_time 10"
) )
frigate_config = FrigateConfig(**self.default_ffmpeg) frigate_config = FrigateConfig(**self.default_ffmpeg)
frigate_config.cameras["back"].create_ffmpeg_cmds()
assert "-some output" in ( assert "-some output" in (
" ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]) " ".join(frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"])
) )

4
frigate/test/test_http.py
View File

@ -344,7 +344,7 @@ class TestHttp(unittest.TestCase):
def test_config(self): def test_config(self):
app = create_fastapi_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config).runtime_config(), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
None, None,
@ -386,7 +386,7 @@ class TestHttp(unittest.TestCase):
stats = Mock(spec=StatsEmitter) stats = Mock(spec=StatsEmitter)
stats.get_latest_stats.return_value = self.test_stats stats.get_latest_stats.return_value = self.test_stats
app = create_fastapi_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config).runtime_config(), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
None, None,

30
frigate/util/builtin.py
View File

@ -9,14 +9,12 @@ import queue
import re import re
import shlex import shlex
import urllib.parse import urllib.parse
from collections import Counter
from collections.abc import Mapping from collections.abc import Mapping
from pathlib import Path from pathlib import Path
from typing import Any, Optional, Tuple from typing import Any, Optional, Tuple
import numpy as np import numpy as np
import pytz import pytz
import yaml
from ruamel.yaml import YAML from ruamel.yaml import YAML
from tzlocal import get_localzone from tzlocal import get_localzone
from zoneinfo import ZoneInfoNotFoundError from zoneinfo import ZoneInfoNotFoundError
@ -89,34 +87,6 @@ def deep_merge(dct1: dict, dct2: dict, override=False, merge_lists=False) -> dic
return merged return merged
def load_config_with_no_duplicates(raw_config) -> dict:
"""Get config ensuring duplicate keys are not allowed."""
# https://stackoverflow.com/a/71751051
# important to use SafeLoader here to avoid RCE
class PreserveDuplicatesLoader(yaml.loader.SafeLoader):
pass
def map_constructor(loader, node, deep=False):
keys = [loader.construct_object(node, deep=deep) for node, _ in node.value]
vals = [loader.construct_object(node, deep=deep) for _, node in node.value]
key_count = Counter(keys)
data = {}
for key, val in zip(keys, vals):
if key_count[key] > 1:
raise ValueError(
f"Config input {key} is defined multiple times for the same field, this is not allowed."
)
else:
data[key] = val
return data
PreserveDuplicatesLoader.add_constructor(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, map_constructor
)
return yaml.load(raw_config, PreserveDuplicatesLoader)
def clean_camera_user_pass(line: str) -> str: def clean_camera_user_pass(line: str) -> str:
"""Removes user and password from line.""" """Removes user and password from line."""
rtsp_cleaned = re.sub(REGEX_RTSP_CAMERA_USER_PASS, "://*:*@", line) rtsp_cleaned = re.sub(REGEX_RTSP_CAMERA_USER_PASS, "://*:*@", line)

5
process_clip.py
View File

@ -280,10 +280,7 @@ def process(path, label, output, debug_path):
json_config["cameras"]["camera"]["ffmpeg"]["inputs"][0]["path"] = c json_config["cameras"]["camera"]["ffmpeg"]["inputs"][0]["path"] = c
frigate_config = FrigateConfig(**json_config) frigate_config = FrigateConfig(**json_config)
runtime_config = frigate_config.runtime_config() process_clip = ProcessClip(c, frame_shape, frigate_config)
runtime_config.cameras["camera"].create_ffmpeg_cmds()
process_clip = ProcessClip(c, frame_shape, runtime_config)
process_clip.load_frames() process_clip.load_frames()
process_clip.process_frames(object_detector, objects_to_track=[label]) process_clip.process_frames(object_detector, objects_to_track=[label])