administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-01 04:16:50 +03:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Refactor detector and model management (#23995)

Browse Source 
* Refactor detector and model management

* Fix model resolution field
This commit is contained in:
Nicolas Mowen
2026-09-12 07:30:04 -06:00
parent fd76eb6c6f
commit 8fe35ace3b
63 changed files with 2052 additions and 1152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/api/app.py
+24 -28
View File
@@ -292,10 +292,6 @@ def config(request: Request):
config: dict[str, dict[str, Any]] = config_obj.model_dump(
mode="json", warnings="none", exclude_none=True
)
config["detectors"] = {
name: detector.model_dump(mode="json", warnings="none", exclude_none=True)
for name, detector in config_obj.detectors.items()
}
# remove environment_vars for non-admin users
if request.headers.get("remote-role") != "admin":
@@ -376,31 +372,28 @@ def config(request: Request):
config["go2rtc"]["streams"][stream_name] = cleaned
config["plus"] = {"enabled": request.app.frigate_config.plus_api.is_active()}
config["model"]["colormap"] = config_obj.model.colormap
config["model"]["all_attributes"] = config_obj.model.all_attributes
config["model"]["non_logo_attributes"] = config_obj.model.non_logo_attributes
# Add model plus data if plus is enabled
if config["plus"]["enabled"]:
model_path = config.get("model", {}).get("path")
if model_path:
model_json_path = FilePath(model_path).with_suffix(".json")
for index, model in enumerate(config_obj.models):
model_dict = config["models"][index]
model_dict["colormap"] = model.colormap
model_dict["all_attributes"] = model.all_attributes
model_dict["non_logo_attributes"] = model.non_logo_attributes
model_dict["labelmap"] = model.merged_labelmap
if not config["plus"]["enabled"]:
continue
# Add model plus data if plus is enabled
model_dict["plus"] = None
if model.path:
model_json_path = FilePath(model.path).with_suffix(".json")
try:
with open(model_json_path) as f:
model_plus_data = json.load(f)
config["model"]["plus"] = model_plus_data
except FileNotFoundError:
config["model"]["plus"] = None
except json.JSONDecodeError:
config["model"]["plus"] = None
else:
config["model"]["plus"] = None
# use merged labelamp
for detector_config in config["detectors"].values():
detector_config["model"]["labelmap"] = (
request.app.frigate_config.model.merged_labelmap
)
model_dict["plus"] = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
pass
return JSONResponse(content=config)
@@ -1360,11 +1353,14 @@ def plusModels(request: Request, filterByCurrentModelDetector: bool = False):
modelList = models["list"]
config: FrigateConfig = request.app.frigate_config
primary_model = config.primary_model
# current model type
modelType = request.app.frigate_config.model.model_type
modelType = primary_model.model_type
# current detectorType for comparing to supportedDetectors
detectorType = list(request.app.frigate_config.detectors.values())[0].type
detectorType = config.devices_for_model(primary_model)[0].detector
validModels = []
frigate/api/media.py
+1 -1
View File
@@ -941,7 +941,7 @@ async def event_snapshot(
timestamp_style=request.app.frigate_config.cameras[
event.camera
].timestamp_style,
colormap=request.app.frigate_config.model.colormap,
colormap=request.app.frigate_config.model_for_camera(event.camera).colormap,
)
except DoesNotExist:
# see if the object is currently being tracked
frigate/app.py
+39 -22
View File
@@ -49,6 +49,8 @@ from frigate.debug_replay import (
DebugReplayManager,
cleanup_replay_cameras,
)
from frigate.detectors.detector_config import SceneEnum
from frigate.detectors.device import build_detector_config, runner_names
from frigate.embeddings import EmbeddingProcess, EmbeddingsContext
from frigate.events.audio import AudioProcessor
from frigate.events.cleanup import EventCleanup
@@ -69,6 +71,7 @@ from frigate.models import (
User,
)
from frigate.object_detection.base import ObjectDetectProcess
from frigate.object_detection.util import detection_frame_size
from frigate.output.output import OutputProcess
from frigate.ptz.autotrack import PtzAutoTrackerThread
from frigate.ptz.onvif import OnvifController
@@ -98,7 +101,9 @@ class FrigateApp:
self.metrics_manager = manager
self.audio_process: mp.Process | None = None
self.stop_event = stop_event
self.detection_queue: Queue = mp.Queue()
self.detection_queues: dict[SceneEnum, Queue] = {
model.scene: mp.Queue() for model in config.models
}
self.detectors: dict[str, ObjectDetectProcess] = {}
self.detection_shms: list[mp.shared_memory.SharedMemory] = []
self.log_queue: Queue = mp.Queue()
@@ -344,20 +349,19 @@ class FrigateApp:
self.dispatcher.profile_manager = self.profile_manager
def start_detectors(self) -> None:
model_cameras: dict[SceneEnum, list[str]] = {
model.scene: [] for model in self.config.models
}
for name in self.config.cameras.keys():
model = self.config.model_for_camera(name)
model_cameras[model.scene].append(name)
try:
largest_frame = max(
[
det.model.height * det.model.width * 3
if det.model is not None
else 320
for det in self.config.detectors.values()
]
)
shm_in = UntrackedSharedMemory(
name=name,
create=True,
size=largest_frame,
size=detection_frame_size(model),
)
except FileExistsError:
shm_in = UntrackedSharedMemory(name=name)
@@ -372,15 +376,26 @@ class FrigateApp:
self.detection_shms.append(shm_in)
self.detection_shms.append(shm_out)
for name, detector_config in self.config.detectors.items():
self.detectors[name] = ObjectDetectProcess(
name,
self.detection_queue,
list(self.config.cameras.keys()),
self.config,
detector_config,
self.stop_event,
)
# a device may be listed more than once to run additional inference
# processes on it, so names are only unique once de-duplicated
all_devices = [
device
for model in self.config.models
for device in self.config.devices_for_model(model)
]
names = iter(runner_names(all_devices))
for model in self.config.models:
for device in self.config.devices_for_model(model):
name = next(names)
self.detectors[name] = ObjectDetectProcess(
name,
self.detection_queues[model.scene],
model_cameras[model.scene],
self.config,
build_detector_config(device, model),
self.stop_event,
)
def start_ptz_autotracker(self) -> None:
self.ptz_autotracker_thread = PtzAutoTrackerThread(
@@ -411,7 +426,7 @@ class FrigateApp:
def start_camera_processor(self) -> None:
self.camera_maintainer = CameraMaintainer(
self.config,
self.detection_queue,
self.detection_queues,
self.detected_frames_queue,
self.camera_metrics,
self.ptz_metrics,
@@ -675,8 +690,10 @@ class FrigateApp:
for detector in self.detectors.values():
detector.stop()
empty_and_close_queue(self.detection_queue)
logger.info("Detection queue closed")
for detection_queue in self.detection_queues.values():
empty_and_close_queue(detection_queue)
logger.info("Detection queues closed")
self.detected_frames_processor.join()
empty_and_close_queue(self.detected_frames_queue)
frigate/camera/activity_manager.py
+2 -1
View File
@@ -18,6 +18,7 @@ from frigate.config.camera.updater import (
CameraConfigUpdateEnum,
CameraConfigUpdateSubscriber,
)
from frigate.detectors.detector_config import NON_LOGO_ATTRIBUTES
logger = logging.getLogger(__name__)
@@ -178,7 +179,7 @@ class CameraActivityManager:
return
for label in camera_config.objects.track:
if label in self.config.model.non_logo_attributes:
if label in NON_LOGO_ATTRIBUTES:
continue
new_count = all_objects[label]
frigate/camera/maintainer.py
+12 -16
View File
@@ -15,7 +15,9 @@ 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
from frigate.util.image import SharedMemoryFrameManager, UntrackedSharedMemory
from frigate.util.object import get_camera_regions_grid
@@ -29,7 +31,7 @@ class CameraMaintainer(threading.Thread):
def __init__(
self,
config: FrigateConfig,
detection_queue: Queue,
detection_queues: dict[SceneEnum, Queue],
detected_frames_queue: Queue,
camera_metrics: DictProxy,
ptz_metrics: dict[str, PTZMetrics],
@@ -38,7 +40,7 @@ class CameraMaintainer(threading.Thread):
):
super().__init__(name="camera_processor")
self.config = config
self.detection_queue = detection_queue
self.detection_queues = detection_queues
self.detected_frames_queue = detected_frames_queue
self.stop_event = stop_event
self.camera_metrics = camera_metrics
@@ -79,10 +81,11 @@ class CameraMaintainer(threading.Thread):
# create or update region grids for each camera
for camera in self.config.cameras.values():
assert camera.name is not None
model = self.config.model_for_camera(camera.name)
self.region_grids[camera.name] = get_camera_regions_grid(
camera.name,
camera.detect,
max(self.config.model.width, self.config.model.height),
max(model.width, model.height),
)
def __calculate_shm_frame_count(self) -> int:
@@ -114,6 +117,7 @@ class CameraMaintainer(threading.Thread):
return
camera_stop_event = self.__ensure_camera_stop_event(name)
model = self.config.model_for_camera(name)
if runtime:
self.camera_metrics[name] = CameraMetrics(self.metrics_manager)
@@ -123,32 +127,24 @@ class CameraMaintainer(threading.Thread):
self.region_grids[name] = get_camera_regions_grid(
name,
config.detect,
max(self.config.model.width, self.config.model.height),
max(model.width, model.height),
)
try:
largest_frame = max(
[
det.model.height * det.model.width * 3
if det.model is not None
else 320
for det in self.config.detectors.values()
]
)
UntrackedSharedMemory(name=f"out-{name}", create=True, size=20 * 6 * 4)
UntrackedSharedMemory(
name=name,
create=True,
size=largest_frame,
size=detection_frame_size(model),
)
except FileExistsError:
pass
camera_process = CameraTracker(
config,
self.config.model,
self.config.model.merged_labelmap,
self.detection_queue,
model,
model.merged_labelmap,
self.detection_queues[model.scene],
self.detected_frames_queue,
self.camera_metrics[name],
self.ptz_metrics[name],
frigate/camera/state.py
+6 -11
View File
@@ -40,6 +40,7 @@ class CameraState:
self.name = name
self.config = config
self.camera_config = config.cameras[name]
self.model = config.model_for_camera(name)
self.frame_manager = frame_manager
self.best_objects: dict[str, TrackedObject] = {}
self.tracked_objects: dict[str, TrackedObject] = {}
@@ -106,9 +107,7 @@ class CameraState:
thickness = 1
else:
thickness = 2
color = self.config.model.colormap.get(
obj["label"], (255, 255, 255)
)
color = self.model.colormap.get(obj["label"], (255, 255, 255))
else:
thickness = 1
color = (255, 0, 0)
@@ -130,9 +129,7 @@ class CameraState:
and obj["frame_time"] == frame_time
):
thickness = 5
color = self.config.model.colormap.get(
obj["label"], (255, 255, 255)
)
color = self.model.colormap.get(obj["label"], (255, 255, 255))
# debug autotracking zooming - show the zoom factor box
if (
@@ -266,9 +263,7 @@ class CameraState:
if draw_options.get("paths"):
for obj in tracked_objects.values():
if obj["frame_time"] == frame_time and obj["path_data"]:
color = self.config.model.colormap.get(
obj["label"], (255, 255, 255)
)
color = self.model.colormap.get(obj["label"], (255, 255, 255))
path_points = [
(
@@ -371,7 +366,7 @@ class CameraState:
for id in new_ids:
logger.debug(f"{self.name}: New tracked object ID: {id}")
new_obj = tracked_objects[id] = TrackedObject(
self.config.model,
self.model,
self.camera_config,
self.config.ui,
self.frame_cache,
@@ -515,7 +510,7 @@ class CameraState:
sub_label = None
if obj.obj_data.get("sub_label"):
if obj.obj_data["sub_label"][0] in self.config.model.all_attributes:
if obj.obj_data["sub_label"][0] in self.model.all_attributes:
label = obj.obj_data["sub_label"][0]
else:
label = f"{object_type}-verified"
frigate/comms/dispatcher.py
+2 -1
View File
@@ -261,10 +261,11 @@ class Dispatcher:
if camera not in self.config.cameras:
return None
model = self.config.model_for_camera(camera)
grid = get_camera_regions_grid(
camera,
self.config.cameras[camera].detect,
max(self.config.model.width, self.config.model.height),
max(model.width, model.height),
)
return grid
frigate/config/camera/detect.py
+7
View File
@@ -1,5 +1,7 @@
from pydantic import Field, model_validator
from frigate.detectors.detector_config import SceneEnum
from ..base import FrigateBaseModel
__all__ = ["DetectConfig", "StationaryConfig", "StationaryMaxFramesConfig"]
@@ -60,6 +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 | None = Field(
default=None,
title="Detect scene",
description="The environment this camera looks at, used to pick which of the configured models runs on it. Defaults to the model with a scene of 'all'.",
)
fps: int = Field(
default=5,
title="Detect FPS",
frigate/config/config.py
+226 -59
View File
@@ -11,7 +11,6 @@ from pydantic import (
BaseModel,
ConfigDict,
Field,
TypeAdapter,
ValidationInfo,
field_validator,
model_validator,
@@ -19,8 +18,9 @@ from pydantic import (
from ruamel.yaml import YAML
from frigate.const import REGEX_JSON
from frigate.detectors import DetectorConfig, ModelConfig
from frigate.detectors.detector_config import BaseDetectorConfig
from frigate.detectors import ModelConfig
from frigate.detectors.detector_config import SceneEnum
from frigate.detectors.device import DeviceParseError, DeviceSpec, parse_device
from frigate.plus import PlusApi
from frigate.util.builtin import (
deep_merge,
@@ -79,9 +79,14 @@ logger = logging.getLogger(__name__)
yaml = YAML()
# Pydantic field default applied when an existing config omits `detectors:`.
# Pydantic field default applied when an existing config omits `models:`.
# Kept as cpu tflite for backwards compatibility with 0.17 configs.
DEFAULT_DETECTORS = {"cpu": {"type": "cpu"}}
DEFAULT_MODELS = [{"devices": ["cpu"]}]
def _default_models() -> list[ModelConfig]:
return [ModelConfig.model_validate(model) for model in DEFAULT_MODELS]
# Used by the openvino branch below and rendered into the new-config YAML
# template so first-time setups default to openvino on CPU.
@@ -93,7 +98,7 @@ DEFAULT_MODEL = {
"path": "/openvino-model/ssdlite_mobilenet_v2.xml",
"labelmap_path": "/openvino-model/coco_91cl_bkgr.txt",
}
NEW_CONFIG_DETECTORS = {"ov": {"type": "openvino", "device": "CPU"}}
NEW_CONFIG_MODELS = [{"devices": ["openvino:CPU"], **DEFAULT_MODEL}]
DEFAULT_DETECT_DIMENSIONS = {"width": 1280, "height": 720}
@@ -109,7 +114,7 @@ DEFAULT_CONFIG = f"""
mqtt:
enabled: False
{_render_default_yaml({"detectors": NEW_CONFIG_DETECTORS, "model": DEFAULT_MODEL})}
{_render_default_yaml({"models": NEW_CONFIG_MODELS})}
cameras: {{}} # No cameras defined, UI wizard should be used
version: {CURRENT_CONFIG_VERSION}
"""
@@ -520,16 +525,11 @@ class FrigateConfig(FrigateBaseModel):
description="User interface preferences such as timezone, time/date formatting, and units.",
)
# Detector config
detectors: dict[str, BaseDetectorConfig] = Field(
default=DEFAULT_DETECTORS,
title="Detector hardware",
description="Configuration for object detectors (CPU, GPU, ONNX backends) and any detector-specific model settings.",
)
model: ModelConfig = Field(
default_factory=ModelConfig,
title="Detection model",
description="Settings to configure a custom object detection model and its input shape.",
# Detection model config
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.",
)
# GenAI config (named provider configs: name -> GenAIConfig)
@@ -644,11 +644,202 @@ class FrigateConfig(FrigateBaseModel):
)
_plus_api: PlusApi
_model_devices: dict[SceneEnum, list[DeviceSpec]]
_camera_models: dict[str, ModelConfig]
_all_attributes: list[str]
_all_attribute_logos: list[str]
_all_attributes_map: dict[str, list[str]]
_all_labels: set[str]
@property
def plus_api(self) -> PlusApi:
return self._plus_api
@property
def all_attributes(self) -> list[str]:
"""Every attribute label across all configured models."""
return self._all_attributes
@property
def all_attribute_logos(self) -> list[str]:
"""Every logo attribute label across all configured models."""
return self._all_attribute_logos
@property
def all_attributes_map(self) -> dict[str, list[str]]:
"""Object label to attribute labels, merged across all configured models."""
return self._all_attributes_map
@property
def all_labels(self) -> set[str]:
"""Every object label across all configured models."""
return self._all_labels
@property
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:
return model
return self.models[0]
def model_for_camera(self, camera_name: str) -> ModelConfig:
"""Get the detection model a camera runs on.
Args:
camera_name: Name of the camera
Returns:
The model matching the camera's detect scene
"""
return self._camera_models[camera_name]
def devices_for_model(self, model: ModelConfig) -> list[DeviceSpec]:
"""Get the parsed hardware devices a model runs on.
Args:
model: One of the configured models
Returns:
The parsed device specs, in config order
"""
return self._model_devices[model.scene]
def _load_model(self, model: ModelConfig, detector: str) -> ModelConfig:
"""Apply detector specific defaults to a model and load its weights and labels.
Args:
model: The configured model
detector: The detector type the model runs on
Returns:
The loaded model
"""
model_config = model.model_dump(exclude_unset=True, warnings="none")
if "path" not in model_config:
if detector == "cpu" or detector.endswith("_tfl"):
model_config["path"] = "/cpu_model.tflite"
elif detector == "edgetpu":
model_config["path"] = "/edgetpu_model.tflite"
elif detector == "openvino":
for default_key, default_value in DEFAULT_MODEL.items():
model_config.setdefault(default_key, default_value)
loaded = ModelConfig.model_validate(model_config)
loaded.check_and_load_plus_model(self.plus_api, detector)
loaded.compute_model_hash()
return loaded
def _load_models(self) -> None:
"""Validate the configured models and load each one."""
if not self.models:
raise ValueError("At least one model must be configured under models")
model_devices: dict[SceneEnum, list[DeviceSpec]] = {}
# device string -> the scene of the model that already claimed it
claimed_devices: dict[str, SceneEnum] = {}
for index, model in enumerate(self.models):
scene = model.scene.value
if model.scene in model_devices:
raise ValueError(
f"Multiple models are configured with a scene of '{scene}'. Each model must use a different scene."
)
if not model.devices:
raise ValueError(
f"Model '{scene}' must list at least one entry under devices."
)
try:
devices = [parse_device(device) for device in model.devices]
except DeviceParseError as err:
raise ValueError(
f"Model '{scene}' has an invalid device: {err}"
) from err
detectors = {device.detector for device in devices}
if len(detectors) > 1:
raise ValueError(
f"Model '{scene}' mixes the {', '.join(sorted(detectors))} detectors. All of a model's devices must use the same detector."
)
for device in devices:
if device.raw in claimed_devices and not device.shareable:
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"
)
raise ValueError(
f"Device '{device.raw}' is used {where}, but it can only run one detection process."
)
claimed_devices[device.raw] = model.scene
self.models[index] = self._load_model(model, devices[0].detector)
model_devices[model.scene] = devices
attributes: set[str] = set()
attribute_logos: set[str] = set()
attributes_map: dict[str, set[str]] = {}
labels: set[str] = set()
for model in self.models:
attributes.update(model.all_attributes)
attribute_logos.update(model.all_attribute_logos)
labels.update(model.merged_labelmap.values())
for label, label_attributes in model.attributes_map.items():
attributes_map.setdefault(label, set()).update(label_attributes)
self._model_devices = model_devices
self._all_attributes = sorted(attributes)
self._all_attribute_logos = sorted(attribute_logos)
self._all_attributes_map = {
label: sorted(label_attributes)
for label, label_attributes in sorted(attributes_map.items())
}
self._all_labels = labels
def _resolve_camera_model(self, name: str, scene: SceneEnum | None) -> ModelConfig:
"""Resolve which model a camera runs on.
Args:
name: Name of the camera
scene: The camera's configured detect scene, if any
Returns:
The model the camera runs on
"""
by_scene = {model.scene: model for model in self.models}
if scene is not None:
model = by_scene.get(scene)
if model is None:
raise ValueError(
f"Camera '{name}' has a detect scene of '{scene.value}', but no model is configured for that scene."
)
return model
default = by_scene.get(SceneEnum.all) or (
self.models[0] if len(self.models) == 1 else None
)
if default is None:
raise ValueError(
f"Camera '{name}' must set detect -> scene, because more than one model is configured and none of them uses a scene of 'all'."
)
return default
@model_validator(mode="after")
def post_validation(self, info: ValidationInfo) -> Self:
# Load plus api from context, if possible.
@@ -693,8 +884,10 @@ class FrigateConfig(FrigateBaseModel):
"'embeddings' in its roles for semantic search."
)
self._load_models()
# set default min_score for object attributes
for attribute in self.model.all_attributes:
for attribute in self.all_attributes:
existing = self.objects.filters.get(attribute)
if existing is None:
self.objects.filters[attribute] = FilterConfig(min_score=0.7)
@@ -744,44 +937,7 @@ class FrigateConfig(FrigateBaseModel):
exclude_unset=True,
)
for key, detector in self.detectors.items():
adapter = TypeAdapter(DetectorConfig)
model_dict = (
detector
if isinstance(detector, dict)
else detector.model_dump(warnings="none")
)
detector_config: BaseDetectorConfig = adapter.validate_python(model_dict)
# users should not set model themselves
if detector_config.model:
logger.warning(
"The model key should be specified at the root level of the config, not under detectors. The nested model key will be ignored."
)
detector_config.model = None
model_config = self.model.model_dump(exclude_unset=True, warnings="none")
if detector_config.model_path:
model_config["path"] = detector_config.model_path
if "path" not in model_config:
if detector_config.type == "cpu" or detector_config.type.endswith(
"_tfl"
):
model_config["path"] = "/cpu_model.tflite"
elif detector_config.type == "edgetpu":
model_config["path"] = "/edgetpu_model.tflite"
elif detector_config.type == "openvino":
for default_key, default_value in DEFAULT_MODEL.items():
model_config.setdefault(default_key, default_value)
model = ModelConfig.model_validate(model_config)
model.check_and_load_plus_model(self.plus_api, detector_config.type)
model.compute_model_hash()
labelmap_objects = model.merged_labelmap.values()
detector_config.model = model
self.detectors[key] = detector_config
self._camera_models = {}
for name, camera in self.cameras.items():
modified_global_config = global_config.copy()
@@ -808,6 +964,9 @@ class FrigateConfig(FrigateBaseModel):
{"name": name, **merged_config}
)
camera_model = self._resolve_camera_model(name, camera_config.detect.scene)
self._camera_models[name] = camera_model
if camera_config.ffmpeg.hwaccel_args == "auto":
camera_config.ffmpeg.hwaccel_args = self.ffmpeg.hwaccel_args
@@ -1028,7 +1187,7 @@ class FrigateConfig(FrigateBaseModel):
verify_profile_overrides_match_base(camera_config)
verify_autotrack_zones(camera_config)
verify_motion_and_detect(camera_config)
verify_objects_track(camera_config, labelmap_objects)
verify_objects_track(camera_config, camera_model.merged_labelmap.values())
verify_lpr_and_face(self, camera_config)
# Validate camera profiles reference top-level profile definitions
@@ -1045,8 +1204,16 @@ class FrigateConfig(FrigateBaseModel):
config.name = name
self.objects.parse_all_objects(self.cameras)
self.model.create_colormap(sorted(self.objects.all_objects))
self.model.check_and_load_plus_model(self.plus_api)
# every model shares one colormap so a label is drawn the same color no
# matter which model detected it, so filter attributes across all models
# rather than letting each model filter with only its own
colored_labels = sorted(
set(self.objects.all_objects) - set(self.all_attributes)
)
for model in self.models:
model.create_colormap(colored_labels)
# Check audio transcription and audio detection requirements
if self.audio_transcription.enabled:
frigate/data_processing/common/license_plate/mixin.py
+1 -1
View File
@@ -72,7 +72,7 @@ class LicensePlateProcessingMixin:
# Object config
self.lp_objects: list[str] = []
for obj, attributes in self.config.model.attributes_map.items():
for obj, attributes in self.config.all_attributes_map.items():
if "license_plate" in attributes:
self.lp_objects.append(obj)
frigate/data_processing/post/review_descriptions.py
+2 -2
View File
@@ -234,8 +234,8 @@ class ReviewDescriptionProcessor(PostProcessorApi):
final_data,
thumbs,
camera_config.review.genai,
list(self.config.model.merged_labelmap.values()),
self.config.model.all_attributes,
sorted(self.config.all_labels),
self.config.all_attributes,
),
).start()
frigate/detectors/detector_config.py
+34 -5
View File
@@ -3,7 +3,7 @@ import json
import logging
import os
from enum import Enum
from typing import Any
from typing import Any, ClassVar
import requests
from pydantic import BaseModel, ConfigDict, Field
@@ -15,6 +15,9 @@ from frigate.util.builtin import generate_color_palette, load_labels
logger = logging.getLogger(__name__)
# attributes that are recognized rather than shown as a logo
NON_LOGO_ATTRIBUTES = ["face", "license_plate"]
class PixelFormatEnum(str, Enum):
rgb = "rgb"
@@ -44,7 +47,27 @@ 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"
class ModelConfig(BaseModel):
scene: SceneEnum = Field(
default=SceneEnum.all,
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.",
)
devices: list[str] = Field(
default_factory=list,
title="Detection hardware",
description="Hardware this model runs on, as '<detector>' or '<detector>:<device>' (for example 'edgetpu:pci:0' or 'openvino:GPU'). Listing the same device more than once runs additional inference processes on it.",
)
path: str | None = Field(
None,
title="Custom object detector model path",
@@ -111,7 +134,7 @@ class ModelConfig(BaseModel):
@property
def non_logo_attributes(self) -> list[str]:
return ["face", "license_plate"]
return NON_LOGO_ATTRIBUTES
@property
def all_attributes(self) -> list[str]:
@@ -201,9 +224,7 @@ class ModelConfig(BaseModel):
unique_attributes.update(attributes)
self._all_attributes = list(unique_attributes)
self._all_attribute_logos = list(
unique_attributes - set(["face", "license_plate"])
)
self._all_attribute_logos = list(unique_attributes - set(NON_LOGO_ATTRIBUTES))
self._merged_labelmap = {
**{int(key): val for key, val in model_info["labelMap"].items()},
@@ -234,6 +255,14 @@ class ModelConfig(BaseModel):
class BaseDetectorConfig(BaseModel):
# how the trailing part of a device string ("openvino:GPU" -> "GPU") maps onto
# this detector's fields, and whether the same device may be listed more than
# once to run additional inference processes against it. Most accelerators
# multiplex fine, so this is opt-out rather than opt-in.
device_spec_field: ClassVar[str] = "device"
device_spec_type: ClassVar[type] = str
shareable: ClassVar[bool] = True
# the type field must be defined in all subclasses
type: str = Field(
default="cpu",
frigate/detectors/detector_types.py
+19 -1
View File
@@ -2,7 +2,7 @@ import importlib
import logging
import pkgutil
from enum import Enum
from typing import Annotated, Union
from typing import Annotated, Union, get_args
from pydantic import Field
@@ -39,3 +39,21 @@ DetectorConfig = Annotated[
Union[tuple(BaseDetectorConfig.__subclasses__())], # noqa: UP007
Field(discriminator="type"),
]
def _discriminator_value(config_class: type[BaseDetectorConfig]) -> str | None:
"""Read the Literal value of a detector config class' type field."""
field = config_class.model_fields.get("type")
if field is None:
return None
values = get_args(field.annotation)
return values[0] if values else None
config_types: dict[str, type[BaseDetectorConfig]] = {
key: config_class
for config_class in BaseDetectorConfig.__subclasses__()
if (key := _discriminator_value(config_class)) is not None
}
frigate/detectors/device.py
+113
View File
@@ -0,0 +1,113 @@
"""Parsing of detection hardware device strings."""
import logging
from dataclasses import dataclass
from pydantic import TypeAdapter, ValidationError
from frigate.detectors.detector_config import BaseDetectorConfig, ModelConfig
from frigate.detectors.detector_types import DetectorConfig, config_types
logger = logging.getLogger(__name__)
_detector_adapter: TypeAdapter[BaseDetectorConfig] = TypeAdapter(DetectorConfig)
@dataclass(frozen=True)
class DeviceSpec:
"""A parsed `<detector>` or `<detector>:<device>` string."""
raw: str
detector: str
device: str | None
@property
def shareable(self) -> bool:
"""Whether this device may be listed more than once."""
return config_types[self.detector].shareable
class DeviceParseError(ValueError):
pass
def parse_device(raw: str) -> DeviceSpec:
"""Parse a device string into its detector type and detector specific device.
Args:
raw: The configured device string, for example 'edgetpu:pci:0'
Returns:
The parsed spec
Raises:
DeviceParseError: If the detector type is unknown or the device is not
valid for that detector
"""
detector, separator, device = raw.partition(":")
if detector not in config_types:
raise DeviceParseError(
f"'{raw}' does not name a known detector. Available detectors are {', '.join(sorted(config_types))}"
)
spec = DeviceSpec(raw=raw, detector=detector, device=device if separator else None)
# surface a bad device now rather than when the detection process starts
build_detector_config(spec, None)
return spec
def build_detector_config(
spec: DeviceSpec, model: ModelConfig | None
) -> BaseDetectorConfig:
"""Build the detector config a device string describes.
Args:
spec: The parsed device spec
model: The model this detector runs, if it has been resolved yet
Returns:
The validated detector config
Raises:
DeviceParseError: If the device is not valid for this detector type
"""
config: dict[str, object] = {"type": spec.detector, "model": model}
if spec.device is not None:
config_class = config_types[spec.detector]
try:
config[config_class.device_spec_field] = config_class.device_spec_type(
spec.device
)
except ValueError as err:
raise DeviceParseError(
f"'{spec.raw}' is not a valid {spec.detector} device: {err}"
) from err
try:
return _detector_adapter.validate_python(config)
except ValidationError as err:
raise DeviceParseError(f"'{spec.raw}' is not a valid device: {err}") from err
def runner_names(devices: list[DeviceSpec]) -> list[str]:
"""Build a unique name for each device, since a shareable device may repeat.
Args:
devices: Every device spec across every configured model, in config order
Returns:
A name per device, suffixed with '#2', '#3', etc. on repeats
"""
names: list[str] = []
seen: dict[str, int] = {}
for spec in devices:
count = seen.get(spec.raw, 0) + 1
seen[spec.raw] = count
names.append(spec.raw if count == 1 else f"{spec.raw}#{count}")
return names
frigate/detectors/plugins/cpu_tfl.py
+4 -1
View File
@@ -1,5 +1,5 @@
import logging
from typing import Literal
from typing import ClassVar, Literal
from pydantic import ConfigDict, Field
@@ -27,6 +27,9 @@ class CpuDetectorConfig(BaseDetectorConfig):
title="CPU",
)
device_spec_field: ClassVar[str] = "num_threads"
device_spec_type: ClassVar[type] = int
type: Literal[DETECTOR_KEY]
num_threads: int = Field(
default=3,
frigate/detectors/plugins/edgetpu_tfl.py
+4 -1
View File
@@ -1,7 +1,7 @@
import logging
import math
import os
from typing import Literal
from typing import ClassVar, Literal
import cv2
import numpy as np
@@ -28,6 +28,9 @@ class EdgeTpuDetectorConfig(BaseDetectorConfig):
title="EdgeTPU",
)
# a TPU can only be opened by one process
shareable: ClassVar[bool] = False
type: Literal[DETECTOR_KEY]
device: str = Field(
default=None,
frigate/detectors/plugins/memryx.py
+4 -1
View File
@@ -5,7 +5,7 @@ import shutil
import urllib.request
import zipfile
from queue import Queue
from typing import Literal
from typing import ClassVar, Literal
import cv2
import numpy as np
@@ -37,6 +37,9 @@ class MemryXDetectorConfig(BaseDetectorConfig):
title="MemryX",
)
# an accelerator can only be opened by one process
shareable: ClassVar[bool] = False
type: Literal[DETECTOR_KEY]
device: str = Field(
default="PCIe",
frigate/detectors/plugins/openvino.py
+1 -1
View File
@@ -28,7 +28,7 @@ class OvDetectorConfig(BaseDetectorConfig):
type: Literal[DETECTOR_KEY]
device: str = Field(
default=None,
default="AUTO",
title="Device Type",
description="The device to use for OpenVINO inference (e.g. 'CPU', 'GPU', 'NPU').",
)
frigate/detectors/plugins/rknn.py
+4 -1
View File
@@ -2,7 +2,7 @@ import logging
import os.path
import re
import urllib.request
from typing import Literal
from typing import ClassVar, Literal
import cv2
import numpy as np
@@ -35,6 +35,9 @@ class RknnDetectorConfig(BaseDetectorConfig):
title="RKNN",
)
device_spec_field: ClassVar[str] = "num_cores"
device_spec_type: ClassVar[type] = int
type: Literal[DETECTOR_KEY]
num_cores: int = Field(
default=0,
frigate/detectors/plugins/tensorrt.py
+3 -1
View File
@@ -14,7 +14,7 @@ try:
except ModuleNotFoundError:
TRT_SUPPORT = False
from typing import Literal
from typing import ClassVar, Literal
from pydantic import ConfigDict, Field
@@ -53,6 +53,8 @@ class TensorRTDetectorConfig(BaseDetectorConfig):
title="TensorRT",
)
device_spec_type: ClassVar[type] = int
type: Literal[DETECTOR_KEY]
device: int = Field(
default=0, title="GPU Device Index", description="The GPU device index to use."
frigate/events/maintainer.py
+5 -8
View File
@@ -159,7 +159,8 @@ class EventProcessor(threading.Thread):
if width is None or height is None:
return
first_detector = list(self.config.detectors.values())[0]
camera_model = self.config.model_for_camera(camera)
camera_detector = self.config.devices_for_model(camera_model)[0].detector
start_time = event_data["start_time"]
end_time = (
@@ -229,13 +230,9 @@ class EventProcessor(threading.Thread):
Event.thumbnail: event_data.get("thumbnail"),
Event.has_clip: event_data["has_clip"],
Event.has_snapshot: event_data["has_snapshot"],
Event.model_hash: first_detector.model.model_hash
if first_detector.model
else None,
Event.model_type: first_detector.model.model_type
if first_detector.model
else None,
Event.detector_type: first_detector.type,
Event.model_hash: camera_model.model_hash,
Event.model_type: camera_model.model_type,
Event.detector_type: camera_detector,
Event.data: {
"box": box,
"region": region,
frigate/object_detection/util.py
+13 -1
View File
@@ -5,7 +5,19 @@ import threading
from numpy import ndarray
from frigate.detectors.detector_config import InputTensorEnum
from frigate.detectors.detector_config import InputTensorEnum, ModelConfig
def detection_frame_size(model: ModelConfig) -> int:
"""Get the shared memory size a camera needs to hand frames to a model.
Args:
model: The model the camera runs on
Returns:
Size in bytes of one model input frame
"""
return model.height * model.width * 3
class RequestStore:
frigate/review/maintainer.py
+2 -2
View File
@@ -481,7 +481,7 @@ class ReviewSegmentMaintainer(threading.Thread):
if not object["sub_label"]:
segment.detections[object["id"]] = object["label"]
elif object["sub_label"][0] in self.config.model.all_attributes:
elif object["sub_label"][0] in self.config.all_attributes:
segment.detections[object["id"]] = object["sub_label"][0]
else:
segment.detections[object["id"]] = f"{object['label']}-verified"
@@ -619,7 +619,7 @@ class ReviewSegmentMaintainer(threading.Thread):
for object in activity.get_all_objects():
if not object["sub_label"]:
detections[object["id"]] = object["label"]
elif object["sub_label"][0] in self.config.model.all_attributes:
elif object["sub_label"][0] in self.config.all_attributes:
detections[object["id"]] = object["sub_label"][0]
else:
detections[object["id"]] = f"{object['label']}-verified"
frigate/stats/util.py
+14 -13
View File
@@ -322,19 +322,20 @@ async def set_gpu_stats(
async def set_npu_usages(config: FrigateConfig, all_stats: dict[str, Any]) -> None:
stats: dict[str, dict] = {}
for detector in config.detectors.values():
if detector.type == "rknn":
# Rockchip NPU usage
rk_usage = get_rockchip_npu_stats()
stats["rockchip"] = rk_usage
elif detector.type == "openvino" and detector.device == "NPU":
# OpenVINO NPU usage
ov_usage = get_openvino_npu_stats()
stats["openvino"] = ov_usage
elif detector.type == "axengine":
# AXERA NPU usage
axcl_usage = get_axcl_npu_stats()
stats["axengine"] = axcl_usage
for model in config.models:
for device in config.devices_for_model(model):
if device.detector == "rknn":
# Rockchip NPU usage
rk_usage = get_rockchip_npu_stats()
stats["rockchip"] = rk_usage
elif device.detector == "openvino" and device.device == "NPU":
# OpenVINO NPU usage
ov_usage = get_openvino_npu_stats()
stats["openvino"] = ov_usage
elif device.detector == "axengine":
# AXERA NPU usage
axcl_usage = get_axcl_npu_stats()
stats["axengine"] = axcl_usage
if stats:
all_stats["npu_usages"] = stats
frigate/test/test_config.py
+160 -37
View File
@@ -11,6 +11,8 @@ 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
@@ -65,49 +67,171 @@ class TestConfig(unittest.TestCase):
def test_config_class(self):
frigate_config = FrigateConfig(**self.minimal)
assert "cpu" in frigate_config.detectors.keys()
assert frigate_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert frigate_config.detectors["cpu"].model.width == 320
model = frigate_config.primary_model
assert model.scene == SceneEnum.all
assert model.width == 320
assert frigate_config.devices_for_model(model)[0].detector == (
DetectorTypeEnum.cpu
)
@patch("frigate.detectors.detector_config.load_labels")
def test_detector_custom_model_path(self, mock_labels):
def test_model_custom_path(self, mock_labels):
mock_labels.return_value = {}
config = {
"detectors": {
"cpu": {
"type": "cpu",
"model_path": "/cpu_model.tflite",
"models": [
# needs to be a file that will exist, doesn't matter what
{"path": "/etc/hosts", "width": 512, "devices": ["openvino:GPU"]},
],
}
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
model = frigate_config.primary_model
assert model.path == "/etc/hosts"
assert model.width == 512
detector_config = build_detector_config(
frigate_config.devices_for_model(model)[0], model
)
assert detector_config.type == DetectorTypeEnum.openvino
assert detector_config.device == "GPU"
assert detector_config.model.path == "/etc/hosts"
@patch("frigate.detectors.detector_config.load_labels")
def test_model_default_paths_per_detector(self, mock_labels):
mock_labels.return_value = {}
for devices, expected in (
(["cpu"], "/cpu_model.tflite"),
(["edgetpu:pci:0"], "/edgetpu_model.tflite"),
(["openvino:CPU"], "/openvino-model/ssdlite_mobilenet_v2.xml"),
):
config = {"models": [{"devices": devices}]}
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
assert frigate_config.primary_model.path == expected
@patch("frigate.detectors.detector_config.load_labels")
def test_camera_picks_model_by_scene(self, mock_labels):
mock_labels.return_value = {}
config = {
"models": [
{"scene": "outdoor", "devices": ["cpu"], "width": 320},
{"scene": "indoor", "devices": ["openvino:CPU"], "width": 300},
],
"cameras": {
"back": {
"detect": {"scene": "outdoor"},
"ffmpeg": {
"inputs": [
{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]},
]
},
},
"edgetpu": {
"type": "edgetpu",
"model_path": "/edgetpu_model.tflite",
},
"openvino": {
"type": "openvino",
"front": {
"detect": {"scene": "indoor"},
"ffmpeg": {
"inputs": [
{"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]},
]
},
},
},
# needs to be a file that will exist, doesn't matter what
"model": {"path": "/etc/hosts", "width": 512},
}
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
assert "cpu" in frigate_config.detectors.keys()
assert "edgetpu" in frigate_config.detectors.keys()
assert "openvino" in frigate_config.detectors.keys()
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").width == 320
assert frigate_config.model_for_camera("front").width == 300
assert frigate_config.detectors["cpu"].type == DetectorTypeEnum.cpu
assert frigate_config.detectors["edgetpu"].type == DetectorTypeEnum.edgetpu
assert frigate_config.detectors["openvino"].type == DetectorTypeEnum.openvino
@patch("frigate.detectors.detector_config.load_labels")
def test_camera_requires_a_scene_without_a_default(self, mock_labels):
mock_labels.return_value = {}
config = {
"models": [
{"scene": "outdoor", "devices": ["cpu"]},
{"scene": "indoor", "devices": ["openvino:CPU"]},
],
}
assert frigate_config.detectors["cpu"].num_threads == 3
assert frigate_config.detectors["edgetpu"].device is None
assert frigate_config.detectors["openvino"].device is None
with self.assertRaises(ValidationError):
FrigateConfig(**(deep_merge(config, self.minimal)))
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"
@patch("frigate.detectors.detector_config.load_labels")
def test_camera_scene_must_match_a_model(self, mock_labels):
mock_labels.return_value = {}
config = {
"models": [{"devices": ["cpu"]}],
"cameras": {
"back": {
"detect": {"scene": "outdoor"},
"ffmpeg": {
"inputs": [
{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]},
]
},
},
},
}
with self.assertRaises(ValidationError):
FrigateConfig(**(deep_merge(config, self.minimal)))
@patch("frigate.detectors.detector_config.load_labels")
def test_models_must_use_unique_scenes(self, mock_labels):
mock_labels.return_value = {}
config = {
"models": [
{"scene": "outdoor", "devices": ["cpu"]},
{"scene": "outdoor", "devices": ["openvino:CPU"]},
],
}
with self.assertRaises(ValidationError):
FrigateConfig(**(deep_merge(config, self.minimal)))
@patch("frigate.detectors.detector_config.load_labels")
def test_model_devices_must_share_a_detector(self, mock_labels):
mock_labels.return_value = {}
config = {"models": [{"devices": ["cpu", "openvino:CPU"]}]}
with self.assertRaises(ValidationError):
FrigateConfig(**(deep_merge(config, self.minimal)))
@patch("frigate.detectors.detector_config.load_labels")
def test_model_requires_a_known_detector(self, mock_labels):
mock_labels.return_value = {}
config = {"models": [{"devices": ["not_a_detector:0"]}]}
with self.assertRaises(ValidationError):
FrigateConfig(**(deep_merge(config, self.minimal)))
@patch("frigate.detectors.detector_config.load_labels")
def test_model_requires_a_device(self, mock_labels):
mock_labels.return_value = {}
config = {"models": [{"devices": []}]}
with self.assertRaises(ValidationError):
FrigateConfig(**(deep_merge(config, self.minimal)))
@patch("frigate.detectors.detector_config.load_labels")
def test_shareable_devices_may_repeat(self, mock_labels):
mock_labels.return_value = {}
config = {"models": [{"devices": ["openvino:GPU", "openvino:GPU"]}]}
frigate_config = FrigateConfig(**(deep_merge(config, self.minimal)))
devices = frigate_config.devices_for_model(frigate_config.primary_model)
assert runner_names(devices) == ["openvino:GPU", "openvino:GPU#2"]
@patch("frigate.detectors.detector_config.load_labels")
def test_exclusive_devices_may_not_repeat(self, mock_labels):
mock_labels.return_value = {}
config = {"models": [{"devices": ["edgetpu:pci:0", "edgetpu:pci:0"]}]}
with self.assertRaises(ValidationError):
FrigateConfig(**(deep_merge(config, self.minimal)))
def test_invalid_mqtt_config(self):
config = {
@@ -1131,7 +1255,7 @@ class TestConfig(unittest.TestCase):
def test_merge_labelmap(self):
config = {
"mqtt": {"host": "mqtt"},
"model": {"labelmap": {7: "truck"}},
"models": [{"labelmap": {7: "truck"}, "devices": ["cpu"]}],
"cameras": {
"back": {
"ffmpeg": {
@@ -1152,7 +1276,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert frigate_config.model.merged_labelmap[7] == "truck"
assert frigate_config.primary_model.merged_labelmap[7] == "truck"
def test_audio_labelmap_inheritance_is_separate_from_model_labelmap(self):
config = deep_merge(
@@ -1199,12 +1323,12 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert frigate_config.model.merged_labelmap[0] == "person"
assert frigate_config.primary_model.merged_labelmap[0] == "person"
def test_default_labelmap(self):
config = {
"mqtt": {"host": "mqtt"},
"model": {"width": 320, "height": 320},
"models": [{"width": 320, "height": 320, "devices": ["cpu"]}],
"cameras": {
"back": {
"ffmpeg": {
@@ -1225,7 +1349,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert frigate_config.model.merged_labelmap[0] == "person"
assert frigate_config.primary_model.merged_labelmap[0] == "person"
def test_plus_labelmap(self):
with open(os.path.join(MODEL_CACHE_DIR, "test"), "w") as f:
@@ -1235,8 +1359,7 @@ class TestConfig(unittest.TestCase):
config = {
"mqtt": {"host": "mqtt"},
"detectors": {"cpu": {"type": "cpu"}},
"model": {"path": "plus://test"},
"models": [{"path": "plus://test", "devices": ["cpu"]}],
"cameras": {
"back": {
"ffmpeg": {
@@ -1257,7 +1380,7 @@ class TestConfig(unittest.TestCase):
}
frigate_config = FrigateConfig(**config)
assert frigate_config.model.merged_labelmap[0] == "amazon"
assert frigate_config.primary_model.merged_labelmap[0] == "amazon"
def test_fails_on_invalid_role(self):
config = {
frigate/test/test_config_migration.py
+225
View File
@@ -0,0 +1,225 @@
"""Tests for migrating detectors and model into the models list."""
import logging
import os
import tempfile
import unittest
from unittest.mock import patch
from ruamel.yaml import YAML
from frigate.util.config import (
CURRENT_CONFIG_VERSION,
migrate_frigate_config,
migrate_models,
)
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.assertNotIn("detectors", migrated)
def test_model_settings_are_carried_over(self):
migrated = migrate_models(
{
"detectors": {"coral": {"type": "edgetpu", "device": "pci:0"}},
"model": {"path": "plus://abc", "width": 320},
}
)
self.assertEqual(
migrated["models"],
[
{
"scene": "all",
"path": "plus://abc",
"width": 320,
"devices": ["edgetpu:pci:0"],
}
],
)
self.assertNotIn("model", migrated)
def test_multiple_corals_become_multiple_devices(self):
migrated = migrate_models(
{
"detectors": {
"coral1": {"type": "edgetpu", "device": "pci:0"},
"coral2": {"type": "edgetpu", "device": "pci:1"},
}
}
)
self.assertEqual(
migrated["models"][0]["devices"], ["edgetpu:pci:0", "edgetpu:pci:1"]
)
def test_several_detectors_on_one_device_stay_separate(self):
# a repeated device is now what running two inference processes on one
# piece of hardware looks like
migrated = migrate_models(
{
"detectors": {
"ov_0": {"type": "openvino", "device": "GPU"},
"ov_1": {"type": "openvino", "device": "GPU"},
}
}
)
self.assertEqual(
migrated["models"][0]["devices"], ["openvino:GPU", "openvino:GPU"]
)
def test_repeated_exclusive_devices_are_collapsed(self):
# two detectors both grabbing the first TPU was never really two TPUs
migrated = migrate_models(
{
"detectors": {
"coral_0": {"type": "edgetpu", "device": "usb"},
"coral_1": {"type": "edgetpu", "device": "usb"},
}
}
)
self.assertEqual(migrated["models"][0]["devices"], ["edgetpu:usb"])
def test_detectors_that_named_the_device_field_differently(self):
migrated = migrate_models(
{
"detectors": {
"rk": {"type": "rknn", "num_cores": 2},
}
}
)
self.assertEqual(migrated["models"][0]["devices"], ["rknn:2"])
def test_empty_edgetpu_device_is_kept(self):
# an empty device selects a native Coral, which is not the same as
# letting the delegate pick
migrated = migrate_models(
{"detectors": {"coral": {"type": "edgetpu", "device": ""}}}
)
self.assertEqual(migrated["models"][0]["devices"], ["edgetpu:"])
def test_model_path_overrides_the_model(self):
migrated = migrate_models(
{
"detectors": {
"coral": {"type": "edgetpu", "model_path": "/custom.tflite"}
},
"model": {"path": "/ignored.tflite", "width": 320},
}
)
self.assertEqual(migrated["models"][0]["path"], "/custom.tflite")
def test_no_detectors_falls_back_to_cpu(self):
migrated = migrate_models({"model": {"width": 320}})
self.assertEqual(migrated["models"][0]["devices"], ["cpu"])
def test_dropped_remote_detector_options_are_logged(self):
with self.assertLogs("frigate.util.config", level=logging.ERROR) as logs:
migrated = migrate_models(
{
"detectors": {
"ds": {
"type": "deepstack",
"api_url": "http://host:5000/v1/vision/detection",
"api_key": "secret",
}
}
}
)
self.assertEqual(
migrated["models"][0]["devices"],
["deepstack:http://host:5000/v1/vision/detection"],
)
self.assertTrue(any("api_key" in message for message in logs.output))
def test_mixed_detector_types_are_logged(self):
with self.assertLogs("frigate.util.config", level=logging.ERROR) as logs:
migrate_models(
{
"detectors": {
"ov": {"type": "openvino", "device": "GPU"},
"coral": {"type": "edgetpu", "device": "pci:0"},
}
}
)
self.assertTrue(any("more than one type" in message for message in logs.output))
def test_other_keys_are_untouched(self):
migrated = migrate_models(
{"mqtt": {"host": "mqtt"}, "detectors": {"cpu": {"type": "cpu"}}}
)
self.assertEqual(migrated["mqtt"], {"host": "mqtt"})
class TestMigrateConfigFile(unittest.TestCase):
"""The full file migration, which is gated on shape as well as version."""
def setUp(self):
self.temp_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.temp_dir.cleanup)
self.config_file = os.path.join(self.temp_dir.name, "config.yml")
patcher = patch("frigate.util.config.CONFIG_DIR", self.temp_dir.name)
patcher.start()
self.addCleanup(patcher.stop)
def _migrate(self, config: str) -> dict:
with open(self.config_file, "w") as f:
f.write(config)
migrate_frigate_config(self.config_file)
with open(self.config_file) as f:
return YAML().load(f)
def test_migrates_a_config_already_stamped_with_the_current_version(self):
# 0.19 is unreleased, so a dev config can be current and still use
# the pre-models keys
migrated = self._migrate(
"mqtt:\n"
" enabled: false\n"
"detectors:\n"
" ov:\n"
" type: openvino\n"
" device: GPU\n"
"cameras: {}\n"
f"version: {CURRENT_CONFIG_VERSION}\n"
)
self.assertEqual(migrated["models"][0]["devices"], ["openvino:GPU"])
self.assertNotIn("detectors", migrated)
def test_a_migrated_config_is_left_alone(self):
migrated = self._migrate(
"mqtt:\n"
" enabled: false\n"
"models:\n"
" - scene: all\n"
" devices:\n"
" - openvino:GPU\n"
"cameras: {}\n"
f"version: {CURRENT_CONFIG_VERSION}\n"
)
self.assertEqual(
migrated["models"], [{"scene": "all", "devices": ["openvino:GPU"]}]
)
self.assertFalse(
os.path.exists(os.path.join(self.temp_dir.name, "backup_config.yaml"))
)
if __name__ == "__main__":
unittest.main(verbosity=2)
frigate/test/test_detector_device.py
+94
View File
@@ -0,0 +1,94 @@
"""Tests for parsing detection hardware device strings."""
import unittest
from frigate.detectors.detector_config import ModelConfig
from frigate.detectors.device import (
DeviceParseError,
build_detector_config,
parse_device,
runner_names,
)
class TestParseDevice(unittest.TestCase):
def test_bare_detector_has_no_device(self):
spec = parse_device("cpu")
self.assertEqual(spec.detector, "cpu")
self.assertIsNone(spec.device)
def test_device_is_everything_after_the_first_colon(self):
spec = parse_device("edgetpu:pci:0")
self.assertEqual(spec.detector, "edgetpu")
self.assertEqual(spec.device, "pci:0")
def test_trailing_colon_keeps_an_empty_device(self):
# an empty edgetpu device selects a native Coral
spec = parse_device("edgetpu:")
self.assertEqual(spec.detector, "edgetpu")
self.assertEqual(spec.device, "")
def test_unknown_detector_is_rejected(self):
with self.assertRaises(DeviceParseError):
parse_device("not_a_detector:0")
def test_device_that_the_detector_cannot_use_is_rejected(self):
# tensorrt takes a gpu index
with self.assertRaises(DeviceParseError):
parse_device("tensorrt:the-fast-one")
class TestBuildDetectorConfig(unittest.TestCase):
def _build(self, raw: str):
return build_detector_config(parse_device(raw), ModelConfig())
def test_device_lands_on_the_detector_field(self):
for raw, expected in (
("edgetpu:usb", "usb"),
("edgetpu:pci:1", "pci:1"),
("openvino:GPU.1", "GPU.1"),
("onnx:CPU", "CPU"),
("memryx:PCIe:0", "PCIe:0"),
):
with self.subTest(raw=raw):
self.assertEqual(self._build(raw).device, expected)
def test_detectors_that_name_the_field_something_else(self):
self.assertEqual(self._build("cpu:4").num_threads, 4)
self.assertEqual(self._build("rknn:2").num_cores, 2)
def test_device_is_coerced_to_the_detector_field_type(self):
self.assertEqual(self._build("tensorrt:1").device, 1)
def test_omitted_device_falls_back_to_the_detector_default(self):
self.assertEqual(self._build("cpu").num_threads, 3)
self.assertEqual(self._build("rknn").num_cores, 0)
self.assertEqual(self._build("openvino").device, "AUTO")
self.assertIsNone(self._build("edgetpu").device)
def test_the_model_is_attached(self):
model = ModelConfig(path="/cpu_model.tflite")
self.assertIs(build_detector_config(parse_device("cpu"), model).model, model)
class TestRunnerNames(unittest.TestCase):
def test_unique_devices_keep_their_name(self):
devices = [parse_device("edgetpu:pci:0"), parse_device("edgetpu:pci:1")]
self.assertEqual(runner_names(devices), ["edgetpu:pci:0", "edgetpu:pci:1"])
def test_repeated_devices_are_numbered(self):
devices = [parse_device("openvino:GPU")] * 3
self.assertEqual(
runner_names(devices),
["openvino:GPU", "openvino:GPU#2", "openvino:GPU#3"],
)
if __name__ == "__main__":
unittest.main(verbosity=2)
frigate/track/object_processing.py
+1 -1
View File
@@ -210,7 +210,7 @@ class TrackedObjectProcessor(threading.Thread):
if obj.obj_data.get("sub_label"):
sub_label = obj.obj_data["sub_label"][0]
if sub_label in self.config.model.all_attribute_logos:
if sub_label in self.config.all_attribute_logos:
self.dispatcher.publish(
f"{camera}/{sub_label}/snapshot",
jpg_bytes,
frigate/util/config.py
+113 -1
View File
@@ -23,6 +23,25 @@ logger = logging.getLogger(__name__)
CURRENT_CONFIG_VERSION = "0.19-0"
DEFAULT_CONFIG_FILE = os.path.join(CONFIG_DIR, "config.yml")
# the detector field that used to hold the device, for detectors that named it
# something other than "device"
DETECTOR_DEVICE_FIELDS = {
"cpu": "num_threads",
"rknn": "num_cores",
"deepstack": "api_url",
"degirum": "location",
"zmq": "endpoint",
}
# detector options that have no equivalent in a device string. The remote
# detectors that use them are being reworked, so they are dropped rather than
# carried over.
DROPPED_DETECTOR_OPTIONS = {
"deepstack": ["api_timeout", "api_key"],
"degirum": ["zoo", "token"],
"zmq": ["request_timeout_ms", "linger_ms"],
}
def resolve_ffmpeg_path(path: str, binary: str = "ffmpeg") -> str:
"""Resolve an ffmpeg version alias or custom path to a binary path.
@@ -87,7 +106,11 @@ def migrate_frigate_config(config_file: str):
previous_version = str(config.get("version", "0.13"))
if previous_version == CURRENT_CONFIG_VERSION:
# 0.19 is unreleased, so a config may already be stamped with the current
# version and still use the pre-models detectors and model keys
needs_models = "detectors" in config or "model" in config
if previous_version == CURRENT_CONFIG_VERSION and not needs_models:
logger.info("frigate config does not need migration...")
return
@@ -155,6 +178,12 @@ def migrate_frigate_config(config_file: str):
yaml.dump(new_config, f)
previous_version = "0.19-0"
if needs_models:
logger.info("Migrating frigate detectors and model to models...")
new_config = migrate_models(new_config)
with open(config_file, "w") as f:
yaml.dump(new_config, f)
logger.info("Finished frigate config migration...")
@@ -708,6 +737,89 @@ def migrate_019_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]
return new_config
def migrate_models(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:
"""Merge the detectors and model keys into a single models list.
Every config before this change ran one model across all of its detectors,
so this always produces exactly one model.
Args:
config: The loaded config
Returns:
The config with a models list in place of detectors and model
"""
# imported lazily so loading the detector plugins is not a cost of importing
# this module
from frigate.detectors.detector_types import config_types
new_config = config.copy()
detectors: dict[str, Any] = new_config.pop("detectors", None) or {}
model: dict[str, Any] = new_config.pop("model", None) or {}
devices: list[str] = []
model_path: str | None = None
for name, detector in detectors.items():
detector = detector or {}
detector_type = detector.get("type", "cpu")
device = detector.get(DETECTOR_DEVICE_FIELDS.get(detector_type, "device"))
device_string = detector_type if device is None else f"{detector_type}:{device}"
# repeating a device now means running an extra inference process on it,
# which is what several detectors on one device used to mean. Only
# collapse repeats of hardware that can serve a single process.
config_class = config_types.get(detector_type)
shareable = config_class.shareable if config_class else True
if shareable or device_string not in devices:
devices.append(device_string)
dropped = [
option
for option in DROPPED_DETECTOR_OPTIONS.get(detector_type, [])
if option in detector
]
if dropped:
logger.error(
"Detector '%s' had the %s options set, which are no longer supported and have been removed",
name,
", ".join(dropped),
)
detector_model_path = detector.get("model_path")
if detector_model_path:
if model_path is None:
model_path = detector_model_path
elif model_path != detector_model_path:
logger.warning(
"Detector '%s' set a different model_path than an earlier detector, using '%s' for the migrated model",
name,
model_path,
)
detector_types = {device.partition(":")[0] for device in devices}
if len(detector_types) > 1:
logger.error(
"Detectors of more than one type (%s) were configured. A model now runs on one detector type, so the migrated config will need to be corrected by hand",
", ".join(sorted(detector_types)),
)
entry: dict[str, Any] = {"scene": "all", **model}
if model_path:
entry["path"] = model_path
# a config with no detectors ran a single cpu detector
entry["devices"] = devices or ["cpu"]
new_config["models"] = [entry]
return new_config
def get_relative_coordinates(
mask: str | list | None,
frame_shape: tuple[int, int],
frigate/util/object_names.py
+3 -3
View File
@@ -47,10 +47,10 @@ def get_categorized_object_names(
"""
tracked_objects = _get_tracked_objects(config, allowed_cameras)
names: dict[str, set[str]] = {}
logos = set(config.model.all_attribute_logos)
logos = set(config.all_attribute_logos)
# 1. detector logo attributes, only for objects that are actually tracked
for label, label_attributes in config.model.attributes_map.items():
for label, label_attributes in config.all_attributes_map.items():
if label not in tracked_objects:
continue
@@ -126,7 +126,7 @@ def _objects_with_attribute(
"""
objects = {
label
for label, label_attributes in config.model.attributes_map.items()
for label, label_attributes in config.all_attributes_map.items()
if attribute in label_attributes and label in tracked_objects
}
frigate/util/schema.py
+9 -38
View File
@@ -2,45 +2,16 @@
from typing import Any
from pydantic import BaseModel, TypeAdapter
from pydantic import BaseModel
def get_config_schema(config_class: type[BaseModel]) -> dict[str, Any]:
"""Get the JSON schema for FrigateConfig.
Args:
config_class: The config model to describe
Returns:
The JSON schema
"""
Returns the JSON schema for FrigateConfig with polymorphic detectors.
This utility patches the FrigateConfig schema to include the full polymorphic
definitions for detectors. By default, Pydantic's schema for Dict[str, BaseDetectorConfig]
only includes the base class fields. This function replaces it with a reference
to the DetectorConfig union, which includes all available detector subclasses.
"""
# Import here to ensure all detector plugins are loaded through the detectors module
from frigate.detectors import DetectorConfig
# Get the base schema for FrigateConfig
schema = config_class.model_json_schema()
# Get the schema for the polymorphic DetectorConfig union
detector_adapter: TypeAdapter = TypeAdapter(DetectorConfig)
detector_schema = detector_adapter.json_schema()
# Ensure $defs exists in FrigateConfig schema
if "$defs" not in schema:
schema["$defs"] = {}
# Merge $defs from DetectorConfig into FrigateConfig schema
# This includes the specific schemas for each detector plugin (OvDetectorConfig, etc.)
if "$defs" in detector_schema:
schema["$defs"].update(detector_schema["$defs"])
# Extract the union schema (oneOf/discriminator) and add it as a definition
detector_union_schema = {k: v for k, v in detector_schema.items() if k != "$defs"}
schema["$defs"]["DetectorConfig"] = detector_union_schema
# Update the 'detectors' property to use the polymorphic DetectorConfig definition
if "detectors" in schema.get("properties", {}):
schema["properties"]["detectors"]["additionalProperties"] = {
"$ref": "#/$defs/DetectorConfig"
}
return schema
return config_class.model_json_schema()