"""Satellite code.""" import array import asyncio import logging import math import time import wave from dataclasses import dataclass from enum import Enum, auto from pathlib import Path from queue import Queue from typing import Callable, Dict, Final, List, Optional, Set, Union from pyring_buffer import RingBuffer from wyoming.asr import Transcript from wyoming.audio import AudioChunk, AudioStart, AudioStop from wyoming.client import AsyncClient from wyoming.error import Error from wyoming.event import Event, async_write_event from wyoming.info import Describe, Info from wyoming.mic import MicProcessAsyncClient from wyoming.ping import Ping, Pong from wyoming.pipeline import PipelineStage, RunPipeline from wyoming.satellite import ( PauseSatellite, RunSatellite, SatelliteConnected, SatelliteDisconnected, StreamingStarted, StreamingStopped, ) from wyoming.snd import Played, SndProcessAsyncClient from wyoming.timer import TimerCancelled, TimerFinished, TimerStarted, TimerUpdated from wyoming.tts import Synthesize from wyoming.vad import VoiceStarted, VoiceStopped from wyoming.wake import Detect, Detection, WakeProcessAsyncClient from .settings import SatelliteSettings from .utils import ( DebugAudioWriter, multiply_volume, normalize_wake_word, run_event_command, wav_to_events, ) from .vad import SileroVad, VoiceCommandSegmenter from .webrtc import WebRtcAudio _LOGGER = logging.getLogger(__name__) _PONG_TIMEOUT: Final = 5 _PING_SEND_DELAY: Final = 2 _WAKE_INFO_TIMEOUT: Final = 2 class State(Enum): NOT_STARTED = auto() STARTING = auto() STARTED = auto() RESTARTING = auto() STOPPING = auto() STOPPED = auto() @dataclass class SoundEvent: event: Event is_tts: bool # ----------------------------------------------------------------------------- class SatelliteBase: """Base class for satellites.""" def __init__(self, settings: SatelliteSettings) -> None: self.settings = settings self.server_id: Optional[str] = None self._state = State.NOT_STARTED self._state_changed = asyncio.Event() self._writer: Optional[asyncio.StreamWriter] = None self._mic_task: Optional[asyncio.Task] = None self._mic_webrtc: Optional[Callable[[bytes], bytes]] = None self._snd_task: Optional[asyncio.Task] = None self._snd_queue: "Optional[asyncio.Queue[SoundEvent]]" = None self._wake_task: Optional[asyncio.Task] = None self._wake_queue: "Optional[asyncio.Queue[Event]]" = None self._event_task: Optional[asyncio.Task] = None self._event_queue: "Optional[asyncio.Queue[Event]]" = None self._ping_server_enabled: bool = False self._pong_received_event = asyncio.Event() self._ping_server_task: Optional[asyncio.Task] = None self.microphone_muted = False self._unmute_microphone_task: Optional[asyncio.Task] = None # Debug audio recording self.wake_audio_writer: Optional[DebugAudioWriter] = None self.stt_audio_writer: Optional[DebugAudioWriter] = None if settings.debug_recording_dir: self.wake_audio_writer = DebugAudioWriter( settings.debug_recording_dir, "wake", ring_buffer_size=(2 * 16000 * 2 * 1), # last 2 sec ) self.stt_audio_writer = DebugAudioWriter( settings.debug_recording_dir, "stt" ) # Names of wake words to listen for self._wake_names: Optional[List[str]] = None if self.settings.wake.names: self._wake_names = self.settings.wake.names @property def is_running(self) -> bool: """True if not stopping/stopped.""" return self._state not in (State.STOPPED,) @property def state(self) -> State: """Get current state.""" return self._state @state.setter def state(self, value: State) -> None: """Set state.""" self._state = value self._state_changed.set() self._state_changed.clear() async def run(self) -> None: """Run main satellite loop.""" while self.is_running: try: if self.state == State.NOT_STARTED: await self._start() elif self.state == State.RESTARTING: await self._restart() elif self.state == State.STOPPING: await self._stop() elif self.state == State.STOPPED: await self.stopped() break else: await self._state_changed.wait() except Exception: if self.is_running: # Automatically restart _LOGGER.exception("Unexpected error running satellite") self.state = State.RESTARTING async def stop(self) -> None: """Stop the satellite""" self.state = State.STOPPING # Wait for STOPPED while self.state != State.STOPPED: await self._state_changed.wait() async def set_server(self, server_id: str, writer: asyncio.StreamWriter) -> None: """Set event writer.""" self.server_id = server_id self._writer = writer _LOGGER.debug("Server set: %s", server_id) await self.trigger_server_connected() async def clear_server(self) -> None: """Remove writer.""" self.server_id = None self._writer = None self._disable_ping() _LOGGER.debug("Server disconnected") await self.trigger_server_disonnected() async def event_to_server(self, event: Event) -> None: """Send an event to the server.""" if self._writer is None: return try: await async_write_event(event, self._writer) except Exception as err: await self.clear_server() if isinstance(err, ConnectionResetError): _LOGGER.warning("Server disconnected unexpectedly") else: _LOGGER.exception("Unexpected error sending event to server") def _enable_ping(self) -> None: self._ping_server_enabled = True self._ping_server_task = asyncio.create_task(self._ping_server(), name="ping") def _disable_ping(self) -> None: self._ping_server_enabled = False if self._ping_server_task is not None: self._ping_server_task.cancel() self._ping_server_task = None async def _ping_server(self) -> None: try: while self.is_running: await asyncio.sleep(_PING_SEND_DELAY) if (self.server_id is None) or (not self._ping_server_enabled): # No server connected continue # Send ping and wait for pong self._pong_received_event.clear() await self.event_to_server(Ping().event()) try: await asyncio.wait_for( self._pong_received_event.wait(), timeout=_PONG_TIMEOUT ) except asyncio.TimeoutError: if self.server_id is None: # No server connected continue _LOGGER.warning("Did not receive ping response within timeout") await self.clear_server() except asyncio.CancelledError: pass except Exception: _LOGGER.exception("Unexpected error in ping server task") # ------------------------------------------------------------------------- async def _start(self) -> None: """Connect to services.""" self.state = State.STARTING await self._connect_to_services() self.state = State.STARTED await self.started() async def started(self) -> None: """Called when satellite has started.""" async def _stop(self) -> None: """Disconnect from services.""" self.server_id = None self._writer = None await self._disconnect_from_services() self._disable_ping() self.state = State.STOPPED async def stopped(self) -> None: """Called when satellite has stopped.""" async def event_from_server(self, event: Event) -> None: """Called when an event is received from the server.""" forward_event = True if Ping.is_type(event.type): # Respond with pong ping = Ping.from_event(event) await self.event_to_server(Pong(text=ping.text).event()) if not self._ping_server_enabled: # Enable pinging self._enable_ping() _LOGGER.debug("Ping enabled") forward_event = False elif Pong.is_type(event.type): # Response from our ping self._pong_received_event.set() forward_event = False elif AudioChunk.is_type(event.type): # TTS audio await self.event_to_snd(event) forward_event = False elif AudioStart.is_type(event.type): # TTS started await self.event_to_snd(event) await self.trigger_tts_start() elif AudioStop.is_type(event.type): # TTS stopped await self.event_to_snd(event) await self.trigger_tts_stop() elif VoiceStarted.is_type(event.type): # STT start await self.trigger_stt_start() elif VoiceStopped.is_type(event.type): # STT stop await self.trigger_stt_stop() elif Transcript.is_type(event.type): # STT text _LOGGER.debug(event) await self.trigger_transcript(Transcript.from_event(event)) elif Synthesize.is_type(event.type): # TTS request _LOGGER.debug(event) await self.trigger_synthesize(Synthesize.from_event(event)) elif Error.is_type(event.type): _LOGGER.warning(event) await self.trigger_error(Error.from_event(event)) elif TimerStarted.is_type(event.type): _LOGGER.debug(event) await self.trigger_timer_started(TimerStarted.from_event(event)) elif TimerUpdated.is_type(event.type): _LOGGER.debug(event) await self.trigger_timer_updated(TimerUpdated.from_event(event)) elif TimerCancelled.is_type(event.type): _LOGGER.debug(event) await self.trigger_timer_cancelled(TimerCancelled.from_event(event)) elif TimerFinished.is_type(event.type): _LOGGER.debug(event) await self.trigger_timer_finished(TimerFinished.from_event(event)) # Forward everything except audio/ping/pong to event service if forward_event: await self.forward_event(event) async def _send_run_pipeline( self, start_stage: Optional[PipelineStage] = None, end_stage: Optional[PipelineStage] = None, wake_word_name: Optional[str] = None, ) -> None: """Sends a RunPipeline event with the correct stages.""" restart_on_end = False if start_stage is None: if self.settings.wake.enabled: # Local wake word detection start_stage = PipelineStage.ASR restart_on_end = False else: # Remote wake word detection start_stage = PipelineStage.WAKE restart_on_end = not self.settings.vad.enabled if end_stage is None: if self.settings.snd.enabled: # Play TTS response end_stage = PipelineStage.TTS else: # No audio output end_stage = PipelineStage.HANDLE run_pipeline = RunPipeline( start_stage=start_stage, end_stage=end_stage, wake_word_name=wake_word_name, restart_on_end=restart_on_end, ).event() _LOGGER.debug(run_pipeline) await self.event_to_server(run_pipeline) await self.forward_event(run_pipeline) async def _restart(self) -> None: """Disconnects from services and restarts loop.""" self.state = State.RESTARTING await self._disconnect_from_services() _LOGGER.debug("Restarting in %s second(s)", self.settings.restart_timeout) await asyncio.sleep(self.settings.restart_timeout) self.state = State.NOT_STARTED async def _connect_to_services(self) -> None: """Connects to configured services.""" if self.settings.mic.enabled: _LOGGER.debug( "Connecting to mic service: %s", self.settings.mic.uri or self.settings.mic.command, ) self._mic_task = asyncio.create_task(self._mic_task_proc(), name="mic") if self.settings.snd.enabled: _LOGGER.debug( "Connecting to snd service: %s", self.settings.snd.uri or self.settings.snd.command, ) self._snd_task = asyncio.create_task(self._snd_task_proc(), name="snd") if self.settings.wake.enabled: _LOGGER.debug( "Connecting to wake service: %s", self.settings.wake.uri or self.settings.wake.command, ) self._wake_task = asyncio.create_task(self._wake_task_proc(), name="wake") if self.settings.event.enabled: _LOGGER.debug( "Connecting to event service: %s", self.settings.event.uri or self.settings.event.command, ) self._event_task = asyncio.create_task( self._event_task_proc(), name="event" ) _LOGGER.info("Connected to services") async def _disconnect_from_services(self) -> None: """Disconnects from running services.""" if self._mic_task is not None: _LOGGER.debug("Stopping microphone service") self._mic_task.cancel() self._mic_task = None if self._snd_task is not None: _LOGGER.debug("Stopping sound service") self._snd_task.cancel() self._snd_task = None if self._wake_task is not None: _LOGGER.debug("Stopping wake service") self._wake_task.cancel() self._wake_task = None if self._event_task is not None: _LOGGER.debug("Stopping event service") self._event_task.cancel() self._event_task = None _LOGGER.debug("Disconnected from services") # ------------------------------------------------------------------------- # Microphone # ------------------------------------------------------------------------- async def event_from_mic( self, event: Event, audio_bytes: Optional[bytes] = None ) -> None: """Called when an event is received from the mic service. For AudioChunk events, the audio_bytes may be set if the audio was already unpacked for preprocessing. Use this to avoid unpacking the event again. """ def _make_mic_client(self) -> Optional[AsyncClient]: """Create client for mic service.""" if self.settings.mic.command: program, *program_args = self.settings.mic.command return MicProcessAsyncClient( rate=self.settings.mic.rate, width=self.settings.mic.width, channels=self.settings.mic.channels, samples_per_chunk=self.settings.mic.samples_per_chunk, program=program, program_args=program_args, ) if self.settings.mic.uri: return AsyncClient.from_uri(self.settings.mic.uri) return None async def _mic_task_proc(self) -> None: """Mic service loop.""" mic_client: Optional[AsyncClient] = None audio_bytes: Optional[bytes] = None if self.settings.mic.needs_webrtc and (self._mic_webrtc is None): _LOGGER.debug("Using webrtc audio enhancements") self._mic_webrtc = WebRtcAudio( self.settings.mic.auto_gain, self.settings.mic.noise_suppression ) async def _disconnect() -> None: try: if mic_client is not None: await mic_client.disconnect() except Exception: pass # ignore disconnect errors while self.is_running: try: if mic_client is None: mic_client = self._make_mic_client() assert mic_client is not None await mic_client.connect() _LOGGER.info( "Connected to mic service: %s", self.settings.mic.uri or self.settings.mic.command, ) event = await mic_client.read_event() if event is None: _LOGGER.warning("Mic service disconnected") await _disconnect() mic_client = None # reconnect await asyncio.sleep(self.settings.mic.reconnect_seconds) continue # Audio processing if ( self.settings.mic.needs_processing or (self.settings.mic.channel_index is not None) and AudioChunk.is_type(event.type) ): chunk = AudioChunk.from_event(event) if self.settings.mic.channel_index is not None: if chunk.width != 2: raise ValueError( "Mic channel index selection requires 16-bit samples" ) # Convert to unsigned 16-bit array to make channel extraction easier audio_array = array.array("H", chunk.audio) audio_bytes = audio_array[ self.settings.mic.channel_index :: chunk.channels ].tobytes() if self.settings.mic.needs_processing: if audio_bytes is None: audio_bytes = chunk.audio audio_bytes = self._process_mic_audio(chunk.audio) event = AudioChunk( rate=chunk.rate, width=chunk.width, channels=( chunk.channels if (self.settings.mic.channel_index is None) else 1 ), audio=audio_bytes, ).event() else: audio_bytes = None await self.event_from_mic(event, audio_bytes) except asyncio.CancelledError: break except Exception: _LOGGER.exception("Unexpected error in mic read task") await _disconnect() mic_client = None # reconnect await asyncio.sleep(self.settings.mic.reconnect_seconds) await _disconnect() def _process_mic_audio(self, audio_bytes: bytes) -> bytes: """Perform audio pre-processing on mic input.""" if self.settings.mic.volume_multiplier != 1.0: audio_bytes = multiply_volume( audio_bytes, self.settings.mic.volume_multiplier ) if self._mic_webrtc is not None: # Noise suppression and auto gain audio_bytes = self._mic_webrtc(audio_bytes) return audio_bytes # ------------------------------------------------------------------------- # Sound # ------------------------------------------------------------------------- async def event_to_snd(self, event: Event, is_tts: bool = True) -> None: """Send an event to the sound service.""" if self._snd_queue is not None: self._snd_queue.put_nowait(SoundEvent(event, is_tts)) def _make_snd_client(self) -> Optional[AsyncClient]: """Create client for snd service.""" if self.settings.snd.command: program, *program_args = self.settings.snd.command return SndProcessAsyncClient( rate=self.settings.snd.rate, width=self.settings.snd.width, channels=self.settings.snd.channels, program=program, program_args=program_args, ) if self.settings.snd.uri: return AsyncClient.from_uri(self.settings.snd.uri) return None async def _snd_task_proc(self) -> None: """Snd service loop.""" snd_client: Optional[AsyncClient] = None async def _disconnect() -> None: try: if snd_client is not None: await snd_client.disconnect() except Exception: pass # ignore disconnect errors while self.is_running: try: if self._snd_queue is None: self._snd_queue = asyncio.Queue() snd_event = await self._snd_queue.get() event = snd_event.event if snd_client is None: snd_client = self._make_snd_client() assert snd_client is not None await snd_client.connect() _LOGGER.info( "Connected to snd service: %s", self.settings.snd.uri or self.settings.snd.command, ) # Audio processing if self.settings.snd.needs_processing and AudioChunk.is_type( event.type ): chunk = AudioChunk.from_event(event) audio_bytes = self._process_snd_audio(chunk.audio) event = AudioChunk( rate=chunk.rate, width=chunk.width, channels=chunk.channels, audio=audio_bytes, ).event() await snd_client.write_event(event) if self.settings.snd.disconnect_after_stop and AudioStop.is_type( event.type ): await _disconnect() if snd_event.is_tts: await self.trigger_played() await self.event_to_server(Played().event()) snd_client = None # reconnect on next event except asyncio.CancelledError: break except Exception: _LOGGER.exception("Unexpected error in snd read task") await _disconnect() snd_client = None # reconnect self._snd_queue = None await asyncio.sleep(self.settings.snd.reconnect_seconds) await _disconnect() def _process_snd_audio(self, audio_bytes: bytes) -> bytes: """Perform audio pre-processing on snd output.""" if self.settings.snd.volume_multiplier != 1.0: audio_bytes = multiply_volume( audio_bytes, self.settings.snd.volume_multiplier ) return audio_bytes async def _play_wav( self, wav_path: Optional[Union[str, Path]], mute_microphone: bool = False ) -> None: """Send WAV as events to sound service.""" if (not wav_path) or (not self.settings.snd.enabled): return try: if mute_microphone: with wave.open(str(wav_path), "rb") as wav_file: seconds_to_mute = wav_file.getnframes() / wav_file.getframerate() seconds_to_mute += self.settings.mic.seconds_to_mute_after_awake_wav _LOGGER.debug("Muting microphone for %s second(s)", seconds_to_mute) self.microphone_muted = True self._unmute_microphone_task = asyncio.create_task( self._unmute_microphone_after(seconds_to_mute) ) for event in wav_to_events( wav_path, samples_per_chunk=self.settings.snd.samples_per_chunk, ): await self.event_to_snd(event, is_tts=False) except Exception: # Unmute in case of an error self.microphone_muted = False raise async def _unmute_microphone_after(self, seconds: float) -> None: await asyncio.sleep(seconds) self.microphone_muted = False _LOGGER.debug("Unmuted microphone") # ------------------------------------------------------------------------- # Wake # ------------------------------------------------------------------------- async def event_from_wake(self, event: Event) -> None: """Called when an event is received from the wake service.""" async def event_to_wake(self, event: Event) -> None: """Send event to the wake service.""" if self._wake_queue is not None: self._wake_queue.put_nowait(event) def _make_wake_client(self) -> Optional[AsyncClient]: """Create client for wake service.""" if self.settings.wake.command: program, *program_args = self.settings.wake.command return WakeProcessAsyncClient( rate=self.settings.wake.rate, width=self.settings.wake.width, channels=self.settings.wake.channels, program=program, program_args=program_args, ) if self.settings.wake.uri: return AsyncClient.from_uri(self.settings.wake.uri) return None async def _wake_task_proc(self) -> None: """Wake service loop.""" wake_client: Optional[AsyncClient] = None to_client_task: Optional[asyncio.Task] = None from_client_task: Optional[asyncio.Task] = None pending: Set[asyncio.Task] = set() async def _disconnect() -> None: nonlocal to_client_task, from_client_task try: if wake_client is not None: await wake_client.disconnect() # Clean up tasks if to_client_task is not None: to_client_task.cancel() to_client_task = None if from_client_task is not None: from_client_task.cancel() from_client_task = None except Exception: pass # ignore disconnect errors while self.is_running: try: if self._wake_queue is None: self._wake_queue = asyncio.Queue() if wake_client is None: wake_client = self._make_wake_client() assert wake_client is not None await wake_client.connect() _LOGGER.info( "Connected to wake service: %s", self.settings.wake.uri or self.settings.wake.command, ) # Reset from_client_task = None to_client_task = None pending = set() self._wake_queue = asyncio.Queue() # Inform wake service of which wake word(s) to detect await self._send_wake_detect() # Read/write in "parallel" if to_client_task is None: # From satellite to wake service to_client_task = asyncio.create_task( self._wake_queue.get(), name="wake_to_client" ) pending.add(to_client_task) if from_client_task is None: # From wake service to satellite from_client_task = asyncio.create_task( wake_client.read_event(), name="wake_from_client" ) pending.add(from_client_task) done, pending = await asyncio.wait( pending, return_when=asyncio.FIRST_COMPLETED ) if to_client_task in done: # Event to go to wake service (audio) assert to_client_task is not None event = to_client_task.result() to_client_task = None await wake_client.write_event(event) if from_client_task in done: # Event from wake service (detection) assert from_client_task is not None event = from_client_task.result() from_client_task = None if event is None: _LOGGER.warning("Wake service disconnected") await _disconnect() wake_client = None # reconnect await asyncio.sleep(self.settings.wake.reconnect_seconds) continue await self.event_from_wake(event) except asyncio.CancelledError: break except Exception: _LOGGER.exception("Unexpected error in wake read task") await _disconnect() wake_client = None # reconnect await asyncio.sleep(self.settings.wake.reconnect_seconds) await _disconnect() async def _send_wake_detect(self) -> None: """Inform wake word service of which wake words to detect.""" await self.event_to_wake(Detect(names=self._wake_names).event()) await self.trigger_detect() # ------------------------------------------------------------------------- # Events # ------------------------------------------------------------------------- async def trigger_server_connected(self) -> None: """Called when connected to server.""" _LOGGER.info("Connected to server") await run_event_command(self.settings.event.connected) await self.forward_event(SatelliteConnected().event()) async def trigger_server_disonnected(self) -> None: """Called when disconnected from server.""" _LOGGER.info("Disconnected from server") await run_event_command(self.settings.event.disconnected) await self.forward_event(SatelliteDisconnected().event()) async def trigger_streaming_start(self) -> None: """Called when audio streaming starts.""" await run_event_command(self.settings.event.streaming_start) await self.forward_event(StreamingStarted().event()) async def trigger_streaming_stop(self) -> None: """Called when audio streaming stops.""" await run_event_command(self.settings.event.streaming_stop) await self.forward_event(StreamingStopped().event()) async def trigger_detect(self) -> None: """Called when wake word detection starts.""" await run_event_command(self.settings.event.detect) async def trigger_detection(self, detection: Detection) -> None: """Called when wake word is detected.""" await run_event_command(self.settings.event.detection, detection.name) await self._play_wav( self.settings.snd.awake_wav, mute_microphone=self.settings.mic.mute_during_awake_wav, ) async def trigger_played(self) -> None: """Called when audio stopped playing""" await run_event_command(self.settings.event.played) await self.forward_event(Played().event()) async def trigger_transcript(self, transcript: Transcript) -> None: """Called when speech-to-text text is received.""" await run_event_command(self.settings.event.transcript, transcript.text) await self._play_wav(self.settings.snd.done_wav) async def trigger_stt_start(self) -> None: """Called when user starts speaking.""" await run_event_command(self.settings.event.stt_start) async def trigger_stt_stop(self) -> None: """Called when user stops speaking.""" await run_event_command(self.settings.event.stt_stop) async def trigger_synthesize(self, synthesize: Synthesize) -> None: """Called when text-to-speech text is received.""" await run_event_command(self.settings.event.synthesize, synthesize.text) async def trigger_tts_start(self) -> None: """Called when text-to-speech audio starts.""" await run_event_command(self.settings.event.tts_start) async def trigger_tts_stop(self) -> None: """Called when text-to-speech audio stops.""" await run_event_command(self.settings.event.tts_stop) async def trigger_error(self, error: Error) -> None: """Called when an error occurs on the server.""" await run_event_command(self.settings.event.error, error.text) async def trigger_timer_started(self, timer_started: TimerStarted) -> None: """Called when timer-started event is received.""" await run_event_command(self.settings.timer.started, timer_started) async def trigger_timer_updated(self, timer_updated: TimerUpdated) -> None: """Called when timer-updated event is received.""" await run_event_command(self.settings.timer.updated, timer_updated) async def trigger_timer_cancelled(self, timer_cancelled: TimerCancelled) -> None: """Called when timer-cancelled event is received.""" await run_event_command(self.settings.timer.cancelled, timer_cancelled.id) async def trigger_timer_finished(self, timer_finished: TimerFinished) -> None: """Called when timer-finished event is received.""" await run_event_command(self.settings.timer.finished, timer_finished.id) for _ in range(self.settings.timer.finished_wav_plays): await self._play_wav( self.settings.timer.finished_wav, mute_microphone=self.settings.mic.mute_during_awake_wav, ) await asyncio.sleep(self.settings.timer.finished_wav_delay) async def forward_event(self, event: Event) -> None: """Forward an event to the event service.""" if self._event_queue is not None: self._event_queue.put_nowait(event) def _make_event_client(self) -> Optional[AsyncClient]: """Create client for event service.""" if self.settings.event.uri: return AsyncClient.from_uri(self.settings.event.uri) return None async def _event_task_proc(self) -> None: """Event service loop.""" event_client: Optional[AsyncClient] = None async def _disconnect() -> None: try: if event_client is not None: await event_client.disconnect() except Exception: pass # ignore disconnect errors while self.is_running: try: if self._event_queue is None: self._event_queue = asyncio.Queue() event = await self._event_queue.get() if event_client is None: event_client = self._make_event_client() assert event_client is not None await event_client.connect() _LOGGER.info( "Connected to event service: %s", self.settings.event.uri or self.settings.event.command, ) await event_client.write_event(event) except asyncio.CancelledError: break except Exception: _LOGGER.exception("Unexpected error in event read task") await _disconnect() event_client = None # reconnect self._event_queue = None await asyncio.sleep(self.settings.event.reconnect_seconds) await _disconnect() # ------------------------------------------------------------------------- # Info # ------------------------------------------------------------------------- async def update_info(self, info: Info) -> None: pass # ----------------------------------------------------------------------------- class AlwaysStreamingSatellite(SatelliteBase): """Satellite that always streams audio.""" def __init__(self, settings: SatelliteSettings) -> None: super().__init__(settings) self.is_streaming = False if settings.vad.enabled: _LOGGER.warning("VAD is enabled but will not be used") if settings.wake.enabled: _LOGGER.warning("Local wake word detection is enabled but will not be used") async def event_from_server(self, event: Event) -> None: await super().event_from_server(event) if RunSatellite.is_type(event.type): self.is_streaming = True _LOGGER.info("Streaming audio") await self._send_run_pipeline() await self.trigger_streaming_start() elif PauseSatellite.is_type(event.type): self.is_streaming = False _LOGGER.info("Satellite paused") elif Detect.is_type(event.type): # Wake word detection started await self.trigger_detect() elif Detection.is_type(event.type): # Start debug recording if self.stt_audio_writer is not None: self.stt_audio_writer.start() # Wake word detected _LOGGER.debug("Wake word detected") await self.trigger_detection(Detection.from_event(event)) elif Transcript.is_type(event.type) or Error.is_type(event.type): # Stop debug recording if self.stt_audio_writer is not None: self.stt_audio_writer.stop() if Transcript.is_type(event.type): # We're always streaming _LOGGER.info("Streaming audio") # Re-trigger streaming start even though we technically don't stop # so the event service can reset LEDs, etc. await self.trigger_streaming_start() async def event_from_mic( self, event: Event, audio_bytes: Optional[bytes] = None ) -> None: if (not self.is_streaming) or self.microphone_muted: return if AudioChunk.is_type(event.type): # Forward to server await self.event_to_server(event) # Debug audio recording if self.stt_audio_writer is not None: if audio_bytes is None: chunk = AudioChunk.from_event(event) audio_bytes = chunk.audio self.stt_audio_writer.write(audio_bytes) # ----------------------------------------------------------------------------- class VadStreamingSatellite(SatelliteBase): """Satellite that only streams after speech is detected.""" def __init__(self, settings: SatelliteSettings) -> None: if not settings.vad.enabled: raise ValueError("VAD is not enabled") super().__init__(settings) self.is_streaming = False self.vad = SileroVad( threshold=settings.vad.threshold, trigger_level=settings.vad.trigger_level ) # Timestamp in the future when we will have timed out (set with # time.monotonic()) self.timeout_seconds: Optional[float] = None # Audio from right before speech starts (circular buffer) self.vad_buffer: Optional[RingBuffer] = None if settings.vad.buffer_seconds > 0: # Assume 16Khz, 16-bit mono samples vad_buffer_bytes = int(math.ceil(settings.vad.buffer_seconds * 16000 * 2)) self.vad_buffer = RingBuffer(maxlen=vad_buffer_bytes) if settings.wake.enabled: _LOGGER.warning("Local wake word detection is enabled but will not be used") self._is_paused = False async def event_from_server(self, event: Event) -> None: await super().event_from_server(event) if RunSatellite.is_type(event.type): self._is_paused = False _LOGGER.info("Waiting for speech") elif Detect.is_type(event.type): # Wake word detection started await self.trigger_detect() elif Detection.is_type(event.type): # Start debug recording if self.stt_audio_writer is not None: self.stt_audio_writer.start() _LOGGER.debug("Wake word detected") await self.trigger_detection(Detection.from_event(event)) elif ( Transcript.is_type(event.type) or Error.is_type(event.type) or PauseSatellite.is_type(event.type) ): if PauseSatellite.is_type(event.type): self._is_paused = True _LOGGER.debug("Satellite paused") self.is_streaming = False # Stop debug recording if self.stt_audio_writer is not None: self.stt_audio_writer.stop() async def event_from_mic( self, event: Event, audio_bytes: Optional[bytes] = None ) -> None: if ( (not AudioChunk.is_type(event.type)) or self.microphone_muted or self._is_paused ): return # Only unpack chunk once chunk: Optional[AudioChunk] = None # Debug audio recording if self.stt_audio_writer is not None: if audio_bytes is None: # Need to unpack chunk = AudioChunk.from_event(event) audio_bytes = chunk.audio self.stt_audio_writer.write(audio_bytes) if ( self.is_streaming and (self.timeout_seconds is not None) and (time.monotonic() >= self.timeout_seconds) ): # Time out during wake word recognition self.is_streaming = False self.timeout_seconds = None # Stop debug recording if self.stt_audio_writer is not None: self.stt_audio_writer.stop() # Stop pipeline await self.event_to_server(AudioStop().event()) _LOGGER.info("Waiting for speech") await self.trigger_streaming_stop() if not self.is_streaming: # Check VAD if audio_bytes is None: if chunk is None: # Need to unpack chunk = AudioChunk.from_event(event) audio_bytes = chunk.audio if not self.vad(audio_bytes): # No speech if self.vad_buffer is not None: self.vad_buffer.put(audio_bytes) return # Speech detected self.is_streaming = True _LOGGER.info("Streaming audio") await self._send_run_pipeline() await self.trigger_streaming_start() if self.settings.vad.wake_word_timeout is not None: # Set future time when we'll stop streaming if the wake word # hasn't been detected. self.timeout_seconds = ( time.monotonic() + self.settings.vad.wake_word_timeout ) else: # No timeout self.timeout_seconds = None if self.vad_buffer is not None: # Send contents of VAD buffer first. This is the audio that was # recorded right before speech was detected. if chunk is None: chunk = AudioChunk.from_event(event) await self.event_to_server( AudioChunk( rate=chunk.rate, width=chunk.width, channels=chunk.channels, audio=self.vad_buffer.getvalue(), ).event() ) self._reset_vad() if self.is_streaming: # Forward to server await self.event_to_server(event) def _reset_vad(self): """Reset state of VAD.""" self.vad(None) if self.vad_buffer is not None: # Clear buffer self.vad_buffer.put(bytes(self.vad_buffer.maxlen)) # ----------------------------------------------------------------------------- class WakeStreamingSatellite(SatelliteBase): """Satellite that waits for local wake word detection before streaming.""" def __init__(self, settings: SatelliteSettings) -> None: if not settings.wake.enabled: raise ValueError("Local wake word detection is not enabled") super().__init__(settings) self.is_streaming = False # Timestamp in the future when the refractory period is over (set with # time.monotonic()). # wake word id -> seconds self.refractory_timestamp: Dict[Optional[str], float] = {} # Local VAD for detecting the end of a voice command. # If this is disabled, the server is expected to do it remotely. self.vad: Optional[SileroVad] = None if settings.vad.enabled: _LOGGER.debug("Loading VAD model") self.vad = SileroVad(threshold=settings.vad.threshold, trigger_level=1) _LOGGER.info("Loaded VAD model") self.segmenter = VoiceCommandSegmenter( silence_seconds=settings.vad.finished_speaking_seconds ) # Used for debug audio recording so both wake and stt WAV files have the # same timestamp. self._debug_recording_timestamp: Optional[int] = None self._is_paused = False self._wake_info: Optional[Info] = None self._wake_info_ready = asyncio.Event() self._run_pipeline_queue: Queue[RunPipeline] = Queue() self._next_run_pipeline: Optional[RunPipeline] = None async def event_from_server(self, event: Event) -> None: # Only check event types once is_run_satellite = False is_pause_satellite = False is_transcript = False is_error = False if RunSatellite.is_type(event.type): is_run_satellite = True self._is_paused = False elif PauseSatellite.is_type(event.type): is_pause_satellite = True elif Detect.is_type(event.type): # Request to change active wake words detect = Detect.from_event(event) self._wake_names = detect.names if not self.is_streaming: await self._send_wake_detect() _LOGGER.debug("Updated wake word names: %s", self._wake_names) elif Transcript.is_type(event.type): is_transcript = True if (self._next_run_pipeline is not None) and ( self._next_run_pipeline.end_stage == PipelineStage.ASR ): # Done with remotely triggered pipeline self._next_run_pipeline = None elif Error.is_type(event.type): is_error = True elif RunPipeline.is_type(event.type): run_pipeline = RunPipeline.from_event(event) if not self.is_streaming: # Run right now self._next_run_pipeline = run_pipeline await self._trigger_pipeline(run_pipeline) return # Run after streaming is finished self._run_pipeline_queue.put(run_pipeline) elif AudioStop.is_type(event.type): if (self._next_run_pipeline is not None) and ( self._next_run_pipeline.end_stage == PipelineStage.TTS ): # Done with remotely triggered pipeline self._next_run_pipeline = None if is_transcript or is_pause_satellite: # Stop streaming before event_from_server is called because it will # play the "done" WAV. self.is_streaming = False # Stop debug recording (stt) if self.stt_audio_writer is not None: self.stt_audio_writer.stop() await super().event_from_server(event) if is_run_satellite or is_transcript or is_error or is_pause_satellite: # Stop streaming self.is_streaming = False if is_pause_satellite: self._is_paused = True _LOGGER.debug("Satellite is paused") else: # Go back to wake word detection or run next pipeline await self.trigger_streaming_stop() # It's possible to be paused in the middle of streaming if self._is_paused: return if self._run_pipeline_queue.empty(): # Back to wake word detection await self._start_wake_word_detection() else: # Next run from queue self._next_run_pipeline = self._run_pipeline_queue.get() await self._trigger_pipeline(self._next_run_pipeline) async def trigger_server_disonnected(self) -> None: await super().trigger_server_disonnected() self.is_streaming = False # Stop debug recording (stt) if self.stt_audio_writer is not None: self.stt_audio_writer.stop() await self.trigger_streaming_stop() async def event_from_mic( self, event: Event, audio_bytes: Optional[bytes] = None ) -> None: if ( (not AudioChunk.is_type(event.type)) or self.microphone_muted or self._is_paused ): return # Debug audio recording if (self.wake_audio_writer is not None) or (self.stt_audio_writer is not None): if audio_bytes is None: chunk = AudioChunk.from_event(event) audio_bytes = chunk.audio if self.wake_audio_writer is not None: self.wake_audio_writer.write(audio_bytes) if self.stt_audio_writer is not None: self.stt_audio_writer.write(audio_bytes) if self.is_streaming: if self.vad is not None: if audio_bytes is None: chunk = AudioChunk.from_event(event) audio_bytes = chunk.audio is_speech = self.vad(audio_bytes) chunk_seconds = (len(audio_bytes) // 2) / 16000 if not self.segmenter.process(chunk_seconds, is_speech): # Send empty audio chunk to indicate command is done event = AudioChunk( rate=16000, width=2, channels=1, audio=bytes() ).event() # Forward to server await self.event_to_server(event) else: # Forward to wake word service await self.event_to_wake(event) async def event_from_wake(self, event: Event, is_fake: bool = False) -> None: if Info.is_type(event.type): self._wake_info = Info.from_event(event) self._wake_info_ready.set() return if self.is_streaming or (self.server_id is None): # Not detecting or no server connected return if not Detection.is_type(event.type): return detection = Detection.from_event(event) if (self._next_run_pipeline is not None) and ( self._next_run_pipeline.end_stage == PipelineStage.WAKE ): # Remote pipeline was triggered with only wake word detection wake_word_name: Optional[str] = None if detection.name: wake_word_name = normalize_wake_word(detection.name) if (not self._next_run_pipeline.wake_word_names) or ( wake_word_name in self._next_run_pipeline.wake_word_names ): self._next_run_pipeline = None await self.event_to_server(event) await self._send_wake_detect() _LOGGER.info("Waiting for wake word") else: _LOGGER.debug( "Got %s but listening for %s", wake_word_name, self._next_run_pipeline.wake_word_names, ) return # Check refractory period to avoid multiple back-to-back detections refractory_timestamp = self.refractory_timestamp.get(detection.name) if (refractory_timestamp is not None) and ( refractory_timestamp > time.monotonic() ): _LOGGER.debug("Wake word detection occurred during refractory period") return # Stop debug recording (wake) if self.wake_audio_writer is not None: self.wake_audio_writer.stop() # Start debug recording (stt) if self.stt_audio_writer is not None: self.stt_audio_writer.start(timestamp=self._debug_recording_timestamp) _LOGGER.debug(detection) self.segmenter.reset() self.is_streaming = True _LOGGER.debug("Streaming audio") if (not is_fake) and (self.settings.wake.refractory_seconds is not None): # Another detection may not occur for this wake word until # refractory period is over. self.refractory_timestamp[detection.name] = ( time.monotonic() + self.settings.wake.refractory_seconds ) else: # No refractory period self.refractory_timestamp.pop(detection.name, None) if not is_fake: # Forward to the server await self.event_to_server(event) start_stage: Optional[PipelineStage] = None end_stage: Optional[PipelineStage] = None if self._next_run_pipeline is not None: # Remote pipeline has been triggered start_stage = self._next_run_pipeline.start_stage end_stage = self._next_run_pipeline.end_stage await self._send_run_pipeline( start_stage=start_stage, end_stage=end_stage, wake_word_name=detection.name, ) if not is_fake: await self.forward_event(event) # forward to event service await self.trigger_detection(Detection.from_event(event)) await self.trigger_streaming_start() async def update_info(self, info: Info) -> None: self._wake_info = None self._wake_info_ready.clear() await self.event_to_wake(Describe().event()) try: await asyncio.wait_for( self._wake_info_ready.wait(), timeout=_WAKE_INFO_TIMEOUT ) if self._wake_info is not None: # Update wake info only info.wake = self._wake_info.wake except asyncio.TimeoutError: _LOGGER.warning("Failed to get info from wake service") async def _start_wake_word_detection(self) -> None: await self._send_wake_detect() _LOGGER.info("Waiting for wake word") # Start debug recording (wake) self._debug_recording_timestamp = time.monotonic_ns() if self.wake_audio_writer is not None: self.wake_audio_writer.start(timestamp=self._debug_recording_timestamp) async def _trigger_pipeline(self, run_pipeline: RunPipeline) -> None: _LOGGER.debug("Pipeline remotely triggered: %s", run_pipeline) if run_pipeline.start_stage == PipelineStage.WAKE: if run_pipeline.wake_word_names: run_pipeline.wake_word_names = [ normalize_wake_word(name) for name in run_pipeline.wake_word_names ] _LOGGER.debug( "Listening once for wake words: %s", run_pipeline.wake_word_names ) await self.event_to_wake(Detect(names=run_pipeline.wake_word_names).event()) elif run_pipeline.start_stage == PipelineStage.ASR: _LOGGER.debug("Listening for voice command once") # Fake a detection await self.event_from_wake(Detection().event(), is_fake=True) elif run_pipeline.start_stage == PipelineStage.TTS: _LOGGER.debug("Speaking once: %s", run_pipeline.announce_text) await self.event_to_server(run_pipeline.event())