administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-08-02 17:12:16 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Semantic Search Triggers (#18969)

Browse Source 
* semantic trigger test

* database and model

* config

* embeddings maintainer and trigger post-processor

* api to create, edit, delete triggers

* frontend and i18n keys

* use thumbnail and description for trigger types

* image picker tweaks

* initial sync

* thumbnail file management

* clean up logs and use saved thumbnail on frontend

* publish mqtt messages

* webpush changes to enable trigger notifications

* add enabled switch

* add triggers from explore

* renaming and deletion fixes

* fix typing

* UI updates and add last triggering event time and link

* log exception instead of return in endpoint

* highlight entry in UI when triggered

* save and delete thumbnails directly

* remove alert action for now and add descriptions

* tweaks

* clean up

* fix types

* docs

* docs tweaks

* docs

* reuse enum
This commit is contained in:
Josh Hawkins
2025-08-16 10:20:33 -05:00
committed by Blake Blackshear
parent 28f816b49a
commit 3609b41217
37 changed files with 2736 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/api/defs/request/events_body.py
+8
View File
@@ -2,6 +2,8 @@ from typing import List, Optional, Union
from pydantic import BaseModel, Field
from frigate.config.classification import TriggerType
class EventsSubLabelBody(BaseModel):
subLabel: str = Field(title="Sub label", max_length=100)
@@ -45,3 +47,9 @@ class EventsDeleteBody(BaseModel):
class SubmitPlusBody(BaseModel):
include_annotation: int = Field(default=1)
class TriggerEmbeddingBody(BaseModel):
type: TriggerType
data: str
threshold: float = Field(default=0.5, ge=0.0, le=1.0)
frigate/api/event.py
+432 -2
View File
@@ -1,5 +1,6 @@
"""Event apis."""
import base64
import datetime
import logging
import os
@@ -10,6 +11,7 @@ from pathlib import Path
from urllib.parse import unquote
import cv2
import numpy as np
from fastapi import APIRouter, Request
from fastapi.params import Depends
from fastapi.responses import JSONResponse
@@ -34,6 +36,7 @@ from frigate.api.defs.request.events_body import (
EventsLPRBody,
EventsSubLabelBody,
SubmitPlusBody,
TriggerEmbeddingBody,
)
from frigate.api.defs.response.event_response import (
EventCreateResponse,
@@ -44,11 +47,12 @@ from frigate.api.defs.response.event_response import (
from frigate.api.defs.response.generic_response import GenericResponse
from frigate.api.defs.tags import Tags
from frigate.comms.event_metadata_updater import EventMetadataTypeEnum
from frigate.const import CLIPS_DIR
from frigate.const import CLIPS_DIR, TRIGGER_DIR
from frigate.embeddings import EmbeddingsContext
from frigate.models import Event, ReviewSegment, Timeline
from frigate.models import Event, ReviewSegment, Timeline, Trigger
from frigate.track.object_processing import TrackedObject
from frigate.util.builtin import get_tz_modifiers
from frigate.util.path import get_event_thumbnail_bytes
logger = logging.getLogger(__name__)
@@ -1264,6 +1268,38 @@ def regenerate_description(
)
@router.post(
"/description/generate",
response_model=GenericResponse,
# dependencies=[Depends(require_role(["admin"]))],
)
def generate_description_embedding(
request: Request,
body: EventsDescriptionBody,
):
new_description = body.description
# If semantic search is enabled, update the index
if request.app.frigate_config.semantic_search.enabled:
context: EmbeddingsContext = request.app.embeddings
if len(new_description) > 0:
result = context.generate_description_embedding(
new_description,
)
return JSONResponse(
content=(
{
"success": True,
"message": f"Embedding for description is {result}"
if result
else "Failed to generate embedding",
}
),
status_code=200,
)
def delete_single_event(event_id: str, request: Request) -> dict:
try:
event = Event.get(Event.id == event_id)
@@ -1412,3 +1448,397 @@ def end_event(request: Request, event_id: str, body: EventsEndBody):
content=({"success": True, "message": "Event successfully ended."}),
status_code=200,
)
@router.post(
"/trigger/embedding",
response_model=dict,
dependencies=[Depends(require_role(["admin"]))],
)
def create_trigger_embedding(
request: Request,
body: TriggerEmbeddingBody,
camera: str,
name: str,
):
try:
if not request.app.frigate_config.semantic_search.enabled:
return JSONResponse(
content={
"success": False,
"message": "Semantic search is not enabled",
},
status_code=400,
)
# Check if trigger already exists
if (
Trigger.select()
.where(Trigger.camera == camera, Trigger.name == name)
.exists()
):
return JSONResponse(
content={
"success": False,
"message": f"Trigger {camera}:{name} already exists",
},
status_code=400,
)
context: EmbeddingsContext = request.app.embeddings
# Generate embedding based on type
embedding = None
if body.type == "description":
embedding = context.generate_description_embedding(body.data)
elif body.type == "thumbnail":
try:
event: Event = Event.get(Event.id == body.data)
except DoesNotExist:
# TODO: check triggers directory for image
return JSONResponse(
content={
"success": False,
"message": f"Failed to fetch event for {body.type} trigger",
},
status_code=400,
)
# Skip the event if not an object
if event.data.get("type") != "object":
return
if thumbnail := get_event_thumbnail_bytes(event):
cursor = context.db.execute_sql(
"""
SELECT thumbnail_embedding FROM vec_thumbnails WHERE id = ?
""",
[body.data],
)
row = cursor.fetchone() if cursor else None
if row:
query_embedding = row[0]
embedding = np.frombuffer(query_embedding, dtype=np.float32)
else:
# Extract valid thumbnail
thumbnail = get_event_thumbnail_bytes(event)
if thumbnail is None:
return JSONResponse(
content={
"success": False,
"message": f"Failed to get thumbnail for {body.data} for {body.type} trigger",
},
status_code=400,
)
embedding = context.generate_image_embedding(
body.data, (base64.b64encode(thumbnail).decode("ASCII"))
)
if embedding is None:
return JSONResponse(
content={
"success": False,
"message": f"Failed to generate embedding for {body.type} trigger",
},
status_code=400,
)
if body.type == "thumbnail":
# Save image to the triggers directory
try:
os.makedirs(os.path.join(TRIGGER_DIR, camera), exist_ok=True)
with open(
os.path.join(TRIGGER_DIR, camera, f"{body.data}.webp"), "wb"
) as f:
f.write(thumbnail)
logger.debug(
f"Writing thumbnail for trigger with data {body.data} in {camera}."
)
except Exception as e:
logger.error(
f"Failed to write thumbnail for trigger with data {body.data} in {camera}: {e}"
)
Trigger.create(
camera=camera,
name=name,
type=body.type,
data=body.data,
threshold=body.threshold,
model=request.app.frigate_config.semantic_search.model,
embedding=np.array(embedding, dtype=np.float32).tobytes(),
triggering_event_id="",
last_triggered=None,
)
return JSONResponse(
content={
"success": True,
"message": f"Trigger created successfully for {camera}:{name}",
},
status_code=200,
)
except Exception as e:
return JSONResponse(
content={
"success": False,
"message": f"Error creating trigger embedding: {str(e)}",
},
status_code=500,
)
@router.put(
"/trigger/embedding/{camera}/{name}",
response_model=dict,
dependencies=[Depends(require_role(["admin"]))],
)
def update_trigger_embedding(
request: Request,
camera: str,
name: str,
body: TriggerEmbeddingBody,
):
try:
if not request.app.frigate_config.semantic_search.enabled:
return JSONResponse(
content={
"success": False,
"message": "Semantic search is not enabled",
},
status_code=400,
)
context: EmbeddingsContext = request.app.embeddings
# Generate embedding based on type
embedding = None
if body.type == "description":
embedding = context.generate_description_embedding(body.data)
elif body.type == "thumbnail":
webp_file = body.data + ".webp"
webp_path = os.path.join(TRIGGER_DIR, camera, webp_file)
try:
event: Event = Event.get(Event.id == body.data)
# Skip the event if not an object
if event.data.get("type") != "object":
return JSONResponse(
content={
"success": False,
"message": f"Event {body.data} is not a tracked object for {body.type} trigger",
},
status_code=400,
)
# Extract valid thumbnail
thumbnail = get_event_thumbnail_bytes(event)
with open(webp_path, "wb") as f:
f.write(thumbnail)
except DoesNotExist:
# check triggers directory for image
if not os.path.exists(webp_path):
return JSONResponse(
content={
"success": False,
"message": f"Failed to fetch event for {body.type} trigger",
},
status_code=400,
)
else:
# Load the image from the triggers directory
with open(webp_path, "rb") as f:
thumbnail = f.read()
embedding = context.generate_image_embedding(
body.data, (base64.b64encode(thumbnail).decode("ASCII"))
)
if embedding is None:
return JSONResponse(
content={
"success": False,
"message": f"Failed to generate embedding for {body.type} trigger",
},
status_code=400,
)
# Check if trigger exists for upsert
trigger = Trigger.get_or_none(Trigger.camera == camera, Trigger.name == name)
if trigger:
# Update existing trigger
if trigger.data != body.data: # Delete old thumbnail only if data changes
try:
os.remove(os.path.join(TRIGGER_DIR, camera, f"{trigger.data}.webp"))
logger.debug(
f"Deleted thumbnail for trigger with data {trigger.data} in {camera}."
)
except Exception as e:
logger.error(
f"Failed to delete thumbnail for trigger with data {trigger.data} in {camera}: {e}"
)
Trigger.update(
data=body.data,
model=request.app.frigate_config.semantic_search.model,
embedding=np.array(embedding, dtype=np.float32).tobytes(),
threshold=body.threshold,
triggering_event_id="",
last_triggered=None,
).where(Trigger.camera == camera, Trigger.name == name).execute()
else:
# Create new trigger (for rename case)
Trigger.create(
camera=camera,
name=name,
type=body.type,
data=body.data,
threshold=body.threshold,
model=request.app.frigate_config.semantic_search.model,
embedding=np.array(embedding, dtype=np.float32).tobytes(),
triggering_event_id="",
last_triggered=None,
)
if body.type == "thumbnail":
# Save image to the triggers directory
try:
os.makedirs(os.path.join(TRIGGER_DIR, camera), exist_ok=True)
with open(
os.path.join(TRIGGER_DIR, camera, f"{body.data}.webp"), "wb"
) as f:
f.write(thumbnail)
logger.debug(
f"Writing thumbnail for trigger with data {body.data} in {camera}."
)
except Exception as e:
logger.error(
f"Failed to write thumbnail for trigger with data {body.data} in {camera}: {e}"
)
return JSONResponse(
content={
"success": True,
"message": f"Trigger updated successfully for {camera}:{name}",
},
status_code=200,
)
except Exception as e:
return JSONResponse(
content={
"success": False,
"message": f"Error updating trigger embedding: {str(e)}",
},
status_code=500,
)
@router.delete(
"/trigger/embedding/{camera}/{name}",
response_model=dict,
dependencies=[Depends(require_role(["admin"]))],
)
def delete_trigger_embedding(
request: Request,
camera: str,
name: str,
):
try:
trigger = Trigger.get_or_none(Trigger.camera == camera, Trigger.name == name)
if trigger is None:
return JSONResponse(
content={
"success": False,
"message": f"Trigger {camera}:{name} not found",
},
status_code=500,
)
deleted = (
Trigger.delete()
.where(Trigger.camera == camera, Trigger.name == name)
.execute()
)
if deleted == 0:
return JSONResponse(
content={
"success": False,
"message": f"Error deleting trigger {camera}:{name}",
},
status_code=401,
)
try:
os.remove(os.path.join(TRIGGER_DIR, camera, f"{trigger.data}.webp"))
logger.debug(
f"Deleted thumbnail for trigger with data {trigger.data} in {camera}."
)
except Exception as e:
logger.error(
f"Failed to delete thumbnail for trigger with data {trigger.data} in {camera}: {e}"
)
return JSONResponse(
content={
"success": True,
"message": f"Trigger deleted successfully for {camera}:{name}",
},
status_code=200,
)
except Exception as e:
return JSONResponse(
content={
"success": False,
"message": f"Error deleting trigger embedding: {str(e)}",
},
status_code=500,
)
@router.get(
"/triggers/status/{camera_name}",
response_model=dict,
dependencies=[Depends(require_role(["admin"]))],
)
def get_triggers_status(
camera_name: str,
):
try:
# Fetch all triggers for the specified camera
triggers = Trigger.select().where(Trigger.camera == camera_name)
# Prepare the response with trigger status
status = {
trigger.name: {
"last_triggered": trigger.last_triggered.timestamp()
if trigger.last_triggered
else None,
"triggering_event_id": trigger.triggering_event_id
if trigger.triggering_event_id
else None,
}
for trigger in triggers
}
if not status:
return JSONResponse(
content={
"success": False,
"message": f"No triggers found for camera {camera_name}",
},
status_code=404,
)
return {"success": True, "triggers": status}
except Exception as ex:
logger.exception(ex)
return JSONResponse(
content=({"success": False, "message": "Error fetching trigger status"}),
status_code=400,
)
frigate/app.py
+6
View File
@@ -39,6 +39,7 @@ from frigate.const import (
MODEL_CACHE_DIR,
RECORD_DIR,
THUMB_DIR,
TRIGGER_DIR,
)
from frigate.data_processing.types import DataProcessorMetrics
from frigate.db.sqlitevecq import SqliteVecQueueDatabase
@@ -56,6 +57,7 @@ from frigate.models import (
Regions,
ReviewSegment,
Timeline,
Trigger,
User,
)
from frigate.object_detection.base import ObjectDetectProcess
@@ -123,6 +125,9 @@ class FrigateApp:
if self.config.face_recognition.enabled:
dirs.append(FACE_DIR)
if self.config.semantic_search.enabled:
dirs.append(TRIGGER_DIR)
for d in dirs:
if not os.path.exists(d) and not os.path.islink(d):
logger.info(f"Creating directory: {d}")
@@ -288,6 +293,7 @@ class FrigateApp:
ReviewSegment,
Timeline,
User,
Trigger,
]
self.db.bind(models)
frigate/comms/webpush.py
+82 -18
View File
@@ -186,6 +186,28 @@ class WebPushClient(Communicator): # type: ignore[misc]
logger.debug(f"Notifications for {camera} are currently suspended.")
return
self.send_alert(decoded)
if topic == "triggers":
decoded = json.loads(payload)
camera = decoded["camera"]
name = decoded["name"]
# ensure notifications are enabled and the specific trigger has
# notification action enabled
if (
not self.config.cameras[camera].notifications.enabled
or name not in self.config.cameras[camera].semantic_search.triggers
or "notification"
not in self.config.cameras[camera]
.semantic_search.triggers[name]
.actions
):
return
if self.is_camera_suspended(camera):
logger.debug(f"Notifications for {camera} are currently suspended.")
return
self.send_trigger(decoded)
elif topic == "notification_test":
if not self.config.notifications.enabled and not any(
cam.notifications.enabled for cam in self.config.cameras.values()
@@ -267,6 +289,23 @@ class WebPushClient(Communicator): # type: ignore[misc]
except Exception as e:
logger.error(f"Error processing notification: {str(e)}")
def _within_cooldown(self, camera: str) -> bool:
now = datetime.datetime.now().timestamp()
if now - self.last_notification_time < self.config.notifications.cooldown:
logger.debug(
f"Skipping notification for {camera} - in global cooldown period"
)
return True
if (
now - self.last_camera_notification_time[camera]
< self.config.cameras[camera].notifications.cooldown
):
logger.debug(
f"Skipping notification for {camera} - in camera-specific cooldown period"
)
return True
return False
def send_notification_test(self) -> None:
if not self.config.notifications.email:
return
@@ -295,24 +334,7 @@ class WebPushClient(Communicator): # type: ignore[misc]
camera: str = payload["after"]["camera"]
current_time = datetime.datetime.now().timestamp()
# Check global cooldown period
if (
current_time - self.last_notification_time
< self.config.notifications.cooldown
):
logger.debug(
f"Skipping notification for {camera} - in global cooldown period"
)
return
# Check camera-specific cooldown period
if (
current_time - self.last_camera_notification_time[camera]
< self.config.cameras[camera].notifications.cooldown
):
logger.debug(
f"Skipping notification for {camera} - in camera-specific cooldown period"
)
if self._within_cooldown(camera):
return
self.check_registrations()
@@ -367,6 +389,48 @@ class WebPushClient(Communicator): # type: ignore[misc]
self.cleanup_registrations()
def send_trigger(self, payload: dict[str, Any]) -> None:
if not self.config.notifications.email:
return
camera: str = payload["camera"]
current_time = datetime.datetime.now().timestamp()
if self._within_cooldown(camera):
return
self.check_registrations()
self.last_camera_notification_time[camera] = current_time
self.last_notification_time = current_time
trigger_type = payload["type"]
event_id = payload["event_id"]
name = payload["name"]
score = payload["score"]
title = f"{name.replace('_', ' ')} triggered on {titlecase(camera.replace('_', ' '))}"
message = f"{titlecase(trigger_type)} trigger fired for {titlecase(camera.replace('_', ' '))} with score {score:.2f}"
image = f"clips/triggers/{camera}/{event_id}.webp"
direct_url = f"/explore?event_id={event_id}"
ttl = 0
logger.debug(f"Sending push notification for {camera}, trigger name {name}")
for user in self.web_pushers:
self.send_push_notification(
user=user,
payload=payload,
title=title,
message=message,
direct_url=direct_url,
image=image,
ttl=ttl,
)
self.cleanup_registrations()
def stop(self) -> None:
logger.info("Closing notification queue")
self.notification_thread.join()
frigate/config/camera/camera.py
+5
View File
@@ -22,6 +22,7 @@ from ..classification import (
AudioTranscriptionConfig,
CameraFaceRecognitionConfig,
CameraLicensePlateRecognitionConfig,
CameraSemanticSearchConfig,
)
from .audio import AudioConfig
from .birdseye import BirdseyeCameraConfig
@@ -91,6 +92,10 @@ class CameraConfig(FrigateBaseModel):
review: ReviewConfig = Field(
default_factory=ReviewConfig, title="Review configuration."
)
semantic_search: CameraSemanticSearchConfig = Field(
default_factory=CameraSemanticSearchConfig,
title="Semantic search configuration.",
)
snapshots: SnapshotsConfig = Field(
default_factory=SnapshotsConfig, title="Snapshot configuration."
)
frigate/config/camera/updater.py
+3
View File
@@ -23,6 +23,7 @@ class CameraConfigUpdateEnum(str, Enum):
record = "record"
remove = "remove" # for removing a camera
review = "review"
semantic_search = "semantic_search" # for semantic search triggers
snapshots = "snapshots"
zones = "zones"
@@ -106,6 +107,8 @@ class CameraConfigUpdateSubscriber:
config.record = updated_config
elif update_type == CameraConfigUpdateEnum.review:
config.review = updated_config
elif update_type == CameraConfigUpdateEnum.semantic_search:
config.semantic_search = updated_config
elif update_type == CameraConfigUpdateEnum.snapshots:
config.snapshots = updated_config
elif update_type == CameraConfigUpdateEnum.zones:
frigate/config/classification.py
+36
View File
@@ -10,6 +10,7 @@ __all__ = [
"CameraLicensePlateRecognitionConfig",
"FaceRecognitionConfig",
"SemanticSearchConfig",
"CameraSemanticSearchConfig",
"LicensePlateRecognitionConfig",
]
@@ -24,6 +25,15 @@ class EnrichmentsDeviceEnum(str, Enum):
CPU = "CPU"
class TriggerType(str, Enum):
THUMBNAIL = "thumbnail"
DESCRIPTION = "description"
class TriggerAction(str, Enum):
NOTIFICATION = "notification"
class AudioTranscriptionConfig(FrigateBaseModel):
enabled: bool = Field(default=False, title="Enable audio transcription.")
language: str = Field(
@@ -113,6 +123,32 @@ class SemanticSearchConfig(FrigateBaseModel):
)
class TriggerConfig(FrigateBaseModel):
enabled: bool = Field(default=True, title="Enable this 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",
default=0.8,
gt=0.0,
le=1.0,
)
actions: Optional[List[TriggerAction]] = Field(
default=[], title="Actions to perform when trigger is matched"
)
model_config = ConfigDict(extra="forbid", protected_namespaces=())
class CameraSemanticSearchConfig(FrigateBaseModel):
triggers: Optional[Dict[str, TriggerConfig]] = Field(
default=None,
title="Trigger actions on tracked objects that match existing thumbnails or descriptions",
)
model_config = ConfigDict(extra="forbid", protected_namespaces=())
class FaceRecognitionConfig(FrigateBaseModel):
enabled: bool = Field(default=False, title="Enable face recognition.")
model_size: str = Field(
frigate/const.py
+1
View File
@@ -11,6 +11,7 @@ EXPORT_DIR = f"{BASE_DIR}/exports"
FACE_DIR = f"{CLIPS_DIR}/faces"
THUMB_DIR = f"{CLIPS_DIR}/thumbs"
RECORD_DIR = f"{BASE_DIR}/recordings"
TRIGGER_DIR = f"{CLIPS_DIR}/triggers"
BIRDSEYE_PIPE = "/tmp/cache/birdseye"
CACHE_DIR = "/tmp/cache"
FRIGATE_LOCALHOST = "http://127.0.0.1:5000"
frigate/data_processing/post/semantic_trigger.py
+233
View File
@@ -0,0 +1,233 @@
"""Post time processor to trigger actions based on similar embeddings."""
import datetime
import json
import logging
import os
from typing import Any
import cv2
import numpy as np
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
from frigate.models import Event, Trigger
from frigate.util.builtin import cosine_distance
from frigate.util.path import get_event_thumbnail_bytes
from ..post.api import PostProcessorApi
from ..types import DataProcessorMetrics
logger = logging.getLogger(__name__)
WRITE_DEBUG_IMAGES = False
class SemanticTriggerProcessor(PostProcessorApi):
def __init__(
self,
db: SqliteVecQueueDatabase,
config: FrigateConfig,
requestor: InterProcessRequestor,
metrics: DataProcessorMetrics,
embeddings,
):
super().__init__(config, metrics, None)
self.db = db
self.embeddings = embeddings
self.requestor = requestor
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
) -> None:
event_id = data["event_id"]
camera = data["camera"]
process_type = data["type"]
if self.config.cameras[camera].semantic_search.triggers is None:
return
triggers = (
Trigger.select(
Trigger.camera,
Trigger.name,
Trigger.data,
Trigger.type,
Trigger.embedding,
Trigger.threshold,
)
.where(Trigger.camera == camera)
.dicts()
.iterator()
)
for trigger in triggers:
if (
trigger["name"]
not in self.config.cameras[camera].semantic_search.triggers
or not self.config.cameras[camera]
.semantic_search.triggers[trigger["name"]]
.enabled
):
logger.debug(
f"Trigger {trigger['name']} is disabled for camera {camera}"
)
continue
logger.debug(
f"Processing {trigger['type']} trigger for {event_id} on {trigger['camera']}: {trigger['name']}"
)
trigger_embedding = np.frombuffer(trigger["embedding"], dtype=np.float32)
# Get embeddings based on type
thumbnail_embedding = None
description_embedding = None
if process_type == "image":
cursor = self.db.execute_sql(
"""
SELECT thumbnail_embedding FROM vec_thumbnails WHERE id = ?
""",
[event_id],
)
row = cursor.fetchone() if cursor else None
if row:
thumbnail_embedding = np.frombuffer(row[0], dtype=np.float32)
if process_type == "text":
cursor = self.db.execute_sql(
"""
SELECT description_embedding FROM vec_descriptions WHERE id = ?
""",
[event_id],
)
row = cursor.fetchone() if cursor else None
if row:
description_embedding = np.frombuffer(row[0], dtype=np.float32)
# Skip processing if we don't have any embeddings
if thumbnail_embedding is None and description_embedding is None:
logger.debug(f"No embeddings found for {event_id}")
return
# Determine which embedding to compare based on trigger type
if (
trigger["type"] in ["text", "thumbnail"]
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"] == "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]
else:
continue
similarity = 1 - normalized_distance
logger.debug(
f"Trigger {trigger['name']} ({trigger['data'] if trigger['type'] == 'text' or trigger['type'] == 'description' else 'image'}): "
f"normalized distance: {normalized_distance:.4f}, "
f"similarity: {similarity:.4f}, threshold: {trigger['threshold']}"
)
# Check if similarity meets threshold
if similarity >= trigger["threshold"]:
logger.info(
f"Trigger {trigger['name']} activated with similarity {similarity:.4f}"
)
# Update the trigger's last_triggered and triggering_event_id
Trigger.update(
last_triggered=datetime.datetime.now(), triggering_event_id=event_id
).where(
Trigger.camera == camera, Trigger.name == trigger["name"]
).execute()
# Always publish MQTT message
self.requestor.send_data(
"triggers",
json.dumps(
{
"name": trigger["name"],
"camera": camera,
"event_id": event_id,
"type": trigger["type"],
"score": similarity,
}
),
)
if (
self.config.cameras[camera]
.semantic_search.triggers[trigger["name"]]
.actions
):
# TODO: handle actions for the trigger
# notifications already handled by webpush
pass
if WRITE_DEBUG_IMAGES:
try:
event: Event = Event.get(Event.id == event_id)
except DoesNotExist:
return
# Skip the event if not an object
if event.data.get("type") != "object":
return
thumbnail_bytes = get_event_thumbnail_bytes(event)
nparr = np.frombuffer(thumbnail_bytes, np.uint8)
thumbnail = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
font_scale = 0.5
font = cv2.FONT_HERSHEY_SIMPLEX
cv2.putText(
thumbnail,
f"{similarity:.4f}",
(10, 30),
font,
fontScale=font_scale,
color=(0, 255, 0),
thickness=2,
)
current_time = int(datetime.datetime.now().timestamp())
cv2.imwrite(
f"debug/frames/trigger-{event_id}_{current_time}.jpg",
thumbnail,
)
def handle_request(self, topic, request_data):
return None
def expire_object(self, object_id, camera):
pass
frigate/embeddings/__init__.py
+12
View File
@@ -296,3 +296,15 @@ class EmbeddingsContext:
return self.requestor.send_data(
EmbeddingsRequestEnum.transcribe_audio.value, {"event": event}
)
def generate_description_embedding(self, text: str) -> None:
return self.requestor.send_data(
EmbeddingsRequestEnum.embed_description.value,
{"id": None, "description": text, "upsert": False},
)
def generate_image_embedding(self, event_id: str, thumbnail: bytes) -> None:
return self.requestor.send_data(
EmbeddingsRequestEnum.embed_thumbnail.value,
{"id": str(event_id), "thumbnail": str(thumbnail), "upsert": False},
)
frigate/embeddings/embeddings.py
+232 -6
View File
@@ -7,21 +7,26 @@ import os
import threading
import time
from numpy import ndarray
import numpy as np
from peewee import DoesNotExist, IntegrityError
from PIL import Image
from playhouse.shortcuts import model_to_dict
from frigate.comms.embeddings_updater import (
EmbeddingsRequestEnum,
)
from frigate.comms.inter_process import InterProcessRequestor
from frigate.config import FrigateConfig
from frigate.config.classification import SemanticSearchModelEnum
from frigate.const import (
CONFIG_DIR,
TRIGGER_DIR,
UPDATE_EMBEDDINGS_REINDEX_PROGRESS,
UPDATE_MODEL_STATE,
)
from frigate.data_processing.types import DataProcessorMetrics
from frigate.db.sqlitevecq import SqliteVecQueueDatabase
from frigate.models import Event
from frigate.models import Event, Trigger
from frigate.types import ModelStatusTypesEnum
from frigate.util.builtin import EventsPerSecond, InferenceSpeed, serialize
from frigate.util.path import get_event_thumbnail_bytes
@@ -167,7 +172,7 @@ class Embeddings:
def embed_thumbnail(
self, event_id: str, thumbnail: bytes, upsert: bool = True
) -> ndarray:
) -> np.ndarray:
"""Embed thumbnail and optionally insert into DB.
@param: event_id in Events DB
@@ -194,7 +199,7 @@ class Embeddings:
def batch_embed_thumbnail(
self, event_thumbs: dict[str, bytes], upsert: bool = True
) -> list[ndarray]:
) -> list[np.ndarray]:
"""Embed thumbnails and optionally insert into DB.
@param: event_thumbs Map of Event IDs in DB to thumbnail bytes in jpg format
@@ -244,7 +249,7 @@ class Embeddings:
def embed_description(
self, event_id: str, description: str, upsert: bool = True
) -> ndarray:
) -> np.ndarray:
start = datetime.datetime.now().timestamp()
embedding = self.text_embedding([description])[0]
@@ -264,7 +269,7 @@ class Embeddings:
def batch_embed_description(
self, event_descriptions: dict[str, str], upsert: bool = True
) -> ndarray:
) -> np.ndarray:
start = datetime.datetime.now().timestamp()
# upsert embeddings one by one to avoid token limit
embeddings = []
@@ -417,3 +422,224 @@ class Embeddings:
with self.reindex_lock:
self.reindex_running = False
self.reindex_thread = None
def sync_triggers(self) -> None:
for camera in self.config.cameras.values():
# Get all existing triggers for this camera
existing_triggers = {
trigger.name: trigger
for trigger in Trigger.select().where(Trigger.camera == camera.name)
}
# Get all configured trigger names
configured_trigger_names = set(camera.semantic_search.triggers or {})
# Create or update triggers from config
for trigger_name, trigger in (
camera.semantic_search.triggers or {}
).items():
if trigger_name in existing_triggers:
existing_trigger = existing_triggers[trigger_name]
needs_embedding_update = False
thumbnail_missing = False
# Check if data has changed or thumbnail is missing for thumbnail type
if trigger.type == "thumbnail":
thumbnail_path = os.path.join(
TRIGGER_DIR, camera.name, f"{trigger.data}.webp"
)
try:
event = Event.get(Event.id == trigger.data)
if event.data.get("type") != "object":
logger.warning(
f"Event {trigger.data} is not a tracked object for {trigger.type} trigger"
)
continue # Skip if not an object
# Check if thumbnail needs to be updated (data changed or missing)
if (
existing_trigger.data != trigger.data
or not os.path.exists(thumbnail_path)
):
thumbnail = get_event_thumbnail_bytes(event)
if not thumbnail:
logger.warning(
f"Unable to retrieve thumbnail for event ID {trigger.data} for {trigger_name}."
)
continue
self.write_trigger_thumbnail(
camera.name, trigger.data, thumbnail
)
thumbnail_missing = True
except DoesNotExist:
logger.warning(
f"Event ID {trigger.data} for trigger {trigger_name} does not exist."
)
continue
# Update existing trigger if data has changed
if (
existing_trigger.type != trigger.type
or existing_trigger.data != trigger.data
or existing_trigger.threshold != trigger.threshold
):
existing_trigger.type = trigger.type
existing_trigger.data = trigger.data
existing_trigger.threshold = trigger.threshold
needs_embedding_update = True
# Check if embedding is missing or needs update
if (
not existing_trigger.embedding
or needs_embedding_update
or thumbnail_missing
):
existing_trigger.embedding = self._calculate_trigger_embedding(
trigger
)
needs_embedding_update = True
if needs_embedding_update:
existing_trigger.save()
else:
# Create new trigger
try:
try:
event: Event = Event.get(Event.id == trigger.data)
except DoesNotExist:
logger.warning(
f"Event ID {trigger.data} for trigger {trigger_name} does not exist."
)
continue
# Skip the event if not an object
if event.data.get("type") != "object":
logger.warning(
f"Event ID {trigger.data} for trigger {trigger_name} is not a tracked object."
)
continue
thumbnail = get_event_thumbnail_bytes(event)
if not thumbnail:
logger.warning(
f"Unable to retrieve thumbnail for event ID {trigger.data} for {trigger_name}."
)
continue
self.write_trigger_thumbnail(
camera.name, trigger.data, thumbnail
)
# Calculate embedding for new trigger
embedding = self._calculate_trigger_embedding(trigger)
Trigger.create(
camera=camera.name,
name=trigger_name,
type=trigger.type,
data=trigger.data,
threshold=trigger.threshold,
model=self.config.semantic_search.model,
embedding=embedding,
triggering_event_id="",
last_triggered=None,
)
except IntegrityError:
pass # Handle duplicate creation attempts
# Remove triggers that are no longer in config
triggers_to_remove = (
set(existing_triggers.keys()) - configured_trigger_names
)
if triggers_to_remove:
Trigger.delete().where(
Trigger.camera == camera.name, Trigger.name.in_(triggers_to_remove)
).execute()
for trigger_name in triggers_to_remove:
self.remove_trigger_thumbnail(camera.name, trigger_name)
def write_trigger_thumbnail(
self, camera: str, event_id: str, thumbnail: bytes
) -> None:
"""Write the thumbnail to the trigger directory."""
try:
os.makedirs(os.path.join(TRIGGER_DIR, camera), exist_ok=True)
with open(os.path.join(TRIGGER_DIR, camera, f"{event_id}.webp"), "wb") as f:
f.write(thumbnail)
logger.debug(
f"Writing thumbnail for trigger with data {event_id} in {camera}."
)
except Exception as e:
logger.error(
f"Failed to write thumbnail for trigger with data {event_id} in {camera}: {e}"
)
def remove_trigger_thumbnail(self, camera: str, event_id: str) -> None:
"""Write the thumbnail to the trigger directory."""
try:
os.remove(os.path.join(TRIGGER_DIR, camera, f"{event_id}.webp"))
logger.debug(
f"Deleted thumbnail for trigger with data {event_id} in {camera}."
)
except Exception as e:
logger.error(
f"Failed to delete thumbnail for trigger with data {event_id} in {camera}: {e}"
)
def _calculate_trigger_embedding(self, trigger) -> bytes:
"""Calculate embedding for a trigger based on its type and data."""
if trigger.type == "description":
logger.debug(f"Generating embedding for trigger description {trigger.name}")
embedding = self.requestor.send_data(
EmbeddingsRequestEnum.embed_description.value,
{"id": None, "description": trigger.data, "upsert": False},
)
return embedding.astype(np.float32).tobytes()
elif trigger.type == "thumbnail":
# For image triggers, trigger.data should be an image ID
# Try to get embedding from vec_thumbnails table first
cursor = self.db.execute_sql(
"SELECT thumbnail_embedding FROM vec_thumbnails WHERE id = ?",
[trigger.data],
)
row = cursor.fetchone() if cursor else None
if row:
return row[0] # Already in bytes format
else:
logger.debug(
f"No thumbnail embedding found for image ID: {trigger.data}, generating from saved trigger thumbnail"
)
try:
with open(
os.path.join(
TRIGGER_DIR, trigger.camera, f"{trigger.data}.webp"
),
"rb",
) as f:
thumbnail = f.read()
except Exception as e:
logger.error(
f"Failed to read thumbnail for trigger {trigger.name} with ID {trigger.data}: {e}"
)
return b""
logger.debug(
f"Generating embedding for trigger thumbnail {trigger.name} with ID {trigger.data}"
)
embedding = self.requestor.send_data(
EmbeddingsRequestEnum.embed_thumbnail.value,
{
"id": str(trigger.data),
"thumbnail": str(thumbnail),
"upsert": False,
},
)
return embedding.astype(np.float32).tobytes()
else:
logger.warning(f"Unknown trigger type: {trigger.type}")
return b""
frigate/embeddings/maintainer.py
+71 -31
View File
@@ -14,7 +14,10 @@ import numpy as np
from peewee import DoesNotExist
from frigate.comms.detections_updater import DetectionSubscriber, DetectionTypeEnum
from frigate.comms.embeddings_updater import EmbeddingsRequestEnum, EmbeddingsResponder
from frigate.comms.embeddings_updater import (
EmbeddingsRequestEnum,
EmbeddingsResponder,
)
from frigate.comms.event_metadata_updater import (
EventMetadataPublisher,
EventMetadataSubscriber,
@@ -46,6 +49,7 @@ from frigate.data_processing.post.audio_transcription import (
from frigate.data_processing.post.license_plate import (
LicensePlatePostProcessor,
)
from frigate.data_processing.post.semantic_trigger import SemanticTriggerProcessor
from frigate.data_processing.real_time.api import RealTimeProcessorApi
from frigate.data_processing.real_time.bird import BirdRealTimeProcessor
from frigate.data_processing.real_time.custom_classification import (
@@ -60,7 +64,7 @@ from frigate.data_processing.types import DataProcessorMetrics, PostProcessDataE
from frigate.db.sqlitevecq import SqliteVecQueueDatabase
from frigate.events.types import EventTypeEnum, RegenerateDescriptionEnum
from frigate.genai import get_genai_client
from frigate.models import Event, Recordings
from frigate.models import Event, Recordings, Trigger
from frigate.types import TrackedObjectUpdateTypesEnum
from frigate.util.builtin import serialize
from frigate.util.image import (
@@ -93,7 +97,11 @@ class EmbeddingMaintainer(threading.Thread):
self.config_updater = CameraConfigUpdateSubscriber(
self.config,
self.config.cameras,
[CameraConfigUpdateEnum.add, CameraConfigUpdateEnum.remove],
[
CameraConfigUpdateEnum.add,
CameraConfigUpdateEnum.remove,
CameraConfigUpdateEnum.semantic_search,
],
)
# Configure Frigate DB
@@ -109,7 +117,7 @@ class EmbeddingMaintainer(threading.Thread):
),
load_vec_extension=True,
)
models = [Event, Recordings]
models = [Event, Recordings, Trigger]
db.bind(models)
if config.semantic_search.enabled:
@@ -119,6 +127,9 @@ class EmbeddingMaintainer(threading.Thread):
if config.semantic_search.reindex:
self.embeddings.reindex()
# Sync semantic search triggers in db with config
self.embeddings.sync_triggers()
# create communication for updating event descriptions
self.requestor = InterProcessRequestor()
@@ -211,6 +222,17 @@ class EmbeddingMaintainer(threading.Thread):
AudioTranscriptionPostProcessor(self.config, self.requestor, metrics)
)
if self.config.semantic_search.enabled:
self.post_processors.append(
SemanticTriggerProcessor(
db,
self.config,
self.requestor,
metrics,
self.embeddings,
)
)
self.stop_event = stop_event
self.tracked_events: dict[str, list[Any]] = {}
self.early_request_sent: dict[str, bool] = {}
@@ -387,33 +409,6 @@ class EmbeddingMaintainer(threading.Thread):
event_id, camera, updated_db = ended
camera_config = self.config.cameras[camera]
# call any defined post processors
for processor in self.post_processors:
if isinstance(processor, LicensePlatePostProcessor):
recordings_available = self.recordings_available_through.get(camera)
if (
recordings_available is not None
and event_id in self.detected_license_plates
and self.config.cameras[camera].type != "lpr"
):
processor.process_data(
{
"event_id": event_id,
"camera": camera,
"recordings_available": self.recordings_available_through[
camera
],
"obj_data": self.detected_license_plates[event_id][
"obj_data"
],
},
PostProcessDataEnum.recording,
)
elif isinstance(processor, AudioTranscriptionPostProcessor):
continue
else:
processor.process_data(event_id, PostProcessDataEnum.event_id)
# expire in realtime processors
for processor in self.realtime_processors:
processor.expire_object(event_id, camera)
@@ -450,6 +445,41 @@ class EmbeddingMaintainer(threading.Thread):
):
self._process_genai_description(event, camera_config, thumbnail)
# call any defined post processors
for processor in self.post_processors:
if isinstance(processor, LicensePlatePostProcessor):
recordings_available = self.recordings_available_through.get(camera)
if (
recordings_available is not None
and event_id in self.detected_license_plates
and self.config.cameras[camera].type != "lpr"
):
processor.process_data(
{
"event_id": event_id,
"camera": camera,
"recordings_available": self.recordings_available_through[
camera
],
"obj_data": self.detected_license_plates[event_id][
"obj_data"
],
},
PostProcessDataEnum.recording,
)
elif isinstance(processor, AudioTranscriptionPostProcessor):
continue
elif isinstance(processor, SemanticTriggerProcessor):
processor.process_data(
{"event_id": event_id, "camera": camera, "type": "image"},
PostProcessDataEnum.tracked_object,
)
else:
processor.process_data(
{"event_id": event_id, "camera": camera},
PostProcessDataEnum.tracked_object,
)
# Delete tracked events based on the event_id
if event_id in self.tracked_events:
del self.tracked_events[event_id]
@@ -658,6 +688,16 @@ class EmbeddingMaintainer(threading.Thread):
if self.config.semantic_search.enabled:
self.embeddings.embed_description(event.id, description)
# Check semantic trigger for this description
for processor in self.post_processors:
if isinstance(processor, SemanticTriggerProcessor):
processor.process_data(
{"event_id": event.id, "camera": event.camera, "type": "text"},
PostProcessDataEnum.tracked_object,
)
else:
continue
logger.debug(
"Generated description for %s (%d images): %s",
event.id,
frigate/models.py
+17
View File
@@ -1,6 +1,8 @@
from peewee import (
BlobField,
BooleanField,
CharField,
CompositeKey,
DateTimeField,
FloatField,
ForeignKeyField,
@@ -132,3 +134,18 @@ class User(Model): # type: ignore[misc]
)
password_hash = CharField(null=False, max_length=120)
notification_tokens = JSONField()
class Trigger(Model): # type: ignore[misc]
camera = CharField(max_length=20)
name = CharField()
type = CharField(max_length=10)
data = TextField()
threshold = FloatField()
model = CharField(max_length=30)
embedding = BlobField()
triggering_event_id = CharField(max_length=30)
last_triggered = DateTimeField()
class Meta:
primary_key = CompositeKey("camera", "name")
frigate/util/builtin.py
+16
View File
@@ -428,3 +428,19 @@ def sanitize_float(value):
if isinstance(value, (int, float)) and not math.isfinite(value):
return 0.0
return value
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
return 1 - cosine_distance(a, b)
def cosine_distance(a: np.ndarray, b: np.ndarray) -> float:
"""Returns cosine distance to match sqlite-vec's calculation."""
dot = np.dot(a, b)
a_mag = np.dot(a, a) # ||a||^2
b_mag = np.dot(b, b) # ||b||^2
if a_mag == 0 or b_mag == 0:
return 1.0
return 1.0 - (dot / (np.sqrt(a_mag) * np.sqrt(b_mag)))