add azure openai genai client

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"
 

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