diff --git a/frigate/config/camera/genai.py b/frigate/config/camera/genai.py index 12ff37adb..21c3d4525 100644 --- a/frigate/config/camera/genai.py +++ b/frigate/config/camera/genai.py @@ -11,6 +11,7 @@ __all__ = ["GenAIConfig", "GenAICameraConfig", "GenAIProviderEnum"] class GenAIProviderEnum(str, Enum): openai = "openai" + azure_openai = "azure_openai" gemini = "gemini" ollama = "ollama" diff --git a/frigate/genai/azure-openai.py b/frigate/genai/azure-openai.py new file mode 100644 index 000000000..155fa2431 --- /dev/null +++ b/frigate/genai/azure-openai.py @@ -0,0 +1,73 @@ +"""Azure OpenAI Provider for Frigate AI.""" + +import base64 +import logging +from typing import Optional +from urllib.parse import parse_qs, urlparse + +from openai import AzureOpenAI + +from frigate.config import GenAIProviderEnum +from frigate.genai import GenAIClient, register_genai_provider + +logger = logging.getLogger(__name__) + + +@register_genai_provider(GenAIProviderEnum.azure_openai) +class OpenAIClient(GenAIClient): + """Generative AI client for Frigate using Azure OpenAI.""" + + provider: AzureOpenAI + + def _init_provider(self): + """Initialize the client.""" + try: + parsed_url = urlparse(self.genai_config.base_url) + query_params = parse_qs(parsed_url.query) + api_version = query_params.get("api-version", [None])[0] + azure_endpoint = f"{parsed_url.scheme}://{parsed_url.netloc}/" + + if not api_version: + logger.warning("Azure OpenAI url is missing API version.") + return None + + except Exception as e: + logger.warning("Error parsing Azure OpenAI url: %s", str(e)) + return None + + return AzureOpenAI( + api_key=self.genai_config.api_key, + api_version=api_version, + azure_endpoint=azure_endpoint, + ) + + def _send(self, prompt: str, images: list[bytes]) -> Optional[str]: + """Submit a request to Azure OpenAI.""" + encoded_images = [base64.b64encode(image).decode("utf-8") for image in images] + try: + result = self.provider.chat.completions.create( + model=self.genai_config.model, + messages=[ + { + "role": "user", + "content": [{"type": "text", "text": prompt}] + + [ + { + "type": "image_url", + "image_url": { + "url": f"data:image/jpeg;base64,{image}", + "detail": "low", + }, + } + for image in encoded_images + ], + }, + ], + timeout=self.timeout, + ) + except Exception as e: + logger.warning("Azure OpenAI returned an error: %s", str(e)) + return None + if len(result.choices) > 0: + return result.choices[0].message.content.strip() + return None