Pull from event DB directly

frigate/api/chat.py

@@ -1,6 +1,5 @@
 """Chat and LLM tool calling APIs."""
 
-import asyncio
 import base64
 import json
 import logging
@@ -1654,8 +1653,7 @@ async def chat_completion(
                     f"Calling LLM (iteration {tool_iterations + 1}/{max_iterations}) "
                     f"with {len(conversation)} message(s) in conversation"
                 )
-                response = await asyncio.to_thread(
-                    genai_client.chat_with_tools,
+                response = genai_client.chat_with_tools(
                     messages=conversation,
                     tools=llm_tools if llm_tools else None,
                     tool_choice="auto",

frigate/data_processing/post/review_descriptions.py

@@ -34,7 +34,7 @@ from frigate.const import (
 from frigate.data_processing.types import PostProcessDataEnum
 from frigate.genai import GenAIClient
 from frigate.genai.manager import GenAIClientManager
-from frigate.models import Recordings, ReviewSegment
+from frigate.models import Event, Recordings, ReviewSegment
 from frigate.util.builtin import EventsPerSecond, InferenceSpeed
 from frigate.util.image import get_image_from_recording
 from frigate.util.recording_coverage import recording_overlap
@@ -439,13 +439,22 @@ class ReviewDescriptionProcessor(PostProcessorApi):
         if camera_config.review.genai.debug_save_thumbnails:
             self.save_debug_recording_frames(review_id, frames)
 
-        self.start_analysis(camera_config, final_data, frames)
+        self.start_analysis(
+            camera_config,
+            final_data,
+            frames,
+            get_named_objects(
+                final_data["data"].get("detections") or [],
+                self.config.all_attributes,
+            ),
+        )
 
     def start_analysis(
         self,
         camera_config: CameraConfig,
         final_data: dict[str, Any],
         frames: list[tuple[bytes, float]],
+        named_objects: list[tuple[str, str]] | None = None,
     ) -> None:
         """Kick off description generation for a review item in the background."""
         thumbs = [frame for frame, _ in frames]
@@ -481,6 +490,7 @@ class ReviewDescriptionProcessor(PostProcessorApi):
                 camera_config.review.genai,
                 sorted(self.config.all_labels),
                 self.config.all_attributes,
+                named_objects,
             ),
         ).start()
 
@@ -741,6 +751,29 @@ def format_classification_state_changes(
     return lines
 
 
+def get_named_objects(
+    detection_ids: list[str], attribute_labels: list[str]
+) -> list[tuple[str, str]]:
+    """Get the label and sub label of each named tracked object in a review item.
+
+    Editing a sub label after a review item ends rebuilds the stored sub_labels
+    without verified_objects, so the stored lists can't be paired by position.
+    """
+    events = (
+        Event.select(Event.label, Event.sub_label)
+        .where(Event.id.in_(detection_ids), Event.sub_label.is_null(False))
+        .order_by(Event.start_time)
+    )
+
+    # attribute sub labels (eg, delivery logos) are not names, which matches
+    # how the review maintainer records them
+    return [
+        (event.label, event.sub_label)
+        for event in events
+        if event.sub_label and event.sub_label not in attribute_labels
+    ]
+
+
 def run_analysis(
     requestor: InterProcessRequestor,
     genai_client: GenAIClient,
@@ -752,6 +785,7 @@ def run_analysis(
     genai_config: GenAIReviewConfig,
     labelmap_objects: list[str],
     attribute_labels: list[str],
+    named_objects: list[tuple[str, str]] | None,
 ) -> None:
     start = datetime.datetime.now().timestamp()
 
@@ -776,11 +810,20 @@ def run_analysis(
     unified_objects = []
 
     objects_list = final_data["data"]["objects"]
-    sub_labels_list = final_data["data"]["sub_labels"]
 
-    for i, verified_label in enumerate(final_data["data"]["verified_objects"]):
-        object_type = verified_label.replace("-verified", "").replace("_", " ")
-        name = titlecase(sub_labels_list[i].replace("_", " "))
+    if named_objects is None:
+        # live review items record a sub label for every verified object
+        named_objects = [
+            (verified_label.replace("-verified", ""), sub_label)
+            for verified_label, sub_label in zip(
+                final_data["data"]["verified_objects"],
+                final_data["data"]["sub_labels"],
+            )
+        ]
+
+    for label, sub_label in named_objects:
+        object_type = label.replace("_", " ")
+        name = titlecase(sub_label.replace("_", " "))
         unified_objects.append(f"{name} ← {object_type}")
 
     for label in objects_list:

frigate/genai/plugins/azure-openai.py

@@ -7,8 +7,7 @@ overrides what is genuinely Azure-specific:
 
 - Client construction: parses ``api-version`` out of the configured
   ``base_url`` query string and instantiates :class:`openai.AzureOpenAI`
-  (or :class:`openai.AsyncAzureOpenAI` for streaming chat) with
-  ``azure_endpoint`` instead of ``base_url``. Raises if the URL is
+  with ``azure_endpoint`` instead of ``base_url``. Raises if the URL is
   malformed; :class:`GenAIClientManager` catches the exception and
   disables the provider.
 - Context size: Azure does not expose a per-model ``max_model_len`` field
@@ -30,10 +29,9 @@ the ``transcribe`` role:
 """
 
 import logging
-from typing import Any
 from urllib.parse import parse_qs, urlparse
 
-from openai import AsyncAzureOpenAI, AzureOpenAI
+from openai import AzureOpenAI
 
 from frigate.config import GenAIProviderEnum
 from frigate.genai import register_genai_provider
@@ -46,8 +44,8 @@ logger = logging.getLogger(__name__)
 class AzureOpenAIClient(OpenAIClient):
     """Generative AI client for Frigate using Azure OpenAI."""
 
-    def _client_options(self) -> dict[str, Any]:
-        """Build the Azure client options from the configured base_url."""
+    def _init_provider(self) -> AzureOpenAI:
+        """Initialize the AzureOpenAI client from the configured base_url."""
         parsed_url = urlparse(self.genai_config.base_url or "")
         query_params = parse_qs(parsed_url.query)
         api_version = query_params.get("api-version", [None])[0]
@@ -55,19 +53,13 @@ class AzureOpenAIClient(OpenAIClient):
         if not api_version:
             raise ValueError("Azure OpenAI base_url is missing api-version.")
 
-        return {
-            "api_key": self.genai_config.api_key,
-            "api_version": api_version,
-            "azure_endpoint": f"{parsed_url.scheme}://{parsed_url.netloc}/",
-        }
+        azure_endpoint = f"{parsed_url.scheme}://{parsed_url.netloc}/"
 
-    def _init_provider(self) -> AzureOpenAI:
-        """Initialize the AzureOpenAI client from the configured base_url."""
-        return AzureOpenAI(**self._client_options())
-
-    def _init_async_provider(self) -> AsyncAzureOpenAI:
-        """Create an async Azure client for streaming chat."""
-        return AsyncAzureOpenAI(**self._client_options())
+        return AzureOpenAI(
+            api_key=self.genai_config.api_key,
+            api_version=api_version,
+            azure_endpoint=azure_endpoint,
+        )
 
     def get_context_size(self) -> int:
         """Azure does not reliably surface per-model context size; use 128K."""

frigate/genai/plugins/openai.py

@@ -7,7 +7,7 @@ from collections.abc import AsyncGenerator
 from typing import Any
 
 from httpx import TimeoutException
-from openai import AsyncOpenAI, OpenAI
+from openai import OpenAI
 
 from frigate.config import GenAIProviderEnum
 from frigate.genai import GenAIClient, register_genai_provider
@@ -15,7 +15,6 @@ from frigate.genai.utils import interleave_images
 
 logger = logging.getLogger(__name__)
 
-
 # gpt-transcribe replaced the singular `language` field with a `languages` array
 # and rejects a request that sends both. Older transcription models
 # (gpt-4o-transcribe, gpt-4o-mini-transcribe, whisper-1) still take the singular
@@ -46,8 +45,8 @@ class OpenAIClient(GenAIClient):
     provider: OpenAI
     context_size: int | None = None
 
-    def _client_options(self) -> dict[str, Any]:
-        """Build the constructor options shared by the sync and async clients.
+    def _init_provider(self) -> OpenAI:
+        """Initialize the client.
 
         Subclasses (e.g. Azure) should raise on configuration errors; the
         manager catches construction failures and disables the provider.
@@ -63,19 +62,7 @@ class OpenAIClient(GenAIClient):
         if self.genai_config.base_url:
             provider_opts["base_url"] = self.genai_config.base_url
 
-        return {"api_key": self.genai_config.api_key, **provider_opts}
-
-    def _init_provider(self) -> OpenAI:
-        """Initialize the client."""
-        return OpenAI(**self._client_options())
-
-    def _init_async_provider(self) -> AsyncOpenAI:
-        """Create an async client for streaming chat.
-
-        A new client is created per stream because its connection pool binds
-        to the event loop that first uses it.
-        """
-        return AsyncOpenAI(**self._client_options())
+        return OpenAI(api_key=self.genai_config.api_key, **provider_opts)
 
     def _send(
         self,
@@ -425,13 +412,9 @@ class OpenAIClient(GenAIClient):
             finish_reason = "stop"
             usage_stats: dict[str, Any] | None = None
 
-            # Exiting these blocks, including when the caller disconnects and
-            # the generator is cancelled, closes the response and its connection
-            async with (
-                self._init_async_provider() as client,
-                await client.chat.completions.create(**request_params) as stream,
-            ):
-                async for chunk in stream:
+            stream = self.provider.chat.completions.create(**request_params)
+
+            for chunk in stream:
                 chunk_usage = getattr(chunk, "usage", None)
                 if chunk_usage is not None:
                     usage_stats = _stats_from_openai_usage(chunk_usage)
@@ -448,9 +431,9 @@ class OpenAIClient(GenAIClient):
 
                 # Extract reasoning deltas (reasoning_content or reasoning,
                 # depending on the server)
-                    reasoning_delta = getattr(
-                        delta, "reasoning_content", None
-                    ) or getattr(delta, "reasoning", None)
+                reasoning_delta = getattr(delta, "reasoning_content", None) or getattr(
+                    delta, "reasoning", None
+                )
                 if reasoning_delta:
                     reasoning_parts.append(reasoning_delta)
                     yield ("reasoning_delta", reasoning_delta)

frigate/test/test_genai_providers.py

@@ -145,72 +145,12 @@ def _openai_chunk(content=None, tool_calls=None, finish_reason=None, usage=None)
     return SimpleNamespace(choices=[choice], usage=usage)
 
 
-class _FakeAsyncOpenAIStream:
-    """Mimics openai.AsyncStream: async iterable and async context manager.
-
-    With ``stall_after`` set, the stream stops producing after that many
-    chunks, like a model that stops sending tokens, and sets ``stalled``.
-    """
-
-    def __init__(self, chunks, stall_after=None):
-        self._chunks = chunks
-        self._stall_after = stall_after
-        self.stalled = asyncio.Event()
-        self.close = AsyncMock()
-
-    async def __aiter__(self):
-        for i, chunk in enumerate(self._chunks):
-            if i == self._stall_after:
-                self.stalled.set()
-                await asyncio.Event().wait()
-            yield chunk
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, *exc):
-        await self.close()
-
-
-class _FakeAsyncOpenAI:
-    """Mimics openai.AsyncOpenAI: async context manager with a completions API."""
-
-    def __init__(self, create):
-        self.chat = SimpleNamespace(completions=SimpleNamespace(create=create))
-        self.close = AsyncMock()
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, *exc):
-        await self.close()
-
-
 class TestOpenAIProvider(unittest.TestCase):
     def _client(self):
         return _make_client(
             "openai", model="gpt-4o", api_key="k", base_url="http://localhost:9999/v1"
         )
 
-    def _streaming_client(self, create):
-        """Build a client whose async completions call is ``create``."""
-        client = self._client()
-        async_client = _FakeAsyncOpenAI(create)
-        client._init_async_provider = MagicMock(return_value=async_client)
-        return client, async_client
-
-    def test_async_client_uses_configured_options(self):
-        from openai import AsyncOpenAI
-
-        client = self._client()
-        async_client = client._init_async_provider()
-        try:
-            self.assertIsInstance(async_client, AsyncOpenAI)
-            self.assertEqual(async_client.base_url, client.provider.base_url)
-            self.assertEqual(async_client.api_key, "k")
-        finally:
-            asyncio.run(async_client.close())
-
     def test_stream_tool_call_arguments_are_dict(self):
         # Arguments arrive split across chunks, as the real API streams them.
         chunks = [
@@ -222,9 +162,8 @@ class TestOpenAIProvider(unittest.TestCase):
             _openai_chunk(tool_calls=[_openai_tc(0, arguments=' "person"}')]),
             _openai_chunk(finish_reason="tool_calls"),
         ]
-        client, _ = self._streaming_client(
-            AsyncMock(return_value=_FakeAsyncOpenAIStream(chunks))
-        )
+        client = self._client()
+        client.provider.chat.completions.create = MagicMock(return_value=iter(chunks))
 
         final = _final_message(_collect(client, SIMPLE_MESSAGES))
         self.assertEqual(final["finish_reason"], "tool_calls")
@@ -238,9 +177,8 @@ class TestOpenAIProvider(unittest.TestCase):
             _openai_chunk(content="lo"),
             _openai_chunk(finish_reason="stop"),
         ]
-        client, _ = self._streaming_client(
-            AsyncMock(return_value=_FakeAsyncOpenAIStream(chunks))
-        )
+        client = self._client()
+        client.provider.chat.completions.create = MagicMock(return_value=iter(chunks))
 
         events = _collect(client, SIMPLE_MESSAGES)
         deltas = [v for (k, v) in events if k == "content_delta"]
@@ -248,84 +186,14 @@ class TestOpenAIProvider(unittest.TestCase):
         self.assertEqual(_final_message(events)["content"], "hello")
 
     def test_multimodal_message_does_not_crash(self):
-        client, _ = self._streaming_client(
-            AsyncMock(
-                return_value=_FakeAsyncOpenAIStream(
-                    [_openai_chunk(content="ok", finish_reason="stop")]
-                )
-            )
+        client = self._client()
+        client.provider.chat.completions.create = MagicMock(
+            return_value=iter([_openai_chunk(content="ok", finish_reason="stop")])
         )
         # Passing the OpenAI-native multimodal list through must not raise.
         final = _final_message(_collect(client, MULTIMODAL_MESSAGES))
         self.assertEqual(final["content"], "ok")
 
-    def test_stream_uses_async_client_and_closes_it(self):
-        # The blocking sync client must never be used on the event loop.
-        stream = _FakeAsyncOpenAIStream(
-            [_openai_chunk(content="ok", finish_reason="stop")]
-        )
-        create = AsyncMock(return_value=stream)
-        client, async_client = self._streaming_client(create)
-        client.provider.chat.completions.create = MagicMock()
-
-        final = _final_message(_collect(client, SIMPLE_MESSAGES))
-
-        self.assertEqual(final["content"], "ok")
-        self.assertTrue(create.await_args.kwargs["stream"])
-        client.provider.chat.completions.create.assert_not_called()
-        stream.close.assert_awaited_once()
-        async_client.close.assert_awaited_once()
-
-    def test_client_closed_when_caller_leaves_while_opening(self):
-        # Cancelling before the response arrives must still close the client.
-        async def _run():
-            opening = asyncio.Event()
-
-            async def hang(**kwargs):
-                opening.set()
-                await asyncio.Event().wait()
-
-            client, async_client = self._streaming_client(hang)
-            gen = client.chat_with_tools_stream(
-                messages=SIMPLE_MESSAGES, tools=None, tool_choice="auto"
-            )
-            task = asyncio.create_task(gen.__anext__())
-            await opening.wait()
-            task.cancel()
-            with self.assertRaises(asyncio.CancelledError):
-                await asyncio.wait_for(task, timeout=1)
-            return async_client
-
-        async_client = asyncio.run(_run())
-        async_client.close.assert_awaited_once()
-
-    def test_stream_closed_when_caller_leaves_mid_stream(self):
-        # Cancelling while waiting on a stalled model must close the stream
-        # right away instead of waiting for the next chunk.
-        async def _run():
-            stream = _FakeAsyncOpenAIStream(
-                [_openai_chunk(content="a"), _openai_chunk(finish_reason="stop")],
-                stall_after=1,
-            )
-            client, async_client = self._streaming_client(
-                AsyncMock(return_value=stream)
-            )
-            gen = client.chat_with_tools_stream(
-                messages=SIMPLE_MESSAGES, tools=None, tool_choice="auto"
-            )
-            first = await gen.__anext__()
-            task = asyncio.create_task(gen.__anext__())
-            await stream.stalled.wait()
-            task.cancel()
-            with self.assertRaises(asyncio.CancelledError):
-                await asyncio.wait_for(task, timeout=1)
-            return first, stream, async_client
-
-        first, stream, async_client = asyncio.run(_run())
-        self.assertEqual(first, ("content_delta", "a"))
-        stream.close.assert_awaited_once()
-        async_client.close.assert_awaited_once()
-
 
 # ---------------------------------------------------------------------------
 # Gemini
@@ -1072,17 +940,6 @@ class TestAzureOpenAITranscribe(unittest.TestCase):
         self.assertIsInstance(client.provider, AzureOpenAI)
         self.assertTrue(client.supports_transcription)
 
-    def test_streaming_routes_through_async_azure_client(self):
-        from openai import AsyncAzureOpenAI
-
-        client = self._client()
-        async_client = client._init_async_provider()
-        try:
-            self.assertIsInstance(async_client, AsyncAzureOpenAI)
-            self.assertEqual(async_client.base_url, client.provider.base_url)
-        finally:
-            asyncio.run(async_client.close())
-
     def test_transcribe_inherited(self):
         client = self._client()
         create = MagicMock(return_value="azure text")

frigate/test/test_review_sub_label_sync.py

@@ -0,0 +1,159 @@
+"""Tests for pairing review sub labels with their verified objects."""
+
+import unittest
+from unittest.mock import MagicMock
+
+from peewee import SqliteDatabase
+
+from frigate.config import FrigateConfig
+from frigate.data_processing.post.review_descriptions import (
+    get_named_objects,
+    run_analysis,
+)
+from frigate.models import Event
+
+CONFIG = """
+mqtt:
+  enabled: False
+cameras:
+  front_door:
+    ffmpeg:
+      inputs:
+        - path: rtsp://10.0.0.1:554/video
+          roles:
+            - detect
+    detect:
+      width: 1920
+      height: 1080
+      fps: 5
+"""
+
+
+def unified_objects(
+    review_data: dict, named_objects: list[tuple[str, str]] | None = None
+) -> list[str]:
+    """Run the review analysis and return the objects sent to the GenAI client."""
+    camera_config = FrigateConfig.parse_yaml(CONFIG).cameras["front_door"]
+    client = MagicMock()
+    client.generate_review_description.return_value = None
+    run_analysis(
+        MagicMock(),
+        client,
+        MagicMock(),
+        camera_config,
+        {"id": "review1", "start_time": 0, "end_time": 10, "data": review_data},
+        [],
+        [],
+        camera_config.review.genai,
+        ["car", "person"],
+        ["amazon"],
+        named_objects,
+    )
+    return client.generate_review_description.call_args.args[0]["unified_objects"]
+
+
+class TestRunAnalysisSubLabels(unittest.TestCase):
+    def test_verified_objects_are_named_by_their_sub_label(self):
+        objects = unified_objects(
+            {
+                "zones": [],
+                "objects": ["person-verified", "car"],
+                "verified_objects": ["person-verified"],
+                "sub_labels": ["nick_m"],
+                "classification_state_changes": [],
+            }
+        )
+
+        self.assertEqual(objects, ["Nick M ← person", "car"])
+
+
+class TestRegeneratedSubLabels(unittest.TestCase):
+    """Review items whose stored lists were rebuilt after a sub label edit."""
+
+    def setUp(self):
+        self.db = SqliteDatabase(":memory:")
+        self.db.bind([Event])
+        self.db.create_tables([Event])
+
+    def tearDown(self):
+        self.db.close()
+
+    def _insert_event(
+        self, id: str, label: str, sub_label: str | None, start_time: float
+    ) -> None:
+        Event.create(
+            id=id,
+            label=label,
+            sub_label=sub_label,
+            camera="front_door",
+            start_time=start_time,
+            end_time=start_time + 10,
+            top_score=0.9,
+            score=0.9,
+            false_positive=False,
+            zones=[],
+            thumbnail="",
+            has_clip=True,
+            has_snapshot=True,
+            region=[],
+            box=[],
+            area=0,
+            retain_indefinitely=False,
+            plus_id="",
+            model_hash="",
+            detector_type="cpu",
+            model_type="ssd",
+            data={},
+        )
+
+    def test_cleared_sub_label_does_not_move_to_another_object(self):
+        self._insert_event("car1", "car", None, 1)
+        self._insert_event("person1", "person", "Nick", 2)
+
+        named = get_named_objects(["car1", "person1"], ["amazon"])
+        objects = unified_objects(
+            {
+                "zones": [],
+                "objects": ["car", "person-verified"],
+                "verified_objects": ["car-verified", "person-verified"],
+                "sub_labels": ["Nick"],
+                "classification_state_changes": [],
+            },
+            named,
+        )
+
+        self.assertEqual(objects, ["Nick ← person", "car"])
+
+    def test_repeated_name_is_kept_for_each_object(self):
+        # sub_labels are deduplicated when rebuilt, which previously raised an
+        # IndexError when the verified objects outnumbered them
+        self._insert_event("person1", "person", "Nick", 1)
+        self._insert_event("person2", "person", "Nick", 2)
+
+        named = get_named_objects(["person1", "person2"], ["amazon"])
+        objects = unified_objects(
+            {
+                "zones": [],
+                "objects": ["person-verified", "person-verified"],
+                "verified_objects": ["person-verified", "person-verified"],
+                "sub_labels": ["Nick"],
+                "classification_state_changes": [],
+            },
+            named,
+        )
+
+        self.assertEqual(objects, ["Nick ← person", "Nick ← person"])
+
+    def test_attribute_sub_labels_are_not_names(self):
+        self._insert_event("car1", "car", "amazon", 1)
+        self._insert_event("person1", "person", "Nick", 2)
+        self._insert_event("other", "person", "Bob", 3)
+
+        self.assertEqual(
+            get_named_objects(["car1", "person1"], ["amazon"]),
+            [("person", "Nick")],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()