diff --git a/homeassistant/components/assist_pipeline/default_pipeline.py b/homeassistant/components/assist_pipeline/default_pipeline.py new file mode 100644 index 000000000000..8dccf8c93ced --- /dev/null +++ b/homeassistant/components/assist_pipeline/default_pipeline.py @@ -0,0 +1,1232 @@ +"""Default implementation of the Assist pipeline engine.""" + +import array +import asyncio +from collections import deque +from collections.abc import AsyncGenerator, AsyncIterable, Callable +from dataclasses import asdict, dataclass, field +import logging +from pathlib import Path +from typing import TYPE_CHECKING, Any, Protocol, cast + +import hass_nabucasa + +from homeassistant.components import conversation, media_player, stt, tts, wake_word +from homeassistant.const import MATCH_ALL, EntityStateAttribute +from homeassistant.core import Context, HomeAssistant, callback +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers import ( + chat_session, + device_registry as dr, + entity_registry as er, + intent, +) +from homeassistant.util.hass_dict import HassKey + +from .audio_enhancer import AudioEnhancer, EnhancedAudioChunk, MicroVadSpeexEnhancer +from .const import ( + ACKNOWLEDGE_PATH, + BYTES_PER_CHUNK, + MS_PER_CHUNK, + SAMPLE_CHANNELS, + SAMPLE_RATE, + SAMPLE_WIDTH, + SAMPLES_PER_CHUNK, +) +from .error import ( + IntentRecognitionError, + PipelineRunValidationError, + SpeechToTextError, + TextToSpeechError, + WakeWordDetectionAborted, + WakeWordDetectionError, + WakeWordTimeoutError, +) +from .models import ( + PIPELINE_STAGE_ORDER, + AudioSettings, + Pipeline, + PipelineEvent, + PipelineEventType, + PipelineStage, + WakeWordSettings, +) +from .vad import AudioBuffer, VoiceActivityTimeout, VoiceCommandSegmenter, chunk_samples + +if TYPE_CHECKING: + from hassil.recognize import RecognizeResult + + from .run import _PipelineProcessorRequest + +_LOGGER = logging.getLogger(__name__) + +KEY_PIPELINE_CONVERSATION_DATA: HassKey[dict[str, PipelineConversationData]] = HassKey( + "pipeline_conversation_data" +) +STREAM_RESPONSE_CHARS = 60 + + +@callback +def _async_local_fallback_intent_filter(result: RecognizeResult) -> bool: + """Filter out intents that are not local fallback.""" + return result.intent.name in ( + intent.INTENT_GET_STATE, + media_player.INTENT_MEDIA_SEARCH_AND_PLAY, + ) + + +class _PipelineController(Protocol): + """Home Assistant services exposed to a pipeline processor.""" + + @property + def hass(self) -> HomeAssistant: + """Return the Home Assistant instance.""" + + @property + def context(self) -> Context: + """Return the run context.""" + + @property + def pipeline(self) -> Pipeline: + """Return the pipeline configuration.""" + + @property + def start_stage(self) -> PipelineStage: + """Return the first stage to process.""" + + @property + def end_stage(self) -> PipelineStage: + """Return the last stage to process.""" + + @property + def language(self) -> str: + """Return the run language.""" + + @property + def tts_audio_output(self) -> str | dict[str, Any] | None: + """Return the requested TTS output options.""" + + @property + def wake_word_settings(self) -> WakeWordSettings | None: + """Return wake word settings.""" + + @property + def audio_settings(self) -> AudioSettings: + """Return audio processing settings.""" + + @property + def device_id(self) -> str | None: + """Return the device associated with the run.""" + + @property + def satellite_id(self) -> str | None: + """Return the satellite associated with the run.""" + + @callback + def process_event(self, event: PipelineEvent) -> None: + """Forward a pipeline event.""" + + @callback + def capture_audio(self, audio_bytes: bytes | None) -> None: + """Capture an audio chunk.""" + + @callback + def start_debug_recording(self, name: str) -> None: + """Start a new debug recording.""" + + @callback + def accept_wake_word(self, wake_word_phrase: str) -> None: + """Apply the duplicate wake-up policy.""" + + +@dataclass +class _DefaultPipelineProcessor: + """Process the stages in a default Assist pipeline.""" + + host: _PipelineController + stt_provider: stt.SpeechToTextEntity | stt.Provider = field(init=False, repr=False) + tts_stream: tts.ResultStream | None = field(init=False, default=None) + wake_word_entity_id: str | None = field(init=False, default=None, repr=False) + wake_word_entity: wake_word.WakeWordDetectionEntity = field(init=False, repr=False) + audio_enhancer: AudioEnhancer | None = field(init=False, default=None) + audio_chunking_buffer: AudioBuffer = field( + init=False, default_factory=lambda: AudioBuffer(BYTES_PER_CHUNK) + ) + _conversation_data: PipelineConversationData | None = field( + init=False, default=None + ) + _intent_agent_only: bool = field(init=False, default=False) + _streamed_response_text: bool = field(init=False, default=False) + _invalidated: bool = field(init=False, default=False) + intent_agent: conversation.AgentInfo | None = field(init=False, default=None) + + def __post_init__(self) -> None: + """Initialize audio processing.""" + if self.host.audio_settings.needs_processor: + self.audio_enhancer = MicroVadSpeexEnhancer( + self.host.audio_settings.auto_gain_dbfs, + self.host.audio_settings.noise_suppression_level, + self.host.audio_settings.is_vad_enabled, + ) + + @property + def response_audio(self) -> tts.ResultStream | None: + """Return the response audio stream.""" + return self.tts_stream + + @property + def supports_streaming_response(self) -> bool | None: + """Return whether response audio can be streamed.""" + if self.tts_stream is None: + return None + if not self.tts_stream.supports_streaming_input: + return False + if self.intent_agent is None: + return None + return self.intent_agent.supports_streaming + + @callback + def invalidate(self) -> None: + """Invalidate this processor's active input.""" + self._invalidated = True + + @callback + def cleanup(self) -> None: + """Clean up resources after a pipeline error.""" + if self.tts_stream is not None: + self.tts_stream.delete() + self.tts_stream = None + + async def prepare_wake_word_detection(self) -> None: + """Prepare wake-word-detection.""" + entity_id = ( + self.host.pipeline.wake_word_entity + or wake_word.async_default_entity(self.host.hass) + ) + if entity_id is None: + raise WakeWordDetectionError( + code="wake-engine-missing", + message="No wake word engine", + ) + + wake_word_entity = wake_word.async_get_wake_word_detection_entity( + self.host.hass, entity_id + ) + if wake_word_entity is None: + raise WakeWordDetectionError( + code="wake-provider-missing", + message=f"No wake-word-detection provider for: {entity_id}", + ) + + self.wake_word_entity_id = entity_id + self.wake_word_entity = wake_word_entity + + async def wake_word_detection( + self, + stream: AsyncIterable[EnhancedAudioChunk], + audio_chunks_for_stt: list[EnhancedAudioChunk], + ) -> wake_word.DetectionResult | None: + """Run wake-word-detection portion of pipeline. Returns detection result.""" + metadata_dict = asdict( + stt.SpeechMetadata( + language="", + 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, + ) + ) + + wake_word_settings = self.host.wake_word_settings or WakeWordSettings() + + # Remove language since it doesn't apply to wake words yet + metadata_dict.pop("language", None) + + self.host.process_event( + PipelineEvent( + PipelineEventType.WAKE_WORD_START, + { + "entity_id": self.wake_word_entity_id, + "metadata": metadata_dict, + "timeout": wake_word_settings.timeout or 0, + }, + ) + ) + + self.host.start_debug_recording(f"00_wake-{self.wake_word_entity_id}") + + wake_word_vad: VoiceActivityTimeout | None = None + if (wake_word_settings.timeout is not None) and ( + wake_word_settings.timeout > 0 + ): + # Use VAD to determine timeout + wake_word_vad = VoiceActivityTimeout(wake_word_settings.timeout) + + # Audio chunk buffer. This audio will be forwarded to speech-to-text + # after wake-word-detection. + num_audio_chunks_to_buffer = int( + (wake_word_settings.audio_seconds_to_buffer * SAMPLE_RATE) + / SAMPLES_PER_CHUNK + ) + + stt_audio_buffer: deque[EnhancedAudioChunk] | None = None + if num_audio_chunks_to_buffer > 0: + stt_audio_buffer = deque(maxlen=num_audio_chunks_to_buffer) + + try: + # Detect wake word(s) + result = await self.wake_word_entity.async_process_audio_stream( + self._wake_word_audio_stream( + audio_stream=stream, + stt_audio_buffer=stt_audio_buffer, + wake_word_vad=wake_word_vad, + ), + self.host.pipeline.wake_word_id, + ) + + if stt_audio_buffer is not None: + # All audio kept from right before the wake word was detected as + # a single chunk. + audio_chunks_for_stt.extend(stt_audio_buffer) + except WakeWordDetectionAborted: + raise + except WakeWordTimeoutError: + _LOGGER.debug("Timeout during wake word detection") + raise + except Exception as src_error: + _LOGGER.exception("Unexpected error during wake-word-detection") + raise WakeWordDetectionError( + code="wake-stream-failed", + message="Unexpected error during wake-word-detection", + ) from src_error + + _LOGGER.debug("wake-word-detection result %s", result) + + if result is None: + wake_word_output: dict[str, Any] = {} + else: + self.host.accept_wake_word(result.wake_word_phrase) + + if result.queued_audio: + # Add audio that was pending at detection. + # + # Because detection occurs *after* the wake word was actually + # spoken, we need to make sure pending audio is forwarded to + # speech-to-text so the user does not have to pause before + # speaking the voice command. + audio_chunks_for_stt.extend( + EnhancedAudioChunk( + audio=chunk_ts[0], + timestamp_ms=chunk_ts[1], + speech_probability=None, + ) + for chunk_ts in result.queued_audio + ) + + wake_word_output = asdict(result) + + # Remove non-JSON fields + wake_word_output.pop("queued_audio", None) + + self.host.process_event( + PipelineEvent( + PipelineEventType.WAKE_WORD_END, + {"wake_word_output": wake_word_output}, + ) + ) + + return result + + async def _wake_word_audio_stream( + self, + audio_stream: AsyncIterable[EnhancedAudioChunk], + stt_audio_buffer: deque[EnhancedAudioChunk] | None, + wake_word_vad: VoiceActivityTimeout | None, + sample_rate: int = SAMPLE_RATE, + sample_width: int = SAMPLE_WIDTH, + ) -> AsyncIterable[tuple[bytes, int]]: + """Yield audio chunks with timestamps (milliseconds since start of stream). + + Adds audio to a ring buffer that will be forwarded to speech-to-text after + detection. Times out if VAD detects enough silence. + """ + async for chunk in audio_stream: + if self._invalidated: + raise WakeWordDetectionAborted + + self.host.capture_audio(chunk.audio) + yield chunk.audio, chunk.timestamp_ms + + # Wake-word-detection occurs *after* the wake word was actually + # spoken. Keeping audio right before detection allows the voice + # command to be spoken immediately after the wake word. + if stt_audio_buffer is not None: + stt_audio_buffer.append(chunk) + + if wake_word_vad is not None: + chunk_seconds = (len(chunk.audio) // sample_width) / sample_rate + if not wake_word_vad.process(chunk_seconds, chunk.speech_probability): + raise WakeWordTimeoutError( + code="wake-word-timeout", message="Wake word was not detected" + ) + + async def prepare_speech_to_text(self, metadata: stt.SpeechMetadata) -> None: + """Prepare speech-to-text.""" + # pipeline.stt_engine can't be None or this function is not called + stt_provider = stt.async_get_speech_to_text_engine( + self.host.hass, + self.host.pipeline.stt_engine, # type: ignore[arg-type] + ) + + if stt_provider is None: + engine = self.host.pipeline.stt_engine + raise SpeechToTextError( + code="stt-provider-missing", + message=f"No speech-to-text provider for: {engine}", + ) + + metadata.language = self.host.pipeline.stt_language or self.host.language + + if not stt_provider.check_metadata(metadata): + raise SpeechToTextError( + code="stt-provider-unsupported-metadata", + message=( + f"Provider {stt_provider.name} does not support input speech " + f"to text metadata {metadata}" + ), + ) + + self.stt_provider = stt_provider + + async def speech_to_text( + self, + metadata: stt.SpeechMetadata, + stream: AsyncIterable[EnhancedAudioChunk], + ) -> str: + """Run speech-to-text portion of pipeline. Returns the spoken text.""" + # Create a background task to prepare the conversation agent + if self.host.end_stage >= PipelineStage.INTENT and self.intent_agent: + self.host.hass.async_create_background_task( + conversation.async_prepare_agent( + self.host.hass, self.intent_agent.id, self.host.language + ), + f"prepare conversation agent {self.intent_agent.id}", + ) + + if isinstance(self.stt_provider, stt.Provider): + engine = self.stt_provider.name + else: + engine = self.stt_provider.entity_id + + self.host.process_event( + PipelineEvent( + PipelineEventType.STT_START, + { + "engine": engine, + "metadata": asdict(metadata), + "audio_processing": asdict(self.stt_provider.audio_processing), + }, + ) + ) + + self.host.start_debug_recording(f"01_stt-{engine}") + + try: + # Transcribe audio stream + stt_vad: VoiceCommandSegmenter | None = None + if ( + self.host.audio_settings.is_vad_enabled + and self.stt_provider.audio_processing.requires_external_vad + ): + stt_vad = VoiceCommandSegmenter( + silence_seconds=self.host.audio_settings.silence_seconds + ) + + result = await self.stt_provider.async_process_audio_stream( + metadata, + self._speech_to_text_stream(audio_stream=stream, stt_vad=stt_vad), + ) + except asyncio.CancelledError, TimeoutError: + raise # expected + except hass_nabucasa.auth.Unauthenticated as src_error: + raise SpeechToTextError( + code="cloud-auth-failed", + message="Home Assistant Cloud authentication failed", + ) from src_error + except Exception as src_error: + _LOGGER.exception("Unexpected error during speech-to-text") + raise SpeechToTextError( + code="stt-stream-failed", + message="Unexpected error during speech-to-text", + ) from src_error + + _LOGGER.debug("speech-to-text result %s", result) + + if result.result != stt.SpeechResultState.SUCCESS: + raise SpeechToTextError( + code="stt-stream-failed", + message="speech-to-text failed", + ) + + if not result.text: + raise SpeechToTextError( + code="stt-no-text-recognized", message="No text recognized" + ) + + self.host.process_event( + PipelineEvent( + PipelineEventType.STT_END, + { + "stt_output": { + "text": result.text, + } + }, + ) + ) + + return result.text + + async def _speech_to_text_stream( + self, + audio_stream: AsyncIterable[EnhancedAudioChunk], + stt_vad: VoiceCommandSegmenter | None, + sample_rate: int = SAMPLE_RATE, + sample_width: int = SAMPLE_WIDTH, + ) -> AsyncGenerator[bytes]: + """Yield audio chunks until VAD detects silence or speech-to-text completes.""" + sent_vad_start = False + async for chunk in audio_stream: + self.host.capture_audio(chunk.audio) + + if stt_vad is not None: + chunk_seconds = (len(chunk.audio) // sample_width) / sample_rate + if not stt_vad.process(chunk_seconds, chunk.speech_probability): + # Silence detected at the end of voice command + self.host.process_event( + PipelineEvent( + PipelineEventType.STT_VAD_END, + {"timestamp": chunk.timestamp_ms}, + ) + ) + break + + if stt_vad.in_command and (not sent_vad_start): + # Speech detected at start of voice command + self.host.process_event( + PipelineEvent( + PipelineEventType.STT_VAD_START, + {"timestamp": chunk.timestamp_ms}, + ) + ) + sent_vad_start = True + + yield chunk.audio + + async def prepare_recognize_intent(self, session: chat_session.ChatSession) -> None: + """Prepare recognizing an intent.""" + self._conversation_data = async_get_pipeline_conversation_data( + self.host.hass, session + ) + + if self._conversation_data.continue_conversation_agent is not None: + agent_info = conversation.async_get_agent_info( + self.host.hass, self._conversation_data.continue_conversation_agent + ) + self._conversation_data.continue_conversation_agent = None + if agent_info is None: + raise IntentRecognitionError( + code="intent-agent-not-found", + message=( + f"Intent recognition engine" + f" {self._conversation_data.continue_conversation_agent}" + " asked for follow-up but is no longer found" + ), + ) + self._intent_agent_only = True + + else: + agent_info = conversation.async_get_agent_info( + self.host.hass, + self.host.pipeline.conversation_engine + or conversation.HOME_ASSISTANT_AGENT, + ) + + if agent_info is None: + engine = self.host.pipeline.conversation_engine or "default" + raise IntentRecognitionError( + code="intent-not-supported", + message=f"Intent recognition engine {engine} is not found", + ) + + self.intent_agent = agent_info + + async def recognize_intent( + self, + intent_input: str, + conversation_id: str, + conversation_extra_system_prompt: str | None, + ) -> tuple[str, bool]: + """Run intent recognition portion of pipeline. + + Returns (speech, all_targets_in_satellite_area). + """ + if self.intent_agent is None or self._conversation_data is None: + raise RuntimeError("Recognize intent was not prepared") + + if self.host.pipeline.conversation_language == MATCH_ALL: + # LLMs support all languages ('*') so use languages from the + # pipeline for intent fallback. + # + # We prioritize the STT and TTS languages because they may be more + # specific, such as "zh-CN" instead of just "zh". This is necessary + # for languages whose intents are split out by region when + # preferring local intent matching. + input_language = ( + self.host.pipeline.stt_language + or self.host.pipeline.tts_language + or self.host.pipeline.language + ) + else: + input_language = self.host.pipeline.conversation_language + + self.host.process_event( + PipelineEvent( + PipelineEventType.INTENT_START, + { + "engine": self.intent_agent.id, + "language": input_language, + "intent_input": intent_input, + "conversation_id": conversation_id, + "device_id": self.host.device_id, + "satellite_id": self.host.satellite_id, + "prefer_local_intents": self.host.pipeline.prefer_local_intents, + }, + ) + ) + + try: + if self.tts_stream and self.tts_stream.supports_streaming_input: + tts_input_stream: asyncio.Queue[str | None] | None = asyncio.Queue() + else: + tts_input_stream = None + chat_log_role = None + delta_character_count = 0 + + @callback + def chat_log_delta_listener( + chat_log: conversation.ChatLog, delta: dict + ) -> None: + """Handle chat log delta.""" + self.host.process_event( + PipelineEvent( + PipelineEventType.INTENT_PROGRESS, + { + "chat_log_delta": delta, + }, + ) + ) + if tts_input_stream is None: + return + + nonlocal chat_log_role + + if role := delta.get("role"): + chat_log_role = role + + # We are only interested in assistant deltas + if chat_log_role != "assistant": + return + + if content := delta.get("content"): + tts_input_stream.put_nowait(content) + + if self._streamed_response_text: + return + + nonlocal delta_character_count + + # Streamed responses are not cached. That's why we + # only start streaming text after we have received + # enough characters that indicates it will be a long + # response or if we have received text, and then a + # tool call. + + # Tool call after we already received text + start_streaming = delta_character_count > 0 and delta.get("tool_calls") + + # Count characters in the content and test if we + # exceed streaming threshold + if not start_streaming and content: + delta_character_count += len(content) + start_streaming = delta_character_count > STREAM_RESPONSE_CHARS + + if not start_streaming: + return + + self._streamed_response_text = True + + self.host.process_event( + PipelineEvent( + PipelineEventType.INTENT_PROGRESS, + { + "tts_start_streaming": True, + }, + ) + ) + + async def tts_input_stream_generator() -> AsyncGenerator[str]: + """Yield TTS input stream.""" + while (tts_input := await tts_input_stream.get()) is not None: + yield tts_input + + # Concatenate all existing queue items + parts = [] + while not tts_input_stream.empty(): + parts.append(tts_input_stream.get_nowait()) + tts_input_stream.put_nowait( + "".join( + # At this point parts is only strings, + # None indicates end of queue + cast(list[str], parts) + ) + ) + + assert self.tts_stream is not None + self.tts_stream.async_set_message_stream(tts_input_stream_generator()) + + user_input = conversation.ConversationInput( + text=intent_input, + context=self.host.context, + conversation_id=conversation_id, + device_id=self.host.device_id, + satellite_id=self.host.satellite_id, + language=input_language, + agent_id=self.intent_agent.id, + extra_system_prompt=conversation_extra_system_prompt, + ) + + with ( + chat_session.async_get_chat_session( + self.host.hass, user_input.conversation_id + ) as session, + conversation.async_get_chat_log( + self.host.hass, + session, + user_input, + chat_log_delta_listener=chat_log_delta_listener, + ) as chat_log, + ): + agent_id = self.intent_agent.id + processed_locally = agent_id == conversation.HOME_ASSISTANT_AGENT + all_targets_in_satellite_area = False + intent_response: intent.IntentResponse | None = None + if not processed_locally and not self._intent_agent_only: + # Sentence triggers override conversation agent + if ( + trigger_response_text + := await conversation.async_handle_sentence_triggers( + self.host.hass, user_input, chat_log + ) + ) is not None: + # Sentence trigger matched + agent_id = "sentence_trigger" + processed_locally = True + intent_response = intent.IntentResponse( + self.host.pipeline.conversation_language + ) + intent_response.async_set_speech(trigger_response_text) + + intent_filter: Callable[[RecognizeResult], bool] | None = None + # If the LLM has API access, we filter out some sentences that are + # interfering with LLM operation. + if ( + intent_agent_state := self.host.hass.states.get( + self.intent_agent.id + ) + ) and intent_agent_state.attributes.get( + EntityStateAttribute.SUPPORTED_FEATURES, 0 + ) & conversation.ConversationEntityFeature.CONTROL: + intent_filter = _async_local_fallback_intent_filter + + # Try local intents + if ( + intent_response is None + and self.host.pipeline.prefer_local_intents + and ( + intent_response := await conversation.async_handle_intents( + self.host.hass, + user_input, + chat_log, + intent_filter=intent_filter, + ) + ) + ): + # Local intent matched + agent_id = conversation.HOME_ASSISTANT_AGENT + processed_locally = True + + # It was already handled, create response and add to chat history + if intent_response is not None: + speech: str = intent_response.speech.get("plain", {}).get( + "speech", "" + ) + chat_log.async_add_assistant_content_without_tools( + conversation.AssistantContent( + agent_id=agent_id, + content=speech, + ) + ) + conversation_result = conversation.ConversationResult( + response=intent_response, + conversation_id=session.conversation_id, + ) + + else: + # Fall back to pipeline conversation agent + conversation_result = await conversation.async_converse( + hass=self.host.hass, + text=user_input.text, + conversation_id=user_input.conversation_id, + device_id=user_input.device_id, + satellite_id=user_input.satellite_id, + context=user_input.context, + language=user_input.language, + agent_id=user_input.agent_id, + extra_system_prompt=user_input.extra_system_prompt, + ) + speech = conversation_result.response.speech.get("plain", {}).get( + "speech", "" + ) + if tts_input_stream and self._streamed_response_text: + tts_input_stream.put_nowait(None) + + if agent_id == conversation.HOME_ASSISTANT_AGENT: + # Check if all targeted entities were in the same area as + # the satellite device. + # If so, the satellite should respond with an acknowledge beep + # instead of a full response. + all_targets_in_satellite_area = ( + self._get_all_targets_in_satellite_area( + conversation_result.response, + self.host.satellite_id, + self.host.device_id, + ) + ) + + except Exception as src_error: + _LOGGER.exception("Unexpected error during intent recognition") + raise IntentRecognitionError( + code="intent-failed", + message="Unexpected error during intent recognition", + ) from src_error + + _LOGGER.debug("conversation result %s", conversation_result) + + self.host.process_event( + PipelineEvent( + PipelineEventType.INTENT_END, + { + "processed_locally": processed_locally, + "intent_output": conversation_result.as_dict(), + }, + ) + ) + + if conversation_result.continue_conversation: + self._conversation_data.continue_conversation_agent = agent_id + + return (speech, all_targets_in_satellite_area) + + def _get_all_targets_in_satellite_area( + self, + intent_response: intent.IntentResponse, + satellite_id: str | None, + device_id: str | None, + ) -> bool: + """Return true if all targeted entities were in the same area as the device.""" + if ( + intent_response.response_type is not intent.IntentResponseType.ACTION_DONE + or not intent_response.matched_states + ): + return False + + entity_registry = er.async_get(self.host.hass) + device_registry = dr.async_get(self.host.hass) + + area_id: str | None = None + + if ( + satellite_id is not None + and (target_entity_entry := entity_registry.async_get(satellite_id)) + is not None + ): + area_id = target_entity_entry.area_id + device_id = target_entity_entry.device_id + + if area_id is None: + if device_id is None: + return False + + device_entry = device_registry.async_get(device_id) + if device_entry is None: + return False + + area_id = dr.async_get_effective_area_id(self.host.hass, device_entry) + if area_id is None: + return False + + for state in intent_response.matched_states: + target_entity_entry = entity_registry.async_get(state.entity_id) + if target_entity_entry is None: + return False + + target_area_id = target_entity_entry.area_id + if target_area_id is None: + if target_entity_entry.device_id is None: + return False + + target_device_entry = device_registry.async_get( + target_entity_entry.device_id + ) + if target_device_entry is None: + return False + + target_area_id = dr.async_get_effective_area_id( + self.host.hass, target_device_entry + ) + + if target_area_id != area_id: + return False + + return True + + async def prepare_text_to_speech(self) -> None: + """Prepare text-to-speech.""" + # pipeline.tts_engine can't be None or this function is not called + engine = cast(str, self.host.pipeline.tts_engine) + + tts_options: dict[str, Any] = {} + if self.host.pipeline.tts_voice is not None: + tts_options[tts.ATTR_VOICE] = self.host.pipeline.tts_voice + + if isinstance(self.host.tts_audio_output, dict): + tts_options.update(self.host.tts_audio_output) + elif isinstance(self.host.tts_audio_output, str): + tts_options[tts.ATTR_PREFERRED_FORMAT] = self.host.tts_audio_output + if self.host.tts_audio_output == "wav": + # 16 Khz, 16-bit mono + tts_options[tts.ATTR_PREFERRED_SAMPLE_RATE] = SAMPLE_RATE + tts_options[tts.ATTR_PREFERRED_SAMPLE_CHANNELS] = SAMPLE_CHANNELS + tts_options[tts.ATTR_PREFERRED_SAMPLE_BYTES] = SAMPLE_WIDTH + + try: + self.tts_stream = tts.async_create_stream( + hass=self.host.hass, + engine=engine, + language=self.host.pipeline.tts_language, + options=tts_options, + ) + except HomeAssistantError as err: + raise TextToSpeechError( + code="tts-not-supported", + message=( + f"Text-to-speech engine {engine} " + f"does not support language {self.host.pipeline.tts_language}" + f" or options {tts_options}:" + f" {err}" + ), + ) from err + + async def text_to_speech( + self, tts_input: str, override_media_path: Path | None = None + ) -> None: + """Run text-to-speech portion of pipeline.""" + assert self.tts_stream is not None + + self.host.process_event( + PipelineEvent( + PipelineEventType.TTS_START, + { + "engine": self.tts_stream.engine, + "language": self.host.pipeline.tts_language, + "voice": self.host.pipeline.tts_voice, + "tts_input": tts_input, + "acknowledge_override": override_media_path is not None, + }, + ) + ) + + if override_media_path: + self.tts_stream.async_override_result(override_media_path) + elif not self._streamed_response_text: + self.tts_stream.async_set_message(tts_input) + + tts_output = { + "media_id": self.tts_stream.media_source_id, + "token": self.tts_stream.token, + "url": self.tts_stream.url, + "mime_type": self.tts_stream.content_type, + } + + self.host.process_event( + PipelineEvent(PipelineEventType.TTS_END, {"tts_output": tts_output}) + ) + + async def process_volume_only( + self, audio_stream: AsyncIterable[bytes] + ) -> AsyncGenerator[EnhancedAudioChunk]: + """Apply volume transformation only with optional chunking. + + No VAD/audio enhancements are applied. + """ + timestamp_ms = 0 + async for chunk in audio_stream: + if self.host.audio_settings.volume_multiplier != 1.0: + chunk = _multiply_volume( + chunk, self.host.audio_settings.volume_multiplier + ) + + for sub_chunk in chunk_samples( + chunk, BYTES_PER_CHUNK, self.audio_chunking_buffer + ): + yield EnhancedAudioChunk( + audio=sub_chunk, + timestamp_ms=timestamp_ms, + speech_probability=None, # no VAD + ) + timestamp_ms += MS_PER_CHUNK + + async def process_enhance_audio( + self, audio_stream: AsyncIterable[bytes] + ) -> AsyncGenerator[EnhancedAudioChunk]: + """Split audio into chunks and apply audio enhancements. + + Applies VAD/noise suppression/auto gain/volume + transformation. + """ + assert self.audio_enhancer is not None + + timestamp_ms = 0 + async for dirty_samples in audio_stream: + if self.host.audio_settings.volume_multiplier != 1.0: + # Static gain + dirty_samples = _multiply_volume( + dirty_samples, self.host.audio_settings.volume_multiplier + ) + + # Split into chunks for audio enhancements/VAD + for dirty_chunk in chunk_samples( + dirty_samples, BYTES_PER_CHUNK, self.audio_chunking_buffer + ): + yield self.audio_enhancer.enhance_chunk(dirty_chunk, timestamp_ms) + timestamp_ms += MS_PER_CHUNK + + async def async_execute(self, request: _PipelineProcessorRequest) -> None: + """Run the configured default pipeline stages.""" + current_stage: PipelineStage | None = self.host.start_stage + stt_audio_buffer: list[EnhancedAudioChunk] = [] + stt_processed_stream: AsyncIterable[EnhancedAudioChunk] | None = None + + if request.stt_stream is not None: + if self.host.audio_settings.needs_processor: + # VAD/noise suppression/auto gain/volume + stt_processed_stream = self.process_enhance_audio(request.stt_stream) + else: + # Volume multiplier only + stt_processed_stream = self.process_volume_only(request.stt_stream) + + if current_stage == PipelineStage.WAKE_WORD: + # wake-word-detection + assert stt_processed_stream is not None + detect_result = await self.wake_word_detection( + stt_processed_stream, stt_audio_buffer + ) + if detect_result is None: + # No wake word. Abort the rest of the pipeline. + return + + current_stage = PipelineStage.STT + + # speech-to-text + intent_input = request.intent_input + if current_stage == PipelineStage.STT: + assert request.stt_metadata is not None + assert stt_processed_stream is not None + + if request.wake_word_phrase is not None: + self.host.accept_wake_word(request.wake_word_phrase) + + stt_input_stream = stt_processed_stream + + if stt_audio_buffer: + # Send audio in the buffer first to speech-to-text, + # then move on to stt_stream. + # This is basically an async itertools.chain. + async def buffer_then_audio_stream() -> AsyncGenerator[ + EnhancedAudioChunk + ]: + # Buffered audio + for chunk in stt_audio_buffer: + yield chunk + + # Streamed audio + assert stt_processed_stream is not None + async for chunk in stt_processed_stream: + yield chunk + + stt_input_stream = buffer_then_audio_stream() + + intent_input = await self.speech_to_text( + request.stt_metadata, + stt_input_stream, + ) + current_stage = PipelineStage.INTENT + + if self.host.end_stage != PipelineStage.STT: + tts_input = request.tts_input + all_targets_in_satellite_area = False + + if current_stage == PipelineStage.INTENT: + # intent-recognition + assert intent_input is not None + ( + tts_input, + all_targets_in_satellite_area, + ) = await self.recognize_intent( + intent_input, + request.session.conversation_id, + request.conversation_extra_system_prompt, + ) + if all_targets_in_satellite_area or tts_input.strip(): + current_stage = PipelineStage.TTS + else: + # Skip TTS + current_stage = PipelineStage.END + + if self.host.end_stage != PipelineStage.INTENT: + # text-to-speech + if current_stage == PipelineStage.TTS: + if all_targets_in_satellite_area: + # Use acknowledge media instead of full response + await self.text_to_speech( + tts_input or "", override_media_path=ACKNOWLEDGE_PATH + ) + else: + assert tts_input is not None + await self.text_to_speech(tts_input) + + async def async_validate(self, request: _PipelineProcessorRequest) -> None: + """Validate pipeline input and prepare the default stages.""" + if self.host.start_stage in (PipelineStage.WAKE_WORD, PipelineStage.STT): + if self.host.pipeline.stt_engine is None: + raise PipelineRunValidationError( + "the pipeline does not support speech-to-text" + ) + if request.stt_metadata is None: + raise PipelineRunValidationError( + "stt_metadata is required for speech-to-text" + ) + if request.stt_stream is None: + raise PipelineRunValidationError( + "stt_stream is required for speech-to-text" + ) + elif self.host.start_stage == PipelineStage.INTENT: + if request.intent_input is None: + raise PipelineRunValidationError( + "intent_input is required for intent recognition" + ) + elif self.host.start_stage == PipelineStage.TTS: + if request.tts_input is None: + raise PipelineRunValidationError( + "tts_input is required for text-to-speech" + ) + if self.host.end_stage == PipelineStage.TTS: + if self.host.pipeline.tts_engine is None: + raise PipelineRunValidationError( + "the pipeline does not support text-to-speech" + ) + + start_stage_index = PIPELINE_STAGE_ORDER.index(self.host.start_stage) + end_stage_index = PIPELINE_STAGE_ORDER.index(self.host.end_stage) + + prepare_tasks = [] + + if ( + start_stage_index + <= PIPELINE_STAGE_ORDER.index(PipelineStage.WAKE_WORD) + <= end_stage_index + ): + prepare_tasks.append(self.prepare_wake_word_detection()) + + if ( + start_stage_index + <= PIPELINE_STAGE_ORDER.index(PipelineStage.STT) + <= end_stage_index + ): + assert request.stt_metadata is not None + prepare_tasks.append(self.prepare_speech_to_text(request.stt_metadata)) + + if ( + start_stage_index + <= PIPELINE_STAGE_ORDER.index(PipelineStage.INTENT) + <= end_stage_index + ): + prepare_tasks.append(self.prepare_recognize_intent(request.session)) + + if prepare_tasks: + await asyncio.gather(*prepare_tasks) + + # Do TTS prepare separately so we don't create a ResultStream if the + # pipeline is invalid. + if ( + start_stage_index + <= PIPELINE_STAGE_ORDER.index(PipelineStage.TTS) + <= end_stage_index + ): + await self.prepare_text_to_speech() + + +def _multiply_volume(chunk: bytes, volume_multiplier: float) -> bytes: + """Multiply 16-bit PCM samples by a constant.""" + + def _clamp(val: float) -> float: + """Clamp to signed 16-bit.""" + return max(-32768, min(32767, val)) + + return array.array( + "h", + (int(_clamp(value * volume_multiplier)) for value in array.array("h", chunk)), + ).tobytes() + + +@dataclass +class PipelineConversationData: + """Hold data for the duration of a conversation.""" + + continue_conversation_agent: str | None = None + """The agent that requested the conversation to be continued.""" + + +@callback +def async_get_pipeline_conversation_data( + hass: HomeAssistant, session: chat_session.ChatSession +) -> PipelineConversationData: + """Get the pipeline data for a specific conversation.""" + all_conversation_data = hass.data.get(KEY_PIPELINE_CONVERSATION_DATA) + if all_conversation_data is None: + all_conversation_data = {} + hass.data[KEY_PIPELINE_CONVERSATION_DATA] = all_conversation_data + + data = all_conversation_data.get(session.conversation_id) + if data is not None: + return data + + @callback + def do_cleanup() -> None: + """Handle cleanup.""" + all_conversation_data.pop(session.conversation_id) + + session.async_on_cleanup(do_cleanup) + data = all_conversation_data[session.conversation_id] = PipelineConversationData() + return data diff --git a/homeassistant/components/assist_pipeline/pipeline.py b/homeassistant/components/assist_pipeline/pipeline.py index fbf307c24820..235634a6be40 100644 --- a/homeassistant/components/assist_pipeline/pipeline.py +++ b/homeassistant/components/assist_pipeline/pipeline.py @@ -1,38 +1,12 @@ """Classes for voice assistant pipelines.""" -import array -import asyncio -from collections import deque -from collections.abc import AsyncGenerator, AsyncIterable, Callable -from dataclasses import asdict, dataclass, field import logging -from pathlib import Path -from queue import Empty, Queue -from threading import Thread -import time -from typing import TYPE_CHECKING, Any, cast, override -import wave +from typing import Any, override -import hass_nabucasa import probatio -from homeassistant.components import ( - conversation, - media_player, - stt, - tts, - wake_word, - websocket_api, -) -from homeassistant.const import MATCH_ALL, EntityStateAttribute -from homeassistant.core import Context, HomeAssistant, callback -from homeassistant.exceptions import HomeAssistantError -from homeassistant.helpers import ( - chat_session, - device_registry as dr, - entity_registry as er, - intent, -) +from homeassistant.components import conversation, stt, tts, websocket_api +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.collection import ( CollectionError, ItemNotFound, @@ -44,39 +18,12 @@ from homeassistant.helpers.singleton import singleton from homeassistant.helpers.storage import Store from homeassistant.helpers.typing import UNDEFINED, UndefinedType, VolDictType from homeassistant.util import language as language_util, ulid as ulid_util -from homeassistant.util.hass_dict import HassKey -from homeassistant.util.limited_size_dict import LimitedSizeDict -from . import models as _models, runtime as _runtime -from .audio_enhancer import AudioEnhancer, EnhancedAudioChunk, MicroVadSpeexEnhancer -from .const import ( - ACKNOWLEDGE_PATH, - BYTES_PER_CHUNK, - CONF_DEBUG_RECORDING_DIR, - DATA_CONFIG, - DATA_LAST_WAKE_UP, - DOMAIN, - MS_PER_CHUNK, - SAMPLE_CHANNELS, - SAMPLE_RATE, - SAMPLE_WIDTH, - SAMPLES_PER_CHUNK, - WAKE_WORD_COOLDOWN, -) -from .error import ( - DuplicateWakeUpDetectedError, - IntentRecognitionError, - InvalidPipelineStagesError, - PipelineError, - PipelineNotFound, - PipelineRunValidationError, - SpeechToTextError, - TextToSpeechError, - WakeWordDetectionAborted, - WakeWordDetectionError, - WakeWordTimeoutError, -) -from .vad import AudioBuffer, VoiceActivityTimeout, VoiceCommandSegmenter, chunk_samples +from . import error as _error, models as _models, run as _run, runtime as _runtime +from .const import DOMAIN + +PipelineError = _error.PipelineError +PipelineNotFound = _error.PipelineNotFound PIPELINE_STAGE_ORDER = _models.PIPELINE_STAGE_ORDER AudioSettings = _models.AudioSettings @@ -87,6 +34,10 @@ PipelineEventType = _models.PipelineEventType PipelineStage = _models.PipelineStage WakeWordSettings = _models.WakeWordSettings +STORED_PIPELINE_RUNS = _run.STORED_PIPELINE_RUNS +PipelineInput = _run.PipelineInput +PipelineRun = _run.PipelineRun + KEY_ASSIST_PIPELINE = _runtime.KEY_ASSIST_PIPELINE AssistDevice = _runtime.AssistDevice DeviceAudioQueue = _runtime.DeviceAudioQueue @@ -94,9 +45,6 @@ PipelineData = _runtime.PipelineData PipelineRunDebug = _runtime.PipelineRunDebug PipelineRuns = _runtime.PipelineRuns -if TYPE_CHECKING: - from hassil.recognize import RecognizeResult - _LOGGER = logging.getLogger(__name__) STORAGE_KEY = f"{DOMAIN}.pipelines" @@ -108,12 +56,6 @@ ENGINE_LANGUAGE_PAIRS = ( ("tts_engine", "tts_language"), ) -KEY_PIPELINE_CONVERSATION_DATA: HassKey[dict[str, PipelineConversationData]] = HassKey( - "pipeline_conversation_data" -) -# Number of response parts to handle before streaming the response -STREAM_RESPONSE_CHARS = 60 - def validate_language(data: dict[str, Any]) -> Any: """Validate language settings.""" @@ -141,19 +83,6 @@ PIPELINE_FIELDS: VolDictType = { probatio.Optional("acknowledge_media_id"): str, } -STORED_PIPELINE_RUNS = 10 - -SAVE_DELAY = 10 - - -@callback -def _async_local_fallback_intent_filter(result: RecognizeResult) -> bool: - """Filter out intents that are not local fallback.""" - return result.intent.name in ( - intent.INTENT_GET_STATE, - media_player.INTENT_MEDIA_SEARCH_AND_PLAY, - ) - @callback def _async_resolve_default_pipeline_settings( @@ -394,1372 +323,6 @@ async def async_update_pipeline( await pipeline_data.pipeline_store.async_update_item(pipeline.id, updates) -@dataclass -class PipelineRun: - """Running context for a pipeline.""" - - hass: HomeAssistant - context: Context - pipeline: Pipeline - start_stage: PipelineStage - end_stage: PipelineStage - event_callback: PipelineEventCallback - language: str = None # type: ignore[assignment] - runner_data: Any | None = None - intent_agent: conversation.AgentInfo | None = None - tts_audio_output: str | dict[str, Any] | None = None - wake_word_settings: WakeWordSettings | None = None - audio_settings: AudioSettings = field(default_factory=AudioSettings) - - id: str = field(default_factory=ulid_util.ulid_now) - stt_provider: stt.SpeechToTextEntity | stt.Provider = field(init=False, repr=False) - tts_stream: tts.ResultStream | None = field(init=False, default=None) - wake_word_entity_id: str | None = field(init=False, default=None, repr=False) - wake_word_entity: wake_word.WakeWordDetectionEntity = field(init=False, repr=False) - - abort_wake_word_detection: bool = field(init=False, default=False) - - debug_recording_thread: Thread | None = None - """Thread that records audio to debug_recording_dir""" - - debug_recording_queue: Queue[str | bytes | None] | None = None - """Queue to communicate with debug recording thread""" - - audio_enhancer: AudioEnhancer | None = None - """VAD/noise suppression/auto gain""" - - audio_chunking_buffer: AudioBuffer = field( - default_factory=lambda: AudioBuffer(BYTES_PER_CHUNK) - ) - """Buffer used when splitting audio into chunks for audio processing""" - - _device_id: str | None = None - """Optional device id set during run start.""" - - _satellite_id: str | None = None - """Optional satellite id set during run start.""" - - _conversation_data: PipelineConversationData | None = None - """Data tied to the conversation ID.""" - - _intent_agent_only = False - """If request should only be handled by agent. - - Ignores sentence triggers and local processing. - """ - - _streamed_response_text = False - """If the conversation agent streamed response text to TTS result.""" - - def __post_init__(self) -> None: - """Set language for pipeline.""" - self.language = self.pipeline.language or self.hass.config.language - - # wake -> stt -> intent -> tts - if PIPELINE_STAGE_ORDER.index(self.end_stage) < PIPELINE_STAGE_ORDER.index( - self.start_stage - ): - raise InvalidPipelineStagesError(self.start_stage, self.end_stage) - - pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE] - if self.pipeline.id not in pipeline_data.pipeline_debug: - pipeline_data.pipeline_debug[self.pipeline.id] = LimitedSizeDict( - size_limit=STORED_PIPELINE_RUNS - ) - pipeline_data.pipeline_debug[self.pipeline.id][self.id] = PipelineRunDebug() - pipeline_data.pipeline_runs.add_run(self) - - # Initialize with audio settings - if self.audio_settings.needs_processor and (self.audio_enhancer is None): - # Default audio enhancer - self.audio_enhancer = MicroVadSpeexEnhancer( - self.audio_settings.auto_gain_dbfs, - self.audio_settings.noise_suppression_level, - self.audio_settings.is_vad_enabled, - ) - - @override - def __eq__(self, other: object) -> bool: - """Compare pipeline runs by id.""" - if isinstance(other, PipelineRun): - return self.id == other.id - - return False - - @callback - def process_event(self, event: PipelineEvent) -> None: - """Log an event and call listener.""" - self.event_callback(event) - pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE] - if self.id not in pipeline_data.pipeline_debug[self.pipeline.id]: - # This run has been evicted from the logged pipeline runs already - return - pipeline_data.pipeline_debug[self.pipeline.id][self.id].events.append(event) - - def start( - self, conversation_id: str, device_id: str | None, satellite_id: str | None - ) -> None: - """Emit run start event.""" - self._device_id = device_id - self._satellite_id = satellite_id - if self.start_stage in (PipelineStage.WAKE_WORD, PipelineStage.STT): - self._start_debug_recording_thread() - - data: dict[str, Any] = { - "pipeline": self.pipeline.id, - "language": self.language, - "conversation_id": conversation_id, - } - if satellite_id is not None: - data["satellite_id"] = satellite_id - if self.runner_data is not None: - data["runner_data"] = self.runner_data - if self.tts_stream: - data["tts_output"] = { - "token": self.tts_stream.token, - "url": self.tts_stream.url, - "mime_type": self.tts_stream.content_type, - "stream_response": ( - self.tts_stream.supports_streaming_input - and self.intent_agent - and self.intent_agent.supports_streaming - ), - } - - self.process_event(PipelineEvent(PipelineEventType.RUN_START, data)) - - async def end(self) -> None: - """Emit run end event.""" - # Signal end of stream to listeners - self._capture_chunk(None) - - # Stop the recording thread before emitting run-end. - # This ensures that files are properly closed if the event handler reads them. - await self._stop_debug_recording_thread() - - self.process_event( - PipelineEvent( - PipelineEventType.RUN_END, - ) - ) - - pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE] - pipeline_data.pipeline_runs.remove_run(self) - - async def prepare_wake_word_detection(self) -> None: - """Prepare wake-word-detection.""" - entity_id = self.pipeline.wake_word_entity or wake_word.async_default_entity( - self.hass - ) - if entity_id is None: - raise WakeWordDetectionError( - code="wake-engine-missing", - message="No wake word engine", - ) - - wake_word_entity = wake_word.async_get_wake_word_detection_entity( - self.hass, entity_id - ) - if wake_word_entity is None: - raise WakeWordDetectionError( - code="wake-provider-missing", - message=f"No wake-word-detection provider for: {entity_id}", - ) - - self.wake_word_entity_id = entity_id - self.wake_word_entity = wake_word_entity - - async def wake_word_detection( - self, - stream: AsyncIterable[EnhancedAudioChunk], - audio_chunks_for_stt: list[EnhancedAudioChunk], - ) -> wake_word.DetectionResult | None: - """Run wake-word-detection portion of pipeline. Returns detection result.""" - metadata_dict = asdict( - stt.SpeechMetadata( - language="", - 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, - ) - ) - - wake_word_settings = self.wake_word_settings or WakeWordSettings() - - # Remove language since it doesn't apply to wake words yet - metadata_dict.pop("language", None) - - self.process_event( - PipelineEvent( - PipelineEventType.WAKE_WORD_START, - { - "entity_id": self.wake_word_entity_id, - "metadata": metadata_dict, - "timeout": wake_word_settings.timeout or 0, - }, - ) - ) - - if self.debug_recording_queue is not None: - self.debug_recording_queue.put_nowait(f"00_wake-{self.wake_word_entity_id}") - - wake_word_vad: VoiceActivityTimeout | None = None - if (wake_word_settings.timeout is not None) and ( - wake_word_settings.timeout > 0 - ): - # Use VAD to determine timeout - wake_word_vad = VoiceActivityTimeout(wake_word_settings.timeout) - - # Audio chunk buffer. This audio will be forwarded to speech-to-text - # after wake-word-detection. - num_audio_chunks_to_buffer = int( - (wake_word_settings.audio_seconds_to_buffer * SAMPLE_RATE) - / SAMPLES_PER_CHUNK - ) - - stt_audio_buffer: deque[EnhancedAudioChunk] | None = None - if num_audio_chunks_to_buffer > 0: - stt_audio_buffer = deque(maxlen=num_audio_chunks_to_buffer) - - try: - # Detect wake word(s) - result = await self.wake_word_entity.async_process_audio_stream( - self._wake_word_audio_stream( - audio_stream=stream, - stt_audio_buffer=stt_audio_buffer, - wake_word_vad=wake_word_vad, - ), - self.pipeline.wake_word_id, - ) - - if stt_audio_buffer is not None: - # All audio kept from right before the wake word was detected as - # a single chunk. - audio_chunks_for_stt.extend(stt_audio_buffer) - except WakeWordDetectionAborted: - raise - except WakeWordTimeoutError: - _LOGGER.debug("Timeout during wake word detection") - raise - except Exception as src_error: - _LOGGER.exception("Unexpected error during wake-word-detection") - raise WakeWordDetectionError( - code="wake-stream-failed", - message="Unexpected error during wake-word-detection", - ) from src_error - - _LOGGER.debug("wake-word-detection result %s", result) - - if result is None: - wake_word_output: dict[str, Any] = {} - else: - # Avoid duplicate detections by checking cooldown - last_wake_up = self.hass.data[DATA_LAST_WAKE_UP].get( - result.wake_word_phrase - ) - if last_wake_up is not None: - sec_since_last_wake_up = time.monotonic() - last_wake_up - if sec_since_last_wake_up < WAKE_WORD_COOLDOWN: - _LOGGER.debug( - "Duplicate wake word detection occurred for %s", - result.wake_word_phrase, - ) - raise DuplicateWakeUpDetectedError(result.wake_word_phrase) - - # Record last wake up time to block duplicate detections - self.hass.data[DATA_LAST_WAKE_UP][result.wake_word_phrase] = ( - time.monotonic() - ) - - if result.queued_audio: - # Add audio that was pending at detection. - # - # Because detection occurs *after* the wake word was actually - # spoken, we need to make sure pending audio is forwarded to - # speech-to-text so the user does not have to pause before - # speaking the voice command. - audio_chunks_for_stt.extend( - EnhancedAudioChunk( - audio=chunk_ts[0], - timestamp_ms=chunk_ts[1], - speech_probability=None, - ) - for chunk_ts in result.queued_audio - ) - - wake_word_output = asdict(result) - - # Remove non-JSON fields - wake_word_output.pop("queued_audio", None) - - self.process_event( - PipelineEvent( - PipelineEventType.WAKE_WORD_END, - {"wake_word_output": wake_word_output}, - ) - ) - - return result - - async def _wake_word_audio_stream( - self, - audio_stream: AsyncIterable[EnhancedAudioChunk], - stt_audio_buffer: deque[EnhancedAudioChunk] | None, - wake_word_vad: VoiceActivityTimeout | None, - sample_rate: int = SAMPLE_RATE, - sample_width: int = SAMPLE_WIDTH, - ) -> AsyncIterable[tuple[bytes, int]]: - """Yield audio chunks with timestamps (milliseconds since start of stream). - - Adds audio to a ring buffer that will be forwarded to speech-to-text after - detection. Times out if VAD detects enough silence. - """ - async for chunk in audio_stream: - if self.abort_wake_word_detection: - raise WakeWordDetectionAborted - - self._capture_chunk(chunk.audio) - yield chunk.audio, chunk.timestamp_ms - - # Wake-word-detection occurs *after* the wake word was actually - # spoken. Keeping audio right before detection allows the voice - # command to be spoken immediately after the wake word. - if stt_audio_buffer is not None: - stt_audio_buffer.append(chunk) - - if wake_word_vad is not None: - chunk_seconds = (len(chunk.audio) // sample_width) / sample_rate - if not wake_word_vad.process(chunk_seconds, chunk.speech_probability): - raise WakeWordTimeoutError( - code="wake-word-timeout", message="Wake word was not detected" - ) - - async def prepare_speech_to_text(self, metadata: stt.SpeechMetadata) -> None: - """Prepare speech-to-text.""" - # pipeline.stt_engine can't be None or this function is not called - stt_provider = stt.async_get_speech_to_text_engine( - self.hass, - self.pipeline.stt_engine, # type: ignore[arg-type] - ) - - if stt_provider is None: - engine = self.pipeline.stt_engine - raise SpeechToTextError( - code="stt-provider-missing", - message=f"No speech-to-text provider for: {engine}", - ) - - metadata.language = self.pipeline.stt_language or self.language - - if not stt_provider.check_metadata(metadata): - raise SpeechToTextError( - code="stt-provider-unsupported-metadata", - message=( - f"Provider {stt_provider.name} does not support input speech " - f"to text metadata {metadata}" - ), - ) - - self.stt_provider = stt_provider - - async def speech_to_text( - self, - metadata: stt.SpeechMetadata, - stream: AsyncIterable[EnhancedAudioChunk], - ) -> str: - """Run speech-to-text portion of pipeline. Returns the spoken text.""" - # Create a background task to prepare the conversation agent - if self.end_stage >= PipelineStage.INTENT and self.intent_agent: - self.hass.async_create_background_task( - conversation.async_prepare_agent( - self.hass, self.intent_agent.id, self.language - ), - f"prepare conversation agent {self.intent_agent.id}", - ) - - if isinstance(self.stt_provider, stt.Provider): - engine = self.stt_provider.name - else: - engine = self.stt_provider.entity_id - - self.process_event( - PipelineEvent( - PipelineEventType.STT_START, - { - "engine": engine, - "metadata": asdict(metadata), - "audio_processing": asdict(self.stt_provider.audio_processing), - }, - ) - ) - - if self.debug_recording_queue is not None: - # New recording - self.debug_recording_queue.put_nowait(f"01_stt-{engine}") - - try: - # Transcribe audio stream - stt_vad: VoiceCommandSegmenter | None = None - if ( - self.audio_settings.is_vad_enabled - and self.stt_provider.audio_processing.requires_external_vad - ): - stt_vad = VoiceCommandSegmenter( - silence_seconds=self.audio_settings.silence_seconds - ) - - result = await self.stt_provider.async_process_audio_stream( - metadata, - self._speech_to_text_stream(audio_stream=stream, stt_vad=stt_vad), - ) - except asyncio.CancelledError, TimeoutError: - raise # expected - except hass_nabucasa.auth.Unauthenticated as src_error: - raise SpeechToTextError( - code="cloud-auth-failed", - message="Home Assistant Cloud authentication failed", - ) from src_error - except Exception as src_error: - _LOGGER.exception("Unexpected error during speech-to-text") - raise SpeechToTextError( - code="stt-stream-failed", - message="Unexpected error during speech-to-text", - ) from src_error - - _LOGGER.debug("speech-to-text result %s", result) - - if result.result != stt.SpeechResultState.SUCCESS: - raise SpeechToTextError( - code="stt-stream-failed", - message="speech-to-text failed", - ) - - if not result.text: - raise SpeechToTextError( - code="stt-no-text-recognized", message="No text recognized" - ) - - self.process_event( - PipelineEvent( - PipelineEventType.STT_END, - { - "stt_output": { - "text": result.text, - } - }, - ) - ) - - return result.text - - async def _speech_to_text_stream( - self, - audio_stream: AsyncIterable[EnhancedAudioChunk], - stt_vad: VoiceCommandSegmenter | None, - sample_rate: int = SAMPLE_RATE, - sample_width: int = SAMPLE_WIDTH, - ) -> AsyncGenerator[bytes]: - """Yield audio chunks until VAD detects silence or speech-to-text completes.""" - sent_vad_start = False - async for chunk in audio_stream: - self._capture_chunk(chunk.audio) - - if stt_vad is not None: - chunk_seconds = (len(chunk.audio) // sample_width) / sample_rate - if not stt_vad.process(chunk_seconds, chunk.speech_probability): - # Silence detected at the end of voice command - self.process_event( - PipelineEvent( - PipelineEventType.STT_VAD_END, - {"timestamp": chunk.timestamp_ms}, - ) - ) - break - - if stt_vad.in_command and (not sent_vad_start): - # Speech detected at start of voice command - self.process_event( - PipelineEvent( - PipelineEventType.STT_VAD_START, - {"timestamp": chunk.timestamp_ms}, - ) - ) - sent_vad_start = True - - yield chunk.audio - - async def prepare_recognize_intent(self, session: chat_session.ChatSession) -> None: - """Prepare recognizing an intent.""" - self._conversation_data = async_get_pipeline_conversation_data( - self.hass, session - ) - - if self._conversation_data.continue_conversation_agent is not None: - agent_info = conversation.async_get_agent_info( - self.hass, self._conversation_data.continue_conversation_agent - ) - self._conversation_data.continue_conversation_agent = None - if agent_info is None: - raise IntentRecognitionError( - code="intent-agent-not-found", - message=( - f"Intent recognition engine" - f" {self._conversation_data.continue_conversation_agent}" - " asked for follow-up but is no longer found" - ), - ) - self._intent_agent_only = True - - else: - agent_info = conversation.async_get_agent_info( - self.hass, - self.pipeline.conversation_engine or conversation.HOME_ASSISTANT_AGENT, - ) - - if agent_info is None: - engine = self.pipeline.conversation_engine or "default" - raise IntentRecognitionError( - code="intent-not-supported", - message=f"Intent recognition engine {engine} is not found", - ) - - self.intent_agent = agent_info - - async def recognize_intent( - self, - intent_input: str, - conversation_id: str, - conversation_extra_system_prompt: str | None, - ) -> tuple[str, bool]: - """Run intent recognition portion of pipeline. - - Returns (speech, all_targets_in_satellite_area). - """ - if self.intent_agent is None or self._conversation_data is None: - raise RuntimeError("Recognize intent was not prepared") - - if self.pipeline.conversation_language == MATCH_ALL: - # LLMs support all languages ('*') so use languages from the - # pipeline for intent fallback. - # - # We prioritize the STT and TTS languages because they may be more - # specific, such as "zh-CN" instead of just "zh". This is necessary - # for languages whose intents are split out by region when - # preferring local intent matching. - input_language = ( - self.pipeline.stt_language - or self.pipeline.tts_language - or self.pipeline.language - ) - else: - input_language = self.pipeline.conversation_language - - self.process_event( - PipelineEvent( - PipelineEventType.INTENT_START, - { - "engine": self.intent_agent.id, - "language": input_language, - "intent_input": intent_input, - "conversation_id": conversation_id, - "device_id": self._device_id, - "satellite_id": self._satellite_id, - "prefer_local_intents": self.pipeline.prefer_local_intents, - }, - ) - ) - - try: - if self.tts_stream and self.tts_stream.supports_streaming_input: - tts_input_stream: asyncio.Queue[str | None] | None = asyncio.Queue() - else: - tts_input_stream = None - chat_log_role = None - delta_character_count = 0 - - @callback - def chat_log_delta_listener( - chat_log: conversation.ChatLog, delta: dict - ) -> None: - """Handle chat log delta.""" - self.process_event( - PipelineEvent( - PipelineEventType.INTENT_PROGRESS, - { - "chat_log_delta": delta, - }, - ) - ) - if tts_input_stream is None: - return - - nonlocal chat_log_role - - if role := delta.get("role"): - chat_log_role = role - - # We are only interested in assistant deltas - if chat_log_role != "assistant": - return - - if content := delta.get("content"): - tts_input_stream.put_nowait(content) - - if self._streamed_response_text: - return - - nonlocal delta_character_count - - # Streamed responses are not cached. That's why we - # only start streaming text after we have received - # enough characters that indicates it will be a long - # response or if we have received text, and then a - # tool call. - - # Tool call after we already received text - start_streaming = delta_character_count > 0 and delta.get("tool_calls") - - # Count characters in the content and test if we - # exceed streaming threshold - if not start_streaming and content: - delta_character_count += len(content) - start_streaming = delta_character_count > STREAM_RESPONSE_CHARS - - if not start_streaming: - return - - self._streamed_response_text = True - - self.process_event( - PipelineEvent( - PipelineEventType.INTENT_PROGRESS, - { - "tts_start_streaming": True, - }, - ) - ) - - async def tts_input_stream_generator() -> AsyncGenerator[str]: - """Yield TTS input stream.""" - while (tts_input := await tts_input_stream.get()) is not None: - yield tts_input - - # Concatenate all existing queue items - parts = [] - while not tts_input_stream.empty(): - parts.append(tts_input_stream.get_nowait()) - tts_input_stream.put_nowait( - "".join( - # At this point parts is only strings, - # None indicates end of queue - cast(list[str], parts) - ) - ) - - assert self.tts_stream is not None - self.tts_stream.async_set_message_stream(tts_input_stream_generator()) - - user_input = conversation.ConversationInput( - text=intent_input, - context=self.context, - conversation_id=conversation_id, - device_id=self._device_id, - satellite_id=self._satellite_id, - language=input_language, - agent_id=self.intent_agent.id, - extra_system_prompt=conversation_extra_system_prompt, - ) - - with ( - chat_session.async_get_chat_session( - self.hass, user_input.conversation_id - ) as session, - conversation.async_get_chat_log( - self.hass, - session, - user_input, - chat_log_delta_listener=chat_log_delta_listener, - ) as chat_log, - ): - agent_id = self.intent_agent.id - processed_locally = agent_id == conversation.HOME_ASSISTANT_AGENT - all_targets_in_satellite_area = False - intent_response: intent.IntentResponse | None = None - if not processed_locally and not self._intent_agent_only: - # Sentence triggers override conversation agent - if ( - trigger_response_text - := await conversation.async_handle_sentence_triggers( - self.hass, user_input, chat_log - ) - ) is not None: - # Sentence trigger matched - agent_id = "sentence_trigger" - processed_locally = True - intent_response = intent.IntentResponse( - self.pipeline.conversation_language - ) - intent_response.async_set_speech(trigger_response_text) - - intent_filter: Callable[[RecognizeResult], bool] | None = None - # If the LLM has API access, we filter out some sentences that are - # interfering with LLM operation. - if ( - intent_agent_state := self.hass.states.get(self.intent_agent.id) - ) and intent_agent_state.attributes.get( - EntityStateAttribute.SUPPORTED_FEATURES, 0 - ) & conversation.ConversationEntityFeature.CONTROL: - intent_filter = _async_local_fallback_intent_filter - - # Try local intents - if ( - intent_response is None - and self.pipeline.prefer_local_intents - and ( - intent_response := await conversation.async_handle_intents( - self.hass, - user_input, - chat_log, - intent_filter=intent_filter, - ) - ) - ): - # Local intent matched - agent_id = conversation.HOME_ASSISTANT_AGENT - processed_locally = True - - # It was already handled, create response and add to chat history - if intent_response is not None: - speech: str = intent_response.speech.get("plain", {}).get( - "speech", "" - ) - chat_log.async_add_assistant_content_without_tools( - conversation.AssistantContent( - agent_id=agent_id, - content=speech, - ) - ) - conversation_result = conversation.ConversationResult( - response=intent_response, - conversation_id=session.conversation_id, - ) - - else: - # Fall back to pipeline conversation agent - conversation_result = await conversation.async_converse( - hass=self.hass, - text=user_input.text, - conversation_id=user_input.conversation_id, - device_id=user_input.device_id, - satellite_id=user_input.satellite_id, - context=user_input.context, - language=user_input.language, - agent_id=user_input.agent_id, - extra_system_prompt=user_input.extra_system_prompt, - ) - speech = conversation_result.response.speech.get("plain", {}).get( - "speech", "" - ) - if tts_input_stream and self._streamed_response_text: - tts_input_stream.put_nowait(None) - - if agent_id == conversation.HOME_ASSISTANT_AGENT: - # Check if all targeted entities were in the same area as - # the satellite device. - # If so, the satellite should respond with an acknowledge beep - # instead of a full response. - all_targets_in_satellite_area = ( - self._get_all_targets_in_satellite_area( - conversation_result.response, - self._satellite_id, - self._device_id, - ) - ) - - except Exception as src_error: - _LOGGER.exception("Unexpected error during intent recognition") - raise IntentRecognitionError( - code="intent-failed", - message="Unexpected error during intent recognition", - ) from src_error - - _LOGGER.debug("conversation result %s", conversation_result) - - self.process_event( - PipelineEvent( - PipelineEventType.INTENT_END, - { - "processed_locally": processed_locally, - "intent_output": conversation_result.as_dict(), - }, - ) - ) - - if conversation_result.continue_conversation: - self._conversation_data.continue_conversation_agent = agent_id - - return (speech, all_targets_in_satellite_area) - - def _get_all_targets_in_satellite_area( - self, - intent_response: intent.IntentResponse, - satellite_id: str | None, - device_id: str | None, - ) -> bool: - """Return true if all targeted entities were in the same area as the device.""" - if ( - intent_response.response_type is not intent.IntentResponseType.ACTION_DONE - or not intent_response.matched_states - ): - return False - - entity_registry = er.async_get(self.hass) - device_registry = dr.async_get(self.hass) - - area_id: str | None = None - - if ( - satellite_id is not None - and (target_entity_entry := entity_registry.async_get(satellite_id)) - is not None - ): - area_id = target_entity_entry.area_id - device_id = target_entity_entry.device_id - - if area_id is None: - if device_id is None: - return False - - device_entry = device_registry.async_get(device_id) - if device_entry is None: - return False - - area_id = dr.async_get_effective_area_id(self.hass, device_entry) - if area_id is None: - return False - - for state in intent_response.matched_states: - target_entity_entry = entity_registry.async_get(state.entity_id) - if target_entity_entry is None: - return False - - target_area_id = target_entity_entry.area_id - if target_area_id is None: - if target_entity_entry.device_id is None: - return False - - target_device_entry = device_registry.async_get( - target_entity_entry.device_id - ) - if target_device_entry is None: - return False - - target_area_id = dr.async_get_effective_area_id( - self.hass, target_device_entry - ) - - if target_area_id != area_id: - return False - - return True - - async def prepare_text_to_speech(self) -> None: - """Prepare text-to-speech.""" - # pipeline.tts_engine can't be None or this function is not called - engine = cast(str, self.pipeline.tts_engine) - - tts_options: dict[str, Any] = {} - if self.pipeline.tts_voice is not None: - tts_options[tts.ATTR_VOICE] = self.pipeline.tts_voice - - if isinstance(self.tts_audio_output, dict): - tts_options.update(self.tts_audio_output) - elif isinstance(self.tts_audio_output, str): - tts_options[tts.ATTR_PREFERRED_FORMAT] = self.tts_audio_output - if self.tts_audio_output == "wav": - # 16 Khz, 16-bit mono - tts_options[tts.ATTR_PREFERRED_SAMPLE_RATE] = SAMPLE_RATE - tts_options[tts.ATTR_PREFERRED_SAMPLE_CHANNELS] = SAMPLE_CHANNELS - tts_options[tts.ATTR_PREFERRED_SAMPLE_BYTES] = SAMPLE_WIDTH - - try: - self.tts_stream = tts.async_create_stream( - hass=self.hass, - engine=engine, - language=self.pipeline.tts_language, - options=tts_options, - ) - except HomeAssistantError as err: - raise TextToSpeechError( - code="tts-not-supported", - message=( - f"Text-to-speech engine {engine} " - f"does not support language {self.pipeline.tts_language}" - f" or options {tts_options}:" - f" {err}" - ), - ) from err - - async def text_to_speech( - self, tts_input: str, override_media_path: Path | None = None - ) -> None: - """Run text-to-speech portion of pipeline.""" - assert self.tts_stream is not None - - self.process_event( - PipelineEvent( - PipelineEventType.TTS_START, - { - "engine": self.tts_stream.engine, - "language": self.pipeline.tts_language, - "voice": self.pipeline.tts_voice, - "tts_input": tts_input, - "acknowledge_override": override_media_path is not None, - }, - ) - ) - - if override_media_path: - self.tts_stream.async_override_result(override_media_path) - elif not self._streamed_response_text: - self.tts_stream.async_set_message(tts_input) - - tts_output = { - "media_id": self.tts_stream.media_source_id, - "token": self.tts_stream.token, - "url": self.tts_stream.url, - "mime_type": self.tts_stream.content_type, - } - - self.process_event( - PipelineEvent(PipelineEventType.TTS_END, {"tts_output": tts_output}) - ) - - def _capture_chunk(self, audio_bytes: bytes | None) -> None: - """Forward audio chunk to various capturing mechanisms.""" - if self.debug_recording_queue is not None: - # Forward to debug WAV file recording - self.debug_recording_queue.put_nowait(audio_bytes) - - if self._device_id is None: - return - - # Forward to device audio capture - pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE] - audio_queue = pipeline_data.device_audio_queues.get(self._device_id) - if audio_queue is None: - return - - try: - audio_queue.queue.put_nowait(audio_bytes) - except asyncio.QueueFull: - audio_queue.overflow = True - _LOGGER.warning("Audio queue full for device %s", self._device_id) - - def _start_debug_recording_thread(self) -> None: - """Start thread to record wake/stt audio if debug_recording_dir is set.""" - assert self.debug_recording_thread is None - - # Directory to save audio for each pipeline run. - # Configured in YAML for assist_pipeline. - if debug_recording_dir := self.hass.data[DATA_CONFIG].get( - CONF_DEBUG_RECORDING_DIR - ): - if self._device_id is None: - # // - run_recording_dir = ( - Path(debug_recording_dir) - / self.pipeline.name - / str(time.monotonic_ns()) - ) - else: - # /// - run_recording_dir = ( - Path(debug_recording_dir) - / self._device_id - / self.pipeline.name - / str(time.monotonic_ns()) - ) - - self.debug_recording_queue = Queue() - self.debug_recording_thread = Thread( - target=_pipeline_debug_recording_thread_proc, - args=(run_recording_dir, self.debug_recording_queue), - daemon=True, - ) - self.debug_recording_thread.start() - - async def _stop_debug_recording_thread(self) -> None: - """Stop recording thread.""" - if (self.debug_recording_thread is None) or ( - self.debug_recording_queue is None - ): - # Not running - return - - # NOTE: Expecting a None to have been put in self.debug_recording_queue - # in self.end() to signal the thread to stop. - - # Wait until the thread has finished to ensure that files are fully written - await self.hass.async_add_executor_job(self.debug_recording_thread.join) - - self.debug_recording_queue = None - self.debug_recording_thread = None - - async def process_volume_only( - self, audio_stream: AsyncIterable[bytes] - ) -> AsyncGenerator[EnhancedAudioChunk]: - """Apply volume transformation only with optional chunking. - - No VAD/audio enhancements are applied. - """ - timestamp_ms = 0 - async for chunk in audio_stream: - if self.audio_settings.volume_multiplier != 1.0: - chunk = _multiply_volume(chunk, self.audio_settings.volume_multiplier) - - for sub_chunk in chunk_samples( - chunk, BYTES_PER_CHUNK, self.audio_chunking_buffer - ): - yield EnhancedAudioChunk( - audio=sub_chunk, - timestamp_ms=timestamp_ms, - speech_probability=None, # no VAD - ) - timestamp_ms += MS_PER_CHUNK - - async def process_enhance_audio( - self, audio_stream: AsyncIterable[bytes] - ) -> AsyncGenerator[EnhancedAudioChunk]: - """Split audio into chunks and apply audio enhancements. - - Applies VAD/noise suppression/auto gain/volume - transformation. - """ - assert self.audio_enhancer is not None - - timestamp_ms = 0 - async for dirty_samples in audio_stream: - if self.audio_settings.volume_multiplier != 1.0: - # Static gain - dirty_samples = _multiply_volume( - dirty_samples, self.audio_settings.volume_multiplier - ) - - # Split into chunks for audio enhancements/VAD - for dirty_chunk in chunk_samples( - dirty_samples, BYTES_PER_CHUNK, self.audio_chunking_buffer - ): - yield self.audio_enhancer.enhance_chunk(dirty_chunk, timestamp_ms) - timestamp_ms += MS_PER_CHUNK - - -def _multiply_volume(chunk: bytes, volume_multiplier: float) -> bytes: - """Multiplies 16-bit PCM samples by a constant.""" - - def _clamp(val: float) -> float: - """Clamp to signed 16-bit.""" - return max(-32768, min(32767, val)) - - return array.array( - "h", - (int(_clamp(value * volume_multiplier)) for value in array.array("h", chunk)), - ).tobytes() - - -def _pipeline_debug_recording_thread_proc( - run_recording_dir: Path, - queue: Queue[str | bytes | None], - message_timeout: float = 5, -) -> None: - wav_writer: wave.Wave_write | None = None - - try: - _LOGGER.debug("Saving wake/stt audio to %s", run_recording_dir) - run_recording_dir.mkdir(parents=True, exist_ok=True) - - while True: - message = queue.get(timeout=message_timeout) - if message is None: - # Stop signal - break - - if isinstance(message, str): - # New WAV file name - if wav_writer is not None: - wav_writer.close() - - wav_path = run_recording_dir / f"{message}.wav" - wav_writer = wave.open(str(wav_path), "wb") - wav_writer.setframerate(SAMPLE_RATE) - wav_writer.setsampwidth(SAMPLE_WIDTH) - wav_writer.setnchannels(SAMPLE_CHANNELS) - elif isinstance(message, bytes): - # Chunk of 16-bit mono audio at 16Khz - if wav_writer is not None: - wav_writer.writeframes(message) - except Empty: - pass # occurs when pipeline has unexpected error - except Exception: - _LOGGER.exception("Unexpected error in debug recording thread") - finally: - if wav_writer is not None: - wav_writer.close() - - -@dataclass(kw_only=True) -class PipelineInput: - """Input to a pipeline run.""" - - run: PipelineRun - - session: chat_session.ChatSession - """Session for the conversation.""" - - stt_metadata: stt.SpeechMetadata | None = None - """Metadata of stt input audio. Required when start_stage = stt.""" - - stt_stream: AsyncIterable[bytes] | None = None - """Input audio for stt. Required when start_stage = stt.""" - - wake_word_phrase: str | None = None - """Optional key used to de-duplicate wake-ups for local wake word detection.""" - - intent_input: str | None = None - """Input for conversation agent. Required when start_stage = intent.""" - - tts_input: str | None = None - """Input for text-to-speech. Required when start_stage = tts.""" - - conversation_extra_system_prompt: str | None = None - """Extra prompt information for the conversation agent.""" - - device_id: str | None = None - """Identifier of the device that is processing the input/output of the pipeline.""" - - satellite_id: str | None = None - """Identifier of the satellite processing the pipeline.""" - - async def execute(self, validate: bool = False) -> None: - """Run pipeline.""" - validation_error: PipelineError | None = None - if validate: - try: - await self.validate() - except PipelineError as err: - validation_error = err - - self.run.start( - conversation_id=self.session.conversation_id, - device_id=self.device_id, - satellite_id=self.satellite_id, - ) - current_stage: PipelineStage | None = self.run.start_stage - - try: - if validation_error is not None: - raise validation_error - - stt_audio_buffer: list[EnhancedAudioChunk] = [] - stt_processed_stream: AsyncIterable[EnhancedAudioChunk] | None = None - - if self.stt_stream is not None: - if self.run.audio_settings.needs_processor: - # VAD/noise suppression/auto gain/volume - stt_processed_stream = self.run.process_enhance_audio( - self.stt_stream - ) - else: - # Volume multiplier only - stt_processed_stream = self.run.process_volume_only(self.stt_stream) - - if current_stage == PipelineStage.WAKE_WORD: - # wake-word-detection - assert stt_processed_stream is not None - detect_result = await self.run.wake_word_detection( - stt_processed_stream, stt_audio_buffer - ) - if detect_result is None: - # No wake word. Abort the rest of the pipeline. - return - - current_stage = PipelineStage.STT - - # speech-to-text - intent_input = self.intent_input - if current_stage == PipelineStage.STT: - assert self.stt_metadata is not None - assert stt_processed_stream is not None - - if self.wake_word_phrase is not None: - # Avoid duplicate wake-ups by checking cooldown - last_wake_up = self.run.hass.data[DATA_LAST_WAKE_UP].get( - self.wake_word_phrase - ) - if last_wake_up is not None: - sec_since_last_wake_up = time.monotonic() - last_wake_up - if sec_since_last_wake_up < WAKE_WORD_COOLDOWN: - _LOGGER.debug( - "Speech-to-text cancelled to avoid" - " duplicate wake-up for %s", - self.wake_word_phrase, - ) - raise DuplicateWakeUpDetectedError(self.wake_word_phrase) - - # Record last wake up time to block duplicate detections - self.run.hass.data[DATA_LAST_WAKE_UP][self.wake_word_phrase] = ( - time.monotonic() - ) - - stt_input_stream = stt_processed_stream - - if stt_audio_buffer: - # Send audio in the buffer first to speech-to-text, - # then move on to stt_stream. - # This is basically an async itertools.chain. - async def buffer_then_audio_stream() -> AsyncGenerator[ - EnhancedAudioChunk - ]: - # Buffered audio - for chunk in stt_audio_buffer: - yield chunk - - # Streamed audio - assert stt_processed_stream is not None - async for chunk in stt_processed_stream: - yield chunk - - stt_input_stream = buffer_then_audio_stream() - - intent_input = await self.run.speech_to_text( - self.stt_metadata, - stt_input_stream, - ) - current_stage = PipelineStage.INTENT - - if self.run.end_stage != PipelineStage.STT: - tts_input = self.tts_input - all_targets_in_satellite_area = False - - if current_stage == PipelineStage.INTENT: - # intent-recognition - assert intent_input is not None - ( - tts_input, - all_targets_in_satellite_area, - ) = await self.run.recognize_intent( - intent_input, - self.session.conversation_id, - self.conversation_extra_system_prompt, - ) - if all_targets_in_satellite_area or tts_input.strip(): - current_stage = PipelineStage.TTS - else: - # Skip TTS - current_stage = PipelineStage.END - - if self.run.end_stage != PipelineStage.INTENT: - # text-to-speech - if current_stage == PipelineStage.TTS: - if all_targets_in_satellite_area: - # Use acknowledge media instead of full response - await self.run.text_to_speech( - tts_input or "", override_media_path=ACKNOWLEDGE_PATH - ) - else: - assert tts_input is not None - await self.run.text_to_speech(tts_input) - - except PipelineError as err: - if self.run.tts_stream: - # Clean up TTS stream - self.run.tts_stream.delete() - self.run.tts_stream = None - - self.run.process_event( - PipelineEvent( - PipelineEventType.ERROR, - {"code": err.code, "message": err.message}, - ) - ) - finally: - # Always end the run since it needs to shut down the debug recording - # thread, etc. - await self.run.end() - - async def validate(self) -> None: - """Validate pipeline input against start stage.""" - if self.run.start_stage in (PipelineStage.WAKE_WORD, PipelineStage.STT): - if self.run.pipeline.stt_engine is None: - raise PipelineRunValidationError( - "the pipeline does not support speech-to-text" - ) - if self.stt_metadata is None: - raise PipelineRunValidationError( - "stt_metadata is required for speech-to-text" - ) - if self.stt_stream is None: - raise PipelineRunValidationError( - "stt_stream is required for speech-to-text" - ) - elif self.run.start_stage == PipelineStage.INTENT: - if self.intent_input is None: - raise PipelineRunValidationError( - "intent_input is required for intent recognition" - ) - elif self.run.start_stage == PipelineStage.TTS: - if self.tts_input is None: - raise PipelineRunValidationError( - "tts_input is required for text-to-speech" - ) - if self.run.end_stage == PipelineStage.TTS: - if self.run.pipeline.tts_engine is None: - raise PipelineRunValidationError( - "the pipeline does not support text-to-speech" - ) - - start_stage_index = PIPELINE_STAGE_ORDER.index(self.run.start_stage) - end_stage_index = PIPELINE_STAGE_ORDER.index(self.run.end_stage) - - prepare_tasks = [] - - if ( - start_stage_index - <= PIPELINE_STAGE_ORDER.index(PipelineStage.WAKE_WORD) - <= end_stage_index - ): - prepare_tasks.append(self.run.prepare_wake_word_detection()) - - if ( - start_stage_index - <= PIPELINE_STAGE_ORDER.index(PipelineStage.STT) - <= end_stage_index - ): - # self.stt_metadata can't be None or we'd raise above - prepare_tasks.append(self.run.prepare_speech_to_text(self.stt_metadata)) # type: ignore[arg-type] - - if ( - start_stage_index - <= PIPELINE_STAGE_ORDER.index(PipelineStage.INTENT) - <= end_stage_index - ): - prepare_tasks.append(self.run.prepare_recognize_intent(self.session)) - - if prepare_tasks: - await asyncio.gather(*prepare_tasks) - - # Do TTS prepare separately so we don't create a ResultStream if the - # pipeline is invalid. - if ( - start_stage_index - <= PIPELINE_STAGE_ORDER.index(PipelineStage.TTS) - <= end_stage_index - ): - await self.run.prepare_text_to_speech() - - class PipelinePreferred(CollectionError): """Raised when attempting to delete the preferred pipelen.""" @@ -2003,37 +566,3 @@ async def async_setup_pipeline_store(hass: HomeAssistant) -> PipelineData: PIPELINE_FIELDS, ).async_setup(hass) return PipelineData(pipeline_store) - - -@dataclass -class PipelineConversationData: - """Hold data for the duration of a conversation.""" - - continue_conversation_agent: str | None = None - """The agent that requested the conversation to be continued.""" - - -@callback -def async_get_pipeline_conversation_data( - hass: HomeAssistant, session: chat_session.ChatSession -) -> PipelineConversationData: - """Get the pipeline data for a specific conversation.""" - all_conversation_data = hass.data.get(KEY_PIPELINE_CONVERSATION_DATA) - if all_conversation_data is None: - all_conversation_data = {} - hass.data[KEY_PIPELINE_CONVERSATION_DATA] = all_conversation_data - - data = all_conversation_data.get(session.conversation_id) - - if data is not None: - return data - - @callback - def do_cleanup() -> None: - """Handle cleanup.""" - all_conversation_data.pop(session.conversation_id) - - session.async_on_cleanup(do_cleanup) - - data = all_conversation_data[session.conversation_id] = PipelineConversationData() - return data diff --git a/homeassistant/components/assist_pipeline/run.py b/homeassistant/components/assist_pipeline/run.py new file mode 100644 index 000000000000..f45245de59f2 --- /dev/null +++ b/homeassistant/components/assist_pipeline/run.py @@ -0,0 +1,454 @@ +"""Run controller for Assist pipelines.""" + +import asyncio +from collections.abc import AsyncIterable +from dataclasses import dataclass, field +import logging +from pathlib import Path +from queue import Empty, Queue +from threading import Thread +import time +from typing import Any, Protocol, override +import wave + +from homeassistant.components import stt +from homeassistant.core import Context, HomeAssistant, callback +from homeassistant.helpers import chat_session +from homeassistant.util import ulid as ulid_util +from homeassistant.util.limited_size_dict import LimitedSizeDict + +from .const import ( + CONF_DEBUG_RECORDING_DIR, + DATA_CONFIG, + DATA_LAST_WAKE_UP, + SAMPLE_CHANNELS, + SAMPLE_RATE, + SAMPLE_WIDTH, + WAKE_WORD_COOLDOWN, +) +from .default_pipeline import _DefaultPipelineProcessor +from .error import ( + DuplicateWakeUpDetectedError, + InvalidPipelineStagesError, + PipelineError, +) +from .models import ( + PIPELINE_STAGE_ORDER, + AudioSettings, + Pipeline, + PipelineEvent, + PipelineEventCallback, + PipelineEventType, + PipelineStage, + WakeWordSettings, +) +from .runtime import KEY_ASSIST_PIPELINE, PipelineRunDebug + +_LOGGER = logging.getLogger(__name__) + +STORED_PIPELINE_RUNS = 10 + + +class _PipelineResponseAudio(Protocol): + """Response audio metadata exposed by a pipeline processor.""" + + @property + def token(self) -> str: + """Return the response audio token.""" + + @property + def url(self) -> str: + """Return the response audio URL.""" + + @property + def content_type(self) -> str: + """Return the response audio content type.""" + + +class _PipelineProcessor(Protocol): + """Implementation boundary for processing a pipeline run.""" + + @property + def response_audio(self) -> _PipelineResponseAudio | None: + """Return the response audio stream.""" + + @property + def supports_streaming_response(self) -> bool | None: + """Return whether response audio can be streamed.""" + + async def async_validate(self, request: _PipelineProcessorRequest) -> None: + """Validate pipeline input and prepare processing resources.""" + + async def async_execute(self, request: _PipelineProcessorRequest) -> None: + """Process pipeline input.""" + + def invalidate(self) -> None: + """Invalidate active processing.""" + + def cleanup(self) -> None: + """Clean up resources after a pipeline error.""" + + +@dataclass(frozen=True, slots=True, kw_only=True) +class _PipelineProcessorRequest: + """Input passed across the private pipeline processor boundary.""" + + session: chat_session.ChatSession + stt_metadata: stt.SpeechMetadata | None = None + stt_stream: AsyncIterable[bytes] | None = None + wake_word_phrase: str | None = None + intent_input: str | None = None + tts_input: str | None = None + conversation_extra_system_prompt: str | None = None + device_id: str | None = None + satellite_id: str | None = None + + +@dataclass +class PipelineRun: + """Home Assistant controller for a pipeline run.""" + + hass: HomeAssistant + context: Context + pipeline: Pipeline + start_stage: PipelineStage + end_stage: PipelineStage + event_callback: PipelineEventCallback + language: str = None # type: ignore[assignment] + runner_data: Any | None = None + tts_audio_output: str | dict[str, Any] | None = None + wake_word_settings: WakeWordSettings | None = None + audio_settings: AudioSettings = field(default_factory=AudioSettings) + + id: str = field(default_factory=ulid_util.ulid_now) + debug_recording_thread: Thread | None = None + """Thread that records audio to debug_recording_dir.""" + debug_recording_queue: Queue[str | bytes | None] | None = None + """Queue to communicate with the debug recording thread.""" + _device_id: str | None = None + _satellite_id: str | None = None + _processor: _PipelineProcessor = field(init=False, repr=False) + _registered: bool = field(init=False, default=False, repr=False) + _started: bool = field(init=False, default=False, repr=False) + _ended: bool = field(init=False, default=False, repr=False) + + def __post_init__(self) -> None: + """Initialize the pipeline controller.""" + self.language = self.pipeline.language or self.hass.config.language + + if PIPELINE_STAGE_ORDER.index(self.end_stage) < PIPELINE_STAGE_ORDER.index( + self.start_stage + ): + raise InvalidPipelineStagesError(self.start_stage, self.end_stage) + + self._processor = _create_pipeline_processor(self) + pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE] + if self.pipeline.id not in pipeline_data.pipeline_debug: + pipeline_data.pipeline_debug[self.pipeline.id] = LimitedSizeDict( + size_limit=STORED_PIPELINE_RUNS + ) + pipeline_data.pipeline_debug[self.pipeline.id][self.id] = PipelineRunDebug() + pipeline_data.pipeline_runs.add_run(self) + self._registered = True + + @override + def __eq__(self, other: object) -> bool: + """Compare pipeline runs by id.""" + if isinstance(other, PipelineRun): + return self.id == other.id + return False + + @property + def device_id(self) -> str | None: + """Return the device associated with the run.""" + return self._device_id + + @property + def satellite_id(self) -> str | None: + """Return the satellite associated with the run.""" + return self._satellite_id + + @callback + def process_event(self, event: PipelineEvent) -> None: + """Log an event and call the listener.""" + self.event_callback(event) + pipeline_data = self.hass.data[KEY_ASSIST_PIPELINE] + if self.id not in pipeline_data.pipeline_debug[self.pipeline.id]: + return + pipeline_data.pipeline_debug[self.pipeline.id][self.id].events.append(event) + + def start( + self, conversation_id: str, device_id: str | None, satellite_id: str | None + ) -> None: + """Emit the run-start event.""" + if self._started: + raise RuntimeError("Pipeline run has already started") + + self._started = True + self._device_id = device_id + self._satellite_id = satellite_id + if self.start_stage in (PipelineStage.WAKE_WORD, PipelineStage.STT): + self._start_debug_recording_thread() + + data: dict[str, Any] = { + "pipeline": self.pipeline.id, + "language": self.language, + "conversation_id": conversation_id, + } + if satellite_id is not None: + data["satellite_id"] = satellite_id + if self.runner_data is not None: + data["runner_data"] = self.runner_data + if (response_audio := self._processor.response_audio) is not None: + data["tts_output"] = { + "token": response_audio.token, + "url": response_audio.url, + "mime_type": response_audio.content_type, + "stream_response": self._processor.supports_streaming_response, + } + self.process_event(PipelineEvent(PipelineEventType.RUN_START, data)) + + async def end(self) -> None: + """Emit the run-end event.""" + if self._ended: + return + + if not self._started: + self._ended = True + self._unregister() + return + + self._ended = True + try: + self.capture_audio(None) + await self._stop_debug_recording_thread() + self.process_event(PipelineEvent(PipelineEventType.RUN_END)) + finally: + self._unregister() + + async def async_validate(self, pipeline_input: PipelineInput) -> None: + """Validate input and prepare the pipeline processor.""" + request = pipeline_input.create_processor_request() + self._set_request_identity(request) + try: + await self._processor.async_validate(request) + except BaseException: + self._cleanup_failed_processor() + self._unregister() + raise + + async def async_execute( + self, pipeline_input: PipelineInput, *, validate: bool = False + ) -> None: + """Run the pipeline processor with the Home Assistant lifecycle.""" + request = pipeline_input.create_processor_request() + validation_error: PipelineError | None = None + self._set_request_identity(request) + + try: + if validate: + try: + await self._processor.async_validate(request) + except PipelineError as err: + validation_error = err + + self.start( + conversation_id=request.session.conversation_id, + device_id=request.device_id, + satellite_id=request.satellite_id, + ) + await self._async_process(request, validation_error) + except PipelineError as err: + self._cleanup_failed_processor() + self.process_event( + PipelineEvent( + PipelineEventType.ERROR, + {"code": err.code, "message": err.message}, + ) + ) + except BaseException: + self._cleanup_failed_processor() + raise + finally: + await self.end() + + async def _async_process( + self, + request: _PipelineProcessorRequest, + validation_error: PipelineError | None, + ) -> None: + """Execute a previously validated processor request.""" + if validation_error is not None: + raise validation_error + await self._processor.async_execute(request) + + @callback + def _set_request_identity(self, request: _PipelineProcessorRequest) -> None: + """Set identity used by controller-provided processor services.""" + self._device_id = request.device_id + self._satellite_id = request.satellite_id + + @callback + def _cleanup_failed_processor(self) -> None: + """Clean up processor resources after an error.""" + self._processor.cleanup() + + @callback + def _unregister(self) -> None: + """Remove this run from the active run registry.""" + if self._registered: + self.hass.data[KEY_ASSIST_PIPELINE].pipeline_runs.remove_run(self) + self._registered = False + + @callback + def invalidate(self) -> None: + """Invalidate active processing for this run.""" + self._processor.invalidate() + + @callback + def accept_wake_word(self, wake_word_phrase: str) -> None: + """Apply Home Assistant's duplicate wake-up policy.""" + last_wake_up = self.hass.data[DATA_LAST_WAKE_UP].get(wake_word_phrase) + if ( + last_wake_up is not None + and (time.monotonic() - last_wake_up) < WAKE_WORD_COOLDOWN + ): + _LOGGER.debug("Duplicate wake-up detected for %s", wake_word_phrase) + raise DuplicateWakeUpDetectedError(wake_word_phrase) + self.hass.data[DATA_LAST_WAKE_UP][wake_word_phrase] = time.monotonic() + + @callback + def capture_audio(self, audio_bytes: bytes | None) -> None: + """Forward an audio chunk to Home Assistant capture mechanisms.""" + if self.debug_recording_queue is not None: + self.debug_recording_queue.put_nowait(audio_bytes) + + if self._device_id is None: + return + audio_queue = self.hass.data[KEY_ASSIST_PIPELINE].device_audio_queues.get( + self._device_id + ) + if audio_queue is None: + return + try: + audio_queue.queue.put_nowait(audio_bytes) + except asyncio.QueueFull: + audio_queue.overflow = True + _LOGGER.warning("Audio queue full for device %s", self._device_id) + + @callback + def start_debug_recording(self, name: str) -> None: + """Start a new debug WAV file if recording is active.""" + if self.debug_recording_queue is not None: + self.debug_recording_queue.put_nowait(name) + + def _start_debug_recording_thread(self) -> None: + """Start the debug recording thread if configured.""" + assert self.debug_recording_thread is None + if debug_recording_dir := self.hass.data[DATA_CONFIG].get( + CONF_DEBUG_RECORDING_DIR + ): + if self._device_id is None: + run_recording_dir = ( + Path(debug_recording_dir) + / self.pipeline.name + / str(time.monotonic_ns()) + ) + else: + run_recording_dir = ( + Path(debug_recording_dir) + / self._device_id + / self.pipeline.name + / str(time.monotonic_ns()) + ) + self.debug_recording_queue = Queue() + self.debug_recording_thread = Thread( + target=_pipeline_debug_recording_thread_proc, + args=(run_recording_dir, self.debug_recording_queue), + daemon=True, + ) + self.debug_recording_thread.start() + + async def _stop_debug_recording_thread(self) -> None: + """Stop the debug recording thread.""" + if self.debug_recording_thread is None or self.debug_recording_queue is None: + return + await self.hass.async_add_executor_job(self.debug_recording_thread.join) + self.debug_recording_queue = None + self.debug_recording_thread = None + + +@dataclass(kw_only=True) +class PipelineInput: + """Input to a pipeline run.""" + + run: PipelineRun + session: chat_session.ChatSession + stt_metadata: stt.SpeechMetadata | None = None + stt_stream: AsyncIterable[bytes] | None = None + wake_word_phrase: str | None = None + intent_input: str | None = None + tts_input: str | None = None + conversation_extra_system_prompt: str | None = None + device_id: str | None = None + satellite_id: str | None = None + + async def execute(self, validate: bool = False) -> None: + """Run pipeline.""" + await self.run.async_execute(self, validate=validate) + + async def validate(self) -> None: + """Validate pipeline input against start stage.""" + await self.run.async_validate(self) + + def create_processor_request(self) -> _PipelineProcessorRequest: + """Create the private request passed to the pipeline processor.""" + return _PipelineProcessorRequest( + session=self.session, + stt_metadata=self.stt_metadata, + stt_stream=self.stt_stream, + wake_word_phrase=self.wake_word_phrase, + intent_input=self.intent_input, + tts_input=self.tts_input, + conversation_extra_system_prompt=self.conversation_extra_system_prompt, + device_id=self.device_id, + satellite_id=self.satellite_id, + ) + + +def _create_pipeline_processor(run: PipelineRun) -> _PipelineProcessor: + """Create the default pipeline processor.""" + return _DefaultPipelineProcessor(run) + + +def _pipeline_debug_recording_thread_proc( + run_recording_dir: Path, + queue: Queue[str | bytes | None], + message_timeout: float = 5, +) -> None: + """Write pipeline audio to debug WAV files.""" + wav_writer: wave.Wave_write | None = None + try: + _LOGGER.debug("Saving wake/stt audio to %s", run_recording_dir) + run_recording_dir.mkdir(parents=True, exist_ok=True) + while True: + message = queue.get(timeout=message_timeout) + if message is None: + break + if isinstance(message, str): + if wav_writer is not None: + wav_writer.close() + wav_path = run_recording_dir / f"{message}.wav" + wav_writer = wave.open(str(wav_path), "wb") + wav_writer.setframerate(SAMPLE_RATE) + wav_writer.setsampwidth(SAMPLE_WIDTH) + wav_writer.setnchannels(SAMPLE_CHANNELS) + elif isinstance(message, bytes) and wav_writer is not None: + wav_writer.writeframes(message) + except Empty: + pass + except Exception: + _LOGGER.exception("Unexpected error in debug recording thread") + finally: + if wav_writer is not None: + wav_writer.close() diff --git a/homeassistant/components/assist_pipeline/runtime.py b/homeassistant/components/assist_pipeline/runtime.py index b38ea32562fa..e04e8e1030a6 100644 --- a/homeassistant/components/assist_pipeline/runtime.py +++ b/homeassistant/components/assist_pipeline/runtime.py @@ -14,7 +14,8 @@ from .const import DOMAIN from .models import PipelineEvent if TYPE_CHECKING: - from .pipeline import PipelineRun, PipelineStorageCollection + from .pipeline import PipelineStorageCollection + from .run import PipelineRun class PipelineRuns: @@ -45,7 +46,7 @@ class PipelineRuns: if pipeline_runs := self._pipeline_runs.get(item_id): # Create a temporary list in case the list is modified while we iterate for pipeline_run in list(pipeline_runs.values()): - pipeline_run.abort_wake_word_detection = True + pipeline_run.invalidate() @dataclass(slots=True) diff --git a/tests/components/assist_pipeline/test_init.py b/tests/components/assist_pipeline/test_init.py index f2a28ab5738d..962a7204bddc 100644 --- a/tests/components/assist_pipeline/test_init.py +++ b/tests/components/assist_pipeline/test_init.py @@ -672,7 +672,7 @@ async def test_pipeline_saved_audio_empty_queue( # Wrap original function to time out immediately _pipeline_debug_recording_thread_proc = ( - assist_pipeline.pipeline._pipeline_debug_recording_thread_proc + assist_pipeline.run._pipeline_debug_recording_thread_proc ) def proc_wrapper(run_recording_dir, queue): @@ -686,7 +686,7 @@ async def test_pipeline_saved_audio_empty_queue( ) with patch( - "homeassistant.components.assist_pipeline.pipeline._pipeline_debug_recording_thread_proc", + "homeassistant.components.assist_pipeline.run._pipeline_debug_recording_thread_proc", proc_wrapper, ): await assist_pipeline.async_pipeline_from_audio_stream( diff --git a/tests/components/assist_pipeline/test_pipeline.py b/tests/components/assist_pipeline/test_pipeline.py index 06a6da97092c..99044dd9437f 100644 --- a/tests/components/assist_pipeline/test_pipeline.py +++ b/tests/components/assist_pipeline/test_pipeline.py @@ -1,6 +1,7 @@ """Websocket tests for Voice Assistant integration.""" from collections.abc import AsyncGenerator, Generator +from dataclasses import FrozenInstanceError from pathlib import Path from typing import Any from unittest.mock import ANY, AsyncMock, Mock, patch @@ -25,6 +26,10 @@ from homeassistant.components.assist_pipeline.const import ( DATA_CONFIG, DOMAIN, ) +from homeassistant.components.assist_pipeline.default_pipeline import ( + _async_local_fallback_intent_filter, + _DefaultPipelineProcessor, +) from homeassistant.components.assist_pipeline.pipeline import ( STORAGE_KEY, STORAGE_VERSION, @@ -34,12 +39,12 @@ from homeassistant.components.assist_pipeline.pipeline import ( PipelineEventType, PipelineStorageCollection, PipelineStore, - _async_local_fallback_intent_filter, async_create_default_pipeline, async_get_pipeline, async_get_pipelines, async_update_pipeline, ) +from homeassistant.components.assist_pipeline.run import _PipelineProcessorRequest from homeassistant.components.llm import LLMTools from homeassistant.const import ATTR_FRIENDLY_NAME, MATCH_ALL from homeassistant.core import Context, HomeAssistant @@ -818,6 +823,55 @@ def test_pipeline_run_equality(hass: HomeAssistant, init_components) -> None: assert run_1 != 1234 +async def test_pipeline_run_delegates_to_processor( + hass: HomeAssistant, + init_components: None, + mock_chat_session: chat_session.ChatSession, +) -> None: + """Test that the run controller delegates processing and owns the lifecycle.""" + events: list[assist_pipeline.PipelineEvent] = [] + processor = Mock( + response_audio=None, + supports_streaming_response=False, + async_validate=AsyncMock(), + async_execute=AsyncMock(), + invalidate=Mock(), + cleanup=Mock(), + ) + + with patch( + "homeassistant.components.assist_pipeline.run._create_pipeline_processor", + return_value=processor, + ): + pipeline_input = assist_pipeline.pipeline.PipelineInput( + intent_input="test input", + session=mock_chat_session, + run=assist_pipeline.pipeline.PipelineRun( + hass, + context=Context(), + pipeline=assist_pipeline.pipeline.async_get_pipeline(hass), + start_stage=assist_pipeline.PipelineStage.INTENT, + end_stage=assist_pipeline.PipelineStage.INTENT, + event_callback=events.append, + ), + ) + + await pipeline_input.execute(validate=True) + + validate_request = processor.async_validate.await_args.args[0] + execute_request = processor.async_execute.await_args.args[0] + assert isinstance(validate_request, _PipelineProcessorRequest) + assert validate_request == execute_request + assert validate_request.session is pipeline_input.session + assert validate_request.intent_input == pipeline_input.intent_input + with pytest.raises(FrozenInstanceError): + validate_request.intent_input = "changed" + assert [event.type for event in events] == [ + PipelineEventType.RUN_START, + PipelineEventType.RUN_END, + ] + + async def test_text_only_run_does_not_start_debug_recording_thread( hass: HomeAssistant, init_components, @@ -882,16 +936,13 @@ async def test_tts_audio_output( await pipeline_input.validate() # Verify TTS audio settings - assert pipeline_input.run.tts_stream.options is not None - assert pipeline_input.run.tts_stream.options.get(tts.ATTR_PREFERRED_FORMAT) == "wav" - assert ( - pipeline_input.run.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_RATE) - == 16000 - ) - assert ( - pipeline_input.run.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_CHANNELS) - == 1 - ) + processor = pipeline_input.run._processor + assert isinstance(processor, _DefaultPipelineProcessor) + assert processor.tts_stream is not None + assert processor.tts_stream.options is not None + assert processor.tts_stream.options.get(tts.ATTR_PREFERRED_FORMAT) == "wav" + assert processor.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_RATE) == 16000 + assert processor.tts_stream.options.get(tts.ATTR_PREFERRED_SAMPLE_CHANNELS) == 1 with patch.object(mock_tts_entity, "get_tts_audio") as mock_get_tts_audio: await pipeline_input.execute() @@ -1089,13 +1140,12 @@ async def test_sentence_trigger_overrides_conversation_agent( start_stage=assist_pipeline.PipelineStage.INTENT, end_stage=assist_pipeline.PipelineStage.INTENT, event_callback=events.append, - intent_agent="test-agent", # not the default agent ), ) # Ensure prepare succeeds with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info", return_value=conversation.AgentInfo( id="test-agent", name="Test Agent", @@ -1105,7 +1155,7 @@ async def test_sentence_trigger_overrides_conversation_agent( await pipeline_input.validate() with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse" + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse" ) as mock_async_converse: await pipeline_input.execute() @@ -1178,7 +1228,7 @@ async def test_prefer_local_intents( # Ensure prepare succeeds with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info", return_value=conversation.AgentInfo( id="test-agent", name="Test Agent", @@ -1188,7 +1238,7 @@ async def test_prefer_local_intents( await pipeline_input.validate() with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse" + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse" ) as mock_async_converse: await pipeline_input.execute() @@ -1247,7 +1297,7 @@ async def test_intent_continue_conversation( # Ensure prepare succeeds with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info", return_value=conversation.AgentInfo( id="test-agent", name="Test Agent", @@ -1260,7 +1310,7 @@ async def test_intent_continue_conversation( response.async_set_speech("For how long?") with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse", return_value=conversation.ConversationResult( response=response, conversation_id=mock_chat_session.conversation_id, @@ -1322,7 +1372,7 @@ async def test_intent_continue_conversation( # Ensure prepare succeeds with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info", return_value=conversation.AgentInfo( id="test-agent", name="Test Agent", @@ -1338,7 +1388,7 @@ async def test_intent_continue_conversation( response.async_set_speech("Timer set for 20 minutes") with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse", return_value=conversation.ConversationResult( response=response, conversation_id=mock_chat_session.conversation_id, @@ -1410,7 +1460,7 @@ async def test_stt_language_used_instead_of_conversation_language( await pipeline_input.validate() with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse", return_value=conversation.ConversationResult( intent.IntentResponse(pipeline.language) ), @@ -1486,7 +1536,7 @@ async def test_tts_language_used_instead_of_conversation_language( await pipeline_input.validate() with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse", return_value=conversation.ConversationResult( intent.IntentResponse(pipeline.language) ), @@ -1562,7 +1612,7 @@ async def test_pipeline_language_used_instead_of_conversation_language( await pipeline_input.validate() with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse", return_value=conversation.ConversationResult( intent.IntentResponse(pipeline.language) ), @@ -1748,7 +1798,7 @@ async def test_chat_log_tts_streaming( mock_tts_entity.async_supports_streaming_input = Mock(return_value=True) with patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info", return_value=conversation.AgentInfo( id="test-agent", name="Test Agent", @@ -1828,7 +1878,7 @@ async def test_chat_log_tts_streaming( return_value=LLMTools(tools=[mock_tool]), ), patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_converse", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_converse", mock_converse, ), ): @@ -1913,7 +1963,7 @@ async def test_acknowledge( await pipeline_input.execute() with patch( - "homeassistant.components.assist_pipeline.PipelineRun.text_to_speech" + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.text_to_speech" ) as text_to_speech: def _reset() -> None: @@ -2108,7 +2158,7 @@ async def test_acknowledge_child_device_inherits_area( await pipeline_input.execute() with patch( - "homeassistant.components.assist_pipeline.PipelineRun.text_to_speech" + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.text_to_speech" ) as text_to_speech: await _run("turn on light 1") @@ -2177,7 +2227,7 @@ async def test_acknowledge_other_agents( with ( patch( - "homeassistant.components.assist_pipeline.pipeline.conversation.async_get_agent_info", + "homeassistant.components.assist_pipeline.default_pipeline.conversation.async_get_agent_info", return_value=conversation.AgentInfo( id="test-agent", name="Test Agent", @@ -2185,16 +2235,16 @@ async def test_acknowledge_other_agents( ), ), patch( - "homeassistant.components.assist_pipeline.PipelineRun.prepare_text_to_speech" + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.prepare_text_to_speech" ), patch( - "homeassistant.components.assist_pipeline.PipelineRun.text_to_speech" + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.text_to_speech" ) as text_to_speech, patch( "homeassistant.components.conversation.async_converse", return_value=None ) as async_converse, patch( - "homeassistant.components.assist_pipeline.PipelineRun._get_all_targets_in_satellite_area" + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor._get_all_targets_in_satellite_area" ) as get_all_targets_in_satellite_area, ): pipeline_input = assist_pipeline.pipeline.PipelineInput( @@ -2266,7 +2316,7 @@ async def test_stt_vad_enabled_based_on_audio_processing( # VAD should be used with ( patch( - "homeassistant.components.assist_pipeline.pipeline.VoiceCommandSegmenter" + "homeassistant.components.assist_pipeline.default_pipeline.VoiceCommandSegmenter" ) as mock_vad, patch( "homeassistant.components.stt.async_get_speech_to_text_engine", @@ -2317,7 +2367,7 @@ async def test_stt_vad_enabled_based_on_audio_processing( # VAD should NOT be used with ( patch( - "homeassistant.components.assist_pipeline.pipeline.VoiceCommandSegmenter" + "homeassistant.components.assist_pipeline.default_pipeline.VoiceCommandSegmenter" ) as mock_vad, patch( "homeassistant.components.stt.async_get_speech_to_text_engine", diff --git a/tests/components/assist_satellite/test_entity.py b/tests/components/assist_satellite/test_entity.py index ec44f2c6b256..d8a596005af0 100644 --- a/tests/components/assist_satellite/test_entity.py +++ b/tests/components/assist_satellite/test_entity.py @@ -199,7 +199,7 @@ async def test_pipeline_validation_error_ends_pipeline( ) with patch( - "homeassistant.components.assist_pipeline.pipeline.PipelineRun.prepare_speech_to_text" + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.prepare_speech_to_text" ): await entity.async_accept_pipeline_from_satellite( object(), # type: ignore[arg-type] @@ -927,7 +927,7 @@ async def test_ask_question( ) async def speech_to_text(self, *args, **kwargs): - self.process_event( + self.host.process_event( PipelineEvent( PipelineEventType.STT_END, {"stt_output": {"text": response_text}} ) @@ -950,10 +950,10 @@ async def test_ask_question( audio_stream = object() with ( patch( - "homeassistant.components.assist_pipeline.pipeline.PipelineRun.prepare_speech_to_text" + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.prepare_speech_to_text" ), patch( - "homeassistant.components.assist_pipeline.pipeline.PipelineRun.speech_to_text", + "homeassistant.components.assist_pipeline.default_pipeline._DefaultPipelineProcessor.speech_to_text", speech_to_text, ), ):