frigate/frigate/genai/manager.py

"""GenAI client manager for Frigate.

Manages GenAI provider clients from Frigate config. Clients are created lazily
on first access so that providers whose roles are never used (e.g. chat when
no chat feature is active) are never initialized.
"""

import logging
from typing import TYPE_CHECKING, Any

from frigate.config import FrigateConfig
from frigate.config.camera.genai import GenAIConfig, GenAIRoleEnum

if TYPE_CHECKING:
    from frigate.genai import GenAIClient

logger = logging.getLogger(__name__)


class GenAIClientManager:
    """Manages GenAI provider clients from Frigate config."""

    def __init__(self, config: FrigateConfig) -> None:
        self._configs: dict[str, GenAIConfig] = {}
        self._role_map: dict[GenAIRoleEnum, str] = {}
        self._clients: dict[str, GenAIClient] = {}
        self.update_config(config)

    def update_config(self, config: FrigateConfig) -> None:
        """Store provider configs and build the role→name mapping.

        Called from __init__ and can be called again when config is reloaded.
        Clients are not created here; they are instantiated lazily on first
        access via a role property or list_models().
        """
        from frigate.genai import PROVIDERS, load_providers

        self._configs = {}
        self._role_map = {}
        self._clients = {}

        if not config.genai:
            return

        load_providers()

        for name, genai_cfg in config.genai.items():
            if not genai_cfg.provider:
                continue
            if genai_cfg.provider not in PROVIDERS:
                logger.warning(
                    "Unknown GenAI provider %s in config, skipping.",
                    genai_cfg.provider,
                )
                continue

            self._configs[name] = genai_cfg

            for role in genai_cfg.roles:
                self._role_map[role] = name

    def _get_client(self, name: str) -> "GenAIClient | None":
        """Return the client for *name*, creating it on first access."""
        if name in self._clients:
            client = self._clients[name]
            client.ensure_provider()
            return client

        from frigate.genai import PROVIDERS

        genai_cfg = self._configs.get(name)
        if not genai_cfg:
            return None

        if not genai_cfg.provider:
            return None

        provider_cls = PROVIDERS.get(genai_cfg.provider)
        if not provider_cls:
            return None

        try:
            client = provider_cls(genai_cfg)
        except Exception as e:
            logger.exception(
                "Failed to create GenAI client for provider %s: %s",
                genai_cfg.provider,
                e,
            )
            return None

        self._clients[name] = client
        return client

    @property
    def chat_client(self) -> "GenAIClient | None":
        """Client configured for the chat role (e.g. chat with function calling)."""
        name = self._role_map.get(GenAIRoleEnum.chat)
        return self._get_client(name) if name else None

    @property
    def description_client(self) -> "GenAIClient | None":
        """Client configured for the descriptions role (e.g. review descriptions, object descriptions)."""
        name = self._role_map.get(GenAIRoleEnum.descriptions)
        return self._get_client(name) if name else None

    @property
    def embeddings_client(self) -> "GenAIClient | None":
        """Client configured for the embeddings role."""
        name = self._role_map.get(GenAIRoleEnum.embeddings)
        return self._get_client(name) if name else None

    @property
    def transcribe_client(self) -> "GenAIClient | None":
        """Client configured for the transcribe role."""
        name = self._role_map.get(GenAIRoleEnum.transcribe)
        return self._get_client(name) if name else None

    def role_info(self) -> dict[str, dict[str, Any]]:
        """Return the model selected for each configured role and its context size.

        Only reads state the client saved when it initialized, so unlike
        list_models() this does not ask the provider for its catalog.
        """
        result: dict[str, dict[str, Any]] = {}

        for role, name in self._role_map.items():
            client = self._get_client(name)

            if not client:
                continue

            result[role.value] = {
                "name": name,
                "model": self._configs[name].model,
                "context_size": client.get_context_size(),
            }

        return result

    def list_models(self) -> dict[str, dict[str, Any]]:
        """Return per-entry model lists and capabilities, keyed by config entry name."""
        result: dict[str, dict[str, Any]] = {}
        for name, genai_cfg in self._configs.items():
            client = self._get_client(name)
            if not client:
                continue
            result[name] = {
                "models": client.list_models(),
                "roles": [r.value for r in genai_cfg.roles],
                "supports_toggleable_thinking": client.supports_toggleable_thinking,
                "supports_embeddings": client.supports_embeddings,
                "supports_transcription": client.supports_transcription,
                # Capabilities of the configured model are above; this maps every
                # model the provider serves to its own, so the UI can react to a
                # model selected but not yet saved. Empty when the provider
                # cannot report capabilities without loading a model.
                "model_capabilities": client.list_model_capabilities(),
            }
        return result