Adam SmithandClaude Opus 5
117780dd68
async_setup_entry does entry.data[CONF_BASE_URL] as a hard index, but base_url is this fork's own addition: an entry created by a build from before the base_url commits (48195c9 era, version 2 minor 7) has no such key. On restart, setup raises KeyError: 'base_url' and the integration fails to load -- reproduced in test first, where async_setup returns False with that KeyError in the log. Fixed with a migration step rather than a defensive .get(), matching the pattern the rest of async_migrate_entry already uses, so the invariant "a loaded entry has base_url" holds for every caller instead of each call site guarding itself: - new step 2.7 -> 2.8 backfills DEFAULT_BASE_URL via dict.setdefault, so an entry that already has a base_url keeps its own endpoint and only a missing one is filled. An install from before this feature was implicitly using api.openai.com, which is exactly DEFAULT_BASE_URL, so the backfill preserves existing behavior rather than repointing anyone. - config_flow MINOR_VERSION 7 -> 8, otherwise the step never runs. test_client_built_with_base_url now also covers the preserve path, since its entry starts at minor 7 with a base_url set and must still reach the client unchanged after migrating. Also updated test_form's `minor_version == 7` assertion to 8; it pins the version new entries are created at, so the bump is expected. The reauth test's MockConfigEntry stays at minor 7 on purpose -- it stands in for an already-existing entry. Local: 138 passed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NxJYWgGJtCw13qUHZtG8ki
877 lines
30 KiB
Python
877 lines
30 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 .url_util import CredentialsInBaseURL, InvalidBaseURL, normalize_base_url
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
STEP_USER_DATA_SCHEMA = vol.Schema(
|
|
{
|
|
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:
|
|
self._async_abort_entries_match(user_input)
|
|
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
|
|
)
|
|
return self.async_create_entry(
|
|
title="ChatGPT",
|
|
data=user_input,
|
|
subentries=[
|
|
{
|
|
"subentry_type": "conversation",
|
|
"data": RECOMMENDED_CONVERSATION_OPTIONS,
|
|
"title": DEFAULT_CONVERSATION_NAME,
|
|
"unique_id": None,
|
|
},
|
|
{
|
|
"subentry_type": "ai_task_data",
|
|
"data": RECOMMENDED_AI_TASK_OPTIONS,
|
|
"title": DEFAULT_AI_TASK_NAME,
|
|
"unique_id": None,
|
|
},
|
|
{
|
|
"subentry_type": "stt",
|
|
"data": RECOMMENDED_STT_OPTIONS,
|
|
"title": DEFAULT_STT_NAME,
|
|
"unique_id": None,
|
|
},
|
|
{
|
|
"subentry_type": "tts",
|
|
"data": RECOMMENDED_TTS_OPTIONS,
|
|
"title": DEFAULT_TTS_NAME,
|
|
"unique_id": None,
|
|
},
|
|
],
|
|
)
|
|
|
|
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#generate-an-api-key",
|
|
},
|
|
)
|
|
|
|
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_USER_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()
|
|
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)
|
|
),
|
|
}
|
|
)
|
|
|
|
step_schema[
|
|
vol.Required(CONF_RECOMMENDED, default=options.get(CONF_RECOMMENDED, False))
|
|
] = bool
|
|
|
|
if user_input is not None:
|
|
if user_input.get(CONF_LLM_HASS_API) is None:
|
|
user_input.pop(CONF_LLM_HASS_API, None)
|
|
|
|
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
|
|
),
|
|
)
|
|
|
|
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 = {
|
|
vol.Optional(
|
|
CONF_CHAT_MODEL,
|
|
default=RECOMMENDED_CHAT_MODEL,
|
|
): str,
|
|
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 user_input.get(CONF_CHAT_MODEL) in UNSUPPORTED_MODELS:
|
|
errors[CONF_CHAT_MODEL] = "model_not_supported"
|
|
|
|
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[CONF_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,
|
|
}
|
|
)
|
|
response = await client.responses.create(
|
|
model=RECOMMENDED_CHAT_MODEL,
|
|
input=[
|
|
{
|
|
"role": "system",
|
|
"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 {})
|
|
|
|
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)
|
|
),
|
|
vol.Optional(
|
|
CONF_CHAT_MODEL, default=RECOMMENDED_STT_MODEL
|
|
): SelectSelector(
|
|
SelectSelectorConfig(
|
|
options=[
|
|
"gpt-4o-transcribe",
|
|
"gpt-4o-mini-transcribe",
|
|
"whisper-1",
|
|
],
|
|
mode=SelectSelectorMode.DROPDOWN,
|
|
custom_value=True,
|
|
)
|
|
),
|
|
}
|
|
)
|
|
|
|
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)
|
|
),
|
|
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,
|
|
)
|