Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-02 19:42:30 +01:00
Code Activity

Add STT support for OpenAI (#162931)

Browse Source 
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
Denis Shulyaka
2026-02-24 19:32:13 +01:00
committed by GitHub
co-authored by Copilot Joostlek
parent ff916a783b
commit 30fffafceb
10 changed files with 860 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/openai_conversation/__init__.py
+20 -1
View File
@@ -50,6 +50,7 @@ from .const import (
CONF_TOP_P,
DEFAULT_AI_TASK_NAME,
DEFAULT_NAME,
DEFAULT_STT_NAME,
DEFAULT_TTS_NAME,
DOMAIN,
LOGGER,
@@ -57,6 +58,7 @@ from .const import (
RECOMMENDED_CHAT_MODEL,
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_REASONING_EFFORT,
RECOMMENDED_STT_OPTIONS,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
RECOMMENDED_TTS_OPTIONS,
@@ -66,7 +68,7 @@ from .entity import async_prepare_files_for_prompt
SERVICE_GENERATE_IMAGE = "generate_image"
SERVICE_GENERATE_CONTENT = "generate_content"
PLATFORMS = (Platform.AI_TASK, Platform.CONVERSATION, Platform.TTS)
PLATFORMS = (Platform.AI_TASK, Platform.CONVERSATION, Platform.STT, Platform.TTS)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
type OpenAIConfigEntry = ConfigEntry[openai.AsyncClient]
@@ -480,6 +482,10 @@ async def async_migrate_entry(hass: HomeAssistant, entry: OpenAIConfigEntry) ->
_add_tts_subentry(hass, entry)
hass.config_entries.async_update_entry(entry, minor_version=5)
if entry.version == 2 and entry.minor_version == 5:
_add_stt_subentry(hass, entry)
hass.config_entries.async_update_entry(entry, minor_version=6)
LOGGER.debug(
"Migration to version %s:%s successful", entry.version, entry.minor_version
)
@@ -500,6 +506,19 @@ def _add_ai_task_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None
)
def _add_stt_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None:
"""Add STT subentry to the config entry."""
hass.config_entries.async_add_subentry(
entry,
ConfigSubentry(
data=MappingProxyType(RECOMMENDED_STT_OPTIONS),
subentry_type="stt",
title=DEFAULT_STT_NAME,
unique_id=None,
),
)
def _add_tts_subentry(hass: HomeAssistant, entry: OpenAIConfigEntry) -> None:
"""Add TTS subentry to the config entry."""
hass.config_entries.async_add_subentry(
homeassistant/components/openai_conversation/config_flow.py
+101 -1
View File
@@ -68,6 +68,8 @@ from .const import (
CONF_WEB_SEARCH_USER_LOCATION,
DEFAULT_AI_TASK_NAME,
DEFAULT_CONVERSATION_NAME,
DEFAULT_STT_NAME,
DEFAULT_STT_PROMPT,
DEFAULT_TTS_NAME,
DOMAIN,
RECOMMENDED_AI_TASK_OPTIONS,
@@ -78,6 +80,8 @@ from .const import (
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_REASONING_EFFORT,
RECOMMENDED_REASONING_SUMMARY,
RECOMMENDED_STT_MODEL,
RECOMMENDED_STT_OPTIONS,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
RECOMMENDED_TTS_OPTIONS,
@@ -117,7 +121,7 @@ class OpenAIConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for OpenAI Conversation."""
VERSION = 2
MINOR_VERSION = 5
MINOR_VERSION = 6
async def async_step_user(
self, user_input: dict[str, Any] | None = None
@@ -158,6 +162,12 @@ class OpenAIConfigFlow(ConfigFlow, domain=DOMAIN):
"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,
@@ -204,6 +214,7 @@ class OpenAIConfigFlow(ConfigFlow, domain=DOMAIN):
return {
"conversation": OpenAISubentryFlowHandler,
"ai_task_data": OpenAISubentryFlowHandler,
"stt": OpenAISubentrySTTFlowHandler,
"tts": OpenAISubentryTTSFlowHandler,
}
@@ -595,6 +606,95 @@ class OpenAISubentryFlowHandler(ConfigSubentryFlow):
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 != 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."""
homeassistant/components/openai_conversation/const.py
+7
View File
@@ -1,6 +1,7 @@
"""Constants for the OpenAI Conversation integration."""
import logging
from typing import Any
from homeassistant.const import CONF_LLM_HASS_API
from homeassistant.helpers import llm
@@ -10,6 +11,7 @@ LOGGER: logging.Logger = logging.getLogger(__package__)
DEFAULT_CONVERSATION_NAME = "OpenAI Conversation"
DEFAULT_AI_TASK_NAME = "OpenAI AI Task"
DEFAULT_STT_NAME = "OpenAI STT"
DEFAULT_TTS_NAME = "OpenAI TTS"
DEFAULT_NAME = "OpenAI Conversation"
@@ -40,6 +42,7 @@ RECOMMENDED_IMAGE_MODEL = "gpt-image-1.5"
RECOMMENDED_MAX_TOKENS = 3000
RECOMMENDED_REASONING_EFFORT = "low"
RECOMMENDED_REASONING_SUMMARY = "auto"
RECOMMENDED_STT_MODEL = "gpt-4o-mini-transcribe"
RECOMMENDED_TEMPERATURE = 1.0
RECOMMENDED_TOP_P = 1.0
RECOMMENDED_TTS_SPEED = 1.0
@@ -48,6 +51,9 @@ RECOMMENDED_WEB_SEARCH = False
RECOMMENDED_WEB_SEARCH_CONTEXT_SIZE = "medium"
RECOMMENDED_WEB_SEARCH_USER_LOCATION = False
RECOMMENDED_WEB_SEARCH_INLINE_CITATIONS = False
DEFAULT_STT_PROMPT = (
"The following conversation is a smart home user talking to Home Assistant."
)
UNSUPPORTED_MODELS: list[str] = [
"o1-mini",
@@ -108,6 +114,7 @@ RECOMMENDED_CONVERSATION_OPTIONS = {
RECOMMENDED_AI_TASK_OPTIONS = {
CONF_RECOMMENDED: True,
}
RECOMMENDED_STT_OPTIONS: dict[str, Any] = {}
RECOMMENDED_TTS_OPTIONS = {
CONF_PROMPT: "",
CONF_CHAT_MODEL: "gpt-4o-mini-tts",
homeassistant/components/openai_conversation/entity.py
+7 -1
View File
@@ -92,6 +92,7 @@ from .const import (
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_REASONING_EFFORT,
RECOMMENDED_REASONING_SUMMARY,
RECOMMENDED_STT_MODEL,
RECOMMENDED_TEMPERATURE,
RECOMMENDED_TOP_P,
RECOMMENDED_VERBOSITY,
@@ -471,7 +472,12 @@ class OpenAIBaseLLMEntity(Entity):
identifiers={(DOMAIN, subentry.subentry_id)},
name=subentry.title,
manufacturer="OpenAI",
model=subentry.data.get(CONF_CHAT_MODEL, RECOMMENDED_CHAT_MODEL),
model=subentry.data.get(
CONF_CHAT_MODEL,
RECOMMENDED_CHAT_MODEL
if subentry.subentry_type != "stt"
else RECOMMENDED_STT_MODEL,
),
entry_type=dr.DeviceEntryType.SERVICE,
)
homeassistant/components/openai_conversation/strings.json
+24
View File
@@ -146,6 +146,30 @@
}
}
},
"stt": {
"abort": {
"entry_not_loaded": "[%key:component::openai_conversation::config_subentries::conversation::abort::entry_not_loaded%]",
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"entry_type": "Speech-to-text",
"initiate_flow": {
"reconfigure": "Reconfigure speech-to-text service",
"user": "Add speech-to-text service"
},
"step": {
"init": {
"data": {
"chat_model": "Model",
"name": "[%key:common::config_flow::data::name%]",
"prompt": "[%key:common::config_flow::data::prompt%]"
},
"data_description": {
"chat_model": "The model to use to transcribe speech.",
"prompt": "Use this prompt to improve the quality of the transcripts. Translate to the pipeline language for best results. See the documentation for more details."
}
}
}
},
"tts": {
"abort": {
"entry_not_loaded": "[%key:component::openai_conversation::config_subentries::conversation::abort::entry_not_loaded%]",
homeassistant/components/openai_conversation/stt.py
+196
View File
@@ -0,0 +1,196 @@
"""Speech to text support for OpenAI."""
from __future__ import annotations
from collections.abc import AsyncIterable
import io
import logging
from typing import TYPE_CHECKING
import wave
from openai import OpenAIError
from homeassistant.components import stt
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
CONF_CHAT_MODEL,
CONF_PROMPT,
DEFAULT_STT_PROMPT,
RECOMMENDED_STT_MODEL,
)
from .entity import OpenAIBaseLLMEntity
if TYPE_CHECKING:
from . import OpenAIConfigEntry
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: OpenAIConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up STT entities."""
for subentry in config_entry.subentries.values():
if subentry.subentry_type != "stt":
continue
async_add_entities(
[OpenAISTTEntity(config_entry, subentry)],
config_subentry_id=subentry.subentry_id,
)
class OpenAISTTEntity(stt.SpeechToTextEntity, OpenAIBaseLLMEntity):
"""OpenAI Speech to text entity."""
@property
def supported_languages(self) -> list[str]:
"""Return a list of supported languages."""
# https://developers.openai.com/api/docs/guides/speech-to-text#supported-languages
# The model may also transcribe the audio in other languages but with lower quality
return [
"af-ZA", # Afrikaans
"ar-SA", # Arabic
"hy-AM", # Armenian
"az-AZ", # Azerbaijani
"be-BY", # Belarusian
"bs-BA", # Bosnian
"bg-BG", # Bulgarian
"ca-ES", # Catalan
"zh-CN", # Chinese (Mandarin)
"hr-HR", # Croatian
"cs-CZ", # Czech
"da-DK", # Danish
"nl-NL", # Dutch
"en-US", # English
"et-EE", # Estonian
"fi-FI", # Finnish
"fr-FR", # French
"gl-ES", # Galician
"de-DE", # German
"el-GR", # Greek
"he-IL", # Hebrew
"hi-IN", # Hindi
"hu-HU", # Hungarian
"is-IS", # Icelandic
"id-ID", # Indonesian
"it-IT", # Italian
"ja-JP", # Japanese
"kn-IN", # Kannada
"kk-KZ", # Kazakh
"ko-KR", # Korean
"lv-LV", # Latvian
"lt-LT", # Lithuanian
"mk-MK", # Macedonian
"ms-MY", # Malay
"mr-IN", # Marathi
"mi-NZ", # Maori
"ne-NP", # Nepali
"no-NO", # Norwegian
"fa-IR", # Persian
"pl-PL", # Polish
"pt-PT", # Portuguese
"ro-RO", # Romanian
"ru-RU", # Russian
"sr-RS", # Serbian
"sk-SK", # Slovak
"sl-SI", # Slovenian
"es-ES", # Spanish
"sw-KE", # Swahili
"sv-SE", # Swedish
"fil-PH", # Tagalog (Filipino)
"ta-IN", # Tamil
"th-TH", # Thai
"tr-TR", # Turkish
"uk-UA", # Ukrainian
"ur-PK", # Urdu
"vi-VN", # Vietnamese
"cy-GB", # Welsh
]
@property
def supported_formats(self) -> list[stt.AudioFormats]:
"""Return a list of supported formats."""
# https://developers.openai.com/api/docs/guides/speech-to-text#transcriptions
return [stt.AudioFormats.WAV, stt.AudioFormats.OGG]
@property
def supported_codecs(self) -> list[stt.AudioCodecs]:
"""Return a list of supported codecs."""
return [stt.AudioCodecs.PCM, stt.AudioCodecs.OPUS]
@property
def supported_bit_rates(self) -> list[stt.AudioBitRates]:
"""Return a list of supported bit rates."""
return [
stt.AudioBitRates.BITRATE_8,
stt.AudioBitRates.BITRATE_16,
stt.AudioBitRates.BITRATE_24,
stt.AudioBitRates.BITRATE_32,
]
@property
def supported_sample_rates(self) -> list[stt.AudioSampleRates]:
"""Return a list of supported sample rates."""
return [
stt.AudioSampleRates.SAMPLERATE_8000,
stt.AudioSampleRates.SAMPLERATE_11000,
stt.AudioSampleRates.SAMPLERATE_16000,
stt.AudioSampleRates.SAMPLERATE_18900,
stt.AudioSampleRates.SAMPLERATE_22000,
stt.AudioSampleRates.SAMPLERATE_32000,
stt.AudioSampleRates.SAMPLERATE_37800,
stt.AudioSampleRates.SAMPLERATE_44100,
stt.AudioSampleRates.SAMPLERATE_48000,
]
@property
def supported_channels(self) -> list[stt.AudioChannels]:
"""Return a list of supported channels."""
return [stt.AudioChannels.CHANNEL_MONO, stt.AudioChannels.CHANNEL_STEREO]
async def async_process_audio_stream(
self, metadata: stt.SpeechMetadata, stream: AsyncIterable[bytes]
) -> stt.SpeechResult:
"""Process an audio stream to STT service."""
audio_bytes = bytearray()
async for chunk in stream:
audio_bytes.extend(chunk)
audio_data = bytes(audio_bytes)
if metadata.format == stt.AudioFormats.WAV:
# Add missing wav header
wav_buffer = io.BytesIO()
with wave.open(wav_buffer, "wb") as wf:
wf.setnchannels(metadata.channel.value)
wf.setsampwidth(metadata.bit_rate.value // 8)
wf.setframerate(metadata.sample_rate.value)
wf.writeframes(audio_data)
audio_data = wav_buffer.getvalue()
options = self.subentry.data
client = self.entry.runtime_data
try:
response = await client.audio.transcriptions.create(
model=options.get(CONF_CHAT_MODEL, RECOMMENDED_STT_MODEL),
file=(f"a.{metadata.format.value}", audio_data),
response_format="json",
language=metadata.language.split("-")[0],
prompt=options.get(CONF_PROMPT, DEFAULT_STT_PROMPT),
)
except OpenAIError:
_LOGGER.exception("Error during STT")
else:
if response.text:
return stt.SpeechResult(
response.text,
stt.SpeechResultState.SUCCESS,
)
return stt.SpeechResult(None, stt.SpeechResultState.ERROR)
tests/components/openai_conversation/conftest.py
+26 -1
View File
@@ -5,6 +5,7 @@ from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
from openai.types import ResponseFormatText
from openai.types.audio import Transcription
from openai.types.responses import (
Response,
ResponseCompletedEvent,
@@ -24,8 +25,10 @@ from homeassistant.components.openai_conversation.const import (
CONF_CHAT_MODEL,
DEFAULT_AI_TASK_NAME,
DEFAULT_CONVERSATION_NAME,
DEFAULT_STT_NAME,
DEFAULT_TTS_NAME,
RECOMMENDED_AI_TASK_OPTIONS,
RECOMMENDED_STT_OPTIONS,
RECOMMENDED_TTS_OPTIONS,
)
from homeassistant.config_entries import ConfigSubentryData
@@ -55,7 +58,7 @@ def mock_config_entry(
"api_key": "bla",
},
version=2,
minor_version=5,
minor_version=6,
subentries_data=[
ConfigSubentryData(
data=mock_conversation_subentry_data,
@@ -69,6 +72,12 @@ def mock_config_entry(
title=DEFAULT_AI_TASK_NAME,
unique_id=None,
),
ConfigSubentryData(
data=RECOMMENDED_STT_OPTIONS,
subentry_type="stt",
title=DEFAULT_STT_NAME,
unique_id=None,
),
ConfigSubentryData(
data=RECOMMENDED_TTS_OPTIONS,
subentry_type="tts",
@@ -219,6 +228,22 @@ def mock_create_stream() -> Generator[AsyncMock]:
yield mock_create
@pytest.fixture
def mock_create_transcription() -> Generator[AsyncMock]:
"""Mock transcription response."""
with patch(
"openai.resources.audio.transcriptions.AsyncTranscriptions.create",
AsyncMock(return_value=""),
) as mock_create:
mock_create.side_effect = lambda *args, **kwargs: (
Transcription(text=mock_create.return_value)
if isinstance(mock_create.return_value, str)
else mock_create.return_value
)
yield mock_create
@pytest.fixture
def mock_create_speech() -> Generator[MagicMock]:
"""Mock stream response."""
tests/components/openai_conversation/test_config_flow.py
+103 -9
View File
@@ -34,12 +34,14 @@ from homeassistant.components.openai_conversation.const import (
CONF_WEB_SEARCH_USER_LOCATION,
DEFAULT_AI_TASK_NAME,
DEFAULT_CONVERSATION_NAME,
DEFAULT_STT_NAME,
DEFAULT_TTS_NAME,
DOMAIN,
RECOMMENDED_AI_TASK_OPTIONS,
RECOMMENDED_CHAT_MODEL,
RECOMMENDED_MAX_TOKENS,
RECOMMENDED_REASONING_SUMMARY,
RECOMMENDED_STT_OPTIONS,
RECOMMENDED_TOP_P,
RECOMMENDED_TTS_OPTIONS,
)
@@ -100,6 +102,12 @@ async def test_form(hass: HomeAssistant) -> None:
"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,
@@ -107,6 +115,8 @@ async def test_form(hass: HomeAssistant) -> None:
"unique_id": None,
},
]
assert result2["version"] == 2
assert result2["minor_version"] == 6
assert len(mock_setup_entry.mock_calls) == 1
@@ -954,8 +964,8 @@ async def test_creating_ai_task_subentry(
) -> None:
"""Test creating an AI task subentry."""
old_subentries = set(mock_config_entry.subentries)
# Original conversation + original ai_task + original tts
assert len(mock_config_entry.subentries) == 3
# Original conversation + ai_task + stt + tts
assert len(mock_config_entry.subentries) == 4
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "ai_task_data"),
@@ -982,8 +992,8 @@ async def test_creating_ai_task_subentry(
}
assert (
len(mock_config_entry.subentries) == 4
) # Original conversation + original tts + original ai_task + new ai_task
len(mock_config_entry.subentries) == 5
) # Original conversation + stt + tts + ai_task + new ai_task
new_subentry_id = list(set(mock_config_entry.subentries) - old_subentries)[0]
new_subentry = mock_config_entry.subentries[new_subentry_id]
@@ -1067,6 +1077,90 @@ async def test_creating_ai_task_subentry_advanced(
}
async def test_creating_stt_subentry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_init_component,
) -> None:
"""Test creating a STT subentry."""
old_subentries = set(mock_config_entry.subentries)
# Original conversation + ai_task + stt + tts
assert len(mock_config_entry.subentries) == 4
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "stt"),
context={"source": config_entries.SOURCE_USER},
)
assert result.get("type") is FlowResultType.FORM
assert result.get("step_id") == "init"
assert not result.get("errors")
result = await hass.config_entries.subentries.async_configure(
result["flow_id"],
{
"name": "Custom STT",
CONF_PROMPT: "Umm, let me think like, hmm… Okay, here’s what I’m, like, thinking.",
CONF_CHAT_MODEL: "gpt-4o-transcribe",
},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "Custom STT"
assert result.get("data") == {
CONF_PROMPT: "Umm, let me think like, hmm… Okay, here’s what I’m, like, thinking.",
CONF_CHAT_MODEL: "gpt-4o-transcribe",
}
assert (
len(mock_config_entry.subentries) == 5
) # Original conversation + ai_task + tts + original stt + new stt
new_subentry_id = list(set(mock_config_entry.subentries) - old_subentries)[0]
new_subentry = mock_config_entry.subentries[new_subentry_id]
assert new_subentry.subentry_type == "stt"
assert new_subentry.title == "Custom STT"
async def test_stt_subentry_not_loaded(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test creating a STT subentry when entry is not loaded."""
# Don't call mock_init_component to simulate not loaded state
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "stt"),
context={"source": config_entries.SOURCE_USER},
)
assert result.get("type") is FlowResultType.ABORT
assert result.get("reason") == "entry_not_loaded"
async def test_stt_reconfigure(
hass: HomeAssistant, mock_config_entry, mock_init_component
) -> None:
"""Test reconfiguring the STT subentry updates prompt and chat model."""
subentry = [
s for s in mock_config_entry.subentries.values() if s.subentry_type == "stt"
][0]
subentry_flow = await mock_config_entry.start_subentry_reconfigure_flow(
hass, subentry.subentry_id
)
options = await hass.config_entries.subentries.async_configure(
subentry_flow["flow_id"],
{
"prompt": "This is a conversation about smart pirate ships.",
"chat_model": "gpt-4o-mini-transcribe-2025-12-15",
},
)
await hass.async_block_till_done()
assert options["type"] is FlowResultType.ABORT
assert options["reason"] == "reconfigure_successful"
assert subentry.data["prompt"] == "This is a conversation about smart pirate ships."
assert subentry.data["chat_model"] == "gpt-4o-mini-transcribe-2025-12-15"
async def test_creating_tts_subentry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
@@ -1074,8 +1168,8 @@ async def test_creating_tts_subentry(
) -> None:
"""Test creating a TTS subentry."""
old_subentries = set(mock_config_entry.subentries)
# Original conversation + original ai_task + original tts
assert len(mock_config_entry.subentries) == 3
# Original conversation + ai_task + stt + tts
assert len(mock_config_entry.subentries) == 4
result = await hass.config_entries.subentries.async_init(
(mock_config_entry.entry_id, "tts"),
@@ -1104,8 +1198,8 @@ async def test_creating_tts_subentry(
}
assert (
len(mock_config_entry.subentries) == 4
) # Original conversation + original ai_task + original tts + new tts
len(mock_config_entry.subentries) == 5
) # Original conversation + ai_task + stt + tts + new tts
new_subentry_id = list(set(mock_config_entry.subentries) - old_subentries)[0]
new_subentry = mock_config_entry.subentries[new_subentry_id]
@@ -1131,7 +1225,7 @@ async def test_tts_subentry_not_loaded(
async def test_tts_reconfigure(
hass: HomeAssistant, mock_config_entry, mock_init_component
) -> None:
"""Test the tts subentry reconfigure flow with."""
"""Test the tts subentry reconfigure flow."""
subentry = [
s for s in mock_config_entry.subentries.values() if s.subentry_type == "tts"
][0]
tests/components/openai_conversation/test_init.py
+163 -21
View File
@@ -21,10 +21,12 @@ from homeassistant.components.openai_conversation import CONF_CHAT_MODEL
from homeassistant.components.openai_conversation.const import (
DEFAULT_AI_TASK_NAME,
DEFAULT_CONVERSATION_NAME,
DEFAULT_STT_NAME,
DEFAULT_TTS_NAME,
DOMAIN,
RECOMMENDED_AI_TASK_OPTIONS,
RECOMMENDED_CONVERSATION_OPTIONS,
RECOMMENDED_STT_OPTIONS,
RECOMMENDED_TTS_OPTIONS,
)
from homeassistant.config_entries import (
@@ -663,21 +665,24 @@ async def test_migration_from_v1(
await hass.async_block_till_done()
assert mock_config_entry.version == 2
assert mock_config_entry.minor_version == 5
assert mock_config_entry.minor_version == 6
assert mock_config_entry.data == {"api_key": "1234"}
assert mock_config_entry.options == {}
assert len(mock_config_entry.subentries) == 3
assert len(mock_config_entry.subentries) == 4
# Find the conversation subentry
# Find the subentries
conversation_subentry = None
ai_task_subentry = None
stt_subentry = None
tts_subentry = None
for subentry in mock_config_entry.subentries.values():
if subentry.subentry_type == "conversation":
conversation_subentry = subentry
elif subentry.subentry_type == "ai_task_data":
ai_task_subentry = subentry
elif subentry.subentry_type == "stt":
stt_subentry = subentry
elif subentry.subentry_type == "tts":
tts_subentry = subentry
assert conversation_subentry is not None
@@ -691,6 +696,11 @@ async def test_migration_from_v1(
assert ai_task_subentry.title == DEFAULT_AI_TASK_NAME
assert ai_task_subentry.subentry_type == "ai_task_data"
assert stt_subentry is not None
assert stt_subentry.unique_id is None
assert stt_subentry.title == DEFAULT_STT_NAME
assert stt_subentry.subentry_type == "stt"
assert tts_subentry is not None
assert tts_subentry.unique_id is None
assert tts_subentry.title == DEFAULT_TTS_NAME
@@ -800,9 +810,9 @@ async def test_migration_from_v1_with_multiple_keys(
for idx, entry in enumerate(entries):
assert entry.version == 2
assert entry.minor_version == 5
assert entry.minor_version == 6
assert not entry.options
assert len(entry.subentries) == 3
assert len(entry.subentries) == 4
conversation_subentry = None
for subentry in entry.subentries.values():
@@ -905,11 +915,11 @@ async def test_migration_from_v1_with_same_keys(
entry = entries[0]
assert entry.version == 2
assert entry.minor_version == 5
assert entry.minor_version == 6
assert not entry.options
assert (
len(entry.subentries) == 4
) # Two conversation subentries + one AI task subentry + one TTS subentry
len(entry.subentries) == 5
) # Two conversation subentries + one AI task subentry + one STT subentry + one TTS subentry
# Check both conversation subentries exist with correct data
conversation_subentries = [
@@ -918,12 +928,16 @@ async def test_migration_from_v1_with_same_keys(
ai_task_subentries = [
sub for sub in entry.subentries.values() if sub.subentry_type == "ai_task_data"
]
stt_subentries = [
sub for sub in entry.subentries.values() if sub.subentry_type == "stt"
]
tts_subentries = [
sub for sub in entry.subentries.values() if sub.subentry_type == "tts"
]
assert len(conversation_subentries) == 2
assert len(ai_task_subentries) == 1
assert len(stt_subentries) == 1
assert len(tts_subentries) == 1
titles = [sub.title for sub in conversation_subentries]
@@ -1113,11 +1127,11 @@ async def test_migration_from_v1_disabled(
assert entry.disabled_by is merged_config_entry_disabled_by
assert entry.version == 2
assert entry.minor_version == (
4 if merged_config_entry_disabled_by is not None else 5
4 if merged_config_entry_disabled_by is not None else 6
)
assert not entry.options
assert entry.title == "OpenAI Conversation"
assert len(entry.subentries) == (3 if entry.minor_version == 4 else 4)
assert len(entry.subentries) == (3 if entry.minor_version == 4 else 5)
conversation_subentries = [
subentry
for subentry in entry.subentries.values()
@@ -1136,14 +1150,23 @@ async def test_migration_from_v1_disabled(
assert len(ai_task_subentries) == 1
assert ai_task_subentries[0].data == RECOMMENDED_AI_TASK_OPTIONS
assert ai_task_subentries[0].title == DEFAULT_AI_TASK_NAME
stt_subentries = [
subentry
for subentry in entry.subentries.values()
if subentry.subentry_type == "stt"
]
tts_subentries = [
subentry
for subentry in entry.subentries.values()
if subentry.subentry_type == "tts"
]
if entry.minor_version == 4:
assert len(stt_subentries) == 0
assert len(tts_subentries) == 0
else:
assert len(stt_subentries) == 1
assert stt_subentries[0].data == RECOMMENDED_STT_OPTIONS
assert stt_subentries[0].title == DEFAULT_STT_NAME
assert len(tts_subentries) == 1
assert tts_subentries[0].data == RECOMMENDED_TTS_OPTIONS
assert tts_subentries[0].title == DEFAULT_TTS_NAME
@@ -1277,10 +1300,10 @@ async def test_migration_from_v2_1(
assert len(entries) == 1
entry = entries[0]
assert entry.version == 2
assert entry.minor_version == 5
assert entry.minor_version == 6
assert not entry.options
assert entry.title == "ChatGPT"
assert len(entry.subentries) == 4 # 2 conversation + 1 AI task + 1 TTS
assert len(entry.subentries) == 5 # 2 conversation + 1 AI task + 1 STT + 1 TTS
conversation_subentries = [
subentry
for subentry in entry.subentries.values()
@@ -1291,6 +1314,11 @@ async def test_migration_from_v2_1(
for subentry in entry.subentries.values()
if subentry.subentry_type == "ai_task_data"
]
stt_subentries = [
subentry
for subentry in entry.subentries.values()
if subentry.subentry_type == "stt"
]
tts_subentries = [
subentry
for subentry in entry.subentries.values()
@@ -1298,6 +1326,7 @@ async def test_migration_from_v2_1(
]
assert len(conversation_subentries) == 2
assert len(ai_task_subentries) == 1
assert len(stt_subentries) == 1
assert len(tts_subentries) == 1
for subentry in conversation_subentries:
assert subentry.subentry_type == "conversation"
@@ -1362,7 +1391,7 @@ async def test_devices(
devices = dr.async_entries_for_config_entry(
device_registry, mock_config_entry.entry_id
)
assert len(devices) == 3 # One for conversation, one for AI task, one for TTS
assert len(devices) == 4 # One for conversation, AI task, STT, and TTS
# Use the first device for snapshot comparison
device = devices[0]
@@ -1419,10 +1448,10 @@ async def test_migration_from_v2_2(
assert len(entries) == 1
entry = entries[0]
assert entry.version == 2
assert entry.minor_version == 5
assert entry.minor_version == 6
assert not entry.options
assert entry.title == "ChatGPT"
assert len(entry.subentries) == 3
assert len(entry.subentries) == 4
# Check conversation subentry is still there
conversation_subentries = [
@@ -1464,7 +1493,7 @@ async def test_migration_from_v2_2(
DeviceEntryDisabler.CONFIG_ENTRY,
RegistryEntryDisabler.CONFIG_ENTRY,
True,
5,
6,
None,
DeviceEntryDisabler.USER,
RegistryEntryDisabler.DEVICE,
@@ -1474,7 +1503,7 @@ async def test_migration_from_v2_2(
DeviceEntryDisabler.USER,
RegistryEntryDisabler.DEVICE,
True,
5,
6,
None,
DeviceEntryDisabler.USER,
RegistryEntryDisabler.DEVICE,
@@ -1484,7 +1513,7 @@ async def test_migration_from_v2_2(
DeviceEntryDisabler.USER,
RegistryEntryDisabler.USER,
True,
5,
6,
None,
DeviceEntryDisabler.USER,
RegistryEntryDisabler.USER,
@@ -1494,7 +1523,7 @@ async def test_migration_from_v2_2(
None,
None,
True,
5,
6,
None,
None,
None,
@@ -1686,10 +1715,10 @@ async def test_migration_from_v2_4(
assert len(entries) == 1
entry = entries[0]
assert entry.version == 2
assert entry.minor_version == 5
assert entry.minor_version == 6
assert not entry.options
assert entry.title == "ChatGPT"
assert len(entry.subentries) == 3
assert len(entry.subentries) == 4
# Check conversation subentry is still there
conversation_subentries = [
@@ -1721,3 +1750,116 @@ async def test_migration_from_v2_4(
tts_subentry = tts_subentries[0]
assert tts_subentry.data == {"chat_model": "gpt-4o-mini-tts", "prompt": ""}
assert tts_subentry.title == "OpenAI TTS"
async def test_migration_from_v2_5(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
) -> None:
"""Test migration from version 2.5."""
# Create a v2.5 config entry with a conversation, AI Task, and TTS subentries
conversation_options = {
"recommended": True,
"llm_hass_api": ["assist"],
"prompt": "You are a helpful assistant",
"chat_model": "gpt-4o-mini",
}
ai_task_options = {
"recommended": True,
"chat_model": "gpt-5-mini",
}
tts_options = {
"prompt": "Be friendly",
"chat_model": "gpt-4o-mini-tts",
}
mock_config_entry = MockConfigEntry(
domain=DOMAIN,
data={"api_key": "1234"},
entry_id="mock_entry_id",
version=2,
minor_version=5,
subentries_data=[
ConfigSubentryData(
data=conversation_options,
subentry_id="mock_id_1",
subentry_type="conversation",
title="ChatGPT",
unique_id=None,
),
ConfigSubentryData(
data=ai_task_options,
subentry_id="mock_id_2",
subentry_type="ai_task_data",
title="OpenAI AI Task",
unique_id=None,
),
ConfigSubentryData(
data=tts_options,
subentry_id="mock_id_3",
subentry_type="tts",
title="OpenAI TTS",
unique_id=None,
),
],
title="ChatGPT",
)
mock_config_entry.add_to_hass(hass)
# Run migration
with patch(
"homeassistant.components.openai_conversation.async_setup_entry",
return_value=True,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entries = hass.config_entries.async_entries(DOMAIN)
assert len(entries) == 1
entry = entries[0]
assert entry.version == 2
assert entry.minor_version == 6
assert not entry.options
assert entry.title == "ChatGPT"
assert len(entry.subentries) == 4
# Check conversation subentry is still there
conversation_subentries = [
subentry
for subentry in entry.subentries.values()
if subentry.subentry_type == "conversation"
]
assert len(conversation_subentries) == 1
conversation_subentry = conversation_subentries[0]
assert conversation_subentry.data == conversation_options
# Check AI Task subentry is still there
ai_task_subentries = [
subentry
for subentry in entry.subentries.values()
if subentry.subentry_type == "ai_task_data"
]
assert len(ai_task_subentries) == 1
ai_task_subentry = ai_task_subentries[0]
assert ai_task_subentry.data == ai_task_options
# Check TTS subentry is still there
tts_subentries = [
subentry
for subentry in entry.subentries.values()
if subentry.subentry_type == "tts"
]
assert len(tts_subentries) == 1
tts_subentry = tts_subentries[0]
assert tts_subentry.data == tts_options
# Check STT subentry was added
stt_subentries = [
subentry
for subentry in entry.subentries.values()
if subentry.subentry_type == "stt"
]
assert len(stt_subentries) == 1
stt_subentry = stt_subentries[0]
assert stt_subentry.data == {}
assert stt_subentry.title == "OpenAI STT"
tests/components/openai_conversation/test_stt.py
+213
View File
@@ -0,0 +1,213 @@
"""Test STT platform of OpenAI Conversation integration."""
from collections.abc import AsyncIterable
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
from openai import RateLimitError
import pytest
from homeassistant.components import stt
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def _async_get_audio_stream(data: bytes) -> AsyncIterable[bytes]:
"""Yield the audio data."""
yield data
@pytest.mark.usefixtures("mock_init_component")
async def test_stt_entity_properties(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test STT entity properties."""
entity: stt.SpeechToTextEntity = hass.data[stt.DOMAIN].get_entity("stt.openai_stt")
assert entity is not None
assert isinstance(entity.supported_languages, list)
assert len(entity.supported_languages)
assert stt.AudioFormats.WAV in entity.supported_formats
assert stt.AudioFormats.OGG in entity.supported_formats
assert stt.AudioCodecs.PCM in entity.supported_codecs
assert stt.AudioCodecs.OPUS in entity.supported_codecs
assert stt.AudioBitRates.BITRATE_8 in entity.supported_bit_rates
assert stt.AudioBitRates.BITRATE_16 in entity.supported_bit_rates
assert stt.AudioBitRates.BITRATE_24 in entity.supported_bit_rates
assert stt.AudioBitRates.BITRATE_32 in entity.supported_bit_rates
assert stt.AudioSampleRates.SAMPLERATE_8000 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_11000 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_16000 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_18900 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_22000 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_32000 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_37800 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_44100 in entity.supported_sample_rates
assert stt.AudioSampleRates.SAMPLERATE_48000 in entity.supported_sample_rates
assert stt.AudioChannels.CHANNEL_MONO in entity.supported_channels
assert stt.AudioChannels.CHANNEL_STEREO in entity.supported_channels
@pytest.mark.usefixtures("mock_init_component")
async def test_stt_process_audio_stream_success_wav(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_create_transcription: AsyncMock,
) -> None:
"""Test STT processing audio stream successfully."""
entity = hass.data[stt.DOMAIN].get_entity("stt.openai_stt")
mock_create_transcription.return_value = "This is a test transcription."
metadata = stt.SpeechMetadata(
language="en-US",
format=stt.AudioFormats.WAV,
codec=stt.AudioCodecs.PCM,
bit_rate=stt.AudioBitRates.BITRATE_16,
sample_rate=stt.AudioSampleRates.SAMPLERATE_16000,
channel=stt.AudioChannels.CHANNEL_MONO,
)
audio_stream = _async_get_audio_stream(b"test_audio_bytes")
wav_buffer = None
mock_wf = MagicMock()
mock_wf.writeframes.side_effect = lambda data: wav_buffer.write(
b"converted_wav_bytes"
)
def mock_open(buffer, mode):
nonlocal wav_buffer
wav_buffer = buffer
mock_cm = MagicMock()
mock_cm.__enter__.return_value = mock_wf
return mock_cm
with patch(
"homeassistant.components.openai_conversation.stt.wave.open",
side_effect=mock_open,
) as mock_wave_open:
result = await entity.async_process_audio_stream(metadata, audio_stream)
assert result.result == stt.SpeechResultState.SUCCESS
assert result.text == "This is a test transcription."
mock_wave_open.assert_called_once()
mock_wf.setnchannels.assert_called_once_with(1)
mock_wf.setsampwidth.assert_called_once_with(2)
mock_wf.setframerate.assert_called_once_with(16000)
mock_create_transcription.assert_called_once()
call_args = mock_create_transcription.call_args
assert call_args.kwargs["model"] == "gpt-4o-mini-transcribe"
contents = call_args.kwargs["file"]
assert contents[0].endswith(".wav")
assert contents[1] == b"converted_wav_bytes"
@pytest.mark.usefixtures("mock_init_component")
async def test_stt_process_audio_stream_success_ogg(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_create_transcription: AsyncMock,
) -> None:
"""Test STT processing audio stream successfully."""
entity = hass.data[stt.DOMAIN].get_entity("stt.openai_stt")
mock_create_transcription.return_value = "This is a test transcription."
metadata = stt.SpeechMetadata(
language="en-US",
format=stt.AudioFormats.OGG,
codec=stt.AudioCodecs.PCM,
bit_rate=stt.AudioBitRates.BITRATE_16,
sample_rate=stt.AudioSampleRates.SAMPLERATE_16000,
channel=stt.AudioChannels.CHANNEL_MONO,
)
audio_stream = _async_get_audio_stream(b"test_audio_bytes")
wav_buffer = None
mock_wf = MagicMock()
mock_wf.writeframes.side_effect = lambda data: wav_buffer.write(
b"converted_wav_bytes"
)
def mock_open(buffer, mode):
nonlocal wav_buffer
wav_buffer = buffer
mock_cm = MagicMock()
mock_cm.__enter__.return_value = mock_wf
return mock_cm
with patch(
"homeassistant.components.openai_conversation.stt.wave.open",
side_effect=mock_open,
) as mock_wave_open:
result = await entity.async_process_audio_stream(metadata, audio_stream)
assert result.result == stt.SpeechResultState.SUCCESS
assert result.text == "This is a test transcription."
mock_wave_open.assert_not_called()
mock_create_transcription.assert_called_once()
call_args = mock_create_transcription.call_args
assert call_args.kwargs["model"] == "gpt-4o-mini-transcribe"
contents = call_args.kwargs["file"]
assert contents[0].endswith(".ogg")
assert contents[1] == b"test_audio_bytes"
@pytest.mark.usefixtures("mock_init_component")
async def test_stt_process_audio_stream_api_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_create_transcription: AsyncMock,
) -> None:
"""Test STT processing audio stream with API errors."""
entity = hass.data[stt.DOMAIN].get_entity("stt.openai_stt")
mock_create_transcription.side_effect = RateLimitError(
response=httpx.Response(status_code=429, request=""),
body=None,
message=None,
)
metadata = stt.SpeechMetadata(
language="en-US",
format=stt.AudioFormats.OGG,
codec=stt.AudioCodecs.OPUS,
bit_rate=stt.AudioBitRates.BITRATE_16,
sample_rate=stt.AudioSampleRates.SAMPLERATE_16000,
channel=stt.AudioChannels.CHANNEL_MONO,
)
audio_stream = _async_get_audio_stream(b"test_audio_bytes")
result = await entity.async_process_audio_stream(metadata, audio_stream)
assert result.result == stt.SpeechResultState.ERROR
assert result.text is None
@pytest.mark.usefixtures("mock_init_component")
async def test_stt_process_audio_stream_empty_response(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_create_transcription: AsyncMock,
) -> None:
"""Test STT processing with an empty response from the API."""
entity = hass.data[stt.DOMAIN].get_entity("stt.openai_stt")
mock_create_transcription.return_value = ""
metadata = stt.SpeechMetadata(
language="en-US",
format=stt.AudioFormats.OGG,
codec=stt.AudioCodecs.OPUS,
bit_rate=stt.AudioBitRates.BITRATE_16,
sample_rate=stt.AudioSampleRates.SAMPLERATE_16000,
channel=stt.AudioChannels.CHANNEL_MONO,
)
audio_stream = _async_get_audio_stream(b"test_audio_bytes")
result = await entity.async_process_audio_stream(metadata, audio_stream)
assert result.result == stt.SpeechResultState.ERROR
assert result.text is None