Migrate object genai configuration (#19437)

* Move genAI object to objects section

* Adjust config propogation behavior

* Refactor genai config usage

* Automatic migration

* Always start the embeddings process

* Always init embeddings

* Config fixes

* Adjust reference config

* Adjust docs

* Formatting

* Fix

frigate/api/event.py

@@ -1239,7 +1239,7 @@ def regenerate_description(
 
     camera_config = request.app.frigate_config.cameras[event.camera]
 
-    if camera_config.genai.enabled or params.force:
+    if camera_config.objects.genai.enabled or params.force:
         request.app.event_metadata_updater.publish(
             (event.id, params.source, params.force),
             EventMetadataTypeEnum.regenerate_description.value,

frigate/app.py

@@ -246,18 +246,7 @@ class FrigateApp:
         logger.info(f"Review process started: {review_segment_process.pid}")
 
     def init_embeddings_manager(self) -> None:
-        genai_cameras = [
-            c for c in self.config.cameras.values() if c.enabled and c.genai.enabled
-        ]
-
-        if (
-            not self.config.semantic_search.enabled
-            and not genai_cameras
-            and not self.config.lpr.enabled
-            and not self.config.face_recognition.enabled
-        ):
-            return
-
+        # always start the embeddings process
         embedding_process = EmbeddingProcess(
             self.config, self.embeddings_metrics, self.stop_event
         )
@@ -309,20 +298,8 @@ class FrigateApp:
             migrate_exports(self.config.ffmpeg, list(self.config.cameras.keys()))
 
     def init_embeddings_client(self) -> None:
-        genai_cameras = [
-            c
-            for c in self.config.cameras.values()
-            if c.enabled_in_config and c.genai.enabled
-        ]
-
-        if (
-            self.config.semantic_search.enabled
-            or self.config.lpr.enabled
-            or genai_cameras
-            or self.config.face_recognition.enabled
-        ):
-            # Create a client for other processes to use
-            self.embeddings = EmbeddingsContext(self.db)
+        # Create a client for other processes to use
+        self.embeddings = EmbeddingsContext(self.db)
 
     def init_inter_process_communicator(self) -> None:
         self.inter_process_communicator = InterProcessCommunicator()

frigate/comms/dispatcher.py

@@ -209,7 +209,7 @@ class Dispatcher:
                     ].onvif.autotracking.enabled,
                     "alerts": self.config.cameras[camera].review.alerts.enabled,
                     "detections": self.config.cameras[camera].review.detections.enabled,
-                    "genai": self.config.cameras[camera].genai.enabled,
+                    "genai": self.config.cameras[camera].objects.genai.enabled,
                 }
 
             self.publish("camera_activity", json.dumps(camera_status))
@@ -744,10 +744,10 @@ class Dispatcher:
 
     def _on_genai_command(self, camera_name: str, payload: str) -> None:
         """Callback for GenAI topic."""
-        genai_settings = self.config.cameras[camera_name].genai
+        genai_settings = self.config.cameras[camera_name].objects.genai
 
         if payload == "ON":
-            if not self.config.cameras[camera_name].genai.enabled_in_config:
+            if not self.config.cameras[camera_name].objects.genai.enabled_in_config:
                 logger.error(
                     "GenAI must be enabled in the config to be turned on via MQTT."
                 )

frigate/comms/mqtt.py

@@ -124,7 +124,7 @@ class MqttClient(Communicator):
             )
             self.publish(
                 f"{camera_name}/genai/state",
-                "ON" if camera.genai.enabled_in_config else "OFF",
+                "ON" if camera.objects.genai.enabled_in_config else "OFF",
                 retain=True,
             )
 

frigate/config/camera/camera.py

@@ -28,7 +28,6 @@ from .audio import AudioConfig
 from .birdseye import BirdseyeCameraConfig
 from .detect import DetectConfig
 from .ffmpeg import CameraFfmpegConfig, CameraInput
-from .genai import GenAICameraConfig
 from .live import CameraLiveConfig
 from .motion import MotionConfig
 from .mqtt import CameraMqttConfig
