fix save_attempts trimming (#24246)

frigate/data_processing/real_time/custom_classification.py

@@ -16,6 +16,7 @@ from frigate.config.classification import CustomClassificationConfig
 from frigate.const import CLIPS_DIR, MODEL_CACHE_DIR
 from frigate.log import suppress_stderr_during
 from frigate.util.builtin import EventsPerSecond, InferenceSpeed, load_labels
+from frigate.util.file import trim_oldest_files
 from frigate.util.image import calculate_region
 from frigate.util.object import box_overlaps
 
@@ -729,16 +730,4 @@ def write_classification_attempt(
     file = os.path.join(folder, f"{event_id}-{timestamp}-{label}-{score}.webp")
     os.makedirs(folder, exist_ok=True)
     cv2.imwrite(file, frame)
-
-    # delete oldest face image if maximum is reached
-    try:
-        files = sorted(
-            filter(lambda f: f.endswith(".webp"), os.listdir(folder)),
-            key=lambda f: os.path.getctime(os.path.join(folder, f)),
-            reverse=True,
-        )
-
-        if len(files) > max_files:
-            os.unlink(os.path.join(folder, files[-1]))
-    except (FileNotFoundError, OSError):
-        pass
+    trim_oldest_files(folder, max_files)

frigate/data_processing/real_time/face.py

@@ -6,7 +6,6 @@ import json
 import logging
 import os
 import shutil
-from pathlib import Path
 from typing import Any
 
 import cv2
@@ -28,6 +27,7 @@ from frigate.data_processing.common.face.recognizer import (
 )
 from frigate.types import TrackedObjectUpdateTypesEnum
 from frigate.util.builtin import EventsPerSecond, InferenceSpeed
+from frigate.util.file import trim_oldest_files
 from frigate.util.image import area
 from frigate.util.path import safe_join, sanitize_path_component
 
@@ -489,13 +489,4 @@ class FaceRealTimeProcessor(RealTimeProcessorApi):
             )
             os.makedirs(folder, exist_ok=True)
             cv2.imwrite(file, frame)
-
-            files = sorted(
-                filter(lambda f: f.endswith(".webp"), os.listdir(folder)),
-                key=lambda f: os.path.getctime(os.path.join(folder, f)),
-                reverse=True,
-            )
-
-            # delete oldest face image if maximum is reached
-            if len(files) > self.config.face_recognition.save_attempts:
-                Path(os.path.join(folder, files[-1])).unlink(missing_ok=True)
+            trim_oldest_files(folder, self.config.face_recognition.save_attempts)

frigate/test/test_file.py

@@ -1,5 +1,6 @@
 import os
 import tempfile
+import time
 from types import SimpleNamespace
 from unittest import TestCase
 from unittest.mock import patch
@@ -86,3 +87,52 @@ class TestFileUtils(TestCase):
                 pass
 
             assert file_util.get_event_thumbnail_bytes(event) is None
+
+
+class TestTrimOldestFiles(TestCase):
+    def _fill(self, folder: str, names: list[str]) -> None:
+        # a short gap keeps the ctime order deterministic
+        for name in names:
+            with open(os.path.join(folder, name), "wb"):
+                pass
+
+            time.sleep(0.01)
+
+    def test_trims_folder_already_over_limit(self):
+        """Verify one call trims a folder that is far over the limit."""
+        with tempfile.TemporaryDirectory() as folder:
+            self._fill(folder, [f"{i:03d}.webp" for i in range(10)])
+
+            file_util.trim_oldest_files(folder, 4)
+
+            assert sorted(os.listdir(folder)) == [
+                "006.webp",
+                "007.webp",
+                "008.webp",
+                "009.webp",
+            ]
+
+    def test_counts_every_listed_image_extension(self):
+        """Verify the trim counts the same images the train listing shows."""
+        with tempfile.TemporaryDirectory() as folder:
+            self._fill(
+                folder, ["notes.txt", "a.jpg", "b.jpeg", "c.png", "d.webp", "e.webp"]
+            )
+
+            file_util.trim_oldest_files(folder, 2)
+
+            assert sorted(os.listdir(folder)) == ["d.webp", "e.webp", "notes.txt"]
+
+    def test_file_removed_during_scan_does_not_skip_trim(self):
+        """Verify a file deleted between listing and stat still trims the rest."""
+        real_listdir = os.listdir
+
+        with tempfile.TemporaryDirectory() as folder:
+            self._fill(folder, [f"{i:03d}.webp" for i in range(10)])
+
+            with patch(
+                "os.listdir", side_effect=lambda p: real_listdir(p) + ["gone.webp"]
+            ):
+                file_util.trim_oldest_files(folder, 4)
+
+            assert len(os.listdir(folder)) == 4

frigate/util/file.py

@@ -269,6 +269,41 @@ def delete_event_thumbnail(event: Event) -> bool:
         return True
 
 
+### Training Images
+
+TRAINING_IMAGE_EXTENSIONS = (".webp", ".png", ".jpg", ".jpeg")
+
+
+def trim_oldest_files(folder: str, max_files: int) -> None:
+    """Delete the oldest training images until at most max_files remain."""
+    try:
+        names = os.listdir(folder)
+    except OSError:
+        return
+
+    files: list[tuple[float, str]] = []
+
+    for name in names:
+        if not name.lower().endswith(TRAINING_IMAGE_EXTENSIONS):
+            continue
+
+        path = os.path.join(folder, name)
+
+        # the UI can move or delete an image between listdir and stat
+        try:
+            files.append((os.path.getctime(path), path))
+        except OSError:
+            continue
+
+    files.sort(reverse=True)
+
+    for _, path in files[max_files:]:
+        try:
+            os.unlink(path)
+        except OSError:
+            logger.debug("Unable to delete training image %s", path)
+
+
 ### File Locking