From bb04e6d49ddc84bd20f89db69c5f21da28e65611 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Fri, 13 Nov 2026 07:45:29 -0600 Subject: [PATCH] Move genai review frame extraction off the embeddings loop (#24959) * move genai review frame extraction off the embeddings loop - Run frame gathering for finished review items on a thread so per-frame ffmpeg calls don't stall the embeddings maintainer - Share one path between live and regenerated review descriptions - Stop padding review start and end times in place, which widened the stored review item - Lock timestamp expiry in EventsPerSecond since update() and eps() now run on different threads * serialize recording frame extraction for review descriptions --- .../post/review_descriptions.py | 119 +++++++++--------- frigate/test/test_genai_processor_sync.py | 58 +++++++++ frigate/util/builtin.py | 9 +- 3 files changed, 126 insertions(+), 60 deletions(-) diff --git a/frigate/data_processing/post/review_descriptions.py b/frigate/data_processing/post/review_descriptions.py index d2f5619e12..c56d4e1961 100644 --- a/frigate/data_processing/post/review_descriptions.py +++ b/frigate/data_processing/post/review_descriptions.py @@ -68,6 +68,9 @@ class ReviewDescriptionProcessor(PostProcessorApi): self.review_desc_dps = EventsPerSecond() self.review_desc_dps.start() + # one review's ffmpeg extraction at a time, as on the maintainer loop + self._frame_extraction_lock = threading.Lock() + def calculate_frame_count( self, camera: str, @@ -179,53 +182,72 @@ class ReviewDescriptionProcessor(PostProcessorApi): ): return - image_source = camera_config.review.genai.image_source - frame_mode = camera_config.review.genai.frame_mode + threading.Thread( + target=self._describe_finished_review, + name=f"review_description_{id}", + daemon=True, + args=( + camera_config, + final_data, + camera_config.review.genai.image_source, + ), + ).start() - if image_source == ImageSourceEnum.recordings: - buffer_extension = get_recording_buffer_extension( - final_data["end_time"] - final_data["start_time"] - ) - final_data["start_time"] -= buffer_extension - final_data["end_time"] += buffer_extension + def _describe_finished_review( + self, + camera_config: CameraConfig, + final_data: dict[str, Any], + image_source: ImageSourceEnum, + preview_fallback: bool = True, + named_objects: list[tuple[str, str]] | None = None, + ) -> None: + """Gather frames for a finished review item and start its description. + Runs off the maintainer loop since recording frames shell out to ffmpeg + once per frame. + """ + camera = final_data["camera"] + id = final_data["id"] + genai_config = camera_config.review.genai + frames: list[tuple[bytes, float]] = [] + + if image_source == ImageSourceEnum.recordings: + buffer_extension = get_recording_buffer_extension( + final_data["end_time"] - final_data["start_time"] + ) + with self._frame_extraction_lock: frames = self.get_recording_frames( camera, - final_data["start_time"], - final_data["end_time"], + final_data["start_time"] - buffer_extension, + final_data["end_time"] + buffer_extension, height=480, # Use 480p for good balance between quality and token usage - frame_mode=frame_mode, + frame_mode=genai_config.frame_mode, ) - if not frames: - # Fallback to preview frames if no recordings available - logger.warning( - f"No recording frames found for {camera}, falling back to preview frames" - ) - frames = self.get_preview_frames_as_bytes( - camera, - final_data["start_time"], - final_data["end_time"], - final_data["thumb_path"], - id, - camera_config.review.genai.debug_save_thumbnails, - frame_mode, - ) - elif camera_config.review.genai.debug_save_thumbnails: + if frames: + if genai_config.debug_save_thumbnails: self.save_debug_recording_frames(id, frames) + elif not preview_fallback: + logger.error("No recording frames are available for review item %s", id) + return else: - # Use preview frames - frames = self.get_preview_frames_as_bytes( + logger.warning( + "No recording frames found for %s, falling back to preview frames", camera, - final_data["start_time"], - final_data["end_time"], - final_data["thumb_path"], - id, - camera_config.review.genai.debug_save_thumbnails, - frame_mode, ) - self.start_analysis(camera_config, final_data, frames) + if not frames: + frames = self.get_preview_frames_as_bytes( + camera, + final_data["start_time"], + final_data["end_time"], + final_data["thumb_path"], + id, + genai_config.debug_save_thumbnails, + genai_config.frame_mode, + ) + + self.start_analysis(camera_config, final_data, frames, named_objects) def handle_request(self, topic: str, request_data: dict[str, Any]) -> str | None: if topic == EmbeddingsRequestEnum.regenerate_review_description.value: @@ -419,31 +441,12 @@ class ReviewDescriptionProcessor(PostProcessorApi): logger.error("Review item %s has not ended yet", review_id) return - buffer_extension = get_recording_buffer_extension( - final_data["end_time"] - final_data["start_time"] - ) - frames = self.get_recording_frames( - str(review.camera), - final_data["start_time"] - buffer_extension, - final_data["end_time"] + buffer_extension, - height=480, - frame_mode=camera_config.review.genai.frame_mode, - ) - - if not frames: - logger.error( - "No recording frames are available for review item %s", review_id - ) - return - - if camera_config.review.genai.debug_save_thumbnails: - self.save_debug_recording_frames(review_id, frames) - - self.start_analysis( + self._describe_finished_review( camera_config, final_data, - frames, - get_named_objects( + ImageSourceEnum.recordings, + preview_fallback=False, + named_objects=get_named_objects( final_data["data"].get("detections") or [], self.config.all_attributes, ), diff --git a/frigate/test/test_genai_processor_sync.py b/frigate/test/test_genai_processor_sync.py index 41e2bf3355..163dce1843 100644 --- a/frigate/test/test_genai_processor_sync.py +++ b/frigate/test/test_genai_processor_sync.py @@ -5,6 +5,7 @@ per-camera gating those processors apply once they exist. """ import sys +import threading import unittest from unittest.mock import MagicMock, patch @@ -19,6 +20,8 @@ for mod in _MOCK_MODULES: if mod not in sys.modules: sys.modules[mod] = MagicMock() +from frigate.config.camera.review import ImageSourceEnum # noqa: E402 + # imported from the maintainer to avoid tripping the circular import between # the maintainer and the processor modules from frigate.embeddings.maintainer import ( # noqa: E402 @@ -211,3 +214,58 @@ class TestObjectDescriptionCameraGating(unittest.TestCase): mock_create_thumbnail.assert_called_once() self.assertEqual(len(processor.tracked_events["1234.5-abcdef"]), 1) + + +class TestReviewDescriptionOffLoop(unittest.TestCase): + """Frame extraction for a finished review must not block process_data.""" + + def test_recording_frames_extracted_off_the_caller_thread(self): + camera = MagicMock() + camera.review.genai.enabled = True + camera.review.genai.alerts = True + camera.review.genai.image_source = ImageSourceEnum.recordings + camera.review.genai.debug_save_thumbnails = False + config = MagicMock() + config.cameras = {"front": camera} + processor = ReviewDescriptionProcessor( + config, MagicMock(), MagicMock(), MagicMock() + ) + + release = threading.Event() + extracted = threading.Event() + analyzed = threading.Event() + + def slow_frames(*args, **kwargs): + release.wait(5) + extracted.set() + return [(b"jpg", 1.0)] + + with ( + patch.object(processor, "get_recording_frames", side_effect=slow_frames), + patch.object( + processor, "start_analysis", side_effect=lambda *a, **k: analyzed.set() + ) as start_analysis, + ): + processor.process_data( + { + "type": "end", + "after": { + "id": "1234.5-abcdef", + "camera": "front", + "severity": "alert", + "start_time": 100.0, + "end_time": 110.0, + }, + }, + PostProcessDataEnum.review, + ) + + self.assertFalse(extracted.is_set()) + release.set() + self.assertTrue(extracted.wait(5)) + + self.assertTrue(analyzed.wait(5)) + start_analysis.assert_called_once() + final_data = start_analysis.call_args.args[1] + self.assertEqual(final_data["start_time"], 100.0) + self.assertEqual(final_data["end_time"], 110.0) diff --git a/frigate/util/builtin.py b/frigate/util/builtin.py index 83ab6a78ca..181e76b166 100644 --- a/frigate/util/builtin.py +++ b/frigate/util/builtin.py @@ -9,6 +9,7 @@ import queue import re import shlex import struct +import threading import time import urllib.parse from collections import deque @@ -42,6 +43,7 @@ class EventsPerSecond: self._max_events = max_events self._last_n_seconds = last_n_seconds self._timestamps: deque[float] = deque(maxlen=max_events) + self._lock = threading.Lock() def start(self) -> None: self._start = time.monotonic() @@ -71,8 +73,11 @@ class EventsPerSecond: # remove aged out timestamps def expire_timestamps(self, now: float) -> None: threshold = now - self._last_n_seconds - while self._timestamps and self._timestamps[0] < threshold: - self._timestamps.popleft() + + # update() and eps() can run on different threads + with self._lock: + while self._timestamps and self._timestamps[0] < threshold: + self._timestamps.popleft() class InferenceSpeed: