migrator and runtime config changes

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

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):

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

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

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"