Compare commits

..
Author SHA1 Message Date
Nicolas Mowen 1a3abd472a Pull from event DB directly 2026-11-04 10:54:50 -07:00
Nicolas Mowen 9fe12ed0b7 Fix reading labels when there are multiple that match 2026-11-03 09:17:26 -07:00
6 changed files with 273 additions and 241 deletions
+1 -3
View File
@@ -1,6 +1,5 @@
"""Chat and LLM tool calling APIs.""" """Chat and LLM tool calling APIs."""
import asyncio
import base64 import base64
import json import json
import logging import logging
@@ -1654,8 +1653,7 @@ async def chat_completion(
f"Calling LLM (iteration {tool_iterations + 1}/{max_iterations}) " f"Calling LLM (iteration {tool_iterations + 1}/{max_iterations}) "
f"with {len(conversation)} message(s) in conversation" f"with {len(conversation)} message(s) in conversation"
) )
response = await asyncio.to_thread( response = genai_client.chat_with_tools(
genai_client.chat_with_tools,
messages=conversation, messages=conversation,
tools=llm_tools if llm_tools else None, tools=llm_tools if llm_tools else None,
tool_choice="auto", tool_choice="auto",
@@ -34,7 +34,7 @@ from frigate.const import (
from frigate.data_processing.types import PostProcessDataEnum from frigate.data_processing.types import PostProcessDataEnum
from frigate.genai import GenAIClient from frigate.genai import GenAIClient
from frigate.genai.manager import GenAIClientManager 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.builtin import EventsPerSecond, InferenceSpeed
from frigate.util.image import get_image_from_recording from frigate.util.image import get_image_from_recording
from frigate.util.recording_coverage import recording_overlap from frigate.util.recording_coverage import recording_overlap
@@ -439,13 +439,22 @@ class ReviewDescriptionProcessor(PostProcessorApi):
if camera_config.review.genai.debug_save_thumbnails: if camera_config.review.genai.debug_save_thumbnails:
self.save_debug_recording_frames(review_id, frames) 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( def start_analysis(
self, self,
camera_config: CameraConfig, camera_config: CameraConfig,
final_data: dict[str, Any], final_data: dict[str, Any],
frames: list[tuple[bytes, float]], frames: list[tuple[bytes, float]],
named_objects: list[tuple[str, str]] | None = None,
) -> None: ) -> None:
"""Kick off description generation for a review item in the background.""" """Kick off description generation for a review item in the background."""
thumbs = [frame for frame, _ in frames] thumbs = [frame for frame, _ in frames]
@@ -481,6 +490,7 @@ class ReviewDescriptionProcessor(PostProcessorApi):
camera_config.review.genai, camera_config.review.genai,
sorted(self.config.all_labels), sorted(self.config.all_labels),
self.config.all_attributes, self.config.all_attributes,
named_objects,
), ),
).start() ).start()
@@ -741,6 +751,29 @@ def format_classification_state_changes(
return lines 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( def run_analysis(
requestor: InterProcessRequestor, requestor: InterProcessRequestor,
genai_client: GenAIClient, genai_client: GenAIClient,
@@ -752,6 +785,7 @@ def run_analysis(
genai_config: GenAIReviewConfig, genai_config: GenAIReviewConfig,
labelmap_objects: list[str], labelmap_objects: list[str],
attribute_labels: list[str], attribute_labels: list[str],
named_objects: list[tuple[str, str]] | None,
) -> None: ) -> None:
start = datetime.datetime.now().timestamp() start = datetime.datetime.now().timestamp()
@@ -776,11 +810,20 @@ def run_analysis(
unified_objects = [] unified_objects = []
objects_list = final_data["data"]["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"]): if named_objects is None:
object_type = verified_label.replace("-verified", "").replace("_", " ") # live review items record a sub label for every verified object
name = titlecase(sub_labels_list[i].replace("_", " ")) 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}") unified_objects.append(f"{name} ← {object_type}")
for label in objects_list: for label in objects_list:
+10 -18
View File
@@ -7,8 +7,7 @@ overrides what is genuinely Azure-specific:
- Client construction: parses ``api-version`` out of the configured - Client construction: parses ``api-version`` out of the configured
``base_url`` query string and instantiates :class:`openai.AzureOpenAI` ``base_url`` query string and instantiates :class:`openai.AzureOpenAI`
(or :class:`openai.AsyncAzureOpenAI` for streaming chat) with with ``azure_endpoint`` instead of ``base_url``. Raises if the URL is
``azure_endpoint`` instead of ``base_url``. Raises if the URL is
malformed; :class:`GenAIClientManager` catches the exception and malformed; :class:`GenAIClientManager` catches the exception and
disables the provider. disables the provider.
- Context size: Azure does not expose a per-model ``max_model_len`` field - Context size: Azure does not expose a per-model ``max_model_len`` field
@@ -30,10 +29,9 @@ the ``transcribe`` role:
""" """
import logging import logging
from typing import Any
from urllib.parse import parse_qs, urlparse from urllib.parse import parse_qs, urlparse
from openai import AsyncAzureOpenAI, AzureOpenAI from openai import AzureOpenAI
from frigate.config import GenAIProviderEnum from frigate.config import GenAIProviderEnum
from frigate.genai import register_genai_provider from frigate.genai import register_genai_provider
@@ -46,8 +44,8 @@ logger = logging.getLogger(__name__)
class AzureOpenAIClient(OpenAIClient): class AzureOpenAIClient(OpenAIClient):
"""Generative AI client for Frigate using Azure OpenAI.""" """Generative AI client for Frigate using Azure OpenAI."""
def _client_options(self) -> dict[str, Any]: def _init_provider(self) -> AzureOpenAI:
"""Build the Azure client options from the configured base_url.""" """Initialize the AzureOpenAI client from the configured base_url."""
parsed_url = urlparse(self.genai_config.base_url or "") parsed_url = urlparse(self.genai_config.base_url or "")
query_params = parse_qs(parsed_url.query) query_params = parse_qs(parsed_url.query)
api_version = query_params.get("api-version", [None])[0] api_version = query_params.get("api-version", [None])[0]
@@ -55,19 +53,13 @@ class AzureOpenAIClient(OpenAIClient):
if not api_version: if not api_version:
raise ValueError("Azure OpenAI base_url is missing api-version.") raise ValueError("Azure OpenAI base_url is missing api-version.")
return { azure_endpoint = f"{parsed_url.scheme}://{parsed_url.netloc}/"
"api_key": self.genai_config.api_key,
"api_version": api_version,
"azure_endpoint": f"{parsed_url.scheme}://{parsed_url.netloc}/",
}
def _init_provider(self) -> AzureOpenAI: return AzureOpenAI(
"""Initialize the AzureOpenAI client from the configured base_url.""" api_key=self.genai_config.api_key,
return AzureOpenAI(**self._client_options()) api_version=api_version,
azure_endpoint=azure_endpoint,
def _init_async_provider(self) -> AsyncAzureOpenAI: )
"""Create an async Azure client for streaming chat."""
return AsyncAzureOpenAI(**self._client_options())
def get_context_size(self) -> int: def get_context_size(self) -> int:
"""Azure does not reliably surface per-model context size; use 128K.""" """Azure does not reliably surface per-model context size; use 128K."""
+47 -64
View File
@@ -7,7 +7,7 @@ from collections.abc import AsyncGenerator
from typing import Any from typing import Any
from httpx import TimeoutException from httpx import TimeoutException
from openai import AsyncOpenAI, OpenAI from openai import OpenAI
from frigate.config import GenAIProviderEnum from frigate.config import GenAIProviderEnum
from frigate.genai import GenAIClient, register_genai_provider from frigate.genai import GenAIClient, register_genai_provider
@@ -15,7 +15,6 @@ from frigate.genai.utils import interleave_images
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# gpt-transcribe replaced the singular `language` field with a `languages` array # gpt-transcribe replaced the singular `language` field with a `languages` array
# and rejects a request that sends both. Older transcription models # and rejects a request that sends both. Older transcription models
# (gpt-4o-transcribe, gpt-4o-mini-transcribe, whisper-1) still take the singular # (gpt-4o-transcribe, gpt-4o-mini-transcribe, whisper-1) still take the singular
@@ -46,8 +45,8 @@ class OpenAIClient(GenAIClient):
provider: OpenAI provider: OpenAI
context_size: int | None = None context_size: int | None = None
def _client_options(self) -> dict[str, Any]: def _init_provider(self) -> OpenAI:
"""Build the constructor options shared by the sync and async clients. """Initialize the client.
Subclasses (e.g. Azure) should raise on configuration errors; the Subclasses (e.g. Azure) should raise on configuration errors; the
manager catches construction failures and disables the provider. manager catches construction failures and disables the provider.
@@ -63,19 +62,7 @@ class OpenAIClient(GenAIClient):
if self.genai_config.base_url: if self.genai_config.base_url:
provider_opts["base_url"] = self.genai_config.base_url provider_opts["base_url"] = self.genai_config.base_url
return {"api_key": self.genai_config.api_key, **provider_opts} return OpenAI(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())
def _send( def _send(
self, self,
@@ -425,61 +412,57 @@ class OpenAIClient(GenAIClient):
finish_reason = "stop" finish_reason = "stop"
usage_stats: dict[str, Any] | None = None usage_stats: dict[str, Any] | None = None
# Exiting these blocks, including when the caller disconnects and stream = self.provider.chat.completions.create(**request_params)
# 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:
chunk_usage = getattr(chunk, "usage", None)
if chunk_usage is not None:
usage_stats = _stats_from_openai_usage(chunk_usage)
if not chunk or not chunk.choices: for chunk in stream:
continue chunk_usage = getattr(chunk, "usage", None)
if chunk_usage is not None:
usage_stats = _stats_from_openai_usage(chunk_usage)
choice = chunk.choices[0] if not chunk or not chunk.choices:
delta = choice.delta continue
# Check for finish reason choice = chunk.choices[0]
if choice.finish_reason: delta = choice.delta
finish_reason = choice.finish_reason
# Extract reasoning deltas (reasoning_content or reasoning, # Check for finish reason
# depending on the server) if choice.finish_reason:
reasoning_delta = getattr( finish_reason = choice.finish_reason
delta, "reasoning_content", None
) or getattr(delta, "reasoning", None)
if reasoning_delta:
reasoning_parts.append(reasoning_delta)
yield ("reasoning_delta", reasoning_delta)
# Extract content deltas # Extract reasoning deltas (reasoning_content or reasoning,
if delta.content: # depending on the server)
content_parts.append(delta.content) reasoning_delta = getattr(delta, "reasoning_content", None) or getattr(
yield ("content_delta", delta.content) delta, "reasoning", None
)
if reasoning_delta:
reasoning_parts.append(reasoning_delta)
yield ("reasoning_delta", reasoning_delta)
# Extract tool calls # Extract content deltas
if delta.tool_calls: if delta.content:
for tc in delta.tool_calls: content_parts.append(delta.content)
idx = tc.index yield ("content_delta", delta.content)
fn = tc.function
if idx not in tool_calls_by_index: # Extract tool calls
tool_calls_by_index[idx] = { if delta.tool_calls:
"id": tc.id or "", for tc in delta.tool_calls:
"name": fn.name if fn and fn.name else "", idx = tc.index
"arguments": "", fn = tc.function
}
t = tool_calls_by_index[idx] if idx not in tool_calls_by_index:
if tc.id: tool_calls_by_index[idx] = {
t["id"] = tc.id "id": tc.id or "",
if fn and fn.name: "name": fn.name if fn and fn.name else "",
t["name"] = fn.name "arguments": "",
if fn and fn.arguments: }
t["arguments"] += fn.arguments
t = tool_calls_by_index[idx]
if tc.id:
t["id"] = tc.id
if fn and fn.name:
t["name"] = fn.name
if fn and fn.arguments:
t["arguments"] += fn.arguments
# Build final message # Build final message
full_content = "".join(content_parts).strip() or None full_content = "".join(content_parts).strip() or None
+7 -150
View File
@@ -145,72 +145,12 @@ def _openai_chunk(content=None, tool_calls=None, finish_reason=None, usage=None)
return SimpleNamespace(choices=[choice], usage=usage) 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): class TestOpenAIProvider(unittest.TestCase):
def _client(self): def _client(self):
return _make_client( return _make_client(
"openai", model="gpt-4o", api_key="k", base_url="http://localhost:9999/v1" "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): def test_stream_tool_call_arguments_are_dict(self):
# Arguments arrive split across chunks, as the real API streams them. # Arguments arrive split across chunks, as the real API streams them.
chunks = [ chunks = [
@@ -222,9 +162,8 @@ class TestOpenAIProvider(unittest.TestCase):
_openai_chunk(tool_calls=[_openai_tc(0, arguments=' "person"}')]), _openai_chunk(tool_calls=[_openai_tc(0, arguments=' "person"}')]),
_openai_chunk(finish_reason="tool_calls"), _openai_chunk(finish_reason="tool_calls"),
] ]
client, _ = self._streaming_client( client = self._client()
AsyncMock(return_value=_FakeAsyncOpenAIStream(chunks)) client.provider.chat.completions.create = MagicMock(return_value=iter(chunks))
)
final = _final_message(_collect(client, SIMPLE_MESSAGES)) final = _final_message(_collect(client, SIMPLE_MESSAGES))
self.assertEqual(final["finish_reason"], "tool_calls") self.assertEqual(final["finish_reason"], "tool_calls")
@@ -238,9 +177,8 @@ class TestOpenAIProvider(unittest.TestCase):
_openai_chunk(content="lo"), _openai_chunk(content="lo"),
_openai_chunk(finish_reason="stop"), _openai_chunk(finish_reason="stop"),
] ]
client, _ = self._streaming_client( client = self._client()
AsyncMock(return_value=_FakeAsyncOpenAIStream(chunks)) client.provider.chat.completions.create = MagicMock(return_value=iter(chunks))
)
events = _collect(client, SIMPLE_MESSAGES) events = _collect(client, SIMPLE_MESSAGES)
deltas = [v for (k, v) in events if k == "content_delta"] 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") self.assertEqual(_final_message(events)["content"], "hello")
def test_multimodal_message_does_not_crash(self): def test_multimodal_message_does_not_crash(self):
client, _ = self._streaming_client( client = self._client()
AsyncMock( client.provider.chat.completions.create = MagicMock(
return_value=_FakeAsyncOpenAIStream( return_value=iter([_openai_chunk(content="ok", finish_reason="stop")])
[_openai_chunk(content="ok", finish_reason="stop")]
)
)
) )
# Passing the OpenAI-native multimodal list through must not raise. # Passing the OpenAI-native multimodal list through must not raise.
final = _final_message(_collect(client, MULTIMODAL_MESSAGES)) final = _final_message(_collect(client, MULTIMODAL_MESSAGES))
self.assertEqual(final["content"], "ok") 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 # Gemini
@@ -1072,17 +940,6 @@ class TestAzureOpenAITranscribe(unittest.TestCase):
self.assertIsInstance(client.provider, AzureOpenAI) self.assertIsInstance(client.provider, AzureOpenAI)
self.assertTrue(client.supports_transcription) 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): def test_transcribe_inherited(self):
client = self._client() client = self._client()
create = MagicMock(return_value="azure text") create = MagicMock(return_value="azure text")
+159
View File
@@ -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()