3e3ece4e56
* Allow passing binary to the WS connection * Expand test coverage * Test non-existing handler * Add text to speech and stages to pipeline * Default to "cloud" TTS when engine is None * Refactor pipeline request to split text/audio * Refactor with PipelineRun * Generate pipeline from language * Clean up * Restore TTS code * Add audio pipeline test * Clean TTS cache in test * Clean up tests and pipeline base class * Stop pylint and pytest magics from fighting * Include mock_get_cache_files * Working on STT * Preparing to test * First successful test * Send handler_id * Allow signaling end of stream using empty payloads * Store handlers in a list * Handle binary handlers raising exceptions * Add stt/tts dependencies to voice_assistant * Include STT audio in pipeline test * Working on tests * Refactoring with stages * Fix tests * Add more tests * Add method docs * Change stt demo/cloud to AsyncIterable * Add pipeline error events * Move handler id to separate message before pipeline * Add test for invalid stage order * Change "finish" to "end" * Use enum --------- Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
"""Support for the demo for speech to text service."""
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterable
|
|
|
|
from homeassistant.components.stt import (
|
|
AudioBitRates,
|
|
AudioChannels,
|
|
AudioCodecs,
|
|
AudioFormats,
|
|
AudioSampleRates,
|
|
Provider,
|
|
SpeechMetadata,
|
|
SpeechResult,
|
|
SpeechResultState,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
|
|
|
SUPPORT_LANGUAGES = ["en", "de"]
|
|
|
|
|
|
async def async_get_engine(
|
|
hass: HomeAssistant,
|
|
config: ConfigType,
|
|
discovery_info: DiscoveryInfoType | None = None,
|
|
) -> Provider:
|
|
"""Set up Demo speech component."""
|
|
return DemoProvider()
|
|
|
|
|
|
class DemoProvider(Provider):
|
|
"""Demo speech API provider."""
|
|
|
|
@property
|
|
def supported_languages(self) -> list[str]:
|
|
"""Return a list of supported languages."""
|
|
return SUPPORT_LANGUAGES
|
|
|
|
@property
|
|
def supported_formats(self) -> list[AudioFormats]:
|
|
"""Return a list of supported formats."""
|
|
return [AudioFormats.WAV]
|
|
|
|
@property
|
|
def supported_codecs(self) -> list[AudioCodecs]:
|
|
"""Return a list of supported codecs."""
|
|
return [AudioCodecs.PCM]
|
|
|
|
@property
|
|
def supported_bit_rates(self) -> list[AudioBitRates]:
|
|
"""Return a list of supported bit rates."""
|
|
return [AudioBitRates.BITRATE_16]
|
|
|
|
@property
|
|
def supported_sample_rates(self) -> list[AudioSampleRates]:
|
|
"""Return a list of supported sample rates."""
|
|
return [AudioSampleRates.SAMPLERATE_16000, AudioSampleRates.SAMPLERATE_44100]
|
|
|
|
@property
|
|
def supported_channels(self) -> list[AudioChannels]:
|
|
"""Return a list of supported channels."""
|
|
return [AudioChannels.CHANNEL_STEREO]
|
|
|
|
async def async_process_audio_stream(
|
|
self, metadata: SpeechMetadata, stream: AsyncIterable[bytes]
|
|
) -> SpeechResult:
|
|
"""Process an audio stream to STT service."""
|
|
|
|
# Read available data
|
|
async for _ in stream:
|
|
pass
|
|
|
|
return SpeechResult("Turn the Kitchen Lights on", SpeechResultState.SUCCESS)
|