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

frigate/api/event.py

@@ -386,9 +386,7 @@ def events_explore(
     limit: int = 10,
     allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter),
 ):
-    if not allowed_cameras:
-        return JSONResponse(content=[])
-
+    # get distinct labels for all events
     distinct_labels = (
         Event.select(Event.label)
         .where(Event.camera << allowed_cameras)
@@ -398,31 +396,13 @@ def events_explore(
 
     label_counts = {}
 
-    explore_columns = (
-        Event.id,
-        Event.camera,
-        Event.label,
-        Event.sub_label,
-        Event.zones,
-        Event.start_time,
-        Event.end_time,
-        Event.has_clip,
-        Event.has_snapshot,
-        Event.plus_id,
-        Event.retain_indefinitely,
-        Event.top_score,
-        Event.false_positive,
-        Event.box,
-        Event.data,
-    )
-
     def event_generator():
         for label_obj in distinct_labels.iterator():
             label = label_obj.label
 
             # get most recent events for this label
             label_events = (
-                Event.select(*explore_columns)
+                Event.select()
                 .where((Event.label == label) & (Event.camera << allowed_cameras))
                 .order_by(Event.start_time.desc())
                 .limit(limit)
@@ -504,18 +484,22 @@ async def event_ids(ids: str, request: Request):
             status_code=400,
         )
 
+    for event_id in ids:
+        try:
+            event = Event.get(Event.id == event_id)
+            await require_camera_access(event.camera, request=request)
+        except DoesNotExist:
+            # we should not fail the entire request if an event is not found
+            continue
+
     try:
-        events = list(Event.select().where(Event.id << ids).dicts().iterator())
+        events = Event.select().where(Event.id << ids).dicts().iterator()
+        return JSONResponse(list(events))
     except Exception:
         return JSONResponse(
             content=({"success": False, "message": "Events not found"}), status_code=400
         )
 
-    for event in events:
-        await require_camera_access(event["camera"], request=request)
-
-    return JSONResponse(events)
-
 
 @router.get(
     "/events/search",

frigate/api/review.py

@@ -9,7 +9,7 @@ import pandas as pd
 from fastapi import APIRouter, Request
 from fastapi.params import Depends
 from fastapi.responses import JSONResponse
-from peewee import Case, DoesNotExist, fn, operator
+from peewee import Case, DoesNotExist, IntegrityError, fn, operator
 from playhouse.shortcuts import model_to_dict
 
 from frigate.api.auth import (
@@ -171,19 +171,11 @@ async def review_ids(request: Request, ids: str):
             status_code=400,
         )
 
-    try:
-        reviews = list(
-            ReviewSegment.select().where(ReviewSegment.id << ids).dicts().iterator()
-        )
-    except Exception:
-        return JSONResponse(
-            content=({"success": False, "message": "Review segments not found"}),
-            status_code=400,
-        )
-
-    found_ids = {r["id"] for r in reviews}
     for review_id in ids:
-        if review_id not in found_ids:
+        try:
+            review = ReviewSegment.get(ReviewSegment.id == review_id)
+            await require_camera_access(review.camera, request=request)
+        except DoesNotExist:
             return JSONResponse(
                 content=(
                     {"success": False, "message": f"Review {review_id} not found"}
@@ -191,10 +183,16 @@ async def review_ids(request: Request, ids: str):
                 status_code=404,
             )
 
-    for review in reviews:
-        await require_camera_access(review["camera"], request=request)
-
-    return JSONResponse(reviews)
+    try:
+        reviews = (
+            ReviewSegment.select().where(ReviewSegment.id << ids).dicts().iterator()
+        )
+        return JSONResponse(list(reviews))
+    except Exception:
+        return JSONResponse(
+            content=({"success": False, "message": "Review segments not found"}),
+            status_code=400,
+        )
 
 
 @router.get(
@@ -491,52 +489,27 @@ async def set_multiple_reviewed(
 
     user_id = current_user["username"]
 
-    reviews = list(
-        ReviewSegment.select(ReviewSegment.id, ReviewSegment.camera).where(
-            ReviewSegment.id << body.ids
-        )
-    )
-
-    for review in reviews:
-        await require_camera_access(review.camera, request=request)
-
-    found_ids = [r.id for r in reviews]
-
-    if found_ids:
-        existing_statuses = list(
-            UserReviewStatus.select().where(
-                (UserReviewStatus.user_id == user_id)
-                & (UserReviewStatus.review_segment << found_ids)
+    for review_id in body.ids:
+        try:
+            review = ReviewSegment.get(ReviewSegment.id == review_id)
+            await require_camera_access(review.camera, request=request)
+            review_status = UserReviewStatus.get(
+                UserReviewStatus.user_id == user_id,
+                UserReviewStatus.review_segment == review_id,
             )
-        )
-
-        status_by_review = {s.review_segment_id: s for s in existing_statuses}
-
-        to_update = []
-        to_create = []
-
-        for review_id in found_ids:
-            if review_id in status_by_review:
-                status = status_by_review[review_id]
-                if status.has_been_reviewed != body.reviewed:
-                    status.has_been_reviewed = body.reviewed
-                    to_update.append(status)
-            else:
-                to_create.append(
-                    {
-                        "user_id": user_id,
-                        "review_segment_id": review_id,
-                        "has_been_reviewed": body.reviewed,
-                    }
+            # Update based on the reviewed parameter
+            if review_status.has_been_reviewed != body.reviewed:
+                review_status.has_been_reviewed = body.reviewed
+                review_status.save()
+        except DoesNotExist:
+            try:
+                UserReviewStatus.create(
+                    user_id=user_id,
+                    review_segment=ReviewSegment.get(id=review_id),
+                    has_been_reviewed=body.reviewed,
                 )
-
-        if to_update:
-            UserReviewStatus.bulk_update(
-                to_update, fields=[UserReviewStatus.has_been_reviewed], batch_size=100
-            )
-
-        if to_create:
-            UserReviewStatus.insert_many(to_create).on_conflict_ignore().execute()
+            except (DoesNotExist, IntegrityError):
+                pass
 
     return JSONResponse(
         content=(

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 = []
-            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
-
+            self._remove_custom_classification_processor(model_name)
             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/test/http_api/test_http_review.py

@@ -497,43 +497,6 @@ class TestHttpReview(BaseTestHttp):
             )
             assert user_review.has_been_reviewed == True
 
-    def test_post_reviews_viewed_concurrent_duplicate_does_not_raise(self):
-        """Regression: concurrent requests marking the same review must not 500.
-
-        Two requests can both SELECT and find no existing status, then both try
-        to INSERT, hitting the unique (user_id, review_segment) constraint.
-        on_conflict_ignore() must silently skip the duplicate instead of raising
-        an IntegrityError (which was previously caught with try/except).
-        """
-        id = "123456.random"
-        with AuthTestClient(self.app):
-            super().insert_mock_review_segment(id)
-
-            # Simulate the first request having already committed its insert.
-            self._insert_user_review_status(id, reviewed=True)
-
-            # Simulate the second concurrent request attempting the same insert.
-            UserReviewStatus.insert_many(
-                [
-                    {
-                        "user_id": self.user_id,
-                        "review_segment_id": id,
-                        "has_been_reviewed": True,
-                    }
-                ]
-            ).on_conflict_ignore().execute()
-
-            # Exactly one row should exist; no exception should have been raised.
-            count = (
-                UserReviewStatus.select()
-                .where(
-                    (UserReviewStatus.user_id == self.user_id)
-                    & (UserReviewStatus.review_segment == id)
-                )
-                .count()
-            )
-            assert count == 1
-
     ####################################################################################################################
     ###################################  POST reviews/delete Endpoint   ################################################
     ####################################################################################################################

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

migrations/036_add_perf_indexes.py

@@ -1,21 +0,0 @@
-"""Peewee migrations -- 036_add_perf_indexes.py.
-
-Adds composite/single-column indexes to speed up single-camera queries
-issued by the web UI.
-
-"""
-
-import peewee as pw
-
-SQL = pw.SQL
-
-
-def migrate(migrator, database, fake=False, **kwargs):
-    migrator.sql(
-        'CREATE INDEX IF NOT EXISTS "event_camera_start_time" '
-        'ON "event" ("camera", "start_time" DESC)'
-    )
-
-
-def rollback(migrator, database, fake=False, **kwargs):
-    migrator.sql('DROP INDEX IF EXISTS "event_camera_start_time"')