mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-16 11:42:47 +03:00
Refactor model scene definitions (#24508)
CI / AMD64 Build (push) Canceled after 0s
CI / AMD64 Smoke Test (push) Canceled after 0s
CI / ARM Build (push) Canceled after 0s
CI / Jetson Jetpack 6 (push) Canceled after 0s
CI / AMD64 Extra Build (push) Canceled after 0s
CI / ARM Extra Build (push) Canceled after 0s
CI / Synaptics Build (push) Canceled after 0s
CI / Assemble and push default build (push) Canceled after 0s
CI / AMD64 Build (push) Canceled after 0s
CI / AMD64 Smoke Test (push) Canceled after 0s
CI / ARM Build (push) Canceled after 0s
CI / Jetson Jetpack 6 (push) Canceled after 0s
CI / AMD64 Extra Build (push) Canceled after 0s
CI / ARM Extra Build (push) Canceled after 0s
CI / Synaptics Build (push) Canceled after 0s
CI / Assemble and push default build (push) Canceled after 0s
* Refactor model scene definitions * Cleanup * Validate model paths in the UI * Handle form validation
This commit is contained in:
+2
-3
@@ -49,7 +49,6 @@ from frigate.debug_replay import (
|
||||
DebugReplayManager,
|
||||
cleanup_replay_cameras,
|
||||
)
|
||||
from frigate.detectors.detector_config import SceneEnum
|
||||
from frigate.detectors.detector_types import api_types
|
||||
from frigate.detectors.device import build_detector_config, runner_names
|
||||
from frigate.embeddings import EmbeddingProcess, EmbeddingsContext
|
||||
@@ -108,7 +107,7 @@ class FrigateApp:
|
||||
self.metrics_manager = manager
|
||||
self.audio_process: mp.Process | None = None
|
||||
self.stop_event = stop_event
|
||||
self.detection_queues: dict[SceneEnum, Queue] = {
|
||||
self.detection_queues: dict[str, Queue] = {
|
||||
model.scene: mp.Queue() for model in config.models
|
||||
}
|
||||
self.detectors: dict[str, ObjectDetectProcess] = {}
|
||||
@@ -395,7 +394,7 @@ class FrigateApp:
|
||||
logger.error("Unable to prepare the %s runtime: %s", detector_type, err)
|
||||
|
||||
def start_detectors(self) -> None:
|
||||
model_cameras: dict[SceneEnum, list[str]] = {
|
||||
model_cameras: dict[str, list[str]] = {
|
||||
model.scene: [] for model in self.config.models
|
||||
}
|
||||
|
||||
|
||||
@@ -15,7 +15,6 @@ from frigate.config.camera.updater import (
|
||||
CameraConfigUpdateSubscriber,
|
||||
)
|
||||
from frigate.const import REPLAY_CAMERA_PREFIX
|
||||
from frigate.detectors.detector_config import SceneEnum
|
||||
from frigate.models import Regions
|
||||
from frigate.object_detection.util import detection_frame_size
|
||||
from frigate.util.builtin import empty_and_close_queue
|
||||
@@ -31,7 +30,7 @@ class CameraMaintainer(threading.Thread):
|
||||
def __init__(
|
||||
self,
|
||||
config: FrigateConfig,
|
||||
detection_queues: dict[SceneEnum, Queue],
|
||||
detection_queues: dict[str, Queue],
|
||||
detected_frames_queue: Queue,
|
||||
camera_metrics: DictProxy,
|
||||
ptz_metrics: dict[str, PTZMetrics],
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from pydantic import Field, model_validator
|
||||
|
||||
from frigate.detectors.detector_config import SceneEnum
|
||||
from frigate.detectors.detector_config import DEFAULT_SCENE, SCENE_PATTERN
|
||||
|
||||
from ..base import FrigateBaseModel
|
||||
|
||||
@@ -62,10 +62,11 @@ class DetectConfig(FrigateBaseModel):
|
||||
title="Detect width",
|
||||
description="Width (pixels) of frames used for the detect stream; leave empty to use the native stream resolution.",
|
||||
)
|
||||
scene: SceneEnum = Field(
|
||||
default=SceneEnum.all,
|
||||
scene: str = Field(
|
||||
default=DEFAULT_SCENE,
|
||||
pattern=SCENE_PATTERN,
|
||||
title="Detect scene",
|
||||
description="The environment this camera looks at, used to pick which of the configured models runs on it. Cameras left on 'all' run the model configured with a scene of 'all'.",
|
||||
description="The environment this camera looks at, used to pick which of the configured models runs on it. Cameras left on 'default' run the model configured with a scene of 'default'.",
|
||||
)
|
||||
fps: int = Field(
|
||||
default=5,
|
||||
|
||||
+103
-22
@@ -19,7 +19,7 @@ from ruamel.yaml import YAML
|
||||
|
||||
from frigate.const import REGEX_JSON
|
||||
from frigate.detectors import ModelConfig
|
||||
from frigate.detectors.detector_config import SceneEnum
|
||||
from frigate.detectors.detector_config import DEFAULT_SCENE
|
||||
from frigate.detectors.device import DeviceParseError, DeviceSpec, parse_device
|
||||
from frigate.plus import PlusApi
|
||||
from frigate.util.builtin import (
|
||||
@@ -35,6 +35,7 @@ from frigate.util.config import (
|
||||
migrate_frigate_config,
|
||||
)
|
||||
from frigate.util.image import create_mask
|
||||
from frigate.util.runtime_deps import sha256_of
|
||||
from frigate.util.services import auto_detect_hwaccel
|
||||
|
||||
from .auth import AuthConfig
|
||||
@@ -536,7 +537,7 @@ class FrigateConfig(FrigateBaseModel):
|
||||
models: list[ModelConfig] = Field(
|
||||
default_factory=_default_models,
|
||||
title="Detection models",
|
||||
description="Object detection models and the hardware each one runs on. Cameras pick a model by matching their detect.scene against a model's scene.",
|
||||
description="Object detection models and the hardware each one runs on. Cameras pick a model by matching their detect.scene against a model's scene, falling back to the 'default' model.",
|
||||
)
|
||||
|
||||
# GenAI config (named provider configs: name -> GenAIConfig)
|
||||
@@ -651,7 +652,9 @@ class FrigateConfig(FrigateBaseModel):
|
||||
)
|
||||
|
||||
_plus_api: PlusApi
|
||||
_model_devices: dict[SceneEnum, list[DeviceSpec]]
|
||||
_model_devices: dict[str, list[DeviceSpec]]
|
||||
# scene -> model, including the scenes of duplicate models folded into another
|
||||
_scene_models: dict[str, ModelConfig]
|
||||
_camera_models: dict[str, ModelConfig]
|
||||
_all_attributes: list[str]
|
||||
_all_attribute_logos: list[str]
|
||||
@@ -686,7 +689,7 @@ class FrigateConfig(FrigateBaseModel):
|
||||
def primary_model(self) -> ModelConfig:
|
||||
"""The model used when no specific camera is in play."""
|
||||
for model in self.models:
|
||||
if model.scene == SceneEnum.all:
|
||||
if model.scene == DEFAULT_SCENE:
|
||||
return model
|
||||
|
||||
return self.models[0]
|
||||
@@ -708,7 +711,7 @@ class FrigateConfig(FrigateBaseModel):
|
||||
|
||||
if model is None:
|
||||
camera = self.cameras.get(camera_name)
|
||||
scene = camera.detect.scene if camera is not None else SceneEnum.all
|
||||
scene = camera.detect.scene if camera is not None else DEFAULT_SCENE
|
||||
model = self._resolve_camera_model(camera_name, scene)
|
||||
self._camera_models[camera_name] = model
|
||||
|
||||
@@ -768,14 +771,14 @@ class FrigateConfig(FrigateBaseModel):
|
||||
if not self.models:
|
||||
raise ValueError("At least one model must be configured under models")
|
||||
|
||||
model_devices: dict[SceneEnum, list[DeviceSpec]] = {}
|
||||
model_devices: dict[str, list[DeviceSpec]] = {}
|
||||
# device string -> the scene of the model that already claimed it
|
||||
claimed_devices: dict[str, SceneEnum] = {}
|
||||
claimed_devices: dict[str, str] = {}
|
||||
|
||||
for index, model in enumerate(self.models):
|
||||
scene = model.scene.value
|
||||
scene = model.scene
|
||||
|
||||
if model.scene in model_devices:
|
||||
if scene in model_devices:
|
||||
raise ValueError(
|
||||
f"Multiple models are configured with a scene of '{scene}'. Each model must use a different scene."
|
||||
)
|
||||
@@ -804,17 +807,20 @@ class FrigateConfig(FrigateBaseModel):
|
||||
other = claimed_devices[device.raw]
|
||||
where = (
|
||||
f"twice by model '{scene}'"
|
||||
if other == model.scene
|
||||
else f"by both the '{other.value}' and '{scene}' models"
|
||||
if other == scene
|
||||
else f"by both the '{other}' and '{scene}' models"
|
||||
)
|
||||
raise ValueError(
|
||||
f"Device '{device.raw}' is used {where}, but it can only run one detection process."
|
||||
)
|
||||
|
||||
claimed_devices[device.raw] = model.scene
|
||||
claimed_devices[device.raw] = scene
|
||||
|
||||
self.models[index] = self._load_model(model, devices[0].detector)
|
||||
model_devices[model.scene] = devices
|
||||
model_devices[scene] = devices
|
||||
|
||||
self._scene_models = {model.scene: model for model in self.models}
|
||||
self._consolidate_duplicate_models(model_devices)
|
||||
|
||||
attributes: set[str] = set()
|
||||
attribute_logos: set[str] = set()
|
||||
@@ -838,36 +844,107 @@ class FrigateConfig(FrigateBaseModel):
|
||||
}
|
||||
self._all_labels = labels
|
||||
|
||||
def _resolve_camera_model(self, name: str, scene: SceneEnum) -> ModelConfig:
|
||||
def _consolidate_duplicate_models(
|
||||
self, model_devices: dict[str, list[DeviceSpec]]
|
||||
) -> None:
|
||||
"""Fold models that load the same model file into a single model.
|
||||
|
||||
Separate scenes for one model only split the same work across separate
|
||||
detection queues, so each device serves fewer cameras and is slower
|
||||
overall than one shared model. The duplicate's devices are moved to the
|
||||
model it duplicates and its scene resolves to that model.
|
||||
|
||||
Args:
|
||||
model_devices: Scene to parsed devices, updated in place
|
||||
"""
|
||||
kept: list[ModelConfig] = []
|
||||
hashes: dict[str, str | None] = {}
|
||||
|
||||
def file_hash(path: str) -> str | None:
|
||||
if path not in hashes:
|
||||
hashes[path] = sha256_of(path) if os.path.isfile(path) else None
|
||||
|
||||
return hashes[path]
|
||||
|
||||
def same_model(a: ModelConfig, b: ModelConfig) -> bool:
|
||||
# a model's devices all share a detector, so folding across
|
||||
# detectors would produce an invalid model
|
||||
if model_devices[a.scene][0].detector != model_devices[b.scene][0].detector:
|
||||
return False
|
||||
|
||||
if not a.path or not b.path:
|
||||
return False
|
||||
|
||||
if os.path.realpath(a.path) == os.path.realpath(b.path):
|
||||
return True
|
||||
|
||||
a_hash = file_hash(a.path)
|
||||
return a_hash is not None and a_hash == file_hash(b.path)
|
||||
|
||||
for model in self.models:
|
||||
original = next((other for other in kept if same_model(other, model)), None)
|
||||
|
||||
if original is None:
|
||||
kept.append(model)
|
||||
continue
|
||||
|
||||
# keep the default model so cameras without a scene still find it
|
||||
if model.scene == DEFAULT_SCENE:
|
||||
kept[kept.index(original)] = model
|
||||
original, model = model, original
|
||||
|
||||
logger.warning(
|
||||
"Models '%s' and '%s' use the same model file, so they have been combined into the '%s' model. Defining one model under several scenes to assign detectors to specific cameras is slower and less efficient than letting every detector serve every camera. Remove the '%s' model and list its devices under the '%s' model instead",
|
||||
original.scene,
|
||||
model.scene,
|
||||
original.scene,
|
||||
model.scene,
|
||||
original.scene,
|
||||
)
|
||||
original.devices = [*original.devices, *model.devices]
|
||||
model_devices[original.scene] = [
|
||||
*model_devices[original.scene],
|
||||
*model_devices.pop(model.scene),
|
||||
]
|
||||
self._scene_models[model.scene] = original
|
||||
|
||||
# anything already folded into the duplicate follows it
|
||||
for scene, target in self._scene_models.items():
|
||||
if target is model:
|
||||
self._scene_models[scene] = original
|
||||
|
||||
self.models = kept
|
||||
|
||||
def _resolve_camera_model(self, name: str, scene: str) -> ModelConfig:
|
||||
"""Resolve which model a camera runs on.
|
||||
|
||||
A camera may name a scene no model is configured for, which is valid as
|
||||
long as an 'all' model is there to fall back to.
|
||||
long as a 'default' model is there to fall back to.
|
||||
|
||||
Args:
|
||||
name: Name of the camera
|
||||
scene: The camera's detect scene, which defaults to 'all'
|
||||
scene: The camera's detect scene, which defaults to 'default'
|
||||
|
||||
Returns:
|
||||
The model the camera runs on
|
||||
"""
|
||||
by_scene = {model.scene: model for model in self.models}
|
||||
model = by_scene.get(scene)
|
||||
model = self._scene_models.get(scene)
|
||||
|
||||
if model is not None:
|
||||
return model
|
||||
|
||||
default = by_scene.get(SceneEnum.all)
|
||||
default = self._scene_models.get(DEFAULT_SCENE)
|
||||
|
||||
if default is None:
|
||||
raise ValueError(
|
||||
f"Camera '{name}' has a detect scene of '{scene.value}', but no model is configured for that scene or for 'all'."
|
||||
f"Camera '{name}' has a detect scene of '{scene}', but no model is configured for that scene or for '{DEFAULT_SCENE}'."
|
||||
)
|
||||
|
||||
logger.warning(
|
||||
"Camera '%s' has a detect scene of '%s', but no model is configured for that scene, so the 'all' model is used",
|
||||
"Camera '%s' has a detect scene of '%s', but no model is configured for that scene, so the '%s' model is used",
|
||||
name,
|
||||
scene.value,
|
||||
scene,
|
||||
DEFAULT_SCENE,
|
||||
)
|
||||
return default
|
||||
|
||||
@@ -998,6 +1075,10 @@ class FrigateConfig(FrigateBaseModel):
|
||||
camera_model = self._resolve_camera_model(name, camera_config.detect.scene)
|
||||
self._camera_models[name] = camera_model
|
||||
|
||||
# point cameras at the model their duplicate scene was folded into
|
||||
if camera_config.detect.scene in self._scene_models:
|
||||
camera_config.detect.scene = camera_model.scene
|
||||
|
||||
if camera_config.ffmpeg.hwaccel_args == "auto":
|
||||
camera_config.ffmpeg.hwaccel_args = self.ffmpeg.hwaccel_args
|
||||
|
||||
|
||||
@@ -47,21 +47,17 @@ class ModelTypeEnum(str, Enum):
|
||||
yologeneric = "yolo-generic"
|
||||
|
||||
|
||||
class SceneEnum(str, Enum):
|
||||
"""The camera environment a detection model is intended for."""
|
||||
|
||||
all = "all"
|
||||
indoor = "indoor"
|
||||
outdoor = "outdoor"
|
||||
indoor_thermal = "indoor_thermal"
|
||||
outdoor_thermal = "outdoor_thermal"
|
||||
# the scene of the model used by cameras that don't name one
|
||||
DEFAULT_SCENE = "default"
|
||||
SCENE_PATTERN = r"^[A-Za-z0-9_-]+$"
|
||||
|
||||
|
||||
class ModelConfig(BaseModel):
|
||||
scene: SceneEnum = Field(
|
||||
default=SceneEnum.all,
|
||||
scene: str = Field(
|
||||
default=DEFAULT_SCENE,
|
||||
pattern=SCENE_PATTERN,
|
||||
title="Model scene",
|
||||
description="The camera environment this model is used for. Cameras select a model by setting detect.scene to a matching value, and 'all' is used by any camera that does not set one.",
|
||||
description="A name for the camera environment this model is used for, such as 'thermal'. Cameras select a model by setting detect.scene to a matching value, and the 'default' model is used by any camera that does not set one.",
|
||||
)
|
||||
devices: list[str] = Field(
|
||||
default_factory=list,
|
||||
|
||||
@@ -9,7 +9,6 @@ from frigate.config.camera.updater import (
|
||||
CameraConfigUpdateSubscriber,
|
||||
)
|
||||
from frigate.const import SUB_CACHE_TAG
|
||||
from frigate.detectors.detector_config import SceneEnum
|
||||
|
||||
|
||||
def _build_scene_frigate_config(scene: str | None) -> FrigateConfig:
|
||||
@@ -119,7 +118,7 @@ class TestRecordUpdateRecreatesFfmpegCmds(unittest.TestCase):
|
||||
subscriber = CameraConfigUpdateSubscriber(
|
||||
config, {}, [CameraConfigUpdateEnum.add, CameraConfigUpdateEnum.remove]
|
||||
)
|
||||
assert config.model_for_camera("front_door").scene == SceneEnum.outdoor
|
||||
assert config.model_for_camera("front_door").scene == "outdoor"
|
||||
|
||||
subscriber.subscriber.check_for_update.side_effect = [
|
||||
("config/cameras/front_door/remove", config.cameras["front_door"]),
|
||||
@@ -136,7 +135,7 @@ class TestRecordUpdateRecreatesFfmpegCmds(unittest.TestCase):
|
||||
]
|
||||
subscriber.check_for_updates()
|
||||
|
||||
assert config.model_for_camera("front_door").scene == SceneEnum.all
|
||||
assert config.model_for_camera("front_door").scene == "default"
|
||||
|
||||
def test_unchanged_record_update_keeps_existing_cmds(self):
|
||||
camera_config = _build_camera_config(sub_enabled=False)
|
||||
|
||||
+125
-7
@@ -1,5 +1,6 @@
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from copy import deepcopy
|
||||
from unittest.mock import patch
|
||||
@@ -11,7 +12,6 @@ from ruamel.yaml.constructor import DuplicateKeyError
|
||||
from frigate.config import BirdseyeModeEnum, FrigateConfig, RetainModeEnum
|
||||
from frigate.const import MODEL_CACHE_DIR
|
||||
from frigate.detectors import DetectorTypeEnum
|
||||
from frigate.detectors.detector_config import SceneEnum
|
||||
from frigate.detectors.device import build_detector_config, runner_names
|
||||
from frigate.util.builtin import deep_merge
|
||||
|
||||
@@ -68,7 +68,7 @@ class TestConfig(unittest.TestCase):
|
||||
def test_config_class(self):
|
||||
frigate_config = FrigateConfig(**self.minimal)
|
||||
model = frigate_config.primary_model
|
||||
assert model.scene == SceneEnum.all
|
||||
assert model.scene == "default"
|
||||
assert model.width == 320
|
||||
assert frigate_config.devices_for_model(model)[0].detector == (
|
||||
DetectorTypeEnum.cpu
|
||||
@@ -140,8 +140,8 @@ class TestConfig(unittest.TestCase):
|
||||
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert frigate_config.model_for_camera("back").scene == SceneEnum.outdoor
|
||||
assert frigate_config.model_for_camera("front").scene == SceneEnum.indoor
|
||||
assert frigate_config.model_for_camera("back").scene == "outdoor"
|
||||
assert frigate_config.model_for_camera("front").scene == "indoor"
|
||||
assert frigate_config.model_for_camera("back").width == 320
|
||||
assert frigate_config.model_for_camera("front").width == 300
|
||||
|
||||
@@ -177,7 +177,7 @@ class TestConfig(unittest.TestCase):
|
||||
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert frigate_config.model_for_camera("back").scene == SceneEnum.all
|
||||
assert frigate_config.model_for_camera("back").scene == "default"
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_model_for_camera_resolves_camera_added_after_parse(self, mock_labels):
|
||||
@@ -205,7 +205,7 @@ class TestConfig(unittest.TestCase):
|
||||
new_config = FrigateConfig(**(deep_merge(deepcopy(config), added)))
|
||||
frigate_config.cameras["new_cam"] = new_config.cameras["new_cam"]
|
||||
|
||||
assert frigate_config.model_for_camera("new_cam").scene == SceneEnum.outdoor
|
||||
assert frigate_config.model_for_camera("new_cam").scene == "outdoor"
|
||||
assert frigate_config.model_for_camera("new_cam").width == 416
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
@@ -221,7 +221,7 @@ class TestConfig(unittest.TestCase):
|
||||
frigate_config = FrigateConfig(**(deep_merge(deepcopy(config), self.minimal)))
|
||||
|
||||
# a caller racing a runtime remove may still name the popped camera
|
||||
assert frigate_config.model_for_camera("removed").scene == SceneEnum.all
|
||||
assert frigate_config.model_for_camera("removed").scene == "default"
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_camera_scene_without_a_model_or_a_default(self, mock_labels):
|
||||
@@ -256,6 +256,124 @@ class TestConfig(unittest.TestCase):
|
||||
with self.assertRaises(ValidationError):
|
||||
FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_scene_names_are_not_a_fixed_list(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
config = {
|
||||
"models": [
|
||||
{"devices": ["cpu"]},
|
||||
{"scene": "garage_thermal", "devices": ["openvino:CPU"]},
|
||||
],
|
||||
"cameras": {"back": {"detect": {"scene": "garage_thermal"}}},
|
||||
}
|
||||
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert frigate_config.model_for_camera("back").scene == "garage_thermal"
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_scene_names_must_be_simple_identifiers(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
config = {"models": [{"scene": "front yard", "devices": ["cpu"]}]}
|
||||
|
||||
with self.assertRaises(ValidationError):
|
||||
FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
def _two_scene_config(self, default_path: str, outdoor_path: str) -> dict:
|
||||
return {
|
||||
"models": [
|
||||
{"path": default_path, "devices": ["openvino:CPU"]},
|
||||
{"scene": "outdoor", "path": outdoor_path, "devices": ["openvino:GPU"]},
|
||||
],
|
||||
"cameras": {"back": {"detect": {"scene": "outdoor"}}},
|
||||
}
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_models_with_the_same_path_are_combined(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
config = self._two_scene_config("/etc/hosts", "/etc/hosts")
|
||||
|
||||
with self.assertLogs("frigate.config.config", level="WARNING") as logs:
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert len(frigate_config.models) == 1
|
||||
model = frigate_config.primary_model
|
||||
assert model.devices == ["openvino:CPU", "openvino:GPU"]
|
||||
assert [d.raw for d in frigate_config.devices_for_model(model)] == [
|
||||
"openvino:CPU",
|
||||
"openvino:GPU",
|
||||
]
|
||||
assert frigate_config.model_for_camera("back") is model
|
||||
assert frigate_config.cameras["back"].detect.scene == "default"
|
||||
assert any("same model file" in line for line in logs.output)
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_models_with_the_same_file_contents_are_combined(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
first = os.path.join(temp_dir, "model.onnx")
|
||||
copy = os.path.join(temp_dir, "renamed.onnx")
|
||||
|
||||
for path in (first, copy):
|
||||
with open(path, "wb") as f:
|
||||
f.write(b"same weights")
|
||||
|
||||
config = self._two_scene_config(first, copy)
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert len(frigate_config.models) == 1
|
||||
assert frigate_config.model_for_camera("back").scene == "default"
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_models_with_different_files_are_kept_apart(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
first = os.path.join(temp_dir, "model.onnx")
|
||||
other = os.path.join(temp_dir, "thermal.onnx")
|
||||
|
||||
for path, contents in ((first, b"visible"), (other, b"thermal")):
|
||||
with open(path, "wb") as f:
|
||||
f.write(contents)
|
||||
|
||||
config = self._two_scene_config(first, other)
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert len(frigate_config.models) == 2
|
||||
assert frigate_config.model_for_camera("back").scene == "outdoor"
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_combined_models_keep_the_default_model(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
config = {
|
||||
"models": [
|
||||
{"scene": "outdoor", "path": "/etc/hosts", "devices": ["openvino:GPU"]},
|
||||
{"path": "/etc/hosts", "devices": ["openvino:CPU"]},
|
||||
],
|
||||
"cameras": {"back": {"detect": {"scene": "outdoor"}}},
|
||||
}
|
||||
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert [model.scene for model in frigate_config.models] == ["default"]
|
||||
assert frigate_config.primary_model.devices == ["openvino:CPU", "openvino:GPU"]
|
||||
assert frigate_config.model_for_camera("back").scene == "default"
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_models_on_different_detectors_are_kept_apart(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
config = {
|
||||
"models": [
|
||||
{"path": "/etc/hosts", "devices": ["openvino:CPU"]},
|
||||
{"scene": "outdoor", "path": "/etc/hosts", "devices": ["onnx"]},
|
||||
],
|
||||
}
|
||||
|
||||
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
|
||||
|
||||
assert len(frigate_config.models) == 2
|
||||
|
||||
@patch("frigate.detectors.detector_config.load_labels")
|
||||
def test_model_devices_must_share_a_detector(self, mock_labels):
|
||||
mock_labels.return_value = {}
|
||||
|
||||
@@ -20,7 +20,7 @@ class TestMigrateModels(unittest.TestCase):
|
||||
def test_single_cpu_detector(self):
|
||||
migrated = migrate_models({"detectors": {"cpu": {"type": "cpu"}}})
|
||||
|
||||
self.assertEqual(migrated["models"], [{"scene": "all", "devices": ["cpu"]}])
|
||||
self.assertEqual(migrated["models"], [{"scene": "default", "devices": ["cpu"]}])
|
||||
self.assertNotIn("detectors", migrated)
|
||||
|
||||
def test_model_settings_are_carried_over(self):
|
||||
@@ -35,7 +35,7 @@ class TestMigrateModels(unittest.TestCase):
|
||||
migrated["models"],
|
||||
[
|
||||
{
|
||||
"scene": "all",
|
||||
"scene": "default",
|
||||
"path": "plus://abc",
|
||||
"width": 320,
|
||||
"devices": ["edgetpu:pci:0"],
|
||||
@@ -262,7 +262,7 @@ class TestMigrateConfigFile(unittest.TestCase):
|
||||
"mqtt:\n"
|
||||
" enabled: false\n"
|
||||
"models:\n"
|
||||
" - scene: all\n"
|
||||
" - scene: default\n"
|
||||
" devices:\n"
|
||||
" - openvino:GPU\n"
|
||||
"cameras: {}\n"
|
||||
@@ -270,7 +270,7 @@ class TestMigrateConfigFile(unittest.TestCase):
|
||||
)
|
||||
|
||||
self.assertEqual(
|
||||
migrated["models"], [{"scene": "all", "devices": ["openvino:GPU"]}]
|
||||
migrated["models"], [{"scene": "default", "devices": ["openvino:GPU"]}]
|
||||
)
|
||||
self.assertFalse(
|
||||
os.path.exists(os.path.join(self.temp_dir.name, "backup_config.yaml"))
|
||||
@@ -298,7 +298,7 @@ class TestMigrateConfigFile(unittest.TestCase):
|
||||
"mqtt:\n"
|
||||
" enabled: false\n"
|
||||
"models:\n"
|
||||
" - scene: all\n"
|
||||
" - scene: default\n"
|
||||
" devices:\n"
|
||||
" - hailo8l:PCIe\n"
|
||||
"cameras: {}\n"
|
||||
|
||||
@@ -978,7 +978,7 @@ def migrate_models(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any
|
||||
", ".join(sorted(detector_types)),
|
||||
)
|
||||
|
||||
entry: dict[str, Any] = {"scene": "all", **model}
|
||||
entry: dict[str, Any] = {"scene": "default", **model}
|
||||
|
||||
if model_path:
|
||||
entry["path"] = model_path
|
||||
|
||||
Reference in New Issue
Block a user