From 22018e6b0543ca30e689bd86d52fad1412e04364 Mon Sep 17 00:00:00 2001 From: Michael Hansen Date: Sun, 10 Dec 2023 16:24:04 -0600 Subject: [PATCH] Working on refactor --- wyoming_satellite/__main__.py | 1206 ++++++++++++++++---------------- wyoming_satellite/satellite.py | 617 ++++++++++++++++ wyoming_satellite/settings.py | 80 +++ wyoming_satellite/utils.py | 28 +- wyoming_satellite/vad.py | 33 + 5 files changed, 1364 insertions(+), 600 deletions(-) create mode 100644 wyoming_satellite/satellite.py create mode 100644 wyoming_satellite/settings.py create mode 100644 wyoming_satellite/vad.py diff --git a/wyoming_satellite/__main__.py b/wyoming_satellite/__main__.py index 9a0dab4..0f99403 100644 --- a/wyoming_satellite/__main__.py +++ b/wyoming_satellite/__main__.py @@ -29,25 +29,26 @@ from wyoming.tts import Synthesize from wyoming.vad import VoiceStarted, VoiceStopped from wyoming.wake import Detect, Detection, WakeProcessAsyncClient -from .utils import AudioBuffer, chunk_samples, multiply_volume +from .satellite import ( + AlwaysStreamingSatellite, + SatelliteBase, + VadStreamingSatellite, + WakeStreamingSatellite, +) +from .settings import ( + EventSettings, + MicSettings, + SatelliteSettings, + SndSettings, + VadSettings, + WakeSettings, +) +from .utils import AudioBuffer, chunk_samples, multiply_volume, run_event_command _LOGGER = logging.getLogger() _DIR = Path(__file__).parent -class SatelliteState(str, Enum): - WAKE_START = "wake-start" - WAKE = "wake" - ASR = "asr" - VAD_RESET = "vad-reset" - VAD_WAIT = "vad-wait" - PLAYING_AUDIO = "playing-audio" - - -class ServerState: - current_client_id: Optional[str] = None - - async def main() -> None: """Main entry point.""" parser = argparse.ArgumentParser() @@ -210,7 +211,6 @@ async def main() -> None: parser.add_argument("--uri", default="stdio://", help="unix:// or tcp://") parser.add_argument("--name", required=True, help="Name of the satellite") parser.add_argument("--area", help="Area name of the satellite") - parser.add_argument("--pipeline", help="Name of server pipeline to run") # Zeroconf parser.add_argument( @@ -271,6 +271,35 @@ async def main() -> None: ) ) + settings = SatelliteSettings( + mic=MicSettings(uri=args.mic_uri, command=_split(args.mic_command)), + vad=VadSettings( + enabled=args.vad, + threshold=args.vad_threshold, + trigger_level=args.vad_trigger_level, + buffer_seconds=args.vad_buffer_seconds, + wake_word_timeout=args.vad_wake_word_timeout, + ), + wake=WakeSettings( + uri=args.wake_uri, + command=_split(args.wake_command), + names=args.wake_word_name, + ), + snd=SndSettings(uri=args.snd_uri, command=_split(args.snd_command)), + event=EventSettings( + uri=args.event_uri, streaming_start=_split(args.streaming_start_command) + ), + ) + + satellite: SatelliteBase + + if settings.wake.enabled: + satellite = WakeStreamingSatellite(settings) + elif settings.vad.enabled: + satellite = VadStreamingSatellite(settings) + else: + satellite = AlwaysStreamingSatellite(settings) + if args.startup_command: await run_event_command(args.startup_command) @@ -278,7 +307,6 @@ async def main() -> None: # Start server server = AsyncServer.from_uri(args.uri) - server_state = ServerState() if (not args.no_zeroconf) and isinstance(server, AsyncTcpServer): from wyoming.zeroconf import register_server @@ -298,12 +326,15 @@ async def main() -> None: args.zeroconf_host, ) + satellite_task = asyncio.create_task(satellite.run()) + try: - await server.run( - partial(SatelliteEventHandler, wyoming_info, server_state, args) - ) + await server.run(partial(SatelliteEventHandler, wyoming_info, satellite, args)) except KeyboardInterrupt: pass + finally: + await satellite.stop() + await satellite_task # ----------------------------------------------------------------------------- @@ -315,7 +346,7 @@ class SatelliteEventHandler(AsyncEventHandler): def __init__( self, wyoming_info: Info, - server_state: ServerState, + satellite: SatelliteBase, cli_args: argparse.Namespace, *args, **kwargs, @@ -325,562 +356,582 @@ class SatelliteEventHandler(AsyncEventHandler): self.cli_args = cli_args self.wyoming_info_event = wyoming_info.event() self.client_id = str(time.monotonic_ns()) + self.satellite = satellite - self.is_running = True - self.state: Optional[Satellite] = None - self.after_tts_state: Optional[Satellite] = None - self.satellite_task: Optional[asyncio.Task] = None + # self.is_running = True + # self.state: Optional[Satellite] = None + # self.after_tts_state: Optional[Satellite] = None + # self.satellite_task: Optional[asyncio.Task] = None - self.mic_volume_multiplier = self.cli_args.mic_volume_multiplier - self.has_mic_audio_processing = needs_webrtc(self.cli_args) or ( - self.mic_volume_multiplier != 1.0 - ) - self.process_webrtc: Optional[Callable[[bytes], bytes]] = None - if needs_webrtc(self.cli_args): - self.process_webrtc = WebRtcAudio( - self.cli_args.auto_gain, self.cli_args.noise_suppression - ) + # self.mic_volume_multiplier = self.cli_args.mic_volume_multiplier + # self.has_mic_audio_processing = needs_webrtc(self.cli_args) or ( + # self.mic_volume_multiplier != 1.0 + # ) + # self.process_webrtc: Optional[Callable[[bytes], bytes]] = None + # if needs_webrtc(self.cli_args): + # self.process_webrtc = WebRtcAudio( + # self.cli_args.auto_gain, self.cli_args.noise_suppression + # ) - self.has_wake = bool(self.cli_args.wake_uri or self.cli_args.wake_command) + # self.has_wake = bool(self.cli_args.wake_uri or self.cli_args.wake_command) - self.has_snd = bool(self.cli_args.snd_uri or self.cli_args.snd_command) - self.snd_client: Optional[AsyncClient] = None - self.snd_volume_multiplier = self.cli_args.snd_volume_multiplier - self.has_snd_audio_processing = self.snd_volume_multiplier != 1.0 + # self.has_snd = bool(self.cli_args.snd_uri or self.cli_args.snd_command) + # self.snd_client: Optional[AsyncClient] = None + # self.snd_volume_multiplier = self.cli_args.snd_volume_multiplier + # self.has_snd_audio_processing = self.snd_volume_multiplier != 1.0 - self.process_vad: Optional[Callable[[Optional[bytes]], bool]] = None - self.vad_buffer: Optional[RingBuffer] = None - if needs_silero(self.cli_args): - self.process_vad = SileroVad( - self.cli_args.vad_threshold, self.cli_args.vad_trigger_level - ) - if self.cli_args.vad_buffer_seconds > 0: - # Assume 16Khz, 16-bit mono samples - vad_buffer_bytes = int( - math.ceil(self.cli_args.vad_buffer_seconds * 16000 * 2) - ) - self.vad_buffer = RingBuffer(vad_buffer_bytes) + # self.process_vad: Optional[Callable[[Optional[bytes]], bool]] = None + # self.vad_buffer: Optional[RingBuffer] = None + # if needs_silero(self.cli_args): + # self.process_vad = SileroVad( + # self.cli_args.vad_threshold, self.cli_args.vad_trigger_level + # ) + # if self.cli_args.vad_buffer_seconds > 0: + # # Assume 16Khz, 16-bit mono samples + # vad_buffer_bytes = int( + # math.ceil(self.cli_args.vad_buffer_seconds * 16000 * 2) + # ) + # self.vad_buffer = RingBuffer(vad_buffer_bytes) - self.events_client: Optional[AsyncClient] = None + # self.events_client: Optional[AsyncClient] = None - self.server_state = server_state - self.server_state.current_client_id = self.client_id + # self.server_state = server_state + # self.server_state.current_client_id = self.client_id _LOGGER.debug("Client connected: %s", self.client_id) + self.satellite.set_server(self.client_id, self.writer) - self._stop_if_new_client_task = asyncio.create_task(self._stop_if_new_client()) + # self._stop_if_new_client_task = asyncio.create_task(self._stop_if_new_client()) - async def _stop_if_new_client(self) -> None: - """Stop this event handler if a new client connects.""" - while self.is_running: - if self.server_state.current_client_id != self.client_id: - self.is_running = False - _LOGGER.debug("Detected new client. Stopping old client.") - else: - await asyncio.sleep(1) + # async def _stop_if_new_client(self) -> None: + # """Stop this event handler if a new client connects.""" + # while self.is_running: + # if self.server_state.current_client_id != self.client_id: + # self.is_running = False + # _LOGGER.debug("Detected new client. Stopping old client.") + # else: + # await asyncio.sleep(1) # ------------------------------------------------------------------------- async def handle_event(self, event: Event) -> bool: """Handle events from the server.""" + if self.satellite.server_id != self.client_id: + # New connection + return False + if Describe.is_type(event.type): await self.write_event(self.wyoming_info_event) _LOGGER.debug("Sent info to client: %s", self.client_id) return True - if RunSatellite.is_type(event.type) and (self.satellite_task is None): - if self.has_wake: - # Local wake word detection - self.state = SatelliteState.WAKE_START - self.satellite_task = asyncio.create_task(self.run_satellite_wake()) - else: - # Remote wake word detection - self.state = SatelliteState.VAD_RESET - self.satellite_task = asyncio.create_task(self.run_satellite()) - elif Transcript.is_type(event.type): - if self.cli_args.done_wav: - self.state = SatelliteState.PLAYING_AUDIO - await self._play_wav(self.cli_args.done_wav) - - # STT transcript - await self._forward_event(event) - _LOGGER.debug("-> %s", event) - - if self.has_wake: - # Ready for next wake word detection - self.state = SatelliteState.WAKE_START - else: - self.state = SatelliteState.VAD_RESET - elif self.has_snd and ( - AudioStart.is_type(event.type) - or AudioChunk.is_type(event.type) - or AudioStop.is_type(event.type) - ): - if AudioStart.is_type(event.type): - # TTS start - await self._ensure_snd_client() - if self.snd_client is not None: - await self.snd_client.connect() - - # State will be restored when audio ends - self.after_tts_state = self.state - self.state = SatelliteState.PLAYING_AUDIO - - await self._forward_event(event) - _LOGGER.debug("-> %s", event) - elif AudioChunk.is_type(event.type) and (self.snd_client is not None): - # Forward to sound service - event = self._process_snd_audio(event) - await self.snd_client.write_event(event) - elif AudioStop.is_type(event.type): - # TTS stop - await self._forward_event(event) - _LOGGER.debug("-> %s", event) - - if self.snd_client is not None: - await self.snd_client.disconnect() - self.snd_client = None - - if self.after_tts_state is not None: - # Restore state after playing audio - self.state = self.after_tts_state - self.after_tts_state = None - elif Detection.is_type(event.type): - if self.cli_args.awake_wav: - last_state = self.state - self.state = SatelliteState.PLAYING_AUDIO - await self._play_wav(self.cli_args.awake_wav) - self.state = last_state - - await self._forward_event(event) - _LOGGER.debug("-> %s", event) - elif ( - Detect.is_type(event.type) - or Transcribe.is_type(event.type) - or VoiceStarted.is_type(event.type) - or VoiceStopped.is_type(event.type) - or Synthesize.is_type(event.type) - ): - # Other client events: - # - Detect for start of wake word detection - # - Transcribe for when STT starts - # - VoiceStarted/VoiceStopped for when user starts/stops speaking - # - Synthesize for TTS text - await self._forward_event(event) - _LOGGER.debug("-> %s", event) - elif Error.is_type(event.type): - await self._forward_event(event) - _LOGGER.warning(event) - else: - _LOGGER.debug("Unexpected event: type=%s, data=%s", event.type, event.data) + await self.satellite.event_from_server(event) return True - async def disconnect(self) -> None: - self.is_running = False - - # ------------------------------------------------------------------------- - # Tasks - # ------------------------------------------------------------------------- - - async def run_satellite(self) -> None: - """Task to read mic input and do remote wake word detection.""" - vad_timeout_sec: Optional[float] = None - - try: - mic_client = self._make_mic_client() - async with mic_client: - while self.is_running: - mic_event = await mic_client.read_event() - if mic_event is None: - _LOGGER.warning("Microphone service disconnected") - break - - if not AudioChunk.is_type(mic_event.type): - continue - - mic_event = self._process_mic_audio(mic_event) - - if self.state == SatelliteState.VAD_RESET: - # Reset vad state - if self.process_vad is not None: - self.process_vad(None) - - if self.vad_buffer is not None: - self.vad_buffer.put(bytes(self.vad_buffer.maxlen)) - - self.state = SatelliteState.VAD_WAIT - if self.process_vad is not None: - _LOGGER.info("Waiting for speech") - - if self.state == SatelliteState.VAD_WAIT: - if self.process_vad is None: - # No VAD - self.state = SatelliteState.ASR - _LOGGER.info("Streaming audio") - - # Start pipeline - await self._run_pipeline(PipelineStage.WAKE) - else: - # Wait for speech - chunk = AudioChunk.from_event(mic_event) - if self.process_vad(chunk.audio): - # Ready to stream - self.state = SatelliteState.WAKE - - # Start pipeline - await self._run_pipeline(PipelineStage.WAKE) - - if self.vad_buffer: - # Send contents of VAD buffer first - await self.write_event( - AudioChunk( - rate=chunk.rate, - width=chunk.width, - channels=chunk.channels, - audio=self.vad_buffer.getvalue(), - ).event() - ) - - if self.cli_args.vad_wake_word_timeout is not None: - # Time in the future when timeout will occur - vad_timeout_sec = ( - time.monotonic() - + self.cli_args.vad_wake_word_timeout - ) - else: - # No timeout - vad_timeout_sec = None - - _LOGGER.info("Streaming audio") - elif self.vad_buffer is not None: - # Save audio right before speech - self.vad_buffer.put(chunk.audio) - - if self.state == SatelliteState.WAKE: - if (vad_timeout_sec is not None) and ( - time.monotonic() >= vad_timeout_sec - ): - # Wake word was not detected within timeout - self.state = SatelliteState.VAD_RESET - - # Cancel current pipeline with empty audio chunk - chunk = AudioChunk.from_event(mic_event) - await self.write_event( - AudioChunk( - rate=chunk.rate, - width=chunk.width, - channels=chunk.channels, - audio=bytes(), - ).event() - ) - continue - - # Remote wake word detection - await self.write_event(mic_event) - elif self.state == SatelliteState.ASR: - # Forward all audio to server - await self.write_event(mic_event) - - except ConnectionResetError: - _LOGGER.info("Server disconnected") - except Exception: - _LOGGER.exception("Unexpected error in run_satellite") - - async def run_satellite_wake(self) -> None: - """Task to read mic input and do local wake word detection.""" - try: - assert self.cli_args.wake_uri or self.cli_args.wake_command - - mic_client = self._make_mic_client() - - if self.cli_args.wake_command: - _LOGGER.debug("Running %s", self.cli_args.wake_command) - program, *program_args = shlex.split(self.cli_args.wake_command) - wake_client = WakeProcessAsyncClient( - rate=self.cli_args.wake_command_rate, - width=self.cli_args.wake_command_width, - channels=self.cli_args.wake_command_channels, - program=program, - program_args=program_args, - ) - else: - _LOGGER.debug("Connecting to %s", self.cli_args.wake_uri) - wake_client = AsyncClient.from_uri(self.cli_args.wake_uri) - - async with mic_client, wake_client: - # Read events in parallel - mic_task = asyncio.create_task(mic_client.read_event()) - wake_task = asyncio.create_task(wake_client.read_event()) - pending = {mic_task, wake_task} - - while self.is_running: - if self.state == SatelliteState.WAKE_START: - # Send detect event - _LOGGER.info("Detecting wake word") - - if self.cli_args.wake_word_name: - # Request specific wake word(s) - detect_event = Detect( - names=self.cli_args.wake_word_name - ).event() - else: - detect_event = Detect().event() - - _LOGGER.debug("<- %s", detect_event) - await wake_client.write_event(detect_event) - await self._forward_event(detect_event) - - self.state = SatelliteState.WAKE - - # Wait for mic/wake events - done, pending = await asyncio.wait( - pending, return_when=asyncio.FIRST_COMPLETED - ) - - if mic_task in done: - mic_event = mic_task.result() - if mic_event is None: - _LOGGER.warning("Microphone service disconnected") - break - - if AudioChunk.is_type(mic_event.type): - mic_event = self._process_mic_audio(mic_event) - - # Forward to wake service - await wake_client.write_event(mic_event) - - if self.state == SatelliteState.ASR: - # Forward to server - await self.write_event(mic_event) - - # Next audio chunk - mic_task = asyncio.create_task(mic_client.read_event()) - pending.add(mic_task) - elif wake_task in done: - wake_event = wake_task.result() - if wake_event is None: - _LOGGER.warning("Wake word service disconnected") - break - - if Detection.is_type(wake_event.type) and ( - self.state == SatelliteState.WAKE - ): - _LOGGER.debug("Wake word detected: %s", wake_event) - - if self.cli_args.awake_wav: - self.state = SatelliteState.PLAYING_AUDIO - await self._play_wav(self.cli_args.awake_wav) - - self.state = SatelliteState.ASR - - # Start pipeline - await self._run_pipeline(PipelineStage.ASR) - - # Forward to server - await self.write_event(wake_event) - - # Forward to client - await self._forward_event(wake_event) - - _LOGGER.info("Streaming audio") - - # Next wake event - wake_task = asyncio.create_task(wake_client.read_event()) - pending.add(wake_task) - except ConnectionResetError: - _LOGGER.info("Server disconnected") - except Exception: - _LOGGER.exception("Unexpected error in run_satellite") - - # ------------------------------------------------------------------------- - - def _make_mic_client(self) -> AsyncClient: - """Create mic client.""" - if self.cli_args.mic_command: - _LOGGER.debug("Running %s", self.cli_args.mic_command) - program, *program_args = shlex.split(self.cli_args.mic_command) - return MicProcessAsyncClient( - rate=self.cli_args.mic_command_rate, - width=self.cli_args.mic_command_width, - channels=self.cli_args.mic_command_channels, - samples_per_chunk=self.cli_args.mic_command_samples_per_chunk, - program=program, - program_args=program_args, - ) - - assert self.cli_args.mic_uri - _LOGGER.debug("Connecting to %s", self.cli_args.mic_uri) - return AsyncClient.from_uri(self.cli_args.mic_uri) - - async def _ensure_snd_client(self) -> None: - """Create snd client if necessary.""" - if (not self.has_snd) or (self.snd_client is not None): - return - - if self.cli_args.snd_command: - _LOGGER.debug("Running %s", self.cli_args.snd_command) - program, *program_args = shlex.split(self.cli_args.snd_command) - self.snd_client = SndProcessAsyncClient( - rate=self.cli_args.snd_command_rate, - width=self.cli_args.snd_command_width, - channels=self.cli_args.snd_command_channels, - program=program, - program_args=program_args, - ) - else: - _LOGGER.debug("Connecting to %s", self.cli_args.snd_uri) - self.snd_client = AsyncClient.from_uri(self.cli_args.snd_uri) - - async def _run_pipeline(self, start_stage: PipelineStage) -> None: - """Tell server to start running the pipeline.""" - await self.write_event( - RunPipeline( - start_stage=start_stage, - end_stage=PipelineStage.TTS if self.has_snd else PipelineStage.HANDLE, - name=self.cli_args.pipeline, - ).event() - ) - - if self.cli_args.streaming_start_command: - await run_event_command(self.cli_args.streaming_start_command) - - def _process_mic_audio(self, audio_event: Event) -> Event: - """Perform microphone audio processing, if necessary.""" - if not self.has_mic_audio_processing: - # No processing needed - return audio_event - - chunk = AudioChunk.from_event(audio_event) - audio_bytes = chunk.audio - - if self.process_webrtc is not None: - audio_bytes = self.process_webrtc(audio_bytes) - - if self.mic_volume_multiplier != 1.0: - audio_bytes = multiply_volume(audio_bytes, self.mic_volume_multiplier) - - return AudioChunk( - rate=chunk.rate, - width=chunk.width, - channels=chunk.channels, - audio=audio_bytes, - ).event() - - def _process_snd_audio(self, audio_event: Event) -> Event: - """Perform output audio processing, if necessary.""" - if not self.has_snd_audio_processing: - # No processing needed - return audio_event - - chunk = AudioChunk.from_event(audio_event) - audio_bytes = chunk.audio - - if self.snd_volume_multiplier != 1.0: - audio_bytes = multiply_volume(audio_bytes, self.snd_volume_multiplier) - - return AudioChunk( - rate=chunk.rate, - width=chunk.width, - channels=chunk.channels, - audio=audio_bytes, - ).event() - - async def _forward_event(self, event: Event) -> None: - """Forward a Wyoming event to a client and run event commands.""" - if self.cli_args.detect_command and Detect.is_type(event.type): - # Wake word detection started - await run_event_command(self.cli_args.detect_command) - elif self.cli_args.detection_command and Detection.is_type(event.type): - # Wake word is detected - detection = Detection.from_event(event) - await run_event_command(self.cli_args.detection_command, detection.name) - elif self.cli_args.transcript_command and Transcript.is_type(event.type): - # STT text is available - transcript = Transcript.from_event(event) - await run_event_command(self.cli_args.transcript_command, transcript.text) - elif self.cli_args.stt_start_command and VoiceStarted.is_type(event.type): - # User starts speaking - await run_event_command(self.cli_args.stt_start_command) - elif self.cli_args.stt_stop_command and VoiceStopped.is_type(event.type): - # User stops speaking - await run_event_command(self.cli_args.stt_stop_command) - elif self.cli_args.synthesize_command and Synthesize.is_type(event.type): - # TTS text is available - synthesize = Synthesize.from_event(event) - await run_event_command(self.cli_args.synthesize_command, synthesize.text) - elif self.cli_args.tts_start_command and AudioStart.is_type(event.type): - # TTS audio start - await run_event_command(self.cli_args.tts_start_command) - elif self.cli_args.tts_stop_command and AudioStop.is_type(event.type): - # TTS audio stop - await run_event_command(self.cli_args.tts_stop_command) - elif self.cli_args.error_command and Error.is_type(event.type): - # Error occurred - error = Error.from_event(event) - await run_event_command(self.cli_args.error_command, error.text) - - if not self.cli_args.event_uri: - # No external service - return - - if self.events_client is None: - self.events_client = AsyncClient.from_uri(self.cli_args.event_uri) - _LOGGER.debug("Connecting to %s", self.cli_args.event_uri) - await self.events_client.connect() - - assert self.events_client is not None - await self.events_client.write_event(event) - - async def _play_wav(self, wav_path: Union[str, Path]) -> None: - """Send WAV file to snd client as audio chunks.""" - await self._ensure_snd_client() - - if self.snd_client is None: - _LOGGER.debug("Cannot play WAV, no snd service") - return - - try: - async with self.snd_client: - for event in self._wav_to_events(wav_path): - await self.snd_client.write_event(event) - except Exception: - _LOGGER.exception("Unexpected error while playing WAV: %s", wav_path) - finally: - self.snd_client = None - - def _wav_to_events( - self, wav_path: Union[str, Path], samples_per_chunk: int = 1024 - ) -> Iterator[Event]: - """Load WAV audio for playback on an event (wake/done).""" - with wave.open(str(wav_path), "rb") as wav_file: - rate = wav_file.getframerate() - width = wav_file.getsampwidth() - channels = wav_file.getnchannels() - - timestamp = 0 - yield AudioStart( - rate=rate, width=width, channels=channels, timestamp=timestamp - ).event() - - audio_bytes = wav_file.readframes(samples_per_chunk) - while audio_bytes: - if self.snd_volume_multiplier != 1.0: - audio_bytes = multiply_volume( - audio_bytes, self.snd_volume_multiplier - ) - - chunk = AudioChunk( - rate=rate, - width=width, - channels=channels, - audio=audio_bytes, - timestamp=timestamp, - ) - yield chunk.event() - timestamp += int(chunk.seconds * 1000) - audio_bytes = wav_file.readframes(samples_per_chunk) - - yield AudioStop(timestamp=timestamp).event() + # async def handle_event(self, event: Event) -> bool: + # """Handle events from the server.""" + # if Describe.is_type(event.type): + # await self.write_event(self.wyoming_info_event) + # _LOGGER.debug("Sent info to client: %s", self.client_id) + # return True + + # if RunSatellite.is_type(event.type) and (self.satellite_task is None): + # if self.has_wake: + # # Local wake word detection + # self.state = SatelliteState.WAKE_START + # self.satellite_task = asyncio.create_task(self.run_satellite_wake()) + # else: + # # Remote wake word detection + # self.state = SatelliteState.VAD_RESET + # self.satellite_task = asyncio.create_task(self.run_satellite()) + # elif Transcript.is_type(event.type): + # if self.cli_args.done_wav: + # self.state = SatelliteState.PLAYING_AUDIO + # await self._play_wav(self.cli_args.done_wav) + + # # STT transcript + # await self._forward_event(event) + # _LOGGER.debug("-> %s", event) + + # if self.has_wake: + # # Ready for next wake word detection + # self.state = SatelliteState.WAKE_START + # else: + # self.state = SatelliteState.VAD_RESET + # elif self.has_snd and ( + # AudioStart.is_type(event.type) + # or AudioChunk.is_type(event.type) + # or AudioStop.is_type(event.type) + # ): + # if AudioStart.is_type(event.type): + # # TTS start + # await self._ensure_snd_client() + # if self.snd_client is not None: + # await self.snd_client.connect() + + # # State will be restored when audio ends + # self.after_tts_state = self.state + # self.state = SatelliteState.PLAYING_AUDIO + + # await self._forward_event(event) + # _LOGGER.debug("-> %s", event) + # elif AudioChunk.is_type(event.type) and (self.snd_client is not None): + # # Forward to sound service + # event = self._process_snd_audio(event) + # await self.snd_client.write_event(event) + # elif AudioStop.is_type(event.type): + # # TTS stop + # await self._forward_event(event) + # _LOGGER.debug("-> %s", event) + + # if self.snd_client is not None: + # await self.snd_client.disconnect() + # self.snd_client = None + + # if self.after_tts_state is not None: + # # Restore state after playing audio + # self.state = self.after_tts_state + # self.after_tts_state = None + # elif Detection.is_type(event.type): + # if self.cli_args.awake_wav: + # last_state = self.state + # self.state = SatelliteState.PLAYING_AUDIO + # await self._play_wav(self.cli_args.awake_wav) + # self.state = last_state + + # await self._forward_event(event) + # _LOGGER.debug("-> %s", event) + # elif ( + # Detect.is_type(event.type) + # or Transcribe.is_type(event.type) + # or VoiceStarted.is_type(event.type) + # or VoiceStopped.is_type(event.type) + # or Synthesize.is_type(event.type) + # ): + # # Other client events: + # # - Detect for start of wake word detection + # # - Transcribe for when STT starts + # # - VoiceStarted/VoiceStopped for when user starts/stops speaking + # # - Synthesize for TTS text + # await self._forward_event(event) + # _LOGGER.debug("-> %s", event) + # elif Error.is_type(event.type): + # await self._forward_event(event) + # _LOGGER.warning(event) + # else: + # _LOGGER.debug("Unexpected event: type=%s, data=%s", event.type, event.data) + + # return True + + # async def disconnect(self) -> None: + # self.is_running = False + + # # ------------------------------------------------------------------------- + # # Tasks + # # ------------------------------------------------------------------------- + + # async def run_satellite(self) -> None: + # """Task to read mic input and do remote wake word detection.""" + # vad_timeout_sec: Optional[float] = None + + # try: + # mic_client = self._make_mic_client() + # async with mic_client: + # while self.is_running: + # mic_event = await mic_client.read_event() + # if mic_event is None: + # _LOGGER.warning("Microphone service disconnected") + # break + + # if not AudioChunk.is_type(mic_event.type): + # continue + + # mic_event = self._process_mic_audio(mic_event) + + # if self.state == SatelliteState.VAD_RESET: + # # Reset vad state + # if self.process_vad is not None: + # self.process_vad(None) + + # if self.vad_buffer is not None: + # self.vad_buffer.put(bytes(self.vad_buffer.maxlen)) + + # self.state = SatelliteState.VAD_WAIT + # if self.process_vad is not None: + # _LOGGER.info("Waiting for speech") + + # if self.state == SatelliteState.VAD_WAIT: + # if self.process_vad is None: + # # No VAD + # self.state = SatelliteState.ASR + # _LOGGER.info("Streaming audio") + + # # Start pipeline + # await self._run_pipeline(PipelineStage.WAKE) + # else: + # # Wait for speech + # chunk = AudioChunk.from_event(mic_event) + # if self.process_vad(chunk.audio): + # # Ready to stream + # self.state = SatelliteState.WAKE + + # # Start pipeline + # await self._run_pipeline(PipelineStage.WAKE) + + # if self.vad_buffer: + # # Send contents of VAD buffer first + # await self.write_event( + # AudioChunk( + # rate=chunk.rate, + # width=chunk.width, + # channels=chunk.channels, + # audio=self.vad_buffer.getvalue(), + # ).event() + # ) + + # if self.cli_args.vad_wake_word_timeout is not None: + # # Time in the future when timeout will occur + # vad_timeout_sec = ( + # time.monotonic() + # + self.cli_args.vad_wake_word_timeout + # ) + # else: + # # No timeout + # vad_timeout_sec = None + + # _LOGGER.info("Streaming audio") + # elif self.vad_buffer is not None: + # # Save audio right before speech + # self.vad_buffer.put(chunk.audio) + + # if self.state == SatelliteState.WAKE: + # if (vad_timeout_sec is not None) and ( + # time.monotonic() >= vad_timeout_sec + # ): + # # Wake word was not detected within timeout + # self.state = SatelliteState.VAD_RESET + + # # Cancel current pipeline with audio stop + # chunk = AudioChunk.from_event(mic_event) + # await self.write_event(AudioStop().event()) + # continue + + # # Remote wake word detection + # await self.write_event(mic_event) + # elif self.state == SatelliteState.ASR: + # # Forward all audio to server + # await self.write_event(mic_event) + + # except ConnectionResetError: + # _LOGGER.info("Server disconnected") + # except Exception: + # _LOGGER.exception("Unexpected error in run_satellite") + + # async def run_satellite_wake(self) -> None: + # """Task to read mic input and do local wake word detection.""" + # try: + # mic_client = self._make_mic_client() + # wake_client = await self._connect_to_wake() + + # async with mic_client, wake_client: + # # Read events in parallel + # mic_task = asyncio.create_task(mic_client.read_event()) + # wake_task = asyncio.create_task(wake_client.read_event()) + # pending = {mic_task, wake_task} + + # while self.is_running: + # if self.state == SatelliteState.WAKE_START: + # # Send detect event + # _LOGGER.info("Detecting wake word") + + # if self.cli_args.wake_word_name: + # # Request specific wake word(s) + # detect_event = Detect( + # names=self.cli_args.wake_word_name + # ).event() + # else: + # detect_event = Detect().event() + + # _LOGGER.debug("<- %s", detect_event) + # await wake_client.write_event(detect_event) + # await self._forward_event(detect_event) + + # self.state = SatelliteState.WAKE + + # # Wait for mic/wake events + # done, pending = await asyncio.wait( + # pending, return_when=asyncio.FIRST_COMPLETED + # ) + + # if mic_task in done: + # mic_event = mic_task.result() + # if mic_event is None: + # _LOGGER.warning("Microphone service disconnected") + # break + + # if AudioChunk.is_type(mic_event.type): + # mic_event = self._process_mic_audio(mic_event) + + # # Forward to wake service + # await wake_client.write_event(mic_event) + + # if self.state == SatelliteState.ASR: + # # Forward to server + # await self.write_event(mic_event) + + # # Next audio chunk + # mic_task = asyncio.create_task(mic_client.read_event()) + # pending.add(mic_task) + # elif wake_task in done: + # wake_event = wake_task.result() + # if wake_event is None: + # _LOGGER.warning("Wake word service disconnected") + # break + + # if Detection.is_type(wake_event.type) and ( + # self.state == SatelliteState.WAKE + # ): + # _LOGGER.debug("Wake word detected: %s", wake_event) + + # if self.cli_args.awake_wav: + # self.state = SatelliteState.PLAYING_AUDIO + # await self._play_wav(self.cli_args.awake_wav) + + # self.state = SatelliteState.ASR + + # # Start pipeline + # await self._run_pipeline(PipelineStage.ASR) + + # # Forward to server + # await self.write_event(wake_event) + + # # Forward to client + # await self._forward_event(wake_event) + + # _LOGGER.info("Streaming audio") + + # # Next wake event + # wake_task = asyncio.create_task(wake_client.read_event()) + # pending.add(wake_task) + # except ConnectionResetError: + # _LOGGER.info("Server disconnected") + # except Exception: + # _LOGGER.exception("Unexpected error in run_satellite") + + # async def _connect_to_wake(self) -> AsyncClient: + # assert self.cli_args.wake_uri or self.cli_args.wake_command + + # if self.cli_args.wake_command: + # _LOGGER.debug("Running %s", self.cli_args.wake_command) + # program, *program_args = shlex.split(self.cli_args.wake_command) + # return WakeProcessAsyncClient( + # rate=self.cli_args.wake_command_rate, + # width=self.cli_args.wake_command_width, + # channels=self.cli_args.wake_command_channels, + # program=program, + # program_args=program_args, + # ) + + # while True: + # try: + # _LOGGER.debug("Connecting to %s", self.cli_args.wake_uri) + # return AsyncClient.from_uri(self.cli_args.wake_uri) + # except (ConnectionError, OSError): + # _LOGGER.exception( + # "Unexpected error connecting to wake service. " + # "Reconnecting in %s second(s)", + # _WAKE_RECONNECT_SECONDS, + # ) + # await asyncio.sleep(_WAKE_RECONNECT_SECONDS) + + # # ------------------------------------------------------------------------- + + # def _make_mic_client(self) -> AsyncClient: + # """Create mic client.""" + # if self.cli_args.mic_command: + # _LOGGER.debug("Running %s", self.cli_args.mic_command) + # program, *program_args = shlex.split(self.cli_args.mic_command) + # return MicProcessAsyncClient( + # rate=self.cli_args.mic_command_rate, + # width=self.cli_args.mic_command_width, + # channels=self.cli_args.mic_command_channels, + # samples_per_chunk=self.cli_args.mic_command_samples_per_chunk, + # program=program, + # program_args=program_args, + # ) + + # assert self.cli_args.mic_uri + # _LOGGER.debug("Connecting to %s", self.cli_args.mic_uri) + # return AsyncClient.from_uri(self.cli_args.mic_uri) + + # async def _ensure_snd_client(self) -> None: + # """Create snd client if necessary.""" + # if (not self.has_snd) or (self.snd_client is not None): + # return + + # if self.cli_args.snd_command: + # _LOGGER.debug("Running %s", self.cli_args.snd_command) + # program, *program_args = shlex.split(self.cli_args.snd_command) + # self.snd_client = SndProcessAsyncClient( + # rate=self.cli_args.snd_command_rate, + # width=self.cli_args.snd_command_width, + # channels=self.cli_args.snd_command_channels, + # program=program, + # program_args=program_args, + # ) + # else: + # _LOGGER.debug("Connecting to %s", self.cli_args.snd_uri) + # self.snd_client = AsyncClient.from_uri(self.cli_args.snd_uri) + + # async def _run_pipeline(self, start_stage: PipelineStage) -> None: + # """Tell server to start running the pipeline.""" + # await self.write_event( + # RunPipeline( + # start_stage=start_stage, + # end_stage=PipelineStage.TTS if self.has_snd else PipelineStage.HANDLE, + # ).event() + # ) + + # if self.cli_args.streaming_start_command: + # await run_event_command(self.cli_args.streaming_start_command) + + # def _process_mic_audio(self, audio_event: Event) -> Event: + # """Perform microphone audio processing, if necessary.""" + # if not self.has_mic_audio_processing: + # # No processing needed + # return audio_event + + # chunk = AudioChunk.from_event(audio_event) + # audio_bytes = chunk.audio + + # if self.process_webrtc is not None: + # audio_bytes = self.process_webrtc(audio_bytes) + + # if self.mic_volume_multiplier != 1.0: + # audio_bytes = multiply_volume(audio_bytes, self.mic_volume_multiplier) + + # return AudioChunk( + # rate=chunk.rate, + # width=chunk.width, + # channels=chunk.channels, + # audio=audio_bytes, + # ).event() + + # def _process_snd_audio(self, audio_event: Event) -> Event: + # """Perform output audio processing, if necessary.""" + # if not self.has_snd_audio_processing: + # # No processing needed + # return audio_event + + # chunk = AudioChunk.from_event(audio_event) + # audio_bytes = chunk.audio + + # if self.snd_volume_multiplier != 1.0: + # audio_bytes = multiply_volume(audio_bytes, self.snd_volume_multiplier) + + # return AudioChunk( + # rate=chunk.rate, + # width=chunk.width, + # channels=chunk.channels, + # audio=audio_bytes, + # ).event() + + # async def _forward_event(self, event: Event) -> None: + # """Forward a Wyoming event to a client and run event commands.""" + # if self.cli_args.detect_command and Detect.is_type(event.type): + # # Wake word detection started + # await run_event_command(self.cli_args.detect_command) + # elif self.cli_args.detection_command and Detection.is_type(event.type): + # # Wake word is detected + # detection = Detection.from_event(event) + # await run_event_command(self.cli_args.detection_command, detection.name) + # elif self.cli_args.transcript_command and Transcript.is_type(event.type): + # # STT text is available + # transcript = Transcript.from_event(event) + # await run_event_command(self.cli_args.transcript_command, transcript.text) + # elif self.cli_args.stt_start_command and VoiceStarted.is_type(event.type): + # # User starts speaking + # await run_event_command(self.cli_args.stt_start_command) + # elif self.cli_args.stt_stop_command and VoiceStopped.is_type(event.type): + # # User stops speaking + # await run_event_command(self.cli_args.stt_stop_command) + # elif self.cli_args.synthesize_command and Synthesize.is_type(event.type): + # # TTS text is available + # synthesize = Synthesize.from_event(event) + # await run_event_command(self.cli_args.synthesize_command, synthesize.text) + # elif self.cli_args.tts_start_command and AudioStart.is_type(event.type): + # # TTS audio start + # await run_event_command(self.cli_args.tts_start_command) + # elif self.cli_args.tts_stop_command and AudioStop.is_type(event.type): + # # TTS audio stop + # await run_event_command(self.cli_args.tts_stop_command) + # elif self.cli_args.error_command and Error.is_type(event.type): + # # Error occurred + # error = Error.from_event(event) + # await run_event_command(self.cli_args.error_command, error.text) + + # if not self.cli_args.event_uri: + # # No external service + # return + + # if self.events_client is None: + # self.events_client = AsyncClient.from_uri(self.cli_args.event_uri) + # _LOGGER.debug("Connecting to %s", self.cli_args.event_uri) + # await self.events_client.connect() + + # assert self.events_client is not None + # await self.events_client.write_event(event) + + # async def _play_wav(self, wav_path: Union[str, Path]) -> None: + # """Send WAV file to snd client as audio chunks.""" + # await self._ensure_snd_client() + + # if self.snd_client is None: + # _LOGGER.debug("Cannot play WAV, no snd service") + # return + + # try: + # async with self.snd_client: + # for event in self._wav_to_events(wav_path): + # await self.snd_client.write_event(event) + # except Exception: + # _LOGGER.exception("Unexpected error while playing WAV: %s", wav_path) + # finally: + # self.snd_client = None + + # def _wav_to_events( + # self, wav_path: Union[str, Path], samples_per_chunk: int = 1024 + # ) -> Iterator[Event]: + # """Load WAV audio for playback on an event (wake/done).""" + # with wave.open(str(wav_path), "rb") as wav_file: + # rate = wav_file.getframerate() + # width = wav_file.getsampwidth() + # channels = wav_file.getnchannels() + + # timestamp = 0 + # yield AudioStart( + # rate=rate, width=width, channels=channels, timestamp=timestamp + # ).event() + + # audio_bytes = wav_file.readframes(samples_per_chunk) + # while audio_bytes: + # if self.snd_volume_multiplier != 1.0: + # audio_bytes = multiply_volume( + # audio_bytes, self.snd_volume_multiplier + # ) + + # chunk = AudioChunk( + # rate=rate, + # width=width, + # channels=channels, + # audio=audio_bytes, + # timestamp=timestamp, + # ) + # yield chunk.event() + # timestamp += int(chunk.seconds * 1000) + # audio_bytes = wav_file.readframes(samples_per_chunk) + + # yield AudioStop(timestamp=timestamp).event() # ----------------------------------------------------------------------------- @@ -906,23 +957,11 @@ def needs_silero(args: argparse.Namespace) -> bool: return args.vad -async def run_event_command(command: str, command_input: Optional[str] = None) -> None: - """Run a custom event command with optional input.""" +def _split(command: Optional[str]) -> Optional[List[str]]: if not command: - return + return None - _LOGGER.debug("Running %s", command) - program, *program_args = shlex.split(command) - proc = await asyncio.create_subprocess_exec( - program, *program_args, stdin=asyncio.subprocess.PIPE - ) - assert proc.stdin is not None - - if command_input: - await proc.communicate(input=command_input.encode("utf-8")) - else: - proc.stdin.close() - await proc.wait() + return shlex.split(command) class WebRtcAudio: @@ -949,37 +988,6 @@ class WebRtcAudio: return clean_chunk -class SileroVad: - """Voice activity detection with silero VAD.""" - - def __init__(self, threshold: float, trigger_level: int) -> None: - from pysilero_vad import SileroVoiceActivityDetector - - self.detector = SileroVoiceActivityDetector() - self.threshold = threshold - self.trigger_level = trigger_level - self._activation = 0 - - def __call__(self, audio_bytes: Optional[bytes]) -> bool: - if audio_bytes is None: - # Reset - self._activation = 0 - self.detector.reset() - return False - - if self.detector(audio_bytes) >= self.threshold: - # Speech detected - self._activation += 1 - if self._activation >= self.trigger_level: - self._activation = 0 - return True - else: - # Silence detected - self._activation = max(0, self._activation - 1) - - return False - - # ----------------------------------------------------------------------------- if __name__ == "__main__": diff --git a/wyoming_satellite/satellite.py b/wyoming_satellite/satellite.py new file mode 100644 index 0000000..93f9d01 --- /dev/null +++ b/wyoming_satellite/satellite.py @@ -0,0 +1,617 @@ +"""Satellite code.""" +import asyncio +import logging +import math +import time +from dataclasses import dataclass +from enum import Enum, auto +from typing import List, Optional, Set + +from pyring_buffer import RingBuffer +from wyoming.asr import Transcript +from wyoming.audio import AudioChunk, AudioStart, AudioStop +from wyoming.client import AsyncClient, AsyncTcpClient +from wyoming.event import Event, async_read_event, async_write_event +from wyoming.mic import MicProcessAsyncClient +from wyoming.pipeline import PipelineStage, RunPipeline +from wyoming.satellite import RunSatellite +from wyoming.snd import SndProcessAsyncClient +from wyoming.wake import Detect, Detection, WakeProcessAsyncClient + +from .settings import SatelliteSettings +from .utils import run_event_command +from .vad import SileroVad + +_LOGGER = logging.getLogger() + + +class State(Enum): + NOT_STARTED = auto() + STARTING = auto() + STARTED = auto() + RESTARTING = auto() + STOPPING = auto() + STOPPED = auto() + + +# ----------------------------------------------------------------------------- + + +class SatelliteBase: + 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._snd_task: Optional[asyncio.Task] = None + self._snd_queue: "Optional[asyncio.Queue[Event]]" = None + self._wake_task: Optional[asyncio.Task] = None + self._wake_queue: "Optional[asyncio.Queue[Event]]" = None + + @property + def is_running(self) -> bool: + return self._state not in (State.STOPPING, State.STOPPED) + + @property + def state(self) -> State: + return self._state + + @state.setter + def state(self, value: State) -> None: + self._state = value + self._state_changed.set() + self._state_changed.clear() + + async def run(self) -> None: + + 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 _start(self) -> None: + self.state = State.STARTING + await self._connect_to_services() + self.state = State.STARTED + await self.started() + + async def started(self) -> None: + pass + + async def _stop(self) -> None: + self.server_id = None + self._writer = None + + await self._disconnect_from_services() + self.state = State.STOPPED + + async def stop(self) -> None: + self.state = State.STOPPING + + # Wait for STOPPED + while self.state != State.STOPPED: + await self._state_changed.wait() + + async def stopped(self) -> None: + pass + + def set_server(self, server_id: str, writer: asyncio.StreamWriter) -> None: + self.server_id = server_id + self._writer = writer + + async def event_from_server(self, event: Event) -> None: + """Called when an event is received from the server.""" + if RunSatellite.is_type(event.type): + if self.settings.wake.enabled: + # Local wake word detection + start_stage = PipelineStage.ASR + else: + # Remote wake word detection + start_stage = PipelineStage.WAKE + + if self.settings.snd.enabled: + # Play TTS response + end_stage = PipelineStage.TTS + else: + # No audio output + end_stage = PipelineStage.INTENT + + await self.event_to_server( + RunPipeline(start_stage=start_stage, end_stage=end_stage).event() + ) + elif ( + AudioChunk.is_type(event.type) + or AudioStart.is_type(event.type) + or AudioStop.is_type(event.type) + ): + # Forward to sound service + await self.event_to_snd(event) + + async def event_to_server(self, event: Event) -> None: + """Send an event to the server.""" + if self._writer is None: + return + + await async_write_event(event, self._writer) + + async def event_to_snd(self, event: Event) -> None: + """Send an event to the sound service.""" + if self._snd_queue is not None: + self._snd_queue.put_nowait(event) + + async def _restart(self) -> None: + 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: + 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()) + + 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()) + + 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()) + + _LOGGER.info("Connected to services") + + async def _disconnect_from_services(self) -> None: + if self._mic_task is not None: + _LOGGER.debug("Stopping microphone") + self._mic_task.cancel() + self._mic_task = None + + if self._snd_task is not None: + _LOGGER.debug("Stopping sound") + self._snd_task.cancel() + self._snd_task = None + + _LOGGER.debug("Disconnected from services") + + # ------------------------------------------------------------------------- + # Microphone + # ------------------------------------------------------------------------- + + async def event_from_mic( + self, event: Event, audio_bytes: Optional[bytes] = None + ) -> None: + pass + + def _make_mic_client(self) -> Optional[AsyncClient]: + 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_client: Optional[AsyncClient] = None + + 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.debug("Connected to mic service") + + 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 + + # TODO: audio processing + + await self.event_from_mic(event) + 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() + + # ------------------------------------------------------------------------- + # Sound + # ------------------------------------------------------------------------- + + def _make_snd_client(self) -> Optional[AsyncClient]: + 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_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() + + event = await self._snd_queue.get() + + if snd_client is None: + snd_client = self._make_snd_client() + assert snd_client is not None + await snd_client.connect() + _LOGGER.debug("Connected to snd service") + + # TODO: audio processing + + await snd_client.write_event(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() + + # ------------------------------------------------------------------------- + # Wake + # ------------------------------------------------------------------------- + + async def event_from_wake(self, event: Event) -> None: + pass + + async def event_to_wake(self, event: Event) -> None: + if self._wake_queue is not None: + self._wake_queue.put_nowait(event) + + def _make_wake_client(self) -> Optional[AsyncClient]: + 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_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: + try: + if wake_client is not None: + await wake_client.disconnect() + 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.debug("Connected to wake service") + + # Reset + from_client_task = None + to_client_task = None + pending = set() + self._wake_queue = asyncio.Queue() + + await self._send_wake_detect() + + if to_client_task is None: + # From satellite to wake service + to_client_task = asyncio.create_task(self._wake_queue.get()) + 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()) + pending.add(from_client_task) + + done, pending = await asyncio.wait( + pending, return_when=asyncio.FIRST_COMPLETED + ) + + if to_client_task in done: + 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: + 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 + + # TODO: audio processing + 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.settings.wake.names).event()) + + # ------------------------------------------------------------------------- + # Events + # ------------------------------------------------------------------------- + + async def event_streaming_start(self) -> None: + await run_event_command(self.settings.event.streaming_start) + + async def event_streaming_stop(self) -> None: + pass + + +# ----------------------------------------------------------------------------- + + +class AlwaysStreamingSatellite(SatelliteBase): + """Satellite that always streams audio.""" + + def __init__(self, settings: SatelliteSettings) -> None: + super().__init__(settings) + self.is_streaming = False + + 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.event_streaming_start() + + async def event_from_mic( + self, event: Event, audio_bytes: Optional[bytes] = None + ) -> None: + if not self.is_streaming: + return + + if AudioChunk.is_type(event.type): + # Forward to server + await self.event_to_server(event) + + +# ----------------------------------------------------------------------------- + + +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 + ) + self.timeout_seconds: Optional[float] = None + 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) + + async def event_from_server(self, event: Event) -> None: + await super().event_from_server(event) + + if RunSatellite.is_type(event.type): + _LOGGER.info("Waiting for speech") + + async def event_from_mic( + self, event: Event, audio_bytes: Optional[bytes] = None + ) -> None: + if not AudioChunk.is_type(event.type): + return + + 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 pipeline + await self.event_to_server(AudioStop().event()) + + _LOGGER.info("Waiting for speech") + await self.event_streaming_stop() + + if not self.is_streaming: + # Check VAD + if audio_bytes 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.event_streaming_start() + + if self.settings.vad.wake_word_timeout is not None: + self.timeout_seconds = ( + time.monotonic() + self.settings.vad.wake_word_timeout + ) + else: + self.timeout_seconds = None + + if self.vad_buffer is not None: + # Send contents of VAD buffer first + 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): + self.vad(None) + if self.vad_buffer is not None: + 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 + + async def event_from_server(self, event: Event) -> None: + await super().event_from_server(event) + + if RunSatellite.is_type(event.type) or Transcript.is_type(event.type): + self.is_streaming = False + await self._send_wake_detect() + _LOGGER.info("Waiting for wake word") + + if Transcript.is_type(event.type): + await self.event_streaming_stop() + + async def event_from_mic( + self, event: Event, audio_bytes: Optional[bytes] = None + ) -> None: + if not AudioChunk.is_type(event.type): + return + + if self.is_streaming: + # 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) -> None: + if self.is_streaming: + return + + if Detection.is_type(event.type): + self.is_streaming = True + _LOGGER.debug("Streaming audio") + await self.event_streaming_start() diff --git a/wyoming_satellite/settings.py b/wyoming_satellite/settings.py new file mode 100644 index 0000000..2e795cb --- /dev/null +++ b/wyoming_satellite/settings.py @@ -0,0 +1,80 @@ +"""Satellite settings.""" +from abc import ABC +from dataclasses import dataclass +from typing import List, Optional + + +@dataclass(frozen=True) +class ServiceSettings(ABC): + uri: Optional[str] = None + command: Optional[List[str]] = None + reconnect_seconds: float = 3.0 + + @property + def enabled(self) -> bool: + return bool(self.uri or self.command) + + +@dataclass(frozen=True) +class MicSettings(ServiceSettings): + volume_multiplier: float = 1.0 + auto_gain: int = 0 + noise_suppression: int = 0 + rate: int = 16000 + width: int = 2 + channels: int = 1 + samples_per_chunk: int = 1024 + + @property + def needs_webrtc(self) -> bool: + return self.enabled and ((self.auto_gain > 0) or (self.noise_suppression > 0)) + + @property + def needs_processing(self) -> bool: + return self.enabled and ((self.volume_multiplier != 1.0) or self.needs_webrtc) + + +@dataclass(frozen=True) +class SndSettings(ServiceSettings): + volume_multiplier: float = 1.0 + awake_wav: Optional[str] = None + done_wav: Optional[str] = None + rate: int = 22050 + width: int = 2 + channels: int = 1 + + @property + def needs_processing(self) -> bool: + return self.enabled and (self.volume_multiplier != 1.0) + + +@dataclass(frozen=True) +class WakeSettings(ServiceSettings): + names: Optional[List[str]] = None + rate: int = 16000 + width: int = 2 + channels: int = 1 + + +@dataclass(frozen=True) +class VadSettings: + enabled: bool = False + threshold: float = 0.5 + trigger_level: int = 1 + buffer_seconds: float = 2.0 + wake_word_timeout: Optional[float] = 5.0 + + +@dataclass(frozen=True) +class EventSettings(ServiceSettings): + streaming_start: Optional[List[str]] = None + + +@dataclass(frozen=True) +class SatelliteSettings: + mic: MicSettings + vad: VadSettings + wake: WakeSettings + snd: SndSettings + event: EventSettings + restart_timeout: float = 5.0 diff --git a/wyoming_satellite/utils.py b/wyoming_satellite/utils.py index e46e062..4c67717 100644 --- a/wyoming_satellite/utils.py +++ b/wyoming_satellite/utils.py @@ -1,6 +1,11 @@ """Utilities for Wyoming satellite.""" import array -from typing import Iterable +import asyncio +import logging +import shlex +from typing import Iterable, List, Optional + +_LOGGER = logging.getLogger() class AudioBuffer: @@ -87,3 +92,24 @@ def chunk_samples( # Capture leftover chunks if rest_samples := samples[next_chunk_idx:]: leftover_chunk_buffer.append(rest_samples) + + +async def run_event_command( + command: Optional[List[str]], command_input: Optional[str] = None +) -> None: + """Run a custom event command with optional input.""" + if not command: + return + + _LOGGER.debug("Running %s", command) + program, *program_args = command + proc = await asyncio.create_subprocess_exec( + program, *program_args, stdin=asyncio.subprocess.PIPE + ) + assert proc.stdin is not None + + if command_input: + await proc.communicate(input=command_input.encode("utf-8")) + else: + proc.stdin.close() + await proc.wait() diff --git a/wyoming_satellite/vad.py b/wyoming_satellite/vad.py new file mode 100644 index 0000000..bd3a970 --- /dev/null +++ b/wyoming_satellite/vad.py @@ -0,0 +1,33 @@ +"""Voice activity detection.""" +from typing import Optional + + +class SileroVad: + """Voice activity detection with silero VAD.""" + + def __init__(self, threshold: float, trigger_level: int) -> None: + from pysilero_vad import SileroVoiceActivityDetector + + self.detector = SileroVoiceActivityDetector() + self.threshold = threshold + self.trigger_level = trigger_level + self._activation = 0 + + def __call__(self, audio_bytes: Optional[bytes]) -> bool: + if audio_bytes is None: + # Reset + self._activation = 0 + self.detector.reset() + return False + + if self.detector(audio_bytes) >= self.threshold: + # Speech detected + self._activation += 1 + if self._activation >= self.trigger_level: + self._activation = 0 + return True + else: + # Silence detected + self._activation = max(0, self._activation - 1) + + return False