@@ -71,9 +70,6 @@ class CameraConfig(FrigateBaseModel):
         default_factory=CameraFaceRecognitionConfig, title="Face recognition config."
     )
     ffmpeg: CameraFfmpegConfig = Field(title="FFmpeg configuration for the camera.")
-    genai: GenAICameraConfig = Field(
-        default_factory=GenAICameraConfig, title="Generative AI configuration."
-    )
     live: CameraLiveConfig = Field(
         default_factory=CameraLiveConfig, title="Live playback settings."
     )

frigate/config/camera/genai.py

@@ -1,12 +1,12 @@
 from enum import Enum
-from typing import Optional, Union
+from typing import Optional
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import Field
 
 from ..base import FrigateBaseModel
 from ..env import EnvString
 
-__all__ = ["GenAIConfig", "GenAICameraConfig", "GenAIProviderEnum"]
+__all__ = ["GenAIConfig", "GenAIProviderEnum"]
 
 
 class GenAIProviderEnum(str, Enum):
@@ -16,70 +16,8 @@ class GenAIProviderEnum(str, Enum):
     ollama = "ollama"
 
 
-class GenAISendTriggersConfig(BaseModel):
-    tracked_object_end: bool = Field(
-        default=True, title="Send once the object is no longer tracked."
-    )
-    after_significant_updates: Optional[int] = Field(
-        default=None,
-        title="Send an early request to generative AI when X frames accumulated.",
-        ge=1,
-    )
-
-
-# uses BaseModel because some global attributes are not available at the camera level
-class GenAICameraConfig(BaseModel):
-    enabled: bool = Field(default=False, title="Enable GenAI for camera.")
-    use_snapshot: bool = Field(
-        default=False, title="Use snapshots for generating descriptions."
-    )
-    prompt: str = Field(
-        default="Analyze the sequence of images containing the {label}. Focus on the likely intent or behavior of the {label} based on its actions and movement, rather than describing its appearance or the surroundings. Consider what the {label} is doing, why, and what it might do next.",
-        title="Default caption prompt.",
-    )
-    object_prompts: dict[str, str] = Field(
-        default_factory=dict, title="Object specific prompts."
-    )
-
-    objects: Union[str, list[str]] = Field(
-        default_factory=list,
-        title="List of objects to run generative AI for.",
-    )
-    required_zones: Union[str, list[str]] = Field(
-        default_factory=list,
-        title="List of required zones to be entered in order to run generative AI.",
-    )
-    debug_save_thumbnails: bool = Field(
-        default=False,
-        title="Save thumbnails sent to generative AI for debugging purposes.",
-    )
-    send_triggers: GenAISendTriggersConfig = Field(
-        default_factory=GenAISendTriggersConfig,
-        title="What triggers to use to send frames to generative AI for a tracked object.",
-    )
-
-    enabled_in_config: Optional[bool] = Field(
-        default=None, title="Keep track of original state of generative AI."
-    )
-
-    @field_validator("required_zones", mode="before")
-    @classmethod
-    def validate_required_zones(cls, v):
-        if isinstance(v, str) and "," not in v:
-            return [v]
-
-        return v
-
-
 class GenAIConfig(FrigateBaseModel):
-    enabled: bool = Field(default=False, title="Enable GenAI.")
-    prompt: str = Field(
-        default="Analyze the sequence of images containing the {label}. Focus on the likely intent or behavior of the {label} based on its actions and movement, rather than describing its appearance or the surroundings. Consider what the {label} is doing, why, and what it might do next.",
-        title="Default caption prompt.",
-    )
-    object_prompts: dict[str, str] = Field(
-        default_factory=dict, title="Object specific prompts."
-    )
+    """Primary GenAI Config to define GenAI Provider."""
 
     api_key: Optional[EnvString] = Field(default=None, title="Provider API key.")
     base_url: Optional[str] = Field(default=None, title="Provider base url.")

frigate/config/camera/objects.py

@@ -1,10 +1,10 @@
 from typing import Any, Optional, Union
 
-from pydantic import Field, PrivateAttr, field_serializer
+from pydantic import Field, PrivateAttr, field_serializer, field_validator
 
 from ..base import FrigateBaseModel
 
