Improve config validation for zones and object masks

frigate/config.py

@@ -46,6 +46,7 @@
     get_ffmpeg_arg_list,
     load_config_with_no_duplicates,
 )
+from frigate.util.config import get_relative_coordinates
 from frigate.util.image import create_mask
 from frigate.util.services import auto_detect_hwaccel, get_video_properties
 
@@ -348,35 +349,7 @@ class RuntimeMotionConfig(MotionConfig):
     def __init__(self, **config):
         frame_shape = config.get("frame_shape", (1, 1))
 
-        mask = config.get("mask", "")
-
-        # masks and zones are saved as relative coordinates
-        # we know if any points are > 1 then it is using the
-        # old native resolution coordinates
-        if mask:
-            if isinstance(mask, list) and any(x > "1.0" for x in mask[0].split(",")):
-                relative_masks = []
-                for m in mask:
-                    points = m.split(",")
-                    relative_masks.append(
-                        ",".join(
-                            [
-                                f"{round(int(points[i]) / frame_shape[1], 3)},{round(int(points[i + 1]) / frame_shape[0], 3)}"
-                                for i in range(0, len(points), 2)
-                            ]
-                        )
-                    )
-
-                mask = relative_masks
-            elif isinstance(mask, str) and any(x > "1.0" for x in mask.split(",")):
-                points = mask.split(",")
-                mask = ",".join(
-                    [
-                        f"{round(int(points[i]) / frame_shape[1], 3)},{round(int(points[i + 1]) / frame_shape[0], 3)}"
-                        for i in range(0, len(points), 2)
-                    ]
-                )
-
+        mask = get_relative_coordinates(config.get("mask", ""), frame_shape)
         config["raw_mask"] = mask
 
         if mask:
@@ -508,34 +481,7 @@ class RuntimeFilterConfig(FilterConfig):
 
     def __init__(self, **config):
         frame_shape = config.get("frame_shape", (1, 1))
-        mask = config.get("mask")
-
-        # masks and zones are saved as relative coordinates
-        # we know if any points are > 1 then it is using the
-        # old native resolution coordinates
-        if mask:
-            if isinstance(mask, list) and any(x > "1.0" for x in mask[0].split(",")):
-                relative_masks = []
-                for m in mask:
-                    points = m.split(",")
-                    relative_masks.append(
-                        ",".join(
-                            [
-                                f"{round(int(points[i]) / frame_shape[1], 3)},{round(int(points[i + 1]) / frame_shape[0], 3)}"
-                                for i in range(0, len(points), 2)
-                            ]
-                        )
-                    )
-
-                mask = relative_masks
-            elif isinstance(mask, str) and any(x > "1.0" for x in mask.split(",")):
-                points = mask.split(",")
-                mask = ",".join(
-                    [
-                        f"{round(int(points[i]) / frame_shape[1], 3)},{round(int(points[i + 1]) / frame_shape[0], 3)}"
-                        for i in range(0, len(points), 2)
-                    ]
-                )
+        mask = get_relative_coordinates(config.get("mask"), frame_shape)
 
         config["raw_mask"] = mask
 
@@ -1231,6 +1177,20 @@ def verify_zone_objects_are_tracked(camera_config: CameraConfig) -> None:
                 )
 
 
