AI vision in home automation

{
 "domain": "openai_conversation_custom",
 "name": "OpenAI Conversation Custom",
 "codeowners": ["@balloob", "@bitcoder"],
 "config_flow": true,
 "dependencies": ["conversation"],
 "documentation": "https://www.home-assistant.io/integrations/openai_conversation",
 "integration_type": "service",
 "iot_class": "cloud_polling",
 "requirements": ["openai==0.27.2", "httpx"],
 "version": "1.0.0-custom-20231124"
}

import os
import numpy as np
import httpx

from .utils import compose_payload, compose_text_payload


class OpenAIConnector:

    def __init__(self, api_key: str):
        if api_key is None:
            raise ValueError("API_KEY is not set")
        self.api_key = api_key

    async def simple_prompt(self, prompt: str, image_path: str = None) -> str:
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        if image_path is None:
            payload = compose_text_payload(prompt=prompt)
        else:
            payload = compose_payload(image_path=image_path, prompt=prompt)

        async with httpx.AsyncClient() as client:
            response = (await client.post("https://api.openai.com/v1/chat/completions",
                                         headers=headers, json=payload, timeout=20)).json()

        return response['choices'][0]['message']['content']

def compose_payload(image_path: str, prompt: str) -> dict:
    """
    Composes a payload dictionary with a base64 encoded image and a text prompt for the GPT-4 Vision model.

    Args:
        image_path (str): The image path to encode and include in the payload.
        prompt (str): The prompt text to accompany the image in the payload.

    Returns:
        dict: A dictionary structured as a payload for the GPT-4 Vision model, including the model name,
              an array of messages each containing a role and content with text and the base64 encoded image,
              and the maximum number of tokens to generate.
    """

    # encode a base64 image from the image path
    with open(image_path, "rb") as image_file:
         encoded_string = base64.b64encode(image_file.read())
         base64_image = encoded_string.decode('utf-8')

    return {
        "model": "gpt-4-vision-preview",
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}"
                        }
                    }
                ]
            }
        ],
        "max_tokens": 600
    }

def compose_text_payload(prompt: str) -> dict:
    """
    Composes a payload dictionary with a base64 encoded image and a text prompt the standard text model.

    Args:
        prompt (str): The prompt text to accompany the image in the payload.

    Returns:
        dict: A dictionary structured as a payload for the GPT-4 Vision model, including the model name,
              an array of messages each containing a role and content with text and the base64 encoded image,
              and the maximum number of tokens to generate.
    """
    return {
        "model": "gpt-4-vision-preview",
        "messages": [
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": prompt
                    }
                ]
            }
        ],
        "max_tokens": 600
    }

async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
    """Set up OpenAI Conversation."""

    async def process_image(call: ServiceCall) -> ServiceResponse:
        """Process an image..."""
        api_key = hass.data[DOMAIN][call.data["config_entry"]]
        client = OpenAIConnector(api_key=api_key)
        prompt = call.data["prompt"]
        image_path = call.data["file_path"]
        response = await client.simple_prompt(prompt=prompt, image_path=image_path)

        return {
            "response": response
        }

    hass.services.async_register(
        DOMAIN,
        SERVICE_PROCESS_IMAGE,
        process_image,
        schema=vol.Schema(
            {
                vol.Required("config_entry"): selector.ConfigEntrySelector(
                    {
                        "integration": DOMAIN,
                    }
                ),
                vol.Required("prompt"): cv.string,
                vol.Required("file_path"): cv.string,
            }
        ),
        supports_response=SupportsResponse.ONLY,
    )
 
    return True
...

process_image:
  fields:
    config_entry:
      required: true
      selector:
        config_entry:
          integration: openai_conversation_custom
    prompt:
      required: true
      selector:
        text:
          multiline: true
    file_path:
      required: true
      selector:
        text:
          multiline: false

As a home security expert looking at the photo of a camera, do you see any security related issues that are signs of vandalism, robbery, broken items, open windows, fire, smoke, or other conditions that can put the house and its inhabitants at risk? Please answer with NORMAL or WARNING or CRITICAL in the first line, by increasing risk level respectively, and then please describe the main risks you have identified.