Fix enabled flag for custom classification models (#23681)

* honor enabled flag for custom classification models

for both startup and dynamically, even though the UI doesn't currently have a way to toggle dynamically

* add test

docker/main/requirements-wheels.txt

@@ -1,4 +1,4 @@
-anyio == 4.14.*
+aiofiles == 24.1.*
 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 await aopen(file_path) as file:
+            async with aiofiles.open(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 await aopen(service_location) as file:
+        async with aiofiles.open(service_location) as file:
             contents = await file.read()
 
         total_lines, log_lines = process_logs(contents, service, start, end)

frigate/api/camera.py

@@ -10,7 +10,6 @@ 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
@@ -1188,17 +1187,15 @@ async def delete_camera(
 
     try:
         with lock:
-            async with await aopen(config_file) as f:
+            with open(config_file) as f:
-                old_raw_config = await f.read()
+                old_raw_config = f.read()
 
             try:
                 yaml = YAML()
                 yaml.indent(mapping=2, sequence=4, offset=2)
 
-                async with await aopen(config_file) as f:
+                with open(config_file) as f:
-                    text = await f.read()
+                    data = yaml.load(f)
-
-                data = yaml.load(text)
 
                 # Remove camera from config
                 if "cameras" in data and camera_name in data["cameras"]:
@@ -1223,17 +1220,17 @@ async def delete_camera(
                     for role_name in empty_roles:
                         del auth["roles"][role_name]
 
-                async with await aopen(config_file, "w") as f:
+                with open(config_file, "w") as f:
                     yaml.dump(data, f)
 
-                async with await aopen(config_file) as f:
+                with open(config_file) as f:
-                    new_raw_config = await f.read()
+                    new_raw_config = f.read()
 
                 try:
                     config = FrigateConfig.parse(new_raw_config)
                 except Exception:
-                    async with await aopen(config_file, "w") as f:
+                    with open(config_file, "w") as f:
-                        await f.write(old_raw_config)
+                        f.write(old_raw_config)
                     logger.exception(
                         "Config error after removing camera %s",
                         camera_name,

frigate/api/event.py

@@ -13,7 +13,6 @@ 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
@@ -1456,10 +1455,10 @@ async def set_attributes(
         dataset_dir = os.path.join(CLIPS_DIR, sanitize_filename(model_key), "dataset")
         available_labels = set()
 
-        if await AsyncPath(dataset_dir).exists():
+        if os.path.exists(dataset_dir):
             for category_name in os.listdir(dataset_dir):
                 category_dir = os.path.join(dataset_dir, category_name)
-                if await AsyncPath(category_dir).is_dir():
+                if os.path.isdir(category_dir):
                     available_labels.add(category_name)
 
         if not available_labels:

frigate/api/media.py

@@ -15,8 +15,6 @@ 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
@@ -499,18 +497,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)
-    async with await aopen(file_path, "w") as file:
+    with open(file_path, "w") as file:
         clip: Recordings
         for clip in recordings:
-            await file.write(f"file '{clip.path}'\n")
+            file.write(f"file '{clip.path}'\n")
 
             # if this is the starting clip, add an inpoint
             if clip.start_time < start_ts:
-                await file.write(f"inpoint {int(start_ts - clip.start_time)}\n")
+                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:
-                await file.write(f"outpoint {int(end_ts - clip.start_time)}\n")
+                file.write(f"outpoint {int(end_ts - clip.start_time)}\n")
 
     if len(file_name) > 1000:
         return JSONResponse(
@@ -1151,8 +1149,8 @@ async def event_snapshot_clean(request: Request, event_id: str, download: bool =
                 )
 
             if image_path.endswith(".webp"):
-                async with await aopen(image_path, "rb") as image_file:
+                with open(image_path, "rb") as image_file:
-                    webp_bytes = await image_file.read()
+                    webp_bytes = image_file.read()
             else:
                 image = load_event_snapshot_image(event, clean_only=True)[0]
                 if image is None:
@@ -1368,7 +1366,7 @@ async def preview_gif(
         # need to generate from existing images
         preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
-        if not await AsyncPath(preview_dir).is_dir():
+        if not os.path.isdir(preview_dir):
             return JSONResponse(
                 content={"success": False, "message": "Preview not found"},
                 status_code=404,
@@ -1557,7 +1555,7 @@ async def preview_mp4(
         # need to generate from existing images
         preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
-        if not await AsyncPath(preview_dir).is_dir():
+        if not os.path.isdir(preview_dir):
             return JSONResponse(
                 content={"success": False, "message": "Preview not found"},
                 status_code=404,
@@ -1635,7 +1633,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((await AsyncPath(path).stat()).st_size),
+        "Content-Length": str(os.path.getsize(path)),
         # nginx: https://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_ignore_headers
         "X-Accel-Redirect": f"/cache/{file_name}",
     }
@@ -1709,10 +1707,10 @@ async def preview_thumbnail(request: Request, file_name: str):
     preview_dir = os.path.join(CACHE_DIR, "preview_frames")
 
     try:
-        async with await aopen(
+        with open(
             os.path.join(preview_dir, safe_file_name_current), "rb"
         ) as image_file:
-            jpg_bytes = await image_file.read()
+            jpg_bytes = 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:
-            await AsyncPath(recording["path"]).unlink(missing_ok=True)
+            Path(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,6 +200,9 @@ 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
@@ -332,6 +335,25 @@ 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:
@@ -339,23 +361,7 @@ class EmbeddingMaintainer(threading.Thread):
         model_name = topic.split("/")[-1]
 
         if model_config is None:
-            remaining = []
+            self._remove_custom_classification_processor(model_name)
-            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}"
             )
@@ -363,6 +369,13 @@ 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(
@@ -702,7 +715,11 @@ class EmbeddingMaintainer(threading.Thread):
             and "license_plate" not in camera_config.objects.track
         )
 
-        if not dedicated_lpr_enabled and len(self.config.classification.custom) == 0:
+        has_enabled_custom = any(
+            c.enabled for c in self.config.classification.custom.values()
+        )
+
+        if not dedicated_lpr_enabled and not has_enabled_custom:
             # no active features that use this data
             return
 

frigate/record/maintainer.py

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

frigate/test/test_classification_enabled.py

@@ -0,0 +1,106 @@
+"""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 AsyncMock, MagicMock, patch
+from unittest.mock import 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,51 +42,38 @@ class TestMaintainer(unittest.IsolatedAsyncioTestCase):
         # One bad file, one good file
         files = ["bad_filename.mp4", "camera@20210101000000+0000.mp4"]
 
-        mock_paths = []
+        with patch("os.listdir", return_value=files):
-        for filename in files:
+            with patch("os.path.isfile", return_value=True):
-            path = MagicMock()
+                with patch(
-            path.name = filename
+                    "frigate.record.maintainer.psutil.process_iter", return_value=[]
-            path.suffix = ".mp4"
+                ):
-            path.is_file = AsyncMock(return_value=True)
+                    with patch("frigate.record.maintainer.logger.warning") as warn:
-            mock_paths.append(path)
+                        # Mock validate_and_move_segment to avoid further logic
+                        maintainer.validate_and_move_segment = MagicMock()
 
-        async def mock_iterdir():
+                        try:
-            for path in mock_paths:
+                            await maintainer.move_files()
-                yield path
+                        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
 
-        with patch("frigate.record.maintainer.AsyncPath") as mock_async_path:
+                        # The bad filename is encountered in multiple loops, but should only warn once.
-            mock_async_path.return_value.iterdir = mock_iterdir
+                        matching = [
-
+                            c
-            with patch(
+                            for c in warn.call_args_list
-                "frigate.record.maintainer.psutil.process_iter", return_value=[]
+                            if c.args
-            ):
+                            and isinstance(c.args[0], str)
-                with patch("frigate.record.maintainer.logger.warning") as warn:
+                            and "Skipping unexpected files in cache" in c.args[0]
-                    # Mock validate_and_move_segment to avoid further logic
+                        ]
-                    maintainer.validate_and_move_segment = MagicMock()
+                        self.assertEqual(
-
+                            1,
-                    try:
+                            len(matching),
-                        await maintainer.move_files()
+                            f"Expected a single warning for unexpected files, got {len(matching)}",
-                    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,20 +2,5 @@
 target-version = "py311"
 
 [tool.ruff.lint]
-ignore = [
+ignore = ["E501","E711","E712","UP031","UP032","UP042","G004"]
-  "ASYNC109", # Async function definition with a timeout parameter
+extend-select = ["I", "UP", "G", "ASYNC210", "B904"]
-  "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
-]