+def verify_required_zones_exist(camera_config: CameraConfig) -> None:
+    for det_zone in camera_config.review.detections.required_zones:
+        if det_zone not in camera_config.zones.keys():
+            raise ValueError(
+                f"Camera {camera_config.name} has a required zone for detections {det_zone} that is not defined."
+            )
+
+    for det_zone in camera_config.review.alerts.required_zones:
+        if det_zone not in camera_config.zones.keys():
+            raise ValueError(
+                f"Camera {camera_config.name} has a required zone for alerts {det_zone} that is not defined."
+            )
+
+
 def verify_autotrack_zones(camera_config: CameraConfig) -> ValueError | None:
     """Verify that required_zones are specified when autotracking is enabled."""
     if (
@@ -1456,9 +1416,15 @@ def runtime_config(self, plus_api: PlusApi = None) -> FrigateConfig:
                             else [filter.mask]
                         )
                     object_mask = (
-                        camera_config.objects.mask
-                        if isinstance(camera_config.objects.mask, list)
-                        else [camera_config.objects.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
 
@@ -1495,6 +1461,7 @@ def runtime_config(self, plus_api: PlusApi = None) -> FrigateConfig:
             verify_recording_retention(camera_config)
             verify_recording_segments_setup_with_reasonable_time(camera_config)
             verify_zone_objects_are_tracked(camera_config)
+            verify_required_zones_exist(camera_config)
             verify_autotrack_zones(camera_config)
             verify_motion_and_detect(camera_config)
 

frigate/events/cleanup.py

@@ -83,7 +83,7 @@ def expire(self, media_type: EventCleanupType) -> list[str]:
                 datetime.datetime.now() - datetime.timedelta(days=expire_days)
             ).timestamp()
             # grab all events after specific time
-            expired_events = (
+            expired_events: list[Event] = (
                 Event.select(
                     Event.id,
                     Event.camera,
@@ -103,12 +103,16 @@ def expire(self, media_type: EventCleanupType) -> list[str]:
                 media_path = Path(
                     f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}"
                 )
-                media_path.unlink(missing_ok=True)
-                if file_extension == "jpg":
-                    media_path = Path(
-                        f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"
-                    )
+
+                try:
                     media_path.unlink(missing_ok=True)
+                    if file_extension == "jpg":
+                        media_path = Path(
+                            f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"
+                        )
+                        media_path.unlink(missing_ok=True)
+                except OSError as e:
+                    logger.warning(f"Unable to delete event images: {e}")
 
             # update the clips attribute for the db entry
             update_query = Event.update(update_params).where(
@@ -163,15 +167,18 @@ def expire(self, media_type: EventCleanupType) -> list[str]:
                     events_to_update.append(event.id)
 
                     if media_type == EventCleanupType.snapshots:
-                        media_name = f"{event.camera}-{event.id}"
-                        media_path = Path(
-                            f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}"
-                        )
-                        media_path.unlink(missing_ok=True)
-                        media_path = Path(
-                            f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"
-                        )
-                        media_path.unlink(missing_ok=True)
+                        try:
+                            media_name = f"{event.camera}-{event.id}"
+                            media_path = Path(
+                                f"{os.path.join(CLIPS_DIR, media_name)}.{file_extension}"
+                            )
+                            media_path.unlink(missing_ok=True)
+                            media_path = Path(
+                                f"{os.path.join(CLIPS_DIR, media_name)}-clean.png"
+                            )
+                            media_path.unlink(missing_ok=True)
+                        except OSError as e:
+                            logger.warning(f"Unable to delete event images: {e}")
 
         # update the clips attribute for the db entry
         Event.update(update_params).where(Event.id << events_to_update).execute()
@@ -195,14 +202,18 @@ def purge_duplicates(self) -> None:
         select distinct id, camera, has_snapshot, has_clip from grouped_events
         where copy_number > 1 and end_time not null;"""
 
-        duplicate_events = Event.raw(duplicate_query)
+        duplicate_events: list[Event] = Event.raw(duplicate_query)
         for event in duplicate_events:
             logger.debug(f"Removing duplicate: {event.id}")
-            media_name = f"{event.camera}-{event.id}"
-            media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}.jpg")
-            media_path.unlink(missing_ok=True)
-            media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.png")
-            media_path.unlink(missing_ok=True)
+
+            try:
+                media_name = f"{event.camera}-{event.id}"
+                media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}.jpg")
+                media_path.unlink(missing_ok=True)
+                media_path = Path(f"{os.path.join(CLIPS_DIR, media_name)}-clean.png")
+                media_path.unlink(missing_ok=True)
+            except OSError as e:
+                logger.warning(f"Unable to delete event images: {e}")
 
         (
             Event.delete()

frigate/util/config.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import shutil
+from typing import Optional, Union
 
 from ruamel.yaml import YAML
 
@@ -131,3 +132,38 @@ def migrate_014(config: dict[str, dict[str, any]]) -> dict[str, dict[str, any]]:
         new_config["cameras"][name] = camera_config
 
     return new_config
+
+
+def get_relative_coordinates(
+    mask: Optional[Union[str, list]], frame_shape: tuple[int, int]
+) -> Union[str, list]:
+    # masks and zones are saved as relative coordinates
+    # we know if any points are > 1 then it is using the
+    # old native resolution coordinates
+    if mask:
+        if isinstance(mask, list) and any(x > "1.0" for x in mask[0].split(",")):
+            relative_masks = []
+            for m in mask:
+                points = m.split(",")
+                relative_masks.append(
+                    ",".join(
+                        [
+                            f"{round(int(points[i]) / frame_shape[1], 3)},{round(int(points[i + 1]) / frame_shape[0], 3)}"
+                            for i in range(0, len(points), 2)
+                        ]
+                    )
+                )
+
+            mask = relative_masks
+        elif isinstance(mask, str) and any(x > "1.0" for x in mask.split(",")):
+            points = mask.split(",")
+            mask = ",".join(
+                [
+                    f"{round(int(points[i]) / frame_shape[1], 3)},{round(int(points[i + 1]) / frame_shape[0], 3)}"
+                    for i in range(0, len(points), 2)
+                ]
+            )
+
+        return mask
+
+    return mask