Add Azure OpenAI as genai provider (#14102)

* add azure openai genai client

* docs

docs/docs/configuration/genai.md

@@ -100,6 +100,28 @@ genai:
   model: gpt-4o
 ```
 
+## Azure OpenAI
+
+Microsoft offers several vision models through Azure OpenAI. A subscription is required.
+
+### Supported Models
+
+You must use a vision capable model with Frigate. Current model variants can be found [in their documentation](https://learn.microsoft.com/en-us/azure/ai-services/openai/concepts/models). At the time of writing, this includes `gpt-4o` and `gpt-4-turbo`.
+
+### Get API Key
+
+To start using Azure OpenAI, you must first [create a resource](https://learn.microsoft.com/azure/cognitive-services/openai/how-to/create-resource?pivots=web-portal#create-a-resource). You'll need your API key and resource URL, which must include the `api-version` parameter (see the example below). The model field is not required in your configuration as the model is part of the deployment name you chose when deploying the resource.
+
+### Configuration
+
+```yaml
+genai:
+  enabled: True
+  provider: openai
+  base_url: https://example-endpoint.openai.azure.com/openai/deployments/gpt-4o/chat/completions?api-version=2023-03-15-preview
+  api_key: "{FRIGATE_OPENAI_API_KEY}"
+```
+
 ## Custom Prompts
 
 Frigate sends multiple frames from the tracked object along with a prompt to your Generative AI provider asking it to generate a description. The default prompt is as follows:

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