diff --git a/frigate/config/camera/mask.py b/frigate/config/camera/mask.py new file mode 100644 index 000000000..caba8aa6e --- /dev/null +++ b/frigate/config/camera/mask.py @@ -0,0 +1,73 @@ +"""Mask configuration for motion and object masks.""" + +from typing import Any, Optional, Union + +from pydantic import Field, field_serializer + +from ..base import FrigateBaseModel + +__all__ = ["MotionMaskConfig", "ObjectMaskConfig"] + + +class MotionMaskConfig(FrigateBaseModel): + """Configuration for a single motion mask.""" + + friendly_name: Optional[str] = Field( + default=None, + title="Motion mask friendly name used in the Frigate UI.", + ) + enabled: bool = Field( + default=True, + title="Enable this motion mask.", + ) + coordinates: Union[str, list[str]] = Field( + default="", + title="Coordinates polygon for the motion mask.", + ) + raw_coordinates: Union[str, list[str]] = "" + + def get_formatted_name(self, mask_id: str) -> str: + """Return the friendly name if set, otherwise return a formatted version of the mask ID.""" + if self.friendly_name: + return self.friendly_name + return mask_id.replace("_", " ").title() + + @field_serializer("coordinates", when_used="json") + def serialize_coordinates(self, value: Any, info): + return self.raw_coordinates + + @field_serializer("raw_coordinates", when_used="json") + def serialize_raw_coordinates(self, value: Any, info): + return None + + +class ObjectMaskConfig(FrigateBaseModel): + """Configuration for a single object mask.""" + + friendly_name: Optional[str] = Field( + default=None, + title="Object mask friendly name used in the Frigate UI.", + ) + enabled: bool = Field( + default=True, + title="Enable this object mask.", + ) + coordinates: Union[str, list[str]] = Field( + default="", + title="Coordinates polygon for the object mask.", + ) + raw_coordinates: Union[str, list[str]] = "" + + def get_formatted_name(self, mask_id: str) -> str: + """Return the friendly name if set, otherwise return a formatted version of the mask ID.""" + if self.friendly_name: + return self.friendly_name + return mask_id.replace("_", " ").title() + + @field_serializer("coordinates", when_used="json") + def serialize_coordinates(self, value: Any, info): + return self.raw_coordinates + + @field_serializer("raw_coordinates", when_used="json") + def serialize_raw_coordinates(self, value: Any, info): + return None diff --git a/frigate/config/camera/motion.py b/frigate/config/camera/motion.py index d39130108..b6877693b 100644 --- a/frigate/config/camera/motion.py +++ b/frigate/config/camera/motion.py @@ -1,8 +1,9 @@ -from typing import Any, Optional, Union +from typing import Any, Optional from pydantic import Field, field_serializer from ..base import FrigateBaseModel +from .mask import MotionMaskConfig __all__ = ["MotionConfig"] @@ -52,8 +53,8 @@ class MotionConfig(FrigateBaseModel): title="Frame height", description="Height in pixels to scale frames to when computing motion.", ) - mask: Union[str, list[str]] = Field( - default="", + mask: dict[str, Optional[MotionMaskConfig]] = Field( + default_factory=dict, title="Mask coordinates", description="Ordered x,y coordinates defining the motion mask polygon used to include/exclude areas.", ) @@ -67,11 +68,15 @@ class MotionConfig(FrigateBaseModel): title="Original motion state", description="Indicates whether motion detection was enabled in the original static configuration.", ) - raw_mask: Union[str, list[str]] = "" + raw_mask: dict[str, Optional[MotionMaskConfig]] = Field( + default_factory=dict, exclude=True + ) @field_serializer("mask", when_used="json") def serialize_mask(self, value: Any, info): - return self.raw_mask + if self.raw_mask: + return self.raw_mask + return value @field_serializer("raw_mask", when_used="json") def serialize_raw_mask(self, value: Any, info): diff --git a/frigate/config/camera/objects.py b/frigate/config/camera/objects.py index 97a4d5b7c..e93778f23 100644 --- a/frigate/config/camera/objects.py +++ b/frigate/config/camera/objects.py @@ -3,6 +3,7 @@ from typing import Any, Optional, Union from pydantic import Field, PrivateAttr, field_serializer, field_validator from ..base import FrigateBaseModel +from .mask import ObjectMaskConfig __all__ = ["ObjectConfig", "GenAIObjectConfig", "FilterConfig"] @@ -41,16 +42,20 @@ class FilterConfig(FrigateBaseModel): title="Minimum confidence", description="Minimum single-frame detection confidence required for the object to be counted.", ) - mask: Optional[Union[str, list[str]]] = Field( - default=None, + mask: dict[str, Optional[ObjectMaskConfig]] = Field( + default_factory=dict, title="Filter mask", description="Polygon coordinates defining where this filter applies within the frame.", ) - raw_mask: Union[str, list[str]] = "" + raw_mask: dict[str, Optional[ObjectMaskConfig]] = Field( + default_factory=dict, exclude=True + ) @field_serializer("mask", when_used="json") def serialize_mask(self, value: Any, info): - return self.raw_mask + if self.raw_mask: + return self.raw_mask + return value @field_serializer("raw_mask", when_used="json") def serialize_raw_mask(self, value: Any, info): @@ -139,11 +144,14 @@ class ObjectConfig(FrigateBaseModel): title="Object filters", description="Filters applied to detected objects to reduce false positives (area, ratio, confidence).", ) - mask: Union[str, list[str]] = Field( - default="", + mask: dict[str, Optional[ObjectMaskConfig]] = Field( + default_factory=dict, title="Object mask", description="Mask polygon used to prevent object detection in specified areas.", ) + raw_mask: dict[str, Optional[ObjectMaskConfig]] = Field( + default_factory=dict, exclude=True + ) genai: GenAIObjectConfig = Field( default_factory=GenAIObjectConfig, title="GenAI object config", @@ -166,3 +174,13 @@ class ObjectConfig(FrigateBaseModel): enabled_labels.update(camera.objects.track) self._all_objects = list(enabled_labels) + + @field_serializer("mask", when_used="json") + def serialize_mask(self, value: Any, info): + if self.raw_mask: + return self.raw_mask + return value + + @field_serializer("raw_mask", when_used="json") + def serialize_raw_mask(self, value: Any, info): + return None diff --git a/frigate/config/config.py b/frigate/config/config.py index 3934976d3..1bd61fbfd 100644 --- a/frigate/config/config.py +++ b/frigate/config/config.py @@ -3,7 +3,7 @@ from __future__ import annotations import json import logging import os -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, Optional import numpy as np from pydantic import ( @@ -93,54 +93,111 @@ stream_info_retriever = StreamInfoRetriever() class RuntimeMotionConfig(MotionConfig): - raw_mask: Union[str, List[str]] = "" - mask: np.ndarray = None + """Runtime version of MotionConfig with rasterized masks.""" + + # The rasterized numpy mask (combination of all enabled masks) + rasterized_mask: np.ndarray = None def __init__(self, **config): frame_shape = config.get("frame_shape", (1, 1)) - mask = get_relative_coordinates(config.get("mask", ""), frame_shape) - config["raw_mask"] = mask - - if mask: - config["mask"] = create_mask(frame_shape, mask) - else: - empty_mask = np.zeros(frame_shape, np.uint8) - empty_mask[:] = 255 - config["mask"] = empty_mask + # Store original mask dict for serialization + original_mask = config.get("mask", {}) + if isinstance(original_mask, dict): + # Process the new dict format - update raw_coordinates for each mask + processed_mask = {} + for mask_id, mask_config in original_mask.items(): + if isinstance(mask_config, dict): + coords = mask_config.get("coordinates", "") + relative_coords = get_relative_coordinates(coords, frame_shape) + mask_config_copy = mask_config.copy() + mask_config_copy["raw_coordinates"] = ( + relative_coords if relative_coords else coords + ) + mask_config_copy["coordinates"] = ( + relative_coords if relative_coords else coords + ) + processed_mask[mask_id] = mask_config_copy + else: + processed_mask[mask_id] = mask_config + config["mask"] = processed_mask + config["raw_mask"] = processed_mask super().__init__(**config) + # Rasterize only enabled masks + enabled_coords = [] + for mask_config in self.mask.values(): + if mask_config.enabled and mask_config.coordinates: + coords = mask_config.coordinates + if isinstance(coords, list): + enabled_coords.extend(coords) + else: + enabled_coords.append(coords) + + if enabled_coords: + self.rasterized_mask = create_mask(frame_shape, enabled_coords) + else: + empty_mask = np.zeros(frame_shape, np.uint8) + empty_mask[:] = 255 + self.rasterized_mask = empty_mask + def dict(self, **kwargs): ret = super().model_dump(**kwargs) - if "mask" in ret: - ret["mask"] = ret["raw_mask"] - ret.pop("raw_mask") + if "rasterized_mask" in ret: + ret.pop("rasterized_mask") return ret - @field_serializer("mask", when_used="json") - def serialize_mask(self, value: Any, info): - return self.raw_mask - - @field_serializer("raw_mask", when_used="json") - def serialize_raw_mask(self, value: Any, info): + @field_serializer("rasterized_mask", when_used="json") + def serialize_rasterized_mask(self, value: Any, info): return None model_config = ConfigDict(arbitrary_types_allowed=True, extra="ignore") class RuntimeFilterConfig(FilterConfig): - mask: Optional[np.ndarray] = None - raw_mask: Optional[Union[str, List[str]]] = None + """Runtime version of FilterConfig with rasterized masks.""" + + # The rasterized numpy mask (combination of all enabled masks) + rasterized_mask: Optional[np.ndarray] = None def __init__(self, **config): frame_shape = config.get("frame_shape", (1, 1)) - mask = get_relative_coordinates(config.get("mask"), frame_shape) - config["raw_mask"] = mask - - if mask is not None: - config["mask"] = create_mask(frame_shape, mask) + # Store original mask dict for serialization + original_mask = config.get("mask", {}) + if isinstance(original_mask, dict): + # Process the new dict format - update raw_coordinates for each mask + processed_mask = {} + for mask_id, mask_config in original_mask.items(): + # Handle both dict and ObjectMaskConfig formats + if hasattr(mask_config, "model_dump"): + # It's an ObjectMaskConfig object + mask_dict = mask_config.model_dump() + coords = mask_dict.get("coordinates", "") + relative_coords = get_relative_coordinates(coords, frame_shape) + mask_dict["raw_coordinates"] = ( + relative_coords if relative_coords else coords + ) + mask_dict["coordinates"] = ( + relative_coords if relative_coords else coords + ) + processed_mask[mask_id] = mask_dict + elif isinstance(mask_config, dict): + coords = mask_config.get("coordinates", "") + relative_coords = get_relative_coordinates(coords, frame_shape) + mask_config_copy = mask_config.copy() + mask_config_copy["raw_coordinates"] = ( + relative_coords if relative_coords else coords + ) + mask_config_copy["coordinates"] = ( + relative_coords if relative_coords else coords + ) + processed_mask[mask_id] = mask_config_copy + else: + processed_mask[mask_id] = mask_config + config["mask"] = processed_mask + config["raw_mask"] = processed_mask # Convert min_area and max_area to pixels if they're percentages if "min_area" in config: @@ -151,13 +208,31 @@ class RuntimeFilterConfig(FilterConfig): super().__init__(**config) + # Rasterize only enabled masks + enabled_coords = [] + for mask_config in self.mask.values(): + if mask_config.enabled and mask_config.coordinates: + coords = mask_config.coordinates + if isinstance(coords, list): + enabled_coords.extend(coords) + else: + enabled_coords.append(coords) + + if enabled_coords: + self.rasterized_mask = create_mask(frame_shape, enabled_coords) + else: + self.rasterized_mask = None + def dict(self, **kwargs): ret = super().model_dump(**kwargs) - if "mask" in ret: - ret["mask"] = ret["raw_mask"] - ret.pop("raw_mask") + if "rasterized_mask" in ret: + ret.pop("rasterized_mask") return ret + @field_serializer("rasterized_mask", when_used="json") + def serialize_rasterized_mask(self, value: Any, info): + return None + model_config = ConfigDict(arbitrary_types_allowed=True, extra="ignore") @@ -715,31 +790,23 @@ class FrigateConfig(FrigateBaseModel): # Apply global object masks and convert masks to numpy array for object, filter in camera_config.objects.filters.items(): + # Merge global object masks with per-object filter masks + merged_mask = dict(filter.mask) # Copy filter-specific masks + + # Add global object masks if they exist if camera_config.objects.mask: - filter_mask = [] - if filter.mask is not None: - filter_mask = ( - filter.mask - if isinstance(filter.mask, list) - else [filter.mask] - ) - object_mask = ( - get_relative_coordinates( - ( - camera_config.objects.mask - if isinstance(camera_config.objects.mask, list) - else [camera_config.objects.mask] - ), - camera_config.frame_shape, - ) - or [] - ) - filter.mask = filter_mask + object_mask + for mask_id, mask_config in camera_config.objects.mask.items(): + # Use a global prefix to avoid key collisions + global_mask_id = f"global_{mask_id}" + merged_mask[global_mask_id] = mask_config # Set runtime filter to create masks camera_config.objects.filters[object] = RuntimeFilterConfig( frame_shape=camera_config.frame_shape, - **filter.model_dump(exclude_unset=True), + mask=merged_mask, + **filter.model_dump( + exclude_unset=True, exclude={"mask", "raw_mask"} + ), ) # Convert motion configuration @@ -750,7 +817,6 @@ class FrigateConfig(FrigateBaseModel): else: camera_config.motion = RuntimeMotionConfig( frame_shape=camera_config.frame_shape, - raw_mask=camera_config.motion.mask, **camera_config.motion.model_dump(exclude_unset=True), ) camera_config.motion.enabled_in_config = camera_config.motion.enabled diff --git a/frigate/util/config.py b/frigate/util/config.py index 62db3c42b..c689d16e4 100644 --- a/frigate/util/config.py +++ b/frigate/util/config.py @@ -434,6 +434,55 @@ def migrate_017_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any] return new_config +def _convert_legacy_mask_to_dict( + mask: Optional[Union[str, list]], mask_type: str = "motion_mask", label: str = "" +) -> dict[str, dict[str, Any]]: + """Convert legacy mask format (str or list[str]) to new dict format. + + Args: + mask: Legacy mask format (string or list of strings) + mask_type: Type of mask for naming ("motion_mask" or "object_mask") + label: Optional label for object masks (e.g., "person") + + Returns: + Dictionary with mask_id as key and mask config as value + """ + if not mask: + return {} + + result = {} + + if isinstance(mask, str): + if mask: + mask_id = f"{mask_type}_1" + friendly_name = ( + f"Object Mask 1 ({label})" + if label + else f"{mask_type.replace('_', ' ').title()} 1" + ) + result[mask_id] = { + "friendly_name": friendly_name, + "enabled": True, + "coordinates": mask, + } + elif isinstance(mask, list): + for i, coords in enumerate(mask): + if coords: + mask_id = f"{mask_type}_{i + 1}" + friendly_name = ( + f"Object Mask {i + 1} ({label})" + if label + else f"{mask_type.replace('_', ' ').title()} {i + 1}" + ) + result[mask_id] = { + "friendly_name": friendly_name, + "enabled": True, + "coordinates": coords, + } + + return result + + def migrate_018_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]: """Handle migrating frigate config to 0.18-0""" new_config = config.copy() @@ -459,7 +508,35 @@ def migrate_018_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any] if not new_config.get("record"): del new_config["record"] - # Remove deprecated sync_recordings and timelapse_args from camera-specific record configs + # Migrate global motion masks + global_motion = new_config.get("motion", {}) + if global_motion and "mask" in global_motion: + mask = global_motion.get("mask") + if mask is not None and not isinstance(mask, dict): + new_config["motion"]["mask"] = _convert_legacy_mask_to_dict( + mask, "motion_mask" + ) + + # Migrate global object masks + global_objects = new_config.get("objects", {}) + if global_objects and "mask" in global_objects: + mask = global_objects.get("mask") + if mask is not None and not isinstance(mask, dict): + new_config["objects"]["mask"] = _convert_legacy_mask_to_dict( + mask, "object_mask" + ) + + # Migrate global object filters masks + if global_objects and "filters" in global_objects: + for obj_name, filter_config in global_objects.get("filters", {}).items(): + if isinstance(filter_config, dict) and "mask" in filter_config: + mask = filter_config.get("mask") + if mask is not None and not isinstance(mask, dict): + new_config["objects"]["filters"][obj_name]["mask"] = ( + _convert_legacy_mask_to_dict(mask, "object_mask", obj_name) + ) + + # Remove deprecated sync_recordings and migrate masks for camera-specific configs for name, camera in config.get("cameras", {}).items(): camera_config: dict[str, dict[str, Any]] = camera.copy() @@ -478,6 +555,34 @@ def migrate_018_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any] if not camera_config.get("record"): del camera_config["record"] + # Migrate camera motion masks + camera_motion = camera_config.get("motion", {}) + if camera_motion and "mask" in camera_motion: + mask = camera_motion.get("mask") + if mask is not None and not isinstance(mask, dict): + camera_config["motion"]["mask"] = _convert_legacy_mask_to_dict( + mask, "motion_mask" + ) + + # Migrate camera global object masks + camera_objects = camera_config.get("objects", {}) + if camera_objects and "mask" in camera_objects: + mask = camera_objects.get("mask") + if mask is not None and not isinstance(mask, dict): + camera_config["objects"]["mask"] = _convert_legacy_mask_to_dict( + mask, "object_mask" + ) + + # Migrate camera object filter masks + if camera_objects and "filters" in camera_objects: + for obj_name, filter_config in camera_objects.get("filters", {}).items(): + if isinstance(filter_config, dict) and "mask" in filter_config: + mask = filter_config.get("mask") + if mask is not None and not isinstance(mask, dict): + camera_config["objects"]["filters"][obj_name]["mask"] = ( + _convert_legacy_mask_to_dict(mask, "object_mask", obj_name) + ) + new_config["cameras"][name] = camera_config new_config["version"] = "0.18-0"