diff --git a/frigate/app.py b/frigate/app.py index f8c8d9f52c..9f8192a31d 100644 --- a/frigate/app.py +++ b/frigate/app.py @@ -103,21 +103,9 @@ class FrigateApp: self.detection_shms: list[mp.shared_memory.SharedMemory] = [] self.log_queue: Queue = mp.Queue() self.camera_metrics: DictProxy = self.metrics_manager.dict() - self.embeddings_metrics: DataProcessorMetrics | None = ( - DataProcessorMetrics( - self.metrics_manager, list(config.classification.custom.keys()) - ) - if ( - config.semantic_search.enabled - or any( - c.objects.genai.enabled or c.review.genai.enabled - for c in config.cameras.values() - ) - or config.lpr.enabled - or config.face_recognition.enabled - or len(config.classification.custom) > 0 - ) - else None + + self.embeddings_metrics = DataProcessorMetrics( + self.metrics_manager, list(config.classification.custom.keys()) ) self.ptz_metrics: dict[str, PTZMetrics] = {} self.processes: dict[str, int] = {} diff --git a/frigate/data_processing/post/object_descriptions.py b/frigate/data_processing/post/object_descriptions.py index 6404b38516..9612ce7be9 100644 --- a/frigate/data_processing/post/object_descriptions.py +++ b/frigate/data_processing/post/object_descriptions.py @@ -63,8 +63,10 @@ class ObjectDescriptionProcessor(PostProcessorApi): """Handle an update to a frame for an object.""" camera_config = self.config.cameras[camera] - # no need to save our own thumbnails if genai is not enabled - # or if the object has become stationary + if not camera_config.objects.genai.enabled: + return + + # no need to save our own thumbnails if the object has become stationary if not data["stationary"]: if data["id"] not in self.tracked_events: self.tracked_events[data["id"]] = [] diff --git a/frigate/embeddings/__init__.py b/frigate/embeddings/__init__.py index bf74df6811..cb50188893 100644 --- a/frigate/embeddings/__init__.py +++ b/frigate/embeddings/__init__.py @@ -33,7 +33,7 @@ class EmbeddingProcess(FrigateProcess): def __init__( self, config: FrigateConfig, - metrics: DataProcessorMetrics | None, + metrics: DataProcessorMetrics, stop_event: MpEvent, ) -> None: super().__init__( diff --git a/frigate/embeddings/maintainer.py b/frigate/embeddings/maintainer.py index a9b9b837b7..a91e162e7d 100644 --- a/frigate/embeddings/maintainer.py +++ b/frigate/embeddings/maintainer.py @@ -78,6 +78,16 @@ logger = logging.getLogger(__name__) MAX_THUMBNAILS = 10 +GENAI_UPDATE_TOPICS = frozenset( + { + CameraConfigUpdateEnum.add.name, + CameraConfigUpdateEnum.objects.name, + CameraConfigUpdateEnum.object_genai.name, + CameraConfigUpdateEnum.review.name, + CameraConfigUpdateEnum.review_genai.name, + } +) + class EmbeddingMaintainer(threading.Thread): """Handle embedding queue and post event updates.""" @@ -85,7 +95,7 @@ class EmbeddingMaintainer(threading.Thread): def __init__( self, config: FrigateConfig, - metrics: DataProcessorMetrics | None, + metrics: DataProcessorMetrics, stop_event: MpEvent, ) -> None: super().__init__(name="embeddings_maintainer") @@ -220,16 +230,6 @@ class EmbeddingMaintainer(threading.Thread): # post processors self.post_processors: list[PostProcessorApi] = [] - if any(c.review.genai.enabled_in_config for c in self.config.cameras.values()): - self.post_processors.append( - ReviewDescriptionProcessor( - self.config, - self.requestor, - self.metrics, - self.genai_manager, - ) - ) - if self.config.lpr.enabled: self.post_processors.append( LicensePlatePostProcessor( @@ -252,9 +252,9 @@ class EmbeddingMaintainer(threading.Thread): ) ) - semantic_trigger_processor: SemanticTriggerProcessor | None = None + self.semantic_trigger_processor: SemanticTriggerProcessor | None = None if self.config.semantic_search.enabled: - semantic_trigger_processor = SemanticTriggerProcessor( + self.semantic_trigger_processor = SemanticTriggerProcessor( db, self.config, self.requestor, @@ -262,9 +262,49 @@ class EmbeddingMaintainer(threading.Thread): metrics, self.embeddings, ) - self.post_processors.append(semantic_trigger_processor) + self.post_processors.append(self.semantic_trigger_processor) - if any(c.objects.genai.enabled_in_config for c in self.config.cameras.values()): + self._sync_genai_processors() + + self.stop_event = stop_event + + # recordings data + self.recordings_available_through: dict[str, float] = {} + + def _sync_genai_processors(self) -> None: + """Create GenAI post processors for cameras that have GenAI enabled. + + Called at startup and again after camera config updates so enabling + GenAI on the first camera does not require a restart. Processors are + never removed once created. + + A profile can turn GenAI on without setting enabled_in_config, so both + flags are checked. + """ + cameras = self.config.cameras.values() + + if any( + c.review.genai.enabled or c.review.genai.enabled_in_config for c in cameras + ) and not any( + isinstance(p, ReviewDescriptionProcessor) for p in self.post_processors + ): + logger.debug("Initializing review description processor") + self.post_processors.append( + ReviewDescriptionProcessor( + self.config, + self.requestor, + self.metrics, + self.genai_manager, + ) + ) + + if any( + c.objects.genai.enabled or c.objects.genai.enabled_in_config + for c in cameras + ) and not any( + isinstance(p, ObjectDescriptionProcessor) for p in self.post_processors + ): + logger.debug("Initializing object description processor") self.post_processors.append( ObjectDescriptionProcessor( self.config, @@ -272,19 +312,21 @@ class EmbeddingMaintainer(threading.Thread): self.requestor, self.metrics, self.genai_manager, - semantic_trigger_processor, + self.semantic_trigger_processor, ) ) - self.stop_event = stop_event + def _check_camera_config_updates(self) -> None: + """Apply camera config updates and register newly enabled processors.""" + updated_topics = self.config_updater.check_for_updates() - # recordings data - self.recordings_available_through: dict[str, float] = {} + if updated_topics.keys() & GENAI_UPDATE_TOPICS: + self._sync_genai_processors() def run(self) -> None: """Maintain a SQLite-vec database for semantic search.""" while not self.stop_event.is_set(): - self.config_updater.check_for_updates() + self._check_camera_config_updates() self._check_enrichment_config_updates() self._process_requests() self._process_updates() diff --git a/frigate/stats/util.py b/frigate/stats/util.py index d3cb81986a..6e20197391 100644 --- a/frigate/stats/util.py +++ b/frigate/stats/util.py @@ -62,7 +62,7 @@ def get_latest_version(config: FrigateConfig) -> str: def stats_init( config: FrigateConfig, camera_metrics: DictProxy, - embeddings_metrics: DataProcessorMetrics | None, + embeddings_metrics: DataProcessorMetrics, detectors: dict[str, ObjectDetectProcess], processes: dict[str, int], ) -> StatsTrackingTypes: diff --git a/frigate/test/test_genai_processor_sync.py b/frigate/test/test_genai_processor_sync.py new file mode 100644 index 0000000000..41e2bf3355 --- /dev/null +++ b/frigate/test/test_genai_processor_sync.py @@ -0,0 +1,213 @@ +"""Tests for GenAI enablement gating in the embeddings maintainer. + +Covers creating post processors when GenAI is enabled at runtime, and the +per-camera gating those processors apply once they exist. +""" + +import sys +import unittest +from unittest.mock import MagicMock, patch + +# Mock TFLite before importing the maintainer +_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() + +# imported from the maintainer to avoid tripping the circular import between +# the maintainer and the processor modules +from frigate.embeddings.maintainer import ( # noqa: E402 + EmbeddingMaintainer, + ObjectDescriptionProcessor, + PostProcessDataEnum, + ReviewDescriptionProcessor, +) + + +class TestGenAIProcessorSync(unittest.TestCase): + """Enabling GenAI on the first camera must not require a restart.""" + + def _make_maintainer( + self, + review: bool = False, + objects: bool = False, + review_in_config: bool | None = None, + objects_in_config: bool | None = None, + ) -> EmbeddingMaintainer: + # Bypass the heavy __init__; only the attributes touched by + # _sync_genai_processors are needed for these tests. + maintainer = EmbeddingMaintainer.__new__(EmbeddingMaintainer) + maintainer.post_processors = [] + maintainer.config = MagicMock() + maintainer.config.cameras = { + "front": self._make_camera( + review, + objects, + review if review_in_config is None else review_in_config, + objects if objects_in_config is None else objects_in_config, + ) + } + maintainer.config_updater = MagicMock() + maintainer.embeddings = None + maintainer.requestor = MagicMock() + maintainer.metrics = MagicMock() + maintainer.genai_manager = MagicMock() + maintainer.semantic_trigger_processor = None + return maintainer + + def _make_camera( + self, + review: bool, + objects: bool, + review_in_config: bool, + objects_in_config: bool, + ) -> MagicMock: + camera = MagicMock() + camera.review.genai.enabled = review + camera.review.genai.enabled_in_config = review_in_config + camera.objects.genai.enabled = objects + camera.objects.genai.enabled_in_config = objects_in_config + return camera + + def _processor_types(self, maintainer: EmbeddingMaintainer) -> list[type]: + return [type(p) for p in maintainer.post_processors] + + def test_no_processors_when_genai_disabled(self): + """A config with no GenAI cameras registers neither processor.""" + maintainer = self._make_maintainer() + + maintainer._sync_genai_processors() + + self.assertEqual(maintainer.post_processors, []) + + def test_review_processor_added_when_enabled_after_startup(self): + """Enabling review GenAI on the first camera registers the processor.""" + maintainer = self._make_maintainer() + maintainer._sync_genai_processors() + + camera = maintainer.config.cameras["front"] + camera.review.genai.enabled = True + camera.review.genai.enabled_in_config = True + maintainer._sync_genai_processors() + + self.assertEqual( + self._processor_types(maintainer), [ReviewDescriptionProcessor] + ) + + def test_object_processor_added_when_enabled_after_startup(self): + """Enabling object GenAI on the first camera registers the processor.""" + maintainer = self._make_maintainer() + maintainer._sync_genai_processors() + + camera = maintainer.config.cameras["front"] + camera.objects.genai.enabled = True + camera.objects.genai.enabled_in_config = True + maintainer._sync_genai_processors() + + self.assertEqual( + self._processor_types(maintainer), [ObjectDescriptionProcessor] + ) + + def test_processor_added_when_only_enabled_by_profile(self): + """A profile enables GenAI without setting enabled_in_config.""" + maintainer = self._make_maintainer( + review=True, objects=True, review_in_config=False, objects_in_config=False + ) + + maintainer._sync_genai_processors() + + self.assertEqual( + self._processor_types(maintainer), + [ReviewDescriptionProcessor, ObjectDescriptionProcessor], + ) + + def test_processors_are_not_duplicated(self): + """Repeated config updates must not register a second processor.""" + maintainer = self._make_maintainer(review=True, objects=True) + + maintainer._sync_genai_processors() + maintainer._sync_genai_processors() + + self.assertEqual( + self._processor_types(maintainer), + [ReviewDescriptionProcessor, ObjectDescriptionProcessor], + ) + + def test_genai_topic_triggers_sync(self): + """A camera config update on a GenAI topic registers the processor.""" + maintainer = self._make_maintainer(review=True) + maintainer.config_updater.check_for_updates.return_value = {"review": ["front"]} + + maintainer._check_camera_config_updates() + + self.assertEqual( + self._processor_types(maintainer), [ReviewDescriptionProcessor] + ) + + def test_unrelated_topic_does_not_sync(self): + """An unrelated camera config update must not register processors.""" + maintainer = self._make_maintainer(review=True) + maintainer.config_updater.check_for_updates.return_value = {"motion": ["front"]} + + maintainer._check_camera_config_updates() + + self.assertEqual(maintainer.post_processors, []) + + +class TestObjectDescriptionCameraGating(unittest.TestCase): + """One camera enabling object descriptions must not enlist the others.""" + + def _make_processor(self, enabled: bool) -> ObjectDescriptionProcessor: + config = MagicMock() + camera = MagicMock() + camera.objects.genai.enabled = enabled + camera.objects.genai.send_triggers.after_significant_updates = None + config.cameras = {"front": camera} + + genai_manager = MagicMock() + genai_manager.description_client = MagicMock() + + return ObjectDescriptionProcessor( + config, None, MagicMock(), MagicMock(), genai_manager, None + ) + + def _update(self, processor: ObjectDescriptionProcessor) -> None: + processor.process_data( + { + "camera": "front", + "data": { + "id": "1234.5-abcdef", + "box": (0, 0, 10, 10), + "stationary": False, + }, + "state": "update", + "yuv_frame": MagicMock(), + }, + PostProcessDataEnum.tracked_object, + ) + + @patch("frigate.data_processing.post.object_descriptions.create_thumbnail") + def test_disabled_camera_collects_no_thumbnails(self, mock_create_thumbnail): + """A camera with object descriptions off does no thumbnail work.""" + processor = self._make_processor(enabled=False) + + self._update(processor) + + mock_create_thumbnail.assert_not_called() + self.assertEqual(processor.tracked_events, {}) + + @patch("frigate.data_processing.post.object_descriptions.create_thumbnail") + def test_enabled_camera_collects_thumbnails(self, mock_create_thumbnail): + """A camera with object descriptions on still collects thumbnails.""" + mock_create_thumbnail.return_value = b"jpg" + processor = self._make_processor(enabled=True) + + self._update(processor) + + mock_create_thumbnail.assert_called_once() + self.assertEqual(len(processor.tracked_events["1234.5-abcdef"]), 1) diff --git a/frigate/types.py b/frigate/types.py index 77bb508451..e5f913d4fb 100644 --- a/frigate/types.py +++ b/frigate/types.py @@ -8,7 +8,7 @@ from frigate.object_detection.base import ObjectDetectProcess class StatsTrackingTypes(TypedDict): camera_metrics: dict[str, CameraMetrics] - embeddings_metrics: DataProcessorMetrics | None + embeddings_metrics: DataProcessorMetrics detectors: dict[str, ObjectDetectProcess] started: int latest_frigate_version: str