Merge b0588a02f9 into da4037eb52

Replaces aiofiles with anyio, because anyio.Path is much more complete
and comparable to the Pathlib API.

docker/main/requirements-wheels.txt

@@ -1,4 +1,4 @@
-aiofiles == 24.1.*
+anyio == 4.14.*
 click == 8.1.*
 # FastAPI
 aiohttp == 3.12.*

frigate/api/app.py

@@ -14,8 +14,8 @@ from io import StringIO
 from pathlib import Path as FilePath
 from typing import Any
 
-import aiofiles
 import ruamel.yaml
+from anyio import open_file as aopen
 from fastapi import APIRouter, Body, Path, Request, Response
 from fastapi.encoders import jsonable_encoder
 from fastapi.params import Depends
@@ -1052,7 +1052,7 @@ async def logs(
         """Asynchronously stream log lines."""
         buffer = ""
         try:
-            async with aiofiles.open(file_path) as file:
+            async with await aopen(file_path) as file:
                 await file.seek(0, 2)
                 while True:
                     line = await file.readline()
@@ -1090,7 +1090,7 @@ async def logs(
 
     # For full logs initially
     try:
-        async with aiofiles.open(service_location) as file:
+        async with await aopen(service_location) as file:
             contents = await file.read()
 
         total_lines, log_lines = process_logs(contents, service, start, end)

frigate/api/camera.py

@@ -10,6 +10,7 @@ from urllib.parse import quote_plus
 
 import httpx
 import requests
+from anyio import open_file as aopen
 from fastapi import APIRouter, Depends, Query, Request, Response
 from fastapi.responses import JSONResponse
 from filelock import FileLock, Timeout
@@ -1187,15 +1188,17 @@ async def delete_camera(
 
     try:
         with lock:
-            with open(config_file) as f:
-                old_raw_config = f.read()
+            async with await aopen(config_file) as f:
+                old_raw_config = await f.read()
 
             try:
                 yaml = YAML()
                 yaml.indent(mapping=2, sequence=4, offset=2)
 
-                with open(config_file) as f:
-                    data = yaml.load(f)
+                async with await aopen(config_file) as f:
+                    text = await f.read()
+
+                data = yaml.load(text)
 
                 # Remove camera from config
                 if "cameras" in data and camera_name in data["cameras"]:
@@ -1220,17 +1223,17 @@ async def delete_camera(
                     for role_name in empty_roles:
                         del auth["roles"][role_name]
 
-                with open(config_file, "w") as f:
+                async with await aopen(config_file, "w") as f:
                     yaml.dump(data, f)
 
-                with open(config_file) as f:
-                    new_raw_config = f.read()
+                async with await aopen(config_file) as f:
+                    new_raw_config = await f.read()
 
                 try:
                     config = FrigateConfig.parse(new_raw_config)
                 except Exception:
-                    with open(config_file, "w") as f:
-                        f.write(old_raw_config)
+                    async with await aopen(config_file, "w") as f:
+                        await f.write(old_raw_config)
                     logger.exception(
                         "Config error after removing camera %s",
                         camera_name,

frigate/api/event.py

@@ -13,6 +13,7 @@ from pathlib import Path
 from urllib.parse import unquote
 
 import numpy as np
+from anyio import Path as AsyncPath
 from fastapi import APIRouter, Request
 from fastapi.params import Depends
 from fastapi.responses import JSONResponse
@@ -1455,10 +1456,10 @@ async def set_attributes(
         dataset_dir = os.path.join(CLIPS_DIR, sanitize_filename(model_key), "dataset")
         available_labels = set()
 
-        if os.path.exists(dataset_dir):
+        if await AsyncPath(dataset_dir).exists():
             for category_name in os.listdir(dataset_dir):
                 category_dir = os.path.join(dataset_dir, category_name)
-                if os.path.isdir(category_dir):
+                if await AsyncPath(category_dir).is_dir():
                     available_labels.add(category_name)
 
         if not available_labels:

frigate/api/media.py

@@ -15,6 +15,8 @@ from urllib.parse import unquote
 import cv2
 import numpy as np
 import pytz
+from anyio import Path as AsyncPath
+from anyio import open_file as aopen
 from fastapi import APIRouter, Depends, Path, Query, Request, Response
 from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
 from pathvalidate import sanitize_filename
@@ -497,18 +499,18 @@ async def recording_clip(
 
     file_name = sanitize_filename(f"playlist_{camera_name}_{start_ts}-{end_ts}.txt")
     file_path = os.path.join(CACHE_DIR, file_name)
-    with open(file_path, "w") as file:
+    async with await aopen(file_path, "w") as file:
         clip: Recordings
         for clip in recordings:
-            file.write(f"file '{clip.path}'\n")
+            await file.write(f"file '{clip.path}'\n")
 
             # if this is the starting clip, add an inpoint
             if clip.start_time < start_ts:
-                file.write(f"inpoint {int(start_ts - clip.start_time)}\n")
+                await file.write(f"inpoint {int(start_ts - clip.start_time)}\n")
 
             # if this is the ending clip, add an outpoint
             if clip.end_time > end_ts:
-                file.write(f"outpoint {int(end_ts - clip.start_time)}\n")
+                await file.write(f"outpoint {int(end_ts - clip.start_time)}\n")
 
     if len(file_name) > 1000:
         return JSONResponse(
@@ -1149,8 +1151,8 @@ async def event_snapshot_clean(request: Request, event_id: str, download: bool =
                 )
 
             if image_path.endswith(".webp"):
-                with open(image_path, "rb") as image_file:
-                    webp_bytes = image_file.read()
+                async with await aopen(image_path, "rb") as image_file:
+                    webp_bytes = await image_file.read()
             else:
                 image = load_event_snapshot_image(event, clean_only=True)[0]
                 if image is None:
@@ -1366,7 +1368,7 @@ async def preview_gif(
         # need to generate from existing images
         preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
-        if not os.path.isdir(preview_dir):
+        if not await AsyncPath(preview_dir).is_dir():
             return JSONResponse(
                 content={"success": False, "message": "Preview not found"},
                 status_code=404,
@@ -1555,7 +1557,7 @@ async def preview_mp4(
         # need to generate from existing images
         preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
-        if not os.path.isdir(preview_dir):
+        if not await AsyncPath(preview_dir).is_dir():
             return JSONResponse(
                 content={"success": False, "message": "Preview not found"},
                 status_code=404,
@@ -1633,7 +1635,7 @@ async def preview_mp4(
         "Content-Description": "File Transfer",
         "Cache-Control": f"private, max-age={_resolve_cache_age(max_cache_age)}",
         "Content-Type": "video/mp4",
-        "Content-Length": str(os.path.getsize(path)),
+        "Content-Length": str((await AsyncPath(path).stat()).st_size),
         # nginx: https://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_ignore_headers
         "X-Accel-Redirect": f"/cache/{file_name}",
     }
@@ -1707,10 +1709,10 @@ async def preview_thumbnail(request: Request, file_name: str):
     preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
     try:
-        with open(
+        async with await aopen(
             os.path.join(preview_dir, safe_file_name_current), "rb"
         ) as image_file:
-            jpg_bytes = image_file.read()
+            jpg_bytes = await image_file.read()
     except FileNotFoundError:
         return JSONResponse(
             content=({"success": False, "message": "Image file not found"}),

frigate/api/record.py

@@ -4,9 +4,9 @@ import datetime as dt
 import logging
 from datetime import datetime, timedelta
 from functools import reduce
-from pathlib import Path
 from urllib.parse import unquote
 
+from anyio import Path as AsyncPath
 from fastapi import APIRouter, Depends, Request
 from fastapi import Path as PathParam
 from fastapi.responses import JSONResponse
@@ -443,7 +443,7 @@ async def delete_recordings(
         recording_ids.append(recording["id"])
 
         try:
-            Path(recording["path"]).unlink(missing_ok=True)
+            await AsyncPath(recording["path"]).unlink(missing_ok=True)
             deleted_count += 1
         except Exception as e:
             logger.error(f"Failed to delete recording file {recording['path']}: {e}")

frigate/embeddings/maintainer.py

@@ -200,9 +200,6 @@ class EmbeddingMaintainer(threading.Thread):
             )
 
         for model_config in self.config.classification.custom.values():
-            if not model_config.enabled:
-                continue
-
             self.realtime_processors.append(
                 CustomStateClassificationProcessor(
                     self.config, model_config, self.requestor, self.metrics
@@ -335,25 +332,6 @@ class EmbeddingMaintainer(threading.Thread):
         for processor in self.post_processors:
             processor.update_config(topic, payload)
 
-    def _remove_custom_classification_processor(self, model_name: str) -> None:
-        """Shut down and drop any running processor for a custom model."""
-        remaining = []
-        for processor in self.realtime_processors:
-            if (
-                isinstance(
-                    processor,
-                    (
-                        CustomStateClassificationProcessor,
-                        CustomObjectClassificationProcessor,
-                    ),
-                )
-                and processor.model_config.name == model_name
-            ):
-                processor.shutdown()
-            else:
-                remaining.append(processor)
-        self.realtime_processors = remaining
-
     def _handle_custom_classification_update(
         self, topic: str, model_config: Any
     ) -> None:
@@ -361,7 +339,23 @@ class EmbeddingMaintainer(threading.Thread):
         model_name = topic.split("/")[-1]
 
         if model_config is None:
-            self._remove_custom_classification_processor(model_name)
+            remaining = []
+            for processor in self.realtime_processors:
+                if (
+                    isinstance(
+                        processor,
+                        (
+                            CustomStateClassificationProcessor,
+                            CustomObjectClassificationProcessor,
+                        ),
+                    )
+                    and processor.model_config.name == model_name
+                ):
+                    processor.shutdown()
+                else:
+                    remaining.append(processor)
+            self.realtime_processors = remaining
+
             logger.info(
                 f"Successfully removed classification processor for model: {model_name}"
             )
@@ -369,13 +363,6 @@ class EmbeddingMaintainer(threading.Thread):
 
         self.config.classification.custom[model_name] = model_config
 
-        # A disabled model must not run; tear down any existing processor and
-        # do not register a new one.
-        if not model_config.enabled:
-            self._remove_custom_classification_processor(model_name)
-            logger.info(f"Disabled classification processor for model: {model_name}")
-            return
-
         # Check if processor already exists
         for processor in self.realtime_processors:
             if isinstance(
@@ -715,11 +702,7 @@ class EmbeddingMaintainer(threading.Thread):
             and "license_plate" not in camera_config.objects.track
         )
 
-        has_enabled_custom = any(
-            c.enabled for c in self.config.classification.custom.values()
-        )
-
-        if not dedicated_lpr_enabled and not has_enabled_custom:
+        if not dedicated_lpr_enabled and len(self.config.classification.custom) == 0:
             # no active features that use this data
             return
 

frigate/record/maintainer.py

@@ -15,6 +15,7 @@ from typing import Any
 
 import numpy as np
 import psutil
+from anyio import Path as AsyncPath
 
 from frigate.comms.detections_updater import DetectionSubscriber, DetectionTypeEnum
 from frigate.comms.inter_process import InterProcessRequestor
@@ -105,11 +106,11 @@ class RecordingMaintainer(threading.Thread):
 
     async def move_files(self) -> None:
         cache_files = [
-            d
-            for d in os.listdir(CACHE_DIR)
-            if os.path.isfile(os.path.join(CACHE_DIR, d))
-            and d.endswith(".mp4")
-            and not d.startswith("preview_")
+            path.name
+            async for path in AsyncPath(CACHE_DIR).iterdir()
+            if await path.is_file()
+            and path.suffix == ".mp4"
+            and not path.name.startswith("preview_")
         ]
 
         # publish newest cached segment per camera (including in use files)
@@ -229,7 +230,7 @@ class RecordingMaintainer(threading.Thread):
                 to_remove = grouped_recordings[camera][:-keep_count]
                 for rec in to_remove:
                     cache_path = rec["cache_path"]
-                    Path(cache_path).unlink(missing_ok=True)
+                    await AsyncPath(cache_path).unlink(missing_ok=True)
                     self.end_time_cache.pop(cache_path, None)
                 grouped_recordings[camera] = grouped_recordings[camera][-keep_count:]
 
@@ -244,7 +245,7 @@ class RecordingMaintainer(threading.Thread):
                 to_remove = grouped_recordings[camera][:-keep_count]
                 for rec in to_remove:
                     cache_path = rec["cache_path"]
-                    Path(cache_path).unlink(missing_ok=True)
+                    await AsyncPath(cache_path).unlink(missing_ok=True)
                     self.end_time_cache.pop(cache_path, None)
                 grouped_recordings[camera] = grouped_recordings[camera][-keep_count:]
 
@@ -634,7 +635,7 @@ class RecordingMaintainer(threading.Thread):
         file_path = os.path.join(directory, file_name)
 
         try:
-            if not os.path.exists(file_path):
+            if not await AsyncPath(file_path).exists():
                 start_frame = datetime.datetime.now().timestamp()
 
                 # add faststart to kept segments to improve metadata reading
@@ -670,7 +671,9 @@ class RecordingMaintainer(threading.Thread):
                     # get the segment size of the cache file
                     # file without faststart is same size
                     segment_size = round(
-                        float(os.path.getsize(cache_path)) / pow(2, 20), 2
+                        float((await AsyncPath(cache_path).stat()).st_size)
+                        / pow(2, 20),
+                        2,
                     )
                 except OSError:
                     segment_size = 0
@@ -698,7 +701,7 @@ class RecordingMaintainer(threading.Thread):
                 }
         except Exception as e:
             logger.error(f"Unable to store recording segment {cache_path}")
-            Path(cache_path).unlink(missing_ok=True)
+            await AsyncPath(cache_path).unlink(missing_ok=True)
             logger.error(e)
 
         # clear end_time cache

frigate/test/test_classification_enabled.py

@@ -1,106 +0,0 @@
-"""Tests that disabled custom classification models are not registered or run."""
-
-import sys
-import unittest
-from unittest.mock import MagicMock
-
-# Mock TFLite before importing the maintainer / classification modules
-_MOCK_MODULES = [
-    "tflite_runtime",
-    "tflite_runtime.interpreter",
-    "ai_edge_litert",
-    "ai_edge_litert.interpreter",
-]
-for mod in _MOCK_MODULES:
-    if mod not in sys.modules:
-        sys.modules[mod] = MagicMock()
-
-from frigate.data_processing.real_time.custom_classification import (  # noqa: E402
-    CustomObjectClassificationProcessor,
-)
-from frigate.embeddings.maintainer import EmbeddingMaintainer  # noqa: E402
-
-
-class TestCustomClassificationEnabledGating(unittest.TestCase):
-    """A model with enabled: false must not keep a processor registered."""
-
-    def _make_maintainer(self) -> EmbeddingMaintainer:
-        # Bypass the heavy __init__; only the attributes touched by the
-        # config update path are needed for these tests.
-        maintainer = EmbeddingMaintainer.__new__(EmbeddingMaintainer)
-        maintainer.realtime_processors = []
-        maintainer.config = MagicMock()
-        maintainer.config.classification.custom = {}
-        maintainer.requestor = MagicMock()
-        maintainer.metrics = MagicMock()
-        maintainer.event_metadata_publisher = MagicMock()
-        return maintainer
-
-    def _make_model_config(self, name: str, enabled: bool) -> MagicMock:
-        model_config = MagicMock()
-        model_config.name = name
-        model_config.enabled = enabled
-        model_config.state_config = None
-        return model_config
-
-    def _make_processor(self, name: str) -> MagicMock:
-        processor = MagicMock(spec=CustomObjectClassificationProcessor)
-        processor.model_config = MagicMock()
-        processor.model_config.name = name
-        return processor
-
-    def test_disabled_update_tears_down_existing_processor(self):
-        """Toggling a running model to disabled shuts down and drops its processor."""
-        maintainer = self._make_maintainer()
-        processor = self._make_processor("atli")
-        maintainer.realtime_processors = [processor]
-
-        maintainer._handle_custom_classification_update(
-            "config/classification/custom/atli",
-            self._make_model_config("atli", enabled=False),
-        )
-
-        processor.shutdown.assert_called_once()
-        self.assertEqual(maintainer.realtime_processors, [])
-
-    def test_disabled_update_does_not_register_processor(self):
-        """A disabled model that has no processor is never registered."""
-        maintainer = self._make_maintainer()
-
-        maintainer._handle_custom_classification_update(
-            "config/classification/custom/atli",
-            self._make_model_config("atli", enabled=False),
-        )
-
-        self.assertEqual(maintainer.realtime_processors, [])
-
-    def test_disabled_update_leaves_other_processors_untouched(self):
-        """Disabling one model must not affect other running processors."""
-        maintainer = self._make_maintainer()
-        other = self._make_processor("simbi")
-        maintainer.realtime_processors = [other]
-
-        maintainer._handle_custom_classification_update(
-            "config/classification/custom/atli",
-            self._make_model_config("atli", enabled=False),
-        )
-
-        other.shutdown.assert_not_called()
-        self.assertEqual(maintainer.realtime_processors, [other])
-
-    def test_removed_model_tears_down_processor(self):
-        """A None payload (model deleted) still shuts down its processor."""
-        maintainer = self._make_maintainer()
-        processor = self._make_processor("atli")
-        maintainer.realtime_processors = [processor]
-
-        maintainer._handle_custom_classification_update(
-            "config/classification/custom/atli", None
-        )
-
-        processor.shutdown.assert_called_once()
-        self.assertEqual(maintainer.realtime_processors, [])
-
-
-if __name__ == "__main__":
-    unittest.main()

frigate/test/test_maintainer.py

@@ -1,7 +1,7 @@
 import datetime
 import sys
 import unittest
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 # Mock complex imports before importing maintainer, saving originals so we can
 # restore them after import and avoid polluting sys.modules for other tests.
@@ -42,38 +42,51 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
         # One bad file, one good file
         files = ["bad_filename.mp4", "camera@20210101000000+0000.mp4"]
 
-        with patch("os.listdir", return_value=files):
-            with patch("os.path.isfile", return_value=True):
-                with patch(
-                    "frigate.record.maintainer.psutil.process_iter", return_value=[]
-                ):
-                    with patch("frigate.record.maintainer.logger.warning") as warn:
-                        # Mock validate_and_move_segment to avoid further logic
-                        maintainer.validate_and_move_segment = MagicMock()
+        mock_paths = []
+        for filename in files:
+            path = MagicMock()
+            path.name = filename
+            path.suffix = ".mp4"
+            path.is_file = AsyncMock(return_value=True)
+            mock_paths.append(path)
 
-                        try:
-                            await maintainer.move_files()
-                        except ValueError as e:
-                            if "not enough values to unpack" in str(e):
-                                self.fail("move_files() crashed on bad filename!")
-                            raise e
-                        except Exception:
-                            # Ignore other errors (like DB connection) as we only care about the unpack crash
-                            pass
+        async def mock_iterdir():
+            for path in mock_paths:
+                yield path
 
-                        # The bad filename is encountered in multiple loops, but should only warn once.
-                        matching = [
-                            c
-                            for c in warn.call_args_list
-                            if c.args
-                            and isinstance(c.args[0], str)
-                            and "Skipping unexpected files in cache" in c.args[0]
-                        ]
-                        self.assertEqual(
-                            1,
-                            len(matching),
-                            f"Expected a single warning for unexpected files, got {len(matching)}",
-                        )
+        with patch("frigate.record.maintainer.AsyncPath") as mock_async_path:
+            mock_async_path.return_value.iterdir = mock_iterdir
+
+            with patch(
+                "frigate.record.maintainer.psutil.process_iter", return_value=[]
+            ):
+                with patch("frigate.record.maintainer.logger.warning") as warn:
+                    # Mock validate_and_move_segment to avoid further logic
+                    maintainer.validate_and_move_segment = MagicMock()
+
+                    try:
+                        await maintainer.move_files()
+                    except ValueError as e:
+                        if "not enough values to unpack" in str(e):
+                            self.fail("move_files() crashed on bad filename!")
+                        raise e
+                    except Exception:
+                        # Ignore other errors (like DB connection) as we only care about the unpack crash
+                        pass
+
+                    # The bad filename is encountered in multiple loops, but should only warn once.
+                    matching = [
+                        c
+                        for c in warn.call_args_list
+                        if c.args
+                        and isinstance(c.args[0], str)
+                        and "Skipping unexpected files in cache" in c.args[0]
+                    ]
+                    self.assertEqual(
+                        1,
+                        len(matching),
+                        f"Expected a single warning for unexpected files, got {len(matching)}",
+                    )
 
     async def test_drops_quiet_segment_when_only_motion_retention(self):
         # Regression: when motion retention is enabled but a segment has no

pyproject.toml

@@ -2,5 +2,20 @@
 target-version = "py311"
 
 [tool.ruff.lint]
-ignore = ["E501","E711","E712","UP031","UP032","UP042","G004"]
-extend-select = ["I", "UP", "G", "ASYNC210", "B904"]
+ignore = [
+  "ASYNC109", # Async function definition with a timeout parameter
+  "E501",     # line-too-long
+  "E711",     # none-comparison
+  "E712",     # true-false-comparison
+  "UP031",    # printf-string-formatting
+  "UP032",    # f-string
+  "UP042",    # replace-str-enum
+  "G004",     # logging-f-string
+]
+extend-select = [
+  "ASYNC", # https://docs.astral.sh/ruff/rules/#flake8-async-async
+  "B904",  # https://docs.astral.sh/ruff/rules/raise-without-from-inside-except/
+  "G",     # https://docs.astral.sh/ruff/rules/#flake8-logging-format-g
+  "I",     # https://docs.astral.sh/ruff/rules/#isort-i
+  "UP",    # https://docs.astral.sh/ruff/rules/#pyupgrade-up
+]