Removed usage of PyYAML for config parsing. (#13883)

* Ignore entire __pycache__ folder instead of individual *.pyc files

* Ignore .mypy_cache in git

* Rework config YAML parsing to use only ruamel.yaml

PyYAML silently overrides keys when encountering duplicates, but ruamel
raises and exception by default. Since we're already using it elsewhere,
dropping PyYAML is an easy choice to make.

* Added EnvString in config to slim down runtime_config()

* Added gitlens to devcontainer

* Automatically call FrigateConfig.runtime_config()

runtime_config needed to be called manually before. Now, it's been
removed, but the same code is run by a pydantic validator.

* Fix handling of missing -segment_time

* Removed type annotation on FrigateConfig's parse

I'd like to keep them, but then mypy complains about some fundamental
errors with how the pydantic model is structured. I'd like to fix it,
but I'd rather work towards moving some of this config to the database.
This commit is contained in:
gtsiam 2024-09-22 18:56:57 +03:00 committed by GitHub
parent 6f2924006c
commit e8763b3697
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 233 additions and 361 deletions

View File

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

3
.gitignore vendored
View File

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

View File

@ -248,7 +248,7 @@ def config_save():
# Validate the config schema
try:
FrigateConfig.parse_raw(new_config)
FrigateConfig.parse_yaml(new_config)
except Exception:
return make_response(
jsonify(
@ -336,7 +336,7 @@ def config_set():
f.close()
# Validate the config schema
try:
config_obj = FrigateConfig.parse_raw(new_raw_config)
config_obj = FrigateConfig.parse_yaml(new_raw_config)
except Exception:
with open(config_file, "w") as f:
f.write(old_raw_config)
@ -361,8 +361,8 @@ def config_set():
json = request.get_json(silent=True) or {}
if json.get("requires_restart", 1) == 0:
current_app.frigate_config = FrigateConfig.runtime_config(
config_obj, current_app.plus_api
current_app.frigate_config = FrigateConfig.parse_object(
config_obj, plus_api=current_app.plus_api
)
return make_response(

View File

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

View File

@ -6,10 +6,11 @@ import os
import shutil
from enum import Enum
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
from pydantic import (
AfterValidator,
BaseModel,
ConfigDict,
Field,
@ -17,8 +18,11 @@ from pydantic import (
ValidationInfo,
field_serializer,
field_validator,
model_validator,
)
from pydantic.fields import PrivateAttr
from ruamel.yaml import YAML
from typing_extensions import Self
from frigate.const import (
ALL_ATTRIBUTE_LABELS,
@ -31,7 +35,7 @@ from frigate.const import (
INCLUDED_FFMPEG_VERSIONS,
MAX_PRE_CAPTURE,
REGEX_CAMERA_NAME,
YAML_EXT,
REGEX_JSON,
)
from frigate.detectors import DetectorConfig, ModelConfig
from frigate.detectors.detector_config import BaseDetectorConfig
@ -41,13 +45,11 @@ from frigate.ffmpeg_presets import (
parse_preset_input,
parse_preset_output_record,
)
from frigate.plus import PlusApi
from frigate.util.builtin import (
deep_merge,
escape_special_characters,
generate_color_palette,
get_ffmpeg_arg_list,
load_config_with_no_duplicates,
)
from frigate.util.config import StreamInfoRetriever, get_relative_coordinates
from frigate.util.image import create_mask
@ -55,6 +57,8 @@ from frigate.util.services import auto_detect_hwaccel
logger = logging.getLogger(__name__)
yaml = YAML()
# TODO: Identify what the default format to display timestamps is
DEFAULT_TIME_FORMAT = "%m/%d/%Y %H:%M:%S"
# German Style:
@ -103,6 +107,13 @@ class DateTimeStyleEnum(str, Enum):
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):
timezone: Optional[str] = Field(default=None, title="Override UI timezone.")
time_format: TimeFormatEnum = Field(
@ -137,7 +148,7 @@ class ProxyConfig(FrigateBaseModel):
logout_url: Optional[str] = Field(
default=None, title="Redirect url for logging out with proxy."
)
auth_secret: Optional[str] = Field(
auth_secret: Optional[EnvString] = Field(
default=None,
title="Secret value for proxy authentication.",
)
@ -208,8 +219,10 @@ class MqttConfig(FrigateBaseModel):
stats_interval: int = Field(
default=60, ge=FREQUENCY_STATS_POINTS, title="MQTT Camera Stats Interval"
)
user: Optional[str] = Field(None, title="MQTT Username")
password: Optional[str] = Field(None, title="MQTT Password", validate_default=True)
user: Optional[EnvString] = Field(None, title="MQTT Username")
password: Optional[EnvString] = Field(
None, title="MQTT Password", validate_default=True
)
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_key: Optional[str] = Field(None, title="MQTT TLS Client Key")
@ -284,8 +297,8 @@ class PtzAutotrackConfig(FrigateBaseModel):
class OnvifConfig(FrigateBaseModel):
host: str = Field(default="", title="Onvif Host")
port: int = Field(default=8000, title="Onvif Port")
user: Optional[str] = Field(None, title="Onvif Username")
password: Optional[str] = Field(None, title="Onvif Password")
user: Optional[EnvString] = Field(None, title="Onvif Username")
password: Optional[EnvString] = Field(None, title="Onvif Password")
autotracking: PtzAutotrackConfig = Field(
default_factory=PtzAutotrackConfig,
title="PTZ auto tracking config.",
@ -756,7 +769,7 @@ class GenAIConfig(FrigateBaseModel):
default=GenAIProviderEnum.openai, title="GenAI provider."
)
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.")
prompt: str = Field(
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):
path: str = Field(title="Camera input path.")
path: EnvString = Field(title="Camera input path.")
roles: List[CameraRoleEnum] = Field(title="Roles assigned to this input.")
global_args: Union[str, List[str]] = Field(
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"):
return
try:
seg_arg_index = record_args.index("-segment_time")
if seg_arg_index < 0:
raise ValueError(
f"Camera {camera_config.name} has no segment_time in recording output args, segment args are required for record."
)
except ValueError:
raise ValueError(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:
raise ValueError(
f"Camera {camera_config.name} has invalid segment_time output arg, segment_time must be 60 or less."
)
raise ValueError(f"Camera {camera_config.name} has invalid segment_time output arg, \
segment_time must be 60 or less.")
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.")
def runtime_config(self, plus_api: PlusApi = None) -> FrigateConfig:
"""Merge camera config with globals."""
config = self.model_copy(deep=True)
# Proxy secret substitution
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)
@model_validator(mode="after")
def post_validation(self, info: ValidationInfo) -> Self:
plus_api = None
if isinstance(info.context, dict):
plus_api = info.context.get("plus_api")
# set notifications state
config.notifications.enabled_in_config = config.notifications.enabled
# GenAI substitution
if config.genai.api_key:
config.genai.api_key = config.genai.api_key.format(**FRIGATE_ENV_VARS)
self.notifications.enabled_in_config = self.notifications.enabled
# set default min_score for object attributes
for attribute in ALL_ATTRIBUTE_LABELS:
if not config.objects.filters.get(attribute):
config.objects.filters[attribute] = FilterConfig(min_score=0.7)
elif config.objects.filters[attribute].min_score == 0.5:
config.objects.filters[attribute].min_score = 0.7
if not self.objects.filters.get(attribute):
self.objects.filters[attribute] = FilterConfig(min_score=0.7)
elif self.objects.filters[attribute].min_score == 0.5:
self.objects.filters[attribute].min_score = 0.7
# auto detect hwaccel args
if config.ffmpeg.hwaccel_args == "auto":
config.ffmpeg.hwaccel_args = auto_detect_hwaccel()
if self.ffmpeg.hwaccel_args == "auto":
self.ffmpeg.hwaccel_args = auto_detect_hwaccel()
# Global config to propagate down to camera level
global_config = config.model_dump(
global_config = self.model_dump(
include={
"audio": ...,
"birdseye": ...,
@ -1533,7 +1531,7 @@ class FrigateConfig(FrigateBaseModel):
exclude_unset=True,
)
for name, camera in config.cameras.items():
for name, camera in self.cameras.items():
merged_config = deep_merge(
camera.model_dump(exclude_unset=True), global_config
)
@ -1542,7 +1540,7 @@ class FrigateConfig(FrigateBaseModel):
)
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:
need_record_fourcc = False and "record" in input.roles
@ -1555,7 +1553,7 @@ class FrigateConfig(FrigateBaseModel):
stream_info = {"width": 0, "height": 0, "fourcc": None}
try:
stream_info = stream_info_retriever.get_stream_info(
config.ffmpeg, input.path
self.ffmpeg, input.path
)
except Exception:
logger.warn(
@ -1607,18 +1605,6 @@ class FrigateConfig(FrigateBaseModel):
if camera_config.detect.stationary.interval is None:
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
camera_config.audio.enabled_in_config = camera_config.audio.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:
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_valid_live_stream_name(config, camera_config)
verify_valid_live_stream_name(self, camera_config)
verify_recording_retention(camera_config)
verify_recording_segments_setup_with_reasonable_time(camera_config)
verify_zone_objects_are_tracked(camera_config)
@ -1694,20 +1684,16 @@ class FrigateConfig(FrigateBaseModel):
verify_autotrack_zones(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
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)
config.model.create_colormap(sorted(enabled_labels))
config.model.check_and_load_plus_model(plus_api)
self.model.create_colormap(sorted(enabled_labels))
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)
model_dict = (
detector
@ -1716,10 +1702,10 @@ class FrigateConfig(FrigateBaseModel):
)
detector_config: DetectorConfig = adapter.validate_python(model_dict)
if detector_config.model is None:
detector_config.model = config.model.model_copy()
detector_config.model = self.model.model_copy()
else:
path = detector_config.model.path
detector_config.model = config.model.model_copy()
detector_config.model = self.model.model_copy()
detector_config.model.path = path
if "path" not in model_dict or len(model_dict.keys()) > 1:
@ -1729,7 +1715,7 @@ class FrigateConfig(FrigateBaseModel):
merged_model = deep_merge(
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:
@ -1743,9 +1729,9 @@ class FrigateConfig(FrigateBaseModel):
plus_api, detector_config.type
)
detector_config.model.compute_model_hash()
config.detectors[key] = detector_config
self.detectors[key] = detector_config
return config
return self
@field_validator("cameras")
@classmethod
@ -1757,18 +1743,42 @@ class FrigateConfig(FrigateBaseModel):
return v
@classmethod
def parse_file(cls, config_file):
with open(config_file) as f:
raw_config = f.read()
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)
def parse_file(cls, config_path, **kwargs):
with open(config_path) as f:
return FrigateConfig.parse(f, **kwargs)
@classmethod
def parse_raw(cls, raw_config):
config = load_config_with_no_duplicates(raw_config)
return cls.model_validate(config)
def parse(cls, config, *, is_json=None, **context):
# If config is a file, read its contents.
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)

View File

@ -1,3 +1,5 @@
import re
CONFIG_DIR = "/config"
DEFAULT_DB_PATH = f"{CONFIG_DIR}/frigate.db"
MODEL_CACHE_DIR = f"{CONFIG_DIR}/model_cache"
@ -7,7 +9,6 @@ RECORD_DIR = f"{BASE_DIR}/recordings"
EXPORT_DIR = f"{BASE_DIR}/exports"
BIRDSEYE_PIPE = "/tmp/cache/birdseye"
CACHE_DIR = "/tmp/cache"
YAML_EXT = (".yaml", ".yml")
FRIGATE_LOCALHOST = "http://127.0.0.1:5000"
PLUS_ENV_VAR = "PLUS_API_KEY"
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_RTSP_CAMERA_USER_PASS = r":\/\/[a-zA-Z0-9_-]+:[\S]+@"
REGEX_HTTP_CAMERA_USER_PASS = r"user=[a-zA-Z0-9_-]+&password=[\S]+"
REGEX_JSON = re.compile(r"^\s*\{")
# Known Driver Names

View File

@ -5,12 +5,12 @@ from unittest.mock import patch
import numpy as np
from pydantic import ValidationError
from ruamel.yaml.constructor import DuplicateKeyError
from frigate.config import BirdseyeModeEnum, FrigateConfig
from frigate.const import MODEL_CACHE_DIR
from frigate.detectors import DetectorTypeEnum
from frigate.plus import PlusApi
from frigate.util.builtin import deep_merge, load_config_with_no_duplicates
from frigate.util.builtin import deep_merge
class TestConfig(unittest.TestCase):
@ -64,12 +64,9 @@ class TestConfig(unittest.TestCase):
def test_config_class(self):
frigate_config = FrigateConfig(**self.minimal)
assert self.minimal == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert "cpu" in runtime_config.detectors.keys()
assert runtime_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert runtime_config.detectors["cpu"].model.width == 320
assert "cpu" in frigate_config.detectors.keys()
assert frigate_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert frigate_config.detectors["cpu"].model.width == 320
@patch("frigate.detectors.detector_config.load_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)))
runtime_config = frigate_config.runtime_config()
assert "cpu" in runtime_config.detectors.keys()
assert "edgetpu" in runtime_config.detectors.keys()
assert "openvino" in runtime_config.detectors.keys()
assert "cpu" in frigate_config.detectors.keys()
assert "edgetpu" in frigate_config.detectors.keys()
assert "openvino" in frigate_config.detectors.keys()
assert runtime_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert runtime_config.detectors["edgetpu"].type == DetectorTypeEnum.edgetpu
assert runtime_config.detectors["openvino"].type == DetectorTypeEnum.openvino
assert frigate_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert frigate_config.detectors["edgetpu"].type == DetectorTypeEnum.edgetpu
assert frigate_config.detectors["openvino"].type == DetectorTypeEnum.openvino
assert runtime_config.detectors["cpu"].num_threads == 3
assert runtime_config.detectors["edgetpu"].device is None
assert runtime_config.detectors["openvino"].device is None
assert frigate_config.detectors["cpu"].num_threads == 3
assert frigate_config.detectors["edgetpu"].device is None
assert frigate_config.detectors["openvino"].device is None
assert runtime_config.model.path == "/etc/hosts"
assert runtime_config.detectors["cpu"].model.path == "/cpu_model.tflite"
assert runtime_config.detectors["edgetpu"].model.path == "/edgetpu_model.tflite"
assert runtime_config.detectors["openvino"].model.path == "/etc/hosts"
assert frigate_config.model.path == "/etc/hosts"
assert frigate_config.detectors["cpu"].model.path == "/cpu_model.tflite"
assert frigate_config.detectors["edgetpu"].model.path == "/edgetpu_model.tflite"
assert frigate_config.detectors["openvino"].model.path == "/etc/hosts"
def test_invalid_mqtt_config(self):
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()
assert "dog" in runtime_config.cameras["back"].objects.track
frigate_config = FrigateConfig(**config)
assert "dog" in frigate_config.cameras["back"].objects.track
def test_override_birdseye(self):
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()
assert not runtime_config.cameras["back"].birdseye.enabled
assert runtime_config.cameras["back"].birdseye.mode is BirdseyeModeEnum.motion
frigate_config = FrigateConfig(**config)
assert not frigate_config.cameras["back"].birdseye.enabled
assert frigate_config.cameras["back"].birdseye.mode is BirdseyeModeEnum.motion
def test_override_birdseye_non_inheritable(self):
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()
assert runtime_config.cameras["back"].birdseye.enabled
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].birdseye.enabled
def test_inherit_birdseye(self):
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()
assert runtime_config.cameras["back"].birdseye.enabled
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].birdseye.enabled
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):
@ -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()
assert "cat" in runtime_config.cameras["back"].objects.track
frigate_config = FrigateConfig(**config)
assert "cat" in frigate_config.cameras["back"].objects.track
def test_default_object_filters(self):
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()
assert "dog" in runtime_config.cameras["back"].objects.filters
frigate_config = FrigateConfig(**config)
assert "dog" in frigate_config.cameras["back"].objects.filters
def test_inherit_object_filters(self):
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()
assert "dog" in runtime_config.cameras["back"].objects.filters
assert runtime_config.cameras["back"].objects.filters["dog"].threshold == 0.7
frigate_config = FrigateConfig(**config)
assert "dog" in frigate_config.cameras["back"].objects.filters
assert frigate_config.cameras["back"].objects.filters["dog"].threshold == 0.7
def test_override_object_filters(self):
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()
assert "dog" in runtime_config.cameras["back"].objects.filters
assert runtime_config.cameras["back"].objects.filters["dog"].threshold == 0.7
frigate_config = FrigateConfig(**config)
assert "dog" in frigate_config.cameras["back"].objects.filters
assert frigate_config.cameras["back"].objects.filters["dog"].threshold == 0.7
def test_global_object_mask(self):
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()
back_camera = runtime_config.cameras["back"]
frigate_config = FrigateConfig(**config)
back_camera = frigate_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
@ -419,7 +397,8 @@ class TestConfig(unittest.TestCase):
},
},
}
frigate_config = FrigateConfig(**config).runtime_config()
frigate_config = FrigateConfig(**config)
assert np.array_equal(
frigate_config.cameras["explicit"].motion.mask,
frigate_config.cameras["relative"].motion.mask,
@ -448,10 +427,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert "-rtsp_transport" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "-rtsp_transport" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
def test_ffmpeg_params_global(self):
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()
assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
frigate_config = FrigateConfig(**config)
assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
def test_ffmpeg_params_camera(self):
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()
assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test" not in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
frigate_config = FrigateConfig(**config)
assert "-re" in frigate_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):
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()
assert "-re" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test" in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test2" not in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test3" not in runtime_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
frigate_config = FrigateConfig(**config)
assert "-re" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test" in frigate_config.cameras["back"].ffmpeg_cmds[0]["cmd"]
assert "test2" not in frigate_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):
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()
assert runtime_config.cameras["back"].record.alerts.retain.days == 20
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].record.alerts.retain.days == 20
def test_roles_listed_twice_throws_error(self):
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(
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):
config = {
@ -699,7 +665,8 @@ class TestConfig(unittest.TestCase):
}
},
}
frigate_config = FrigateConfig(**config).runtime_config()
frigate_config = FrigateConfig(**config)
assert np.array_equal(
frigate_config.cameras["back"].zones["explicit"].contour,
frigate_config.cameras["back"].zones["relative"].contour,
@ -729,10 +696,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
ffmpeg_cmds = runtime_config.cameras["back"].ffmpeg_cmds
ffmpeg_cmds = frigate_config.cameras["back"].ffmpeg_cmds
assert len(ffmpeg_cmds) == 1
assert "clips" not in ffmpeg_cmds[0]["roles"]
@ -760,10 +724,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert runtime_config.cameras["back"].detect.max_disappeared == 5 * 5
assert frigate_config.cameras["back"].detect.max_disappeared == 5 * 5
def test_motion_frame_height_wont_go_below_120(self):
config = {
@ -788,10 +749,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert runtime_config.cameras["back"].motion.frame_height == 100
assert frigate_config.cameras["back"].motion.frame_height == 100
def test_motion_contour_area_dynamic(self):
config = {
@ -816,10 +774,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert round(runtime_config.cameras["back"].motion.contour_area) == 10
assert round(frigate_config.cameras["back"].motion.contour_area) == 10
def test_merge_labelmap(self):
config = {
@ -845,10 +800,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert runtime_config.model.merged_labelmap[7] == "truck"
assert frigate_config.model.merged_labelmap[7] == "truck"
def test_default_labelmap_empty(self):
config = {
@ -873,10 +825,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert runtime_config.model.merged_labelmap[0] == "person"
assert frigate_config.model.merged_labelmap[0] == "person"
def test_default_labelmap(self):
config = {
@ -902,10 +851,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config()
assert runtime_config.model.merged_labelmap[0] == "person"
assert frigate_config.model.merged_labelmap[0] == "person"
def test_plus_labelmap(self):
with open("/config/model_cache/test", "w") as f:
@ -936,10 +882,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert config == frigate_config.model_dump(exclude_unset=True)
runtime_config = frigate_config.runtime_config(PlusApi())
assert runtime_config.model.merged_labelmap[0] == "amazon"
assert frigate_config.model.merged_labelmap[0] == "amazon"
def test_fails_on_invalid_role(self):
config = {
@ -996,8 +939,7 @@ class TestConfig(unittest.TestCase):
},
}
frigate_config = FrigateConfig(**config)
self.assertRaises(ValueError, lambda: frigate_config.runtime_config())
self.assertRaises(ValueError, lambda: FrigateConfig(**config))
def test_works_on_missing_role_multiple_cams(self):
config = {
@ -1044,8 +986,7 @@ class TestConfig(unittest.TestCase):
},
}
frigate_config = FrigateConfig(**config)
frigate_config.runtime_config()
FrigateConfig(**config)
def test_global_detect(self):
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()
assert runtime_config.cameras["back"].detect.max_disappeared == 1
assert runtime_config.cameras["back"].detect.height == 1080
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].detect.max_disappeared == 1
assert frigate_config.cameras["back"].detect.height == 1080
def test_default_detect(self):
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()
assert runtime_config.cameras["back"].detect.max_disappeared == 25
assert runtime_config.cameras["back"].detect.height == 720
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].detect.max_disappeared == 25
assert frigate_config.cameras["back"].detect.height == 720
def test_global_detect_merge(self):
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()
assert runtime_config.cameras["back"].detect.max_disappeared == 1
assert runtime_config.cameras["back"].detect.height == 1080
assert runtime_config.cameras["back"].detect.width == 1920
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].detect.max_disappeared == 1
assert frigate_config.cameras["back"].detect.height == 1080
assert frigate_config.cameras["back"].detect.width == 1920
def test_global_snapshots(self):
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()
assert runtime_config.cameras["back"].snapshots.enabled
assert runtime_config.cameras["back"].snapshots.height == 100
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].snapshots.enabled
assert frigate_config.cameras["back"].snapshots.height == 100
def test_default_snapshots(self):
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()
assert runtime_config.cameras["back"].snapshots.bounding_box
assert runtime_config.cameras["back"].snapshots.quality == 70
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].snapshots.bounding_box
assert frigate_config.cameras["back"].snapshots.quality == 70
def test_global_snapshots_merge(self):
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()
assert runtime_config.cameras["back"].snapshots.bounding_box is False
assert runtime_config.cameras["back"].snapshots.height == 150
assert runtime_config.cameras["back"].snapshots.enabled
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].snapshots.bounding_box is False
assert frigate_config.cameras["back"].snapshots.height == 150
assert frigate_config.cameras["back"].snapshots.enabled
def test_global_jsmpeg(self):
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()
assert runtime_config.cameras["back"].live.quality == 4
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].live.quality == 4
def test_default_live(self):
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()
assert runtime_config.cameras["back"].live.quality == 8
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].live.quality == 8
def test_global_live_merge(self):
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()
assert runtime_config.cameras["back"].live.quality == 7
assert runtime_config.cameras["back"].live.height == 480
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].live.quality == 7
assert frigate_config.cameras["back"].live.height == 480
def test_global_timestamp_style(self):
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()
assert runtime_config.cameras["back"].timestamp_style.position == "bl"
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].timestamp_style.position == "bl"
def test_default_timestamp_style(self):
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()
assert runtime_config.cameras["back"].timestamp_style.position == "tl"
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].timestamp_style.position == "tl"
def test_global_timestamp_style_merge(self):
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()
assert runtime_config.cameras["back"].timestamp_style.position == "bl"
assert runtime_config.cameras["back"].timestamp_style.thickness == 4
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].timestamp_style.position == "bl"
assert frigate_config.cameras["back"].timestamp_style.thickness == 4
def test_allow_retain_to_be_a_decimal(self):
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()
assert runtime_config.cameras["back"].snapshots.retain.default == 1.5
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].snapshots.retain.default == 1.5
def test_fails_on_bad_camera_name(self):
config = {
@ -1451,11 +1366,7 @@ class TestConfig(unittest.TestCase):
},
}
frigate_config = FrigateConfig(**config)
self.assertRaises(
ValidationError, lambda: frigate_config.runtime_config().cameras
)
self.assertRaises(ValidationError, lambda: FrigateConfig(**config).cameras)
def test_fails_on_bad_segment_time(self):
config = {
@ -1483,11 +1394,9 @@ class TestConfig(unittest.TestCase):
},
}
frigate_config = FrigateConfig(**config)
self.assertRaises(
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):
@ -1519,9 +1428,7 @@ class TestConfig(unittest.TestCase):
},
}
frigate_config = FrigateConfig(**config)
self.assertRaises(ValueError, lambda: frigate_config.runtime_config().cameras)
self.assertRaises(ValueError, lambda: FrigateConfig(**config).cameras)
def test_fails_duplicate_keys(self):
raw_config = """
@ -1537,7 +1444,7 @@ class TestConfig(unittest.TestCase):
"""
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):
@ -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()
assert "dog" in runtime_config.cameras["back"].objects.filters
assert runtime_config.cameras["back"].objects.filters["dog"].min_ratio == 0.2
assert runtime_config.cameras["back"].objects.filters["dog"].max_ratio == 10.1
frigate_config = FrigateConfig(**config)
assert "dog" in frigate_config.cameras["back"].objects.filters
assert frigate_config.cameras["back"].objects.filters["dog"].min_ratio == 0.2
assert frigate_config.cameras["back"].objects.filters["dog"].max_ratio == 10.1
def test_valid_movement_weights(self):
config = {
@ -1591,10 +1496,9 @@ class TestConfig(unittest.TestCase):
}
},
}
frigate_config = FrigateConfig(**config)
runtime_config = frigate_config.runtime_config()
assert runtime_config.cameras["back"].onvif.autotracking.movement_weights == [
frigate_config = FrigateConfig(**config)
assert frigate_config.cameras["back"].onvif.autotracking.movement_weights == [
"0.0",
"1.0",
"1.23",

View File

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

View File

@ -345,7 +345,7 @@ class TestHttp(unittest.TestCase):
def test_config(self):
app = create_app(
FrigateConfig(**self.minimal_config).runtime_config(),
FrigateConfig(**self.minimal_config),
self.db,
None,
None,
@ -363,7 +363,7 @@ class TestHttp(unittest.TestCase):
def test_recordings(self):
app = create_app(
FrigateConfig(**self.minimal_config).runtime_config(),
FrigateConfig(**self.minimal_config),
self.db,
None,
None,
@ -385,7 +385,7 @@ class TestHttp(unittest.TestCase):
stats = Mock(spec=StatsEmitter)
stats.get_latest_stats.return_value = self.test_stats
app = create_app(
FrigateConfig(**self.minimal_config).runtime_config(),
FrigateConfig(**self.minimal_config),
self.db,
None,
None,

View File

@ -9,14 +9,12 @@ import queue
import re
import shlex
import urllib.parse
from collections import Counter
from collections.abc import Mapping
from pathlib import Path
from typing import Any, Optional, Tuple
import numpy as np
import pytz
import yaml
from ruamel.yaml import YAML
from tzlocal import get_localzone
from zoneinfo import ZoneInfoNotFoundError
@ -89,34 +87,6 @@ def deep_merge(dct1: dict, dct2: dict, override=False, merge_lists=False) -> dic
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:
"""Removes user and password from line."""
rtsp_cleaned = re.sub(REGEX_RTSP_CAMERA_USER_PASS, "://*:*@", line)

View File

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