-__all__ = ["ObjectConfig", "FilterConfig"]
+__all__ = ["ObjectConfig", "GenAIObjectConfig", "FilterConfig"]
 
 
 DEFAULT_TRACKED_OBJECTS = ["person"]
@@ -49,12 +49,69 @@ class FilterConfig(FrigateBaseModel):
         return None
 
 
+class GenAIObjectTriggerConfig(FrigateBaseModel):
+    tracked_object_end: bool = Field(
+        default=True, title="Send once the object is no longer tracked."
+    )
+    after_significant_updates: Optional[int] = Field(
+        default=None,
+        title="Send an early request to generative AI when X frames accumulated.",
+        ge=1,
+    )
+
+
+class GenAIObjectConfig(FrigateBaseModel):
+    enabled: bool = Field(default=False, title="Enable GenAI for camera.")
+    use_snapshot: bool = Field(
+        default=False, title="Use snapshots for generating descriptions."
+    )
+    prompt: str = Field(
+        default="Analyze the sequence of images containing the {label}. Focus on the likely intent or behavior of the {label} based on its actions and movement, rather than describing its appearance or the surroundings. Consider what the {label} is doing, why, and what it might do next.",
+        title="Default caption prompt.",
+    )
+    object_prompts: dict[str, str] = Field(
+        default_factory=dict, title="Object specific prompts."
+    )
+
+    objects: Union[str, list[str]] = Field(
+        default_factory=list,
+        title="List of objects to run generative AI for.",
+    )
+    required_zones: Union[str, list[str]] = Field(
+        default_factory=list,
+        title="List of required zones to be entered in order to run generative AI.",
+    )
+    debug_save_thumbnails: bool = Field(
+        default=False,
+        title="Save thumbnails sent to generative AI for debugging purposes.",
+    )
+    send_triggers: GenAIObjectTriggerConfig = Field(
+        default_factory=GenAIObjectTriggerConfig,
+        title="What triggers to use to send frames to generative AI for a tracked object.",
+    )
+    enabled_in_config: Optional[bool] = Field(
+        default=None, title="Keep track of original state of generative AI."
+    )
+
+    @field_validator("required_zones", mode="before")
+    @classmethod
+    def validate_required_zones(cls, v):
+        if isinstance(v, str) and "," not in v:
+            return [v]
+
+        return v
+
+
 class ObjectConfig(FrigateBaseModel):
     track: list[str] = Field(default=DEFAULT_TRACKED_OBJECTS, title="Objects to track.")
     filters: dict[str, FilterConfig] = Field(
         default_factory=dict, title="Object filters."
     )
     mask: Union[str, list[str]] = Field(default="", title="Object mask.")
+    genai: GenAIObjectConfig = Field(
+        default_factory=GenAIObjectConfig,
+        title="Config for using genai to analyze objects.",
+    )
     _all_objects: list[str] = PrivateAttr()
 
     @property

frigate/config/camera/updater.py

@@ -99,7 +99,7 @@ class CameraConfigUpdateSubscriber:
         elif update_type == CameraConfigUpdateEnum.enabled:
             config.enabled = updated_config
         elif update_type == CameraConfigUpdateEnum.genai:
-            config.genai = updated_config
+            config.objects.genai = updated_config
         elif update_type == CameraConfigUpdateEnum.motion:
             config.motion = updated_config
         elif update_type == CameraConfigUpdateEnum.notifications:

frigate/config/config.py

@@ -352,6 +352,11 @@ class FrigateConfig(FrigateBaseModel):
         default_factory=ModelConfig, title="Detection model configuration."
     )
 
