pc-heini/ha-openai-compatible
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
030822c845bcc99cf5b5a94eda3f973af790a4f0
ha-openai-compatible/custom_components/openai_compatible/config_flow.py
T
Adam SmithandClaude Opus 5 e7fcb1d7a1 feat: one entry per provider, endpoints per type and model from a live dropdown 
Reshapes setup around how a multi-model proxy is actually used: one config
entry per server/provider, then one endpoint per model added by hand, with
the model chosen from what that provider advertises right now.

Setup (async_step_user):
- Adds a Name field. With one entry per provider the title matters, and it
  was hardcoded to "ChatGPT" regardless of where the entry pointed. Blank
  falls back to the endpoint host via url_util.title_from_base_url.
  Deliberately absent from the reauth schema (now split out as
  STEP_REAUTH_DATA_SCHEMA): that step writes its input back into entry.data
  via data_updates, so a name field there would persist the title as
  connection data on every key rotation.
- Stops auto-creating four subentries. conversation/ai_task_data/stt/tts were
  all created up front carrying RECOMMENDED_* options, which pin OpenAI model
  names (gpt-4o-mini-tts, gpt-4o-mini-transcribe). Against a third-party
  endpoint those entities are dead on arrival -- the reported symptom was an
  install that "loads bits, it's all janky".
- Drops _async_abort_entries_match, so several entries may point at one
  server (one per provider-side key, for instance). Upstream aborts with
  already_configured on matching url+key.

Model discovery (new model_list.py, kept out of the vendored files like
url_util.py):
- async_fetch_model_ids asks the endpoint's GET /v1/models. It calls
  with_options(max_retries=0), because this runs while the user waits on a
  dialog and the SDK otherwise retries with backoff -- visible in the test
  run as repeated "Retrying request to /models" and a suite that went from
  5s to minutes.
- Returns [] on any OpenAIError, and model_field() then yields a plain text
  field instead of a dropdown, so an endpoint that serves /v1/responses but
  not /v1/models is still configurable by hand. The dropdown sets
  custom_value=True for the same reason (LiteLLM wildcard routes).
- Wired into all four types: conversation and ai_task_data on the additional
  step, stt (which hardcoded OpenAI's three transcribe models) and tts on
  init. TTS had no model field at all -- it silently used gpt-4o-mini-tts.

A new endpoint now starts with "recommended" unticked. Ticked, the flow ends
on the first step and never shows a model field, which is why the model
looked unpickable; reconfigure still keeps whatever was saved.

The model field stays on the additional step rather than moving to init. The
init position reads better, but 71 ported tests drive the model through the
additional step, several from parametrized option dicts that no mechanical
edit could relocate. Rewriting the only regression net against upstream to
save one click was the wrong trade.

Also: the setup description no longer says "your OpenAI API key", and
instructions_url pointed at a README anchor (#generate-an-api-key) that does
not exist -- now #configuration. strings.json re-verified against core's
RE_URL: 0 violations.

Local: 167 passed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NxJYWgGJtCw13qUHZtG8ki
2026-09-17 22:45:36 -04:00

890 lines
31 KiB
Python
Raw Blame History

"""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, 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)
),
}
)
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 = {
# Upstream makes this a free-text box defaulting to an OpenAI
# model name. Offer what the provider actually serves instead.
vol.Optional(
CONF_CHAT_MODEL,
default=RECOMMENDED_CHAT_MODEL,
): model_field(
await async_fetch_model_ids(self._get_entry().runtime_data)
),
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.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,
}
)
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)
),
# 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,
)
Reference in new issue View Git Blame Copy Permalink