"""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, )