+    # GenAI config
+    genai: GenAIConfig = Field(
+        default_factory=GenAIConfig, title="Generative AI configuration."
+    )
+
     # Camera config
     cameras: Dict[str, CameraConfig] = Field(title="Camera configuration.")
     audio: AudioConfig = Field(
@@ -366,9 +371,6 @@ class FrigateConfig(FrigateBaseModel):
     ffmpeg: FfmpegConfig = Field(
         default_factory=FfmpegConfig, title="Global FFmpeg configuration."
     )
-    genai: GenAIConfig = Field(
-        default_factory=GenAIConfig, title="Generative AI configuration."
-    )
     live: CameraLiveConfig = Field(
         default_factory=CameraLiveConfig, title="Live playback settings."
     )
@@ -458,7 +460,6 @@ class FrigateConfig(FrigateBaseModel):
                 "live": ...,
                 "objects": ...,
                 "review": ...,
-                "genai": ...,
                 "motion": ...,
                 "notifications": ...,
                 "detect": ...,
@@ -606,7 +607,9 @@ class FrigateConfig(FrigateBaseModel):
             camera_config.review.detections.enabled_in_config = (
                 camera_config.review.detections.enabled
             )
-            camera_config.genai.enabled_in_config = camera_config.genai.enabled
+            camera_config.objects.genai.enabled_in_config = (
+                camera_config.objects.genai.enabled
+            )
 
             # Add default filters
             object_keys = camera_config.objects.track

frigate/embeddings/maintainer.py

@@ -30,7 +30,7 @@ from frigate.comms.recordings_updater import (
     RecordingsDataTypeEnum,
 )
 from frigate.comms.review_updater import ReviewDataSubscriber
-from frigate.config import FrigateConfig
+from frigate.config import CameraConfig, FrigateConfig
 from frigate.config.camera.camera import CameraTypeEnum
 from frigate.config.camera.updater import (
     CameraConfigUpdateEnum,
@@ -329,7 +329,10 @@ class EmbeddingMaintainer(threading.Thread):
         camera_config = self.config.cameras[camera]
 
         # no need to process updated objects if face recognition, lpr, genai are disabled
-        if not camera_config.genai.enabled and len(self.realtime_processors) == 0:
+        if (
+            not camera_config.objects.genai.enabled
+            and len(self.realtime_processors) == 0
+        ):
             return
 
         # Create our own thumbnail based on the bounding box and the frame time
@@ -367,23 +370,23 @@ class EmbeddingMaintainer(threading.Thread):
         # check if we're configured to send an early request after a minimum number of updates received
         if (
             self.genai_client is not None
-            and camera_config.genai.send_triggers.after_significant_updates
+            and camera_config.objects.genai.send_triggers.after_significant_updates
         ):
             if (
                 len(self.tracked_events.get(data["id"], []))
-                >= camera_config.genai.send_triggers.after_significant_updates
+                >= camera_config.objects.genai.send_triggers.after_significant_updates
                 and data["id"] not in self.early_request_sent
             ):
                 if data["has_clip"] and data["has_snapshot"]:
                     event: Event = Event.get(Event.id == data["id"])
 
                     if (
-                        not camera_config.genai.objects
-                        or event.label in camera_config.genai.objects
+                        not camera_config.objects.genai.objects
+                        or event.label in camera_config.objects.genai.objects
                     ) and (
-                        not camera_config.genai.required_zones
+                        not camera_config.objects.genai.required_zones
                         or set(data["entered_zones"])
-                        & set(camera_config.genai.required_zones)
+                        & set(camera_config.objects.genai.required_zones)
                     ):
                         logger.debug(f"{camera} sending early request to GenAI")
 
@@ -436,16 +439,17 @@ class EmbeddingMaintainer(threading.Thread):
 
                 # Run GenAI
                 if (
-                    camera_config.genai.enabled
-                    and camera_config.genai.send_triggers.tracked_object_end
+                    camera_config.objects.genai.enabled
+                    and camera_config.objects.genai.send_triggers.tracked_object_end
                     and self.genai_client is not None
                     and (
-                        not camera_config.genai.objects
-                        or event.label in camera_config.genai.objects
+                        not camera_config.objects.genai.objects
+                        or event.label in camera_config.objects.genai.objects
                     )
                     and (
-                        not camera_config.genai.required_zones
-                        or set(event.zones) & set(camera_config.genai.required_zones)
+                        not camera_config.objects.genai.required_zones
+                        or set(event.zones)
+                        & set(camera_config.objects.genai.required_zones)
                     )
                 ):
                     self._process_genai_description(event, camera_config, thumbnail)
@@ -624,8 +628,10 @@ class EmbeddingMaintainer(threading.Thread):
 
         self.embeddings.embed_thumbnail(event_id, thumbnail)
 
-    def _process_genai_description(self, event, camera_config, thumbnail) -> None:
-        if event.has_snapshot and camera_config.genai.use_snapshot:
+    def _process_genai_description(
+        self, event: Event, camera_config: CameraConfig, thumbnail
+    ) -> None:
+        if event.has_snapshot and camera_config.objects.genai.use_snapshot:
             snapshot_image = self._read_and_crop_snapshot(event, camera_config)
             if not snapshot_image:
                 return
@@ -637,7 +643,7 @@ class EmbeddingMaintainer(threading.Thread):
 
         embed_image = (
             [snapshot_image]
-            if event.has_snapshot and camera_config.genai.use_snapshot
+            if event.has_snapshot and camera_config.objects.genai.use_snapshot
             else (
                 [data["thumbnail"] for data in self.tracked_events[event.id]]
                 if num_thumbnails > 0
@@ -645,7 +651,7 @@ class EmbeddingMaintainer(threading.Thread):
             )
         )
 
-        if camera_config.genai.debug_save_thumbnails and num_thumbnails > 0:
+        if camera_config.objects.genai.debug_save_thumbnails and num_thumbnails > 0:
             logger.debug(f"Saving {num_thumbnails} thumbnails for event {event.id}")
 
             Path(os.path.join(CLIPS_DIR, f"genai-requests/{event.id}")).mkdir(
@@ -775,7 +781,7 @@ class EmbeddingMaintainer(threading.Thread):
             return
 
         camera_config = self.config.cameras[event.camera]
-        if not camera_config.genai.enabled and not force:
+        if not camera_config.objects.genai.enabled and not force:
             logger.error(f"GenAI not enabled for camera {event.camera}")
             return
 

frigate/genai/__init__.py

@@ -40,9 +40,9 @@ class GenAIClient:
         event: Event,
     ) -> Optional[str]:
         """Generate a description for the frame."""
-        prompt = camera_config.genai.object_prompts.get(
+        prompt = camera_config.objects.genai.object_prompts.get(
             event.label,
-            camera_config.genai.prompt,
+            camera_config.objects.genai.prompt,
         ).format(**model_to_dict(event))
         logger.debug(f"Sending images to genai provider with prompt: {prompt}")
         return self._send(prompt, thumbnails)
@@ -58,16 +58,10 @@ class GenAIClient:
 
 def get_genai_client(config: FrigateConfig) -> Optional[GenAIClient]:
     """Get the GenAI client."""
-    genai_config = config.genai
-    genai_cameras = [
-        c for c in config.cameras.values() if c.enabled and c.genai.enabled
-    ]
-
-    if genai_cameras or genai_config.enabled:
-        load_providers()
-        provider = PROVIDERS.get(genai_config.provider)
-        if provider:
-            return provider(genai_config)
+    load_providers()
+    provider = PROVIDERS.get(config.genai.provider)
+    if provider:
+        return provider(config.genai)
 
     return None
 

frigate/util/config.py

@@ -371,6 +371,22 @@ def migrate_017_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]
 
         del new_config["record"]["retain"]
 
+    # migrate global genai to new objects config
+    global_genai = config.get("genai", {})
+
+    if global_genai:
+        new_genai_config = {}
+        new_object_config = config.get("objects", {})
+        new_object_config["genai"] = {}
+
+        for key in global_genai.keys():
+            if key not in ["provider", "base_url", "api_key"]:
+                new_object_config["genai"][key] = global_genai[key]
+            else:
+                new_genai_config[key] = global_genai[key]
+
+        config["genai"] = new_genai_config
+
     for name, camera in config.get("cameras", {}).items():
         camera_config: dict[str, dict[str, Any]] = camera.copy()
         camera_record_retain = camera_config.get("record", {}).get("retain")
@@ -392,6 +408,13 @@ def migrate_017_0(config: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]
 
             del camera_config["record"]["retain"]
 
+        camera_genai = camera_config.get("genai", {})
+
+        if camera_genai:
+            new_object_config = config.get("objects", {})
+            new_object_config["genai"] = camera_genai
+            del camera_config["genai"]
+
         new_config["cameras"][name] = camera_config
 
     new_config["version"] = "0.17-0"