use thumbnail and description for trigger types

This commit is contained in:
Josh Hawkins 2025-06-30 08:58:13 -05:00
parent e759557a4f
commit a17e202a9e
7 changed files with 42 additions and 72 deletions

View File

@ -48,6 +48,6 @@ class SubmitPlusBody(BaseModel):
class TriggerEmbeddingBody(BaseModel):
type: Literal["text", "image", "both"]
type: Literal["thumbnail", "description"]
data: str
threshold: float = Field(default=0.5, ge=0.0, le=1.0)

View File

@ -1479,9 +1479,9 @@ def create_trigger_embedding(
context: EmbeddingsContext = request.app.embeddings
# Generate embedding based on type
embedding = None
if body.type == "text" or body.type == "both":
if body.type == "description":
embedding = context.generate_description_embedding(body.data)
elif body.type == "image":
elif body.type == "thumbnail":
try:
event: Event = Event.get(Event.id == body.data)
except DoesNotExist:
@ -1584,9 +1584,9 @@ def update_trigger_embedding(
context: EmbeddingsContext = request.app.embeddings
# Generate embedding based on type
embedding = None
if body.type == "text" or body.type == "both":
if body.type == "description":
embedding = context.generate_description_embedding(body.data)
elif body.type == "image":
elif body.type == "thumbnail":
try:
event: Event = Event.get(Event.id == body.data)
except DoesNotExist:

View File

@ -26,9 +26,8 @@ class EnrichmentsDeviceEnum(str, Enum):
class TriggerType(str, Enum):
TEXT = "text"
IMAGE = "image"
BOTH = "both"
THUMBNAIL = "thumbnail"
DESCRIPTION = "description"
class TriggerAction(str, Enum):
@ -126,7 +125,7 @@ class SemanticSearchConfig(FrigateBaseModel):
class TriggerConfig(FrigateBaseModel):
type: TriggerType = Field(default=TriggerType.TEXT, title="Type of trigger")
type: TriggerType = Field(default=TriggerType.DESCRIPTION, title="Type of trigger")
data: str = Field(title="Trigger content (text phrase or image ID)")
threshold: float = Field(
title="Confidence score required to run the trigger.",

View File

@ -1,7 +1,9 @@
"""Real time processor to trigger alerts by matching embeddings."""
import datetime
import json
import logging
import os
from typing import Any
import cv2
@ -10,6 +12,7 @@ from peewee import DoesNotExist
from frigate.comms.inter_process import InterProcessRequestor
from frigate.config import FrigateConfig
from frigate.const import CONFIG_DIR
from frigate.data_processing.types import PostProcessDataEnum
from frigate.db.sqlitevecq import SqliteVecQueueDatabase
from frigate.embeddings.util import ZScoreNormalization
@ -41,6 +44,16 @@ class SemanticTriggerProcessor(PostProcessorApi):
self.trigger_embeddings: list[np.ndarray] = []
self.thumb_stats = ZScoreNormalization()
self.desc_stats = ZScoreNormalization()
# load stats from disk
try:
with open(os.path.join(CONFIG_DIR, ".search_stats.json"), "r") as f:
data = json.loads(f.read())
self.thumb_stats.from_dict(data["thumb_stats"])
self.desc_stats.from_dict(data["desc_stats"])
except FileNotFoundError:
pass
def process_data(
self, data: dict[str, Any], data_type: PostProcessDataEnum
@ -105,60 +118,22 @@ class SemanticTriggerProcessor(PostProcessorApi):
trigger_embedding = np.frombuffer(trigger["embedding"], dtype=np.float32)
# Determine which embedding to compare based on trigger type
if trigger["type"] == "image" and thumbnail_embedding is not None:
if trigger["type"] == "text" and thumbnail_embedding is not None:
data_embedding = thumbnail_embedding
normalized_distance = self.thumb_stats.normalize(
[cosine_distance(data_embedding, trigger_embedding)],
save_stats=False,
)[0]
elif trigger["type"] == "text" and description_embedding is not None:
data_embedding = description_embedding
elif trigger["type"] == "thumbnail" and thumbnail_embedding is not None:
data_embedding = thumbnail_embedding
normalized_distance = cosine_distance(data_embedding, trigger_embedding)
elif trigger["type"] == "both":
# For "both" type triggers, check both embeddings and use the best match
similarities = []
similarity_sources = [] # Track which embedding produced each similarity
elif trigger["type"] == "description" and description_embedding is not None:
data_embedding = description_embedding
normalized_distance = self.desc_stats.normalize(
[cosine_distance(data_embedding, trigger_embedding)],
save_stats=False,
)[0]
if thumbnail_embedding is not None:
thumb_distance = cosine_distance(
thumbnail_embedding, trigger_embedding
)
thumb_normalized = self.thumb_stats.normalize(
[thumb_distance], save_stats=False
)[0]
thumb_similarity = 1 - thumb_normalized
similarities.append(thumb_similarity)
similarity_sources.append("thumbnail")
if description_embedding is not None:
desc_distance = cosine_distance(
description_embedding, trigger_embedding
)
desc_similarity = 1 - desc_distance
similarities.append(desc_similarity)
similarity_sources.append("description")
if not similarities:
continue # Skip if no valid embeddings
# Find the best similarity and its source
max_similarity_idx = similarities.index(max(similarities))
similarity = similarities[max_similarity_idx]
selected_source = similarity_sources[max_similarity_idx]
normalized_distance = 1 - similarity
# Debug log showing all similarities and which was selected
if len(similarities) > 1:
logger.debug(
f"Both embeddings available for trigger '{trigger['name']}': "
f"thumbnail={similarities[0]:.4f}, description={similarities[1]:.4f}, "
f"selected={selected_source} with similarity={similarity:.4f}"
)
else:
logger.debug(
f"Single embedding available for trigger '{trigger['name']}': "
f"{selected_source}={similarity:.4f}"
)
else:
# Skip trigger if embedding type doesn't match available data
continue
@ -166,7 +141,7 @@ class SemanticTriggerProcessor(PostProcessorApi):
similarity = 1 - normalized_distance
logger.debug(
f"Trigger for {trigger['data'] if trigger['type'] == 'text' else 'image/both'} "
f"Trigger for {trigger['data'] if trigger['type'] == 'text' or trigger['type'] == 'description' else 'image'} "
f"(camera: {trigger['camera']}): normalized: {normalized_distance:.4f}, "
f"similarity: {similarity:.4f}, threshold: {trigger['threshold']}"
)

View File

@ -663,9 +663,8 @@
"deleteTrigger": "Delete Trigger"
},
"type": {
"image": "Image",
"text": "Text",
"both": "Both"
"thumbnail": "Thumbnail",
"description": "Description"
},
"actions": {
"alert": "Mark as Alert",

View File

@ -86,7 +86,7 @@ export default function CreateTriggerDialog({
!existingTriggerNames.includes(value) || value === trigger?.name,
t("triggers.dialog.form.name.error.alreadyExists"),
),
type: z.enum(["image", "text", "both"]),
type: z.enum(["thumbnail", "description"]),
data: z.string().min(1, t("triggers.dialog.form.content.error.required")),
threshold: z
.number()
@ -102,7 +102,7 @@ export default function CreateTriggerDialog({
mode: "onChange",
defaultValues: {
name: trigger?.name ?? "",
type: trigger?.type ?? "text",
type: trigger?.type ?? "description",
data: trigger?.data ?? "",
threshold: trigger?.threshold ?? 0.5,
actions: trigger?.actions ?? ["alert"],
@ -129,7 +129,7 @@ export default function CreateTriggerDialog({
if (!show) {
form.reset({
name: "",
type: "text",
type: "description",
data: "",
threshold: 0.5,
actions: ["alert"],
@ -214,14 +214,11 @@ export default function CreateTriggerDialog({
</SelectTrigger>
</FormControl>
<SelectContent>
<SelectItem value="image">
{t("triggers.type.image")}
<SelectItem value="thumbnail">
{t("triggers.type.thumbnail")}
</SelectItem>
<SelectItem value="text">
{t("triggers.type.text")}
</SelectItem>
<SelectItem value="both">
{t("triggers.type.both")}
<SelectItem value="description">
{t("triggers.type.description")}
</SelectItem>
</SelectContent>
</Select>
@ -238,7 +235,7 @@ export default function CreateTriggerDialog({
<FormLabel>
{t("triggers.dialog.form.content.title")}
</FormLabel>
{form.watch("type") === "image" ? (
{form.watch("type") === "thumbnail" ? (
<FormControl>
<ImagePicker
selectedImageId={field.value}

View File

@ -1,4 +1,4 @@
export type TriggerType = "image" | "text" | "both";
export type TriggerType = "thumbnail" | "description";
export type TriggerAction = "alert" | "notification";
export type Trigger = {