pc-heiniandClaude Opus 5.5
84f4bcda55
- Send the location question as a user message. A request with only a
system message has no user content, which endpoints in front of Anthropic
models reject ("text content blocks must be non-empty").
- Leave unknown fields out of the web search user_location. After a failed
lookup there is no city or region, and the empty strings made every
request fail ("user_location.city: String should have at least 1
character").
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
914 lines
33 KiB
Python
914 lines
33 KiB
Python
"""Config flow for OpenAI Conversation integration."""
|
|
|
|
from collections.abc import Mapping
|
|
import json
|
|
import logging
|
|
from typing import Any, override
|
|
|
|
import openai
|
|
from probatio import to_openapi
|
|
import voluptuous as vol
|
|
|
|
from homeassistant.components.zone import ENTITY_ID_HOME
|
|
from homeassistant.config_entries import (
|
|
SOURCE_REAUTH,
|
|
ConfigEntry,
|
|
ConfigEntryState,
|
|
ConfigFlow,
|
|
ConfigFlowResult,
|
|
ConfigSubentryFlow,
|
|
SubentryFlowResult,
|
|
)
|
|
from homeassistant.const import (
|
|
CONF_API_KEY,
|
|
CONF_LLM_HASS_API,
|
|
CONF_NAME,
|
|
CONF_PROMPT,
|
|
EntityStateAttribute,
|
|
)
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers import llm
|
|
from homeassistant.helpers.httpx_client import get_async_client
|
|
from homeassistant.helpers.selector import (
|
|
NumberSelector,
|
|
NumberSelectorConfig,
|
|
SelectOptionDict,
|
|
SelectSelector,
|
|
SelectSelectorConfig,
|
|
SelectSelectorMode,
|
|
TemplateSelector,
|
|
TextSelector,
|
|
TextSelectorConfig,
|
|
TextSelectorType,
|
|
)
|
|
from homeassistant.helpers.typing import VolDictType
|
|
|
|
from .const import (
|
|
CONF_BASE_URL,
|
|
CONF_CHAT_MODEL,
|
|
CONF_CODE_INTERPRETER,
|
|
CONF_IMAGE_MODEL,
|
|
CONF_MAX_TOKENS,
|
|
CONF_PRO_MODE,
|
|
CONF_REASONING_EFFORT,
|
|
CONF_REASONING_SUMMARY,
|
|
CONF_RECOMMENDED,
|
|
CONF_SERVICE_TIER,
|
|
CONF_STORE_RESPONSES,
|
|
CONF_TEMPERATURE,
|
|
CONF_TOP_P,
|
|
CONF_TTS_SPEED,
|
|
CONF_VERBOSITY,
|
|
CONF_WEB_SEARCH,
|
|
CONF_WEB_SEARCH_CITY,
|
|
CONF_WEB_SEARCH_CONTEXT_SIZE,
|
|
CONF_WEB_SEARCH_COUNTRY,
|
|
CONF_WEB_SEARCH_INLINE_CITATIONS,
|
|
CONF_WEB_SEARCH_REGION,
|
|
CONF_WEB_SEARCH_TIMEZONE,
|
|
CONF_WEB_SEARCH_USER_LOCATION,
|
|
DEFAULT_AI_TASK_NAME,
|
|
DEFAULT_BASE_URL,
|
|
DEFAULT_CONVERSATION_NAME,
|
|
DEFAULT_STT_NAME,
|
|
DEFAULT_STT_PROMPT,
|
|
DEFAULT_TTS_NAME,
|
|
DOMAIN,
|
|
PLACEHOLDER_API_KEY,
|
|
RECOMMENDED_AI_TASK_OPTIONS,
|
|
RECOMMENDED_CHAT_MODEL,
|
|
RECOMMENDED_CODE_INTERPRETER,
|
|
RECOMMENDED_CONVERSATION_OPTIONS,
|
|
RECOMMENDED_IMAGE_MODEL,
|
|
RECOMMENDED_MAX_TOKENS,
|
|
RECOMMENDED_PRO_MODE,
|
|
RECOMMENDED_REASONING_EFFORT,
|
|
RECOMMENDED_REASONING_SUMMARY,
|
|
RECOMMENDED_SERVICE_TIER,
|
|
RECOMMENDED_STORE_RESPONSES,
|
|
RECOMMENDED_STT_MODEL,
|
|
RECOMMENDED_STT_OPTIONS,
|
|
RECOMMENDED_TEMPERATURE,
|
|
RECOMMENDED_TOP_P,
|
|
RECOMMENDED_TTS_OPTIONS,
|
|
RECOMMENDED_TTS_SPEED,
|
|
RECOMMENDED_VERBOSITY,
|
|
RECOMMENDED_WEB_SEARCH,
|
|
RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE,
|
|
RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS,
|
|
RECOMMENDED_WEB_SEARCH_USER_LOCATION,
|
|
UNSUPPORTED_CODE_INTERPRETER_MODELS,
|
|
UNSUPPORTED_FLEX_SERVICE_TIERS_MODELS,
|
|
UNSUPPORTED_IMAGE_MODELS,
|
|
UNSUPPORTED_MODELS,
|
|
UNSUPPORTED_PRIORITY_SERVICE_TIERS_MODELS,
|
|
UNSUPPORTED_WEB_SEARCH_MODELS,
|
|
)
|
|
from .model_list import async_fetch_model_ids, default_model, model_field
|
|
from .url_util import (
|
|
CredentialsInBaseURL,
|
|
InvalidBaseURL,
|
|
normalize_base_url,
|
|
title_from_base_url,
|
|
)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
STEP_REAUTH_DATA_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Required(CONF_BASE_URL, default=DEFAULT_BASE_URL): str,
|
|
vol.Optional(CONF_API_KEY, default=""): str,
|
|
}
|
|
)
|
|
|
|
# The user step additionally names the provider, since there is one entry per
|
|
# server and the host alone is a poor label. Reauth deliberately reuses the
|
|
# schema *without* the name: that step writes its input straight back into
|
|
# entry.data via data_updates, so a name field there would persist the title
|
|
# as connection data on every key rotation.
|
|
STEP_USER_DATA_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Optional(CONF_NAME, default=""): str,
|
|
vol.Required(CONF_BASE_URL, default=DEFAULT_BASE_URL): str,
|
|
vol.Optional(CONF_API_KEY, default=""): str,
|
|
}
|
|
)
|
|
|
|
|
|
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
|
|
"""Validate the user input allows us to connect.
|
|
|
|
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
|
|
"""
|
|
data[CONF_BASE_URL] = normalize_base_url(data[CONF_BASE_URL])
|
|
client = openai.AsyncOpenAI(
|
|
api_key=data.get(CONF_API_KEY) or PLACEHOLDER_API_KEY,
|
|
base_url=data[CONF_BASE_URL],
|
|
http_client=get_async_client(hass),
|
|
)
|
|
await client.models.list(timeout=10.0)
|
|
|
|
|
|
class OpenAIConfigFlow(ConfigFlow, domain=DOMAIN):
|
|
"""Handle a config flow for OpenAI Conversation."""
|
|
|
|
VERSION = 2
|
|
MINOR_VERSION = 8
|
|
|
|
@override
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Handle the initial step."""
|
|
|
|
errors: dict[str, str] = {}
|
|
|
|
if user_input is not None:
|
|
# Deliberately no _async_abort_entries_match() here: upstream
|
|
# aborts when url+key match an existing entry, which blocks
|
|
# legitimate setups such as one entry per provider-side key, or a
|
|
# second entry against the same server kept for a different
|
|
# purpose. Duplicate entries are the user's call to make.
|
|
try:
|
|
await validate_input(self.hass, user_input)
|
|
except CredentialsInBaseURL:
|
|
errors["base"] = "credentials_in_url"
|
|
except InvalidBaseURL:
|
|
errors["base"] = "invalid_base_url"
|
|
except openai.APIConnectionError:
|
|
errors["base"] = "cannot_connect"
|
|
except openai.AuthenticationError:
|
|
errors["base"] = "invalid_auth"
|
|
except Exception:
|
|
_LOGGER.exception("Unexpected exception")
|
|
errors["base"] = "unknown"
|
|
else:
|
|
if self.source == SOURCE_REAUTH:
|
|
return self.async_update_reload_and_abort(
|
|
self._get_reauth_entry(), data_updates=user_input
|
|
)
|
|
name = (user_input.pop(CONF_NAME, "") or "").strip()
|
|
return self.async_create_entry(
|
|
title=name or title_from_base_url(user_input[CONF_BASE_URL]),
|
|
data=user_input,
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
STEP_USER_DATA_SCHEMA, user_input
|
|
),
|
|
errors=errors,
|
|
description_placeholders={
|
|
"instructions_url": "https://github.com/adamjs83/ha-openai-compatible#configuration",
|
|
},
|
|
)
|
|
|
|
async def async_step_reauth(
|
|
self, entry_data: Mapping[str, Any]
|
|
) -> ConfigFlowResult:
|
|
"""Perform reauth upon an API authentication error."""
|
|
return await self.async_step_reauth_confirm()
|
|
|
|
async def async_step_reauth_confirm(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> ConfigFlowResult:
|
|
"""Dialog that informs the user that reauth is required."""
|
|
if not user_input:
|
|
return self.async_show_form(
|
|
step_id="reauth_confirm",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
STEP_REAUTH_DATA_SCHEMA, self._get_reauth_entry().data
|
|
),
|
|
)
|
|
|
|
return await self.async_step_user(user_input)
|
|
|
|
@classmethod
|
|
@callback
|
|
@override
|
|
def async_get_supported_subentry_types(
|
|
cls, config_entry: ConfigEntry
|
|
) -> dict[str, type[ConfigSubentryFlow]]:
|
|
"""Return subentries supported by this integration."""
|
|
return {
|
|
"conversation": OpenAISubentryFlowHandler,
|
|
"ai_task_data": OpenAISubentryFlowHandler,
|
|
"stt": OpenAISubentrySTTFlowHandler,
|
|
"tts": OpenAISubentryTTSFlowHandler,
|
|
}
|
|
|
|
|
|
class OpenAISubentryFlowHandler(ConfigSubentryFlow):
|
|
"""Flow for managing OpenAI subentries."""
|
|
|
|
options: dict[str, Any]
|
|
|
|
@property
|
|
def _is_new(self) -> bool:
|
|
"""Return if this is a new subentry."""
|
|
return self.source == "user"
|
|
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Add a subentry."""
|
|
if self._subentry_type == "ai_task_data":
|
|
self.options = RECOMMENDED_AI_TASK_OPTIONS.copy()
|
|
else:
|
|
self.options = RECOMMENDED_CONVERSATION_OPTIONS.copy()
|
|
# Start a new endpoint with "recommended" unticked. Ticked (upstream's
|
|
# default) creates the subentry straight from this first step, which
|
|
# silently pins RECOMMENDED_CHAT_MODEL -- an OpenAI model name that a
|
|
# third-party provider will not serve -- and never shows the model
|
|
# field at all. Unticked, the next step asks which model to use.
|
|
# Reconfigure is untouched: it keeps whatever the subentry was saved
|
|
# with, via async_step_reconfigure.
|
|
self.options[CONF_RECOMMENDED] = False
|
|
return await self.async_step_init()
|
|
|
|
async def async_step_reconfigure(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Handle reconfiguration of a subentry."""
|
|
self.options = self._get_reconfigure_subentry().data.copy()
|
|
return await self.async_step_init()
|
|
|
|
async def async_step_init(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Manage initial options."""
|
|
# abort if entry is not loaded
|
|
if self._get_entry().state is not ConfigEntryState.LOADED:
|
|
return self.async_abort(reason="entry_not_loaded")
|
|
|
|
options = self.options
|
|
|
|
hass_apis: list[SelectOptionDict] = [
|
|
SelectOptionDict(
|
|
label=api.name,
|
|
value=api.id,
|
|
)
|
|
for api in llm.async_get_apis(self.hass)
|
|
]
|
|
if suggested_llm_apis := options.get(CONF_LLM_HASS_API):
|
|
if isinstance(suggested_llm_apis, str):
|
|
suggested_llm_apis = [suggested_llm_apis]
|
|
valid_apis = {api.id for api in llm.async_get_apis(self.hass)}
|
|
options[CONF_LLM_HASS_API] = [
|
|
api for api in suggested_llm_apis if api in valid_apis
|
|
]
|
|
|
|
step_schema: VolDictType = {}
|
|
|
|
if self._is_new:
|
|
if self._subentry_type == "ai_task_data":
|
|
default_name = DEFAULT_AI_TASK_NAME
|
|
else:
|
|
default_name = DEFAULT_CONVERSATION_NAME
|
|
step_schema[vol.Required(CONF_NAME, default=default_name)] = str
|
|
|
|
if self._subentry_type == "conversation":
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_PROMPT,
|
|
description={
|
|
"suggested_value": options.get(
|
|
CONF_PROMPT, llm.DEFAULT_INSTRUCTIONS_PROMPT
|
|
)
|
|
},
|
|
): TemplateSelector(),
|
|
vol.Optional(CONF_LLM_HASS_API): SelectSelector(
|
|
SelectSelectorConfig(options=hass_apis, multiple=True)
|
|
),
|
|
}
|
|
)
|
|
|
|
# The model belongs on the first screen. Picking one is the entire
|
|
# point of adding an endpoint, and behind the "recommended" checkbox
|
|
# it was unreachable: ticked, this step creates the subentry and the
|
|
# model silently stays RECOMMENDED_CHAT_MODEL -- an OpenAI name a
|
|
# third-party provider will not serve. entity.py reads
|
|
# CONF_CHAT_MODEL regardless of the recommended flag, so a model
|
|
# chosen here takes effect on either path.
|
|
model_ids = await async_fetch_model_ids(self._get_entry().runtime_data)
|
|
step_schema[
|
|
vol.Optional(CONF_CHAT_MODEL, default=default_model(model_ids))
|
|
] = model_field(model_ids)
|
|
|
|
step_schema[
|
|
vol.Required(CONF_RECOMMENDED, default=options.get(CONF_RECOMMENDED, False))
|
|
] = bool
|
|
|
|
errors: dict[str, str] = {}
|
|
|
|
if user_input is not None:
|
|
if user_input.get(CONF_LLM_HASS_API) is None:
|
|
user_input.pop(CONF_LLM_HASS_API, None)
|
|
|
|
# Checked here rather than on the next step, because the
|
|
# recommended path creates the subentry from this one and would
|
|
# otherwise skip the check entirely.
|
|
if user_input.get(CONF_CHAT_MODEL) in UNSUPPORTED_MODELS:
|
|
errors[CONF_CHAT_MODEL] = "model_not_supported"
|
|
|
|
if not errors:
|
|
if user_input[CONF_RECOMMENDED]:
|
|
if self._is_new:
|
|
return self.async_create_entry(
|
|
title=user_input.pop(CONF_NAME),
|
|
data=user_input,
|
|
)
|
|
return self.async_update_and_abort(
|
|
self._get_entry(),
|
|
self._get_reconfigure_subentry(),
|
|
data=user_input,
|
|
)
|
|
|
|
options.update(user_input)
|
|
if CONF_LLM_HASS_API in options and CONF_LLM_HASS_API not in user_input:
|
|
options.pop(CONF_LLM_HASS_API)
|
|
return await self.async_step_additional()
|
|
|
|
return self.async_show_form(
|
|
step_id="init",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
vol.Schema(step_schema), options
|
|
),
|
|
errors=errors,
|
|
)
|
|
|
|
async def async_step_additional(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Manage additional options."""
|
|
options = self.options
|
|
errors: dict[str, str] = {}
|
|
|
|
step_schema: VolDictType = {
|
|
# CONF_CHAT_MODEL is asked on the init step, so that it is
|
|
# reachable without unticking "recommended" first.
|
|
vol.Optional(
|
|
CONF_MAX_TOKENS,
|
|
default=RECOMMENDED_MAX_TOKENS,
|
|
): int,
|
|
vol.Optional(
|
|
CONF_TOP_P,
|
|
default=RECOMMENDED_TOP_P,
|
|
): NumberSelector(NumberSelectorConfig(min=0, max=1, step=0.05)),
|
|
vol.Optional(
|
|
CONF_TEMPERATURE,
|
|
default=RECOMMENDED_TEMPERATURE,
|
|
): NumberSelector(NumberSelectorConfig(min=0, max=2, step=0.05)),
|
|
vol.Optional(
|
|
CONF_STORE_RESPONSES,
|
|
default=RECOMMENDED_STORE_RESPONSES,
|
|
): bool,
|
|
}
|
|
|
|
if user_input is not None:
|
|
options.update(user_input)
|
|
|
|
if not errors:
|
|
return await self.async_step_model()
|
|
|
|
return self.async_show_form(
|
|
step_id="additional",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
vol.Schema(step_schema), options
|
|
),
|
|
errors=errors,
|
|
)
|
|
|
|
async def async_step_model(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Manage model-specific options."""
|
|
options = self.options
|
|
errors: dict[str, str] = {}
|
|
|
|
step_schema: VolDictType = {}
|
|
|
|
model = options.get(CONF_CHAT_MODEL, RECOMMENDED_CHAT_MODEL)
|
|
|
|
if not model.startswith(tuple(UNSUPPORTED_CODE_INTERPRETER_MODELS)):
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_CODE_INTERPRETER,
|
|
default=RECOMMENDED_CODE_INTERPRETER,
|
|
): bool,
|
|
}
|
|
)
|
|
elif CONF_CODE_INTERPRETER in options:
|
|
options.pop(CONF_CODE_INTERPRETER)
|
|
|
|
if reasoning_options := self._get_reasoning_options(model):
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_REASONING_EFFORT,
|
|
default=RECOMMENDED_REASONING_EFFORT,
|
|
): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=reasoning_options,
|
|
translation_key=CONF_REASONING_EFFORT,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
),
|
|
}
|
|
)
|
|
elif CONF_REASONING_EFFORT in options:
|
|
options.pop(CONF_REASONING_EFFORT)
|
|
|
|
if model.startswith("gpt-5.6"):
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_PRO_MODE,
|
|
default=RECOMMENDED_PRO_MODE,
|
|
): bool,
|
|
}
|
|
)
|
|
elif CONF_PRO_MODE in options:
|
|
options.pop(CONF_PRO_MODE)
|
|
|
|
if model.startswith("gpt-5"):
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_VERBOSITY,
|
|
default=RECOMMENDED_VERBOSITY,
|
|
): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=["low", "medium", "high"],
|
|
translation_key=CONF_VERBOSITY,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
),
|
|
}
|
|
)
|
|
elif CONF_VERBOSITY in options:
|
|
options.pop(CONF_VERBOSITY)
|
|
|
|
if model.startswith(("o", "gpt-5")):
|
|
reasoning_summary_options = ["off", "auto", "concise", "detailed"]
|
|
if model.startswith("o"):
|
|
reasoning_summary_options.remove("concise")
|
|
stored_summary = options.get(
|
|
CONF_REASONING_SUMMARY, RECOMMENDED_REASONING_SUMMARY
|
|
)
|
|
if stored_summary not in reasoning_summary_options:
|
|
stored_summary = RECOMMENDED_REASONING_SUMMARY
|
|
options[CONF_REASONING_SUMMARY] = stored_summary
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_REASONING_SUMMARY,
|
|
default=stored_summary,
|
|
): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=reasoning_summary_options,
|
|
translation_key=CONF_REASONING_SUMMARY,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
),
|
|
}
|
|
)
|
|
elif CONF_REASONING_SUMMARY in options:
|
|
options.pop(CONF_REASONING_SUMMARY)
|
|
|
|
service_tiers = self._get_service_tiers(model)
|
|
if "flex" in service_tiers or "priority" in service_tiers:
|
|
step_schema[
|
|
vol.Optional(
|
|
CONF_SERVICE_TIER,
|
|
default=RECOMMENDED_SERVICE_TIER,
|
|
)
|
|
] = SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=service_tiers,
|
|
translation_key=CONF_SERVICE_TIER,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
)
|
|
else:
|
|
options.pop(CONF_SERVICE_TIER, None)
|
|
if options.get(CONF_SERVICE_TIER) not in service_tiers:
|
|
options.pop(CONF_SERVICE_TIER, None)
|
|
|
|
if self._subentry_type == "conversation" and not model.startswith(
|
|
tuple(UNSUPPORTED_WEB_SEARCH_MODELS)
|
|
):
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_WEB_SEARCH,
|
|
default=RECOMMENDED_WEB_SEARCH,
|
|
): bool,
|
|
vol.Optional(
|
|
CONF_WEB_SEARCH_CONTEXT_SIZE,
|
|
default=RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE,
|
|
): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=["low", "medium", "high"],
|
|
translation_key=CONF_WEB_SEARCH_CONTEXT_SIZE,
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
),
|
|
vol.Optional(
|
|
CONF_WEB_SEARCH_USER_LOCATION,
|
|
default=RECOMMENDED_WEB_SEARCH_USER_LOCATION,
|
|
): bool,
|
|
vol.Optional(
|
|
CONF_WEB_SEARCH_INLINE_CITATIONS,
|
|
default=RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS,
|
|
): bool,
|
|
}
|
|
)
|
|
elif CONF_WEB_SEARCH in options:
|
|
options = {
|
|
k: v
|
|
for k, v in options.items()
|
|
if k
|
|
not in (
|
|
CONF_WEB_SEARCH,
|
|
CONF_WEB_SEARCH_CONTEXT_SIZE,
|
|
CONF_WEB_SEARCH_USER_LOCATION,
|
|
CONF_WEB_SEARCH_CITY,
|
|
CONF_WEB_SEARCH_REGION,
|
|
CONF_WEB_SEARCH_COUNTRY,
|
|
CONF_WEB_SEARCH_TIMEZONE,
|
|
CONF_WEB_SEARCH_INLINE_CITATIONS,
|
|
)
|
|
}
|
|
|
|
if self._subentry_type == "ai_task_data" and not model.startswith(
|
|
tuple(UNSUPPORTED_IMAGE_MODELS)
|
|
):
|
|
step_schema[
|
|
vol.Optional(CONF_IMAGE_MODEL, default=RECOMMENDED_IMAGE_MODEL)
|
|
] = SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=[
|
|
"gpt-image-2",
|
|
"gpt-image-1.5",
|
|
"gpt-image-1",
|
|
"gpt-image-1-mini",
|
|
],
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
)
|
|
)
|
|
|
|
if user_input is not None:
|
|
if user_input.get(CONF_WEB_SEARCH):
|
|
if user_input.get(CONF_REASONING_EFFORT) == "minimal":
|
|
errors[CONF_WEB_SEARCH] = "web_search_minimal_reasoning"
|
|
if user_input.get(CONF_WEB_SEARCH_USER_LOCATION) and not errors:
|
|
user_input.update(await self._get_location_data())
|
|
else:
|
|
options.pop(CONF_WEB_SEARCH_CITY, None)
|
|
options.pop(CONF_WEB_SEARCH_REGION, None)
|
|
options.pop(CONF_WEB_SEARCH_COUNTRY, None)
|
|
options.pop(CONF_WEB_SEARCH_TIMEZONE, None)
|
|
if (
|
|
user_input.get(CONF_CODE_INTERPRETER)
|
|
and user_input.get(CONF_REASONING_EFFORT) == "minimal"
|
|
):
|
|
errors[CONF_CODE_INTERPRETER] = "code_interpreter_minimal_reasoning"
|
|
|
|
options.update(user_input)
|
|
if not errors:
|
|
if self._is_new:
|
|
return self.async_create_entry(
|
|
title=options.pop(CONF_NAME),
|
|
data=options,
|
|
)
|
|
return self.async_update_and_abort(
|
|
self._get_entry(),
|
|
self._get_reconfigure_subentry(),
|
|
data=options,
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="model",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
vol.Schema(step_schema), options
|
|
),
|
|
errors=errors,
|
|
)
|
|
|
|
def _get_reasoning_options(self, model: str) -> list[str]:
|
|
"""Get reasoning effort options based on model."""
|
|
if not model.startswith(("o", "gpt-5")) or model.startswith("gpt-5-pro"):
|
|
return []
|
|
|
|
models_reasoning_map: dict[str | tuple[str, ...], list[str]] = {
|
|
"gpt-5.6": ["none", "low", "medium", "high", "xhigh", "max"],
|
|
("gpt-5.2-pro", "gpt-5.4-pro", "gpt-5.5-pro"): ["medium", "high", "xhigh"],
|
|
("gpt-5.2", "gpt-5.3", "gpt-5.4", "gpt-5.5"): [
|
|
"none",
|
|
"low",
|
|
"medium",
|
|
"high",
|
|
"xhigh",
|
|
],
|
|
"gpt-5.1": ["none", "low", "medium", "high"],
|
|
"gpt-5": ["minimal", "low", "medium", "high"],
|
|
"": ["low", "medium", "high"], # The default case
|
|
}
|
|
|
|
for prefix, options in models_reasoning_map.items():
|
|
if model.startswith(prefix):
|
|
return options
|
|
return [] # pragma: no cover
|
|
|
|
def _get_service_tiers(self, model: str) -> list[str]:
|
|
"""Get service tier options based on model."""
|
|
service_tiers = ["auto"]
|
|
|
|
if not model.startswith(tuple(UNSUPPORTED_FLEX_SERVICE_TIERS_MODELS)):
|
|
service_tiers.append("flex")
|
|
|
|
service_tiers.append("default")
|
|
|
|
if not model.startswith(tuple(UNSUPPORTED_PRIORITY_SERVICE_TIERS_MODELS)):
|
|
service_tiers.append("priority")
|
|
|
|
return service_tiers
|
|
|
|
async def _get_location_data(self) -> dict[str, str]:
|
|
"""Get approximate location data of the user."""
|
|
location_data: dict[str, str] = {}
|
|
zone_home = self.hass.states.get(ENTITY_ID_HOME)
|
|
if zone_home is not None:
|
|
client = openai.AsyncOpenAI(
|
|
api_key=self._get_entry().data.get(CONF_API_KEY) or PLACEHOLDER_API_KEY,
|
|
base_url=self._get_entry().data[CONF_BASE_URL],
|
|
http_client=get_async_client(self.hass),
|
|
)
|
|
location_schema = vol.Schema(
|
|
{
|
|
vol.Optional(
|
|
CONF_WEB_SEARCH_CITY,
|
|
description=(
|
|
"Free text input for the city, e.g. `San Francisco`"
|
|
),
|
|
): str,
|
|
vol.Optional(
|
|
CONF_WEB_SEARCH_REGION,
|
|
description="Free text input for the region, e.g. `California`",
|
|
): str,
|
|
}
|
|
)
|
|
# Ask the subentry's own model: the endpoint may not serve
|
|
# RECOMMENDED_CHAT_MODEL (an OpenAI name). A failed lookup only
|
|
# costs the city/region hint, so it must not block saving.
|
|
try:
|
|
response = await client.responses.create(
|
|
model=self.options.get(CONF_CHAT_MODEL, RECOMMENDED_CHAT_MODEL),
|
|
# A user message, not a system one: endpoints in front of
|
|
# Anthropic models reject a request without user content.
|
|
input=[
|
|
{
|
|
"role": "user",
|
|
"content": "Where are the following coordinates located: "
|
|
f"({zone_home.attributes[EntityStateAttribute.LATITUDE]},"
|
|
f" {zone_home.attributes[EntityStateAttribute.LONGITUDE]})?",
|
|
}
|
|
],
|
|
text={
|
|
"format": {
|
|
"type": "json_schema",
|
|
"name": "approximate_location",
|
|
"description": "Approximate location data of the user "
|
|
"for refined web search results",
|
|
"schema": to_openapi(location_schema),
|
|
"strict": False,
|
|
}
|
|
},
|
|
store=self.options.get(
|
|
CONF_STORE_RESPONSES, RECOMMENDED_STORE_RESPONSES
|
|
),
|
|
)
|
|
location_data = location_schema(json.loads(response.output_text) or {})
|
|
except (openai.OpenAIError, ValueError, vol.Invalid) as err:
|
|
_LOGGER.warning(
|
|
"Could not look up the home location for web search: %s", err
|
|
)
|
|
|
|
if self.hass.config.country:
|
|
location_data[CONF_WEB_SEARCH_COUNTRY] = self.hass.config.country
|
|
location_data[CONF_WEB_SEARCH_TIMEZONE] = self.hass.config.time_zone
|
|
|
|
_LOGGER.debug("Location data: %s", location_data)
|
|
|
|
return location_data
|
|
|
|
|
|
class OpenAISubentrySTTFlowHandler(ConfigSubentryFlow):
|
|
"""Flow for managing OpenAI STT subentries."""
|
|
|
|
options: dict[str, Any]
|
|
|
|
@property
|
|
def _is_new(self) -> bool:
|
|
"""Return if this is a new subentry."""
|
|
return self.source == "user"
|
|
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Add a subentry."""
|
|
self.options = RECOMMENDED_STT_OPTIONS.copy()
|
|
return await self.async_step_init()
|
|
|
|
async def async_step_reconfigure(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Handle reconfiguration of a subentry."""
|
|
self.options = self._get_reconfigure_subentry().data.copy()
|
|
return await self.async_step_init()
|
|
|
|
async def async_step_init(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Manage initial options."""
|
|
# abort if entry is not loaded
|
|
if self._get_entry().state is not ConfigEntryState.LOADED:
|
|
return self.async_abort(reason="entry_not_loaded")
|
|
|
|
options = self.options
|
|
errors: dict[str, str] = {}
|
|
|
|
step_schema: VolDictType = {}
|
|
|
|
if self._is_new:
|
|
step_schema[vol.Required(CONF_NAME, default=DEFAULT_STT_NAME)] = str
|
|
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(
|
|
CONF_PROMPT,
|
|
description={
|
|
"suggested_value": options.get(CONF_PROMPT, DEFAULT_STT_PROMPT)
|
|
},
|
|
): TextSelector(
|
|
TextSelectorConfig(multiline=True, type=TextSelectorType.TEXT)
|
|
),
|
|
# Upstream hardcodes OpenAI's three transcribe models here;
|
|
# ask the provider instead.
|
|
vol.Optional(
|
|
CONF_CHAT_MODEL, default=RECOMMENDED_STT_MODEL
|
|
): model_field(
|
|
await async_fetch_model_ids(self._get_entry().runtime_data)
|
|
),
|
|
}
|
|
)
|
|
|
|
if user_input is not None:
|
|
options.update(user_input)
|
|
if not errors:
|
|
if self._is_new:
|
|
return self.async_create_entry(
|
|
title=options.pop(CONF_NAME),
|
|
data=options,
|
|
)
|
|
return self.async_update_and_abort(
|
|
self._get_entry(),
|
|
self._get_reconfigure_subentry(),
|
|
data=options,
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="init",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
vol.Schema(step_schema), options
|
|
),
|
|
errors=errors,
|
|
)
|
|
|
|
|
|
class OpenAISubentryTTSFlowHandler(ConfigSubentryFlow):
|
|
"""Flow for managing OpenAI TTS subentries."""
|
|
|
|
options: dict[str, Any]
|
|
|
|
@property
|
|
def _is_new(self) -> bool:
|
|
"""Return if this is a new subentry."""
|
|
return self.source == "user"
|
|
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Add a subentry."""
|
|
self.options = RECOMMENDED_TTS_OPTIONS.copy()
|
|
return await self.async_step_init()
|
|
|
|
async def async_step_reconfigure(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Handle reconfiguration of a subentry."""
|
|
self.options = self._get_reconfigure_subentry().data.copy()
|
|
return await self.async_step_init()
|
|
|
|
async def async_step_init(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> SubentryFlowResult:
|
|
"""Manage initial options."""
|
|
# abort if entry is not loaded
|
|
if self._get_entry().state is not ConfigEntryState.LOADED:
|
|
return self.async_abort(reason="entry_not_loaded")
|
|
|
|
options = self.options
|
|
errors: dict[str, str] = {}
|
|
|
|
step_schema: VolDictType = {}
|
|
|
|
if self._is_new:
|
|
step_schema[vol.Required(CONF_NAME, default=DEFAULT_TTS_NAME)] = str
|
|
|
|
step_schema.update(
|
|
{
|
|
vol.Optional(CONF_PROMPT): TextSelector(
|
|
TextSelectorConfig(multiline=True, type=TextSelectorType.TEXT)
|
|
),
|
|
# Upstream offers no model field for TTS at all -- it silently
|
|
# uses RECOMMENDED_TTS_OPTIONS' gpt-4o-mini-tts, which a
|
|
# third-party endpoint is unlikely to serve.
|
|
vol.Optional(
|
|
CONF_CHAT_MODEL,
|
|
default=RECOMMENDED_TTS_OPTIONS[CONF_CHAT_MODEL],
|
|
): model_field(
|
|
await async_fetch_model_ids(self._get_entry().runtime_data)
|
|
),
|
|
vol.Optional(
|
|
CONF_TTS_SPEED, default=RECOMMENDED_TTS_SPEED
|
|
): NumberSelector(NumberSelectorConfig(min=0.25, max=4.0, step=0.01)),
|
|
}
|
|
)
|
|
|
|
if user_input is not None:
|
|
options.update(user_input)
|
|
if not errors:
|
|
if self._is_new:
|
|
return self.async_create_entry(
|
|
title=options.pop(CONF_NAME),
|
|
data=options,
|
|
)
|
|
return self.async_update_and_abort(
|
|
self._get_entry(),
|
|
self._get_reconfigure_subentry(),
|
|
data=options,
|
|
)
|
|
|
|
return self.async_show_form(
|
|
step_id="init",
|
|
data_schema=self.add_suggested_values_to_schema(
|
|
vol.Schema(step_schema), options
|
|
),
|
|
errors=errors,
|
|
)
|