diff --git a/requirements.txt b/requirements.txt index 9aa282d..2667ffe 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ -wyoming==1.3.0 +wyoming==1.4.0 zeroconf==0.88.0 pyring-buffer==1.0.0 diff --git a/sounds/awake.wav b/sounds/awake.wav new file mode 100644 index 0000000..8cb6535 Binary files /dev/null and b/sounds/awake.wav differ diff --git a/sounds/done.wav b/sounds/done.wav new file mode 100644 index 0000000..d76ec5b Binary files /dev/null and b/sounds/done.wav differ diff --git a/wyoming_satellite/__main__.py b/wyoming_satellite/__main__.py index f5f2b07..b4b30e5 100644 --- a/wyoming_satellite/__main__.py +++ b/wyoming_satellite/__main__.py @@ -2,18 +2,22 @@ import argparse import asyncio import logging +import math import shlex import sys import time import uuid +import wave from enum import Enum from functools import partial from pathlib import Path -from typing import Callable, Final, Optional +from typing import Callable, Final, Iterator, Optional, Union +from pyring_buffer import RingBuffer from wyoming.asr import Transcribe, Transcript from wyoming.audio import AudioChunk, AudioStart, AudioStop from wyoming.client import AsyncClient +from wyoming.error import Error from wyoming.event import Event from wyoming.info import Attribution, Describe, Info, Satellite from wyoming.mic import MicProcessAsyncClient @@ -34,6 +38,8 @@ _DIR = Path(__file__).parent class SatelliteState(str, Enum): WAKE = "wake" ASR = "asr" + VAD_RESET = "vad-reset" + VAD_WAIT = "vad-wait" async def main() -> None: @@ -67,6 +73,7 @@ async def main() -> None: default=1024, help="Sample per chunk for mic-command (default: 1024)", ) + parser.add_argument("--mic-volume-multiplier", type=float, default=1.0) # Sound output parser.add_argument("--snd-uri", help="URI of Wyoming sound service") @@ -89,6 +96,7 @@ async def main() -> None: default=1, help="Sample channels of snd-command (default: 1)", ) + parser.add_argument("--snd-volume-multiplier", type=float, default=1.0) # Local wake word detection parser.add_argument("--wake-uri", help="URI of Wyoming wake word detection service") @@ -120,24 +128,31 @@ async def main() -> None: # Voice activity detector parser.add_argument( - "--vad", choices=("disabled", "webrtcvad", "silero"), default="disabled" + "--vad", action="store_true", help="Wait for speech before streaming audio" ) parser.add_argument("--vad-threshold", type=float, default=0.5) - parser.add_argument("--vad-trigger-level", type=int, default=3) - parser.add_argument("--vad-buffer-chunks", type=int, default=40) + parser.add_argument("--vad-trigger-level", type=int, default=1) + parser.add_argument("--vad-buffer-seconds", type=float, default=1) # Audio enhancement (requires webrtc-noise-audio) parser.add_argument( "--noise-suppression", type=int, default=0, choices=(0, 1, 2, 3, 4) ) parser.add_argument("--auto-gain", type=int, default=0, choices=list(range(32))) - parser.add_argument("--volume-multiplier", type=float, default=1.0) # External event handlers parser.add_argument( "--event-uri", help="URI of Wyoming service to forward events to" ) + # Sounds + parser.add_argument( + "--awake-wav", help="WAV file to play when wake word is detected" + ) + parser.add_argument( + "--done-wav", help="WAV file to play when voice command is done" + ) + # Satellite details parser.add_argument("--uri", default="stdio://", help="unix:// or tcp://") parser.add_argument("--name", required=True, help="Name of the satellite") @@ -175,6 +190,17 @@ async def main() -> None: _LOGGER.fatal("Install extras for silero") sys.exit(1) + if args.awake_wav and (not Path(args.awake_wav).is_file()): + _LOGGER.fatal("%s does not exist", args.awake_wav) + sys.exit(1) + + if args.done_wav and (not Path(args.done_wav).is_file()): + _LOGGER.fatal("%s does not exist", args.done_wav) + sys.exit(1) + + if args.vad and (args.wake_uri or args.wake_command): + _LOGGER.warning("VAD is not used with local wake word detection") + logging.basicConfig(level=logging.DEBUG if args.debug else logging.INFO) _LOGGER.debug(args) @@ -232,24 +258,42 @@ class SatelliteEventHandler(AsyncEventHandler): self.wyoming_info_event = wyoming_info.event() self.client_id = str(time.monotonic_ns()) + self.is_running = True self.state: Optional[Satellite] = None self.satellite_task: Optional[asyncio.Task] = None - self.volume_multiplier = self.cli_args.volume_multiplier - self.has_audio_processing = needs_webrtc(self.cli_args) or ( - self.volume_multiplier != 1.0 - ) - 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.events_client: Optional[AsyncClient] = None - self.is_running = True + 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_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.events_client: Optional[AsyncClient] = None + _LOGGER.debug("Client connected: %s", self.client_id) async def handle_event(self, event: Event) -> bool: @@ -261,35 +305,22 @@ class SatelliteEventHandler(AsyncEventHandler): if RunSatellite.is_type(event.type) and (self.satellite_task is None): if self.has_wake: # Local wake word detection - await self.write_event( - RunPipeline( - start_stage=PipelineStage.ASR, - end_stage=PipelineStage.TTS - if self.has_snd - else PipelineStage.HANDLE, - name=self.cli_args.pipeline, - ).event() - ) self.state = SatelliteState.WAKE self.satellite_task = asyncio.create_task(self.run_satellite_wake()) else: # Remote wake word detection - await self.write_event( - RunPipeline( - start_stage=PipelineStage.WAKE, - end_stage=PipelineStage.TTS - if self.has_snd - else PipelineStage.HANDLE, - name=self.cli_args.pipeline, - ).event() - ) - self.state = None + self.state = SatelliteState.VAD_RESET self.satellite_task = asyncio.create_task(self.run_satellite()) elif Transcript.is_type(event.type): - if self.state == SatelliteState.ASR: + if self.cli_args.done_wav: + await self._play_wav(self.cli_args.done_wav) + + if self.has_wake: # Ready for next wake word detection self.state = SatelliteState.WAKE _LOGGER.info("Detecting wake word") + else: + self.state = SatelliteState.VAD_RESET # STT transcript await self.forward_event(event) @@ -299,38 +330,32 @@ class SatelliteEventHandler(AsyncEventHandler): or AudioChunk.is_type(event.type) or AudioStop.is_type(event.type) ): - if self.snd_client is None: - 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) + 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() - await self.snd_client.connect() - - assert self.snd_client is not None - - try: - # Forward to sound service - await self.snd_client.write_event(event) - except Exception: - _LOGGER.exception("Unexpected error while sending snd audio") - - # Reconnect - self.snd_client = None - - if AudioStart.is_type(event.type) or AudioStop.is_type(event.type): - # TTS start/stop await self.forward_event(event) _LOGGER.debug(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(event) + + if self.snd_client is not None: + await self.snd_client.disconnect() + self.snd_client = None + elif Detection.is_type(event.type): + if self.cli_args.awake_wav: + await self._play_wav(self.cli_args.awake_wav) + + await self.forward_event(event) + _LOGGER.debug(event) elif ( Detect.is_type(event.type) or Detection.is_type(event.type) @@ -341,12 +366,14 @@ class SatelliteEventHandler(AsyncEventHandler): ): # Other client events: # - Detect for start of wake word detection - # - Detection for when wake word is detected # - Transcribe for when STT starts # - VoiceStarted/VoiceStopped for when user starts/stops speaking # - Synthesize for TTS text await self.forward_event(event) _LOGGER.debug(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) @@ -369,21 +396,100 @@ class SatelliteEventHandler(AsyncEventHandler): _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: + 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_satellite(self) -> None: try: mic_client = self._make_mic_client() async with mic_client: - _LOGGER.info("Streaming audio") - while True: mic_event = await mic_client.read_event() if mic_event is None: _LOGGER.warning("Microphone service disconnected") break - if AudioChunk.is_type(mic_event.type): - mic_event = self._process_audio(mic_event) + 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.state == SatelliteState.VAD_WAIT: + if self.process_vad is None: + # No VAD + self.state = SatelliteState.ASR + _LOGGER.info("Streaming audio") + + # Start pipeline + await self.write_event( + RunPipeline( + start_stage=PipelineStage.WAKE, + end_stage=PipelineStage.TTS + if self.has_snd + else PipelineStage.HANDLE, + name=self.cli_args.pipeline, + ).event() + ) + else: + # Wait for speech + chunk = AudioChunk.from_event(mic_event) + if self.process_vad(chunk.audio): + # Ready to stream + self.state = SatelliteState.ASR + + # Start pipeline + await self.write_event( + RunPipeline( + start_stage=PipelineStage.WAKE, + end_stage=PipelineStage.TTS + if self.has_snd + else PipelineStage.HANDLE, + name=self.cli_args.pipeline, + ).event() + ) + + 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() + ) + + _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.ASR: # Forward all audio to server await self.write_event(mic_event) @@ -440,7 +546,7 @@ class SatelliteEventHandler(AsyncEventHandler): break if AudioChunk.is_type(mic_event.type): - mic_event = self._process_audio(mic_event) + mic_event = self._process_mic_audio(mic_event) # Forward to wake service await wake_client.write_event(mic_event) @@ -464,6 +570,17 @@ class SatelliteEventHandler(AsyncEventHandler): _LOGGER.debug("Wake word detected: %s", wake_event) self.state = SatelliteState.ASR + # Start pipeline + await self.write_event( + RunPipeline( + start_stage=PipelineStage.ASR, + end_stage=PipelineStage.TTS + if self.has_snd + else PipelineStage.HANDLE, + name=self.cli_args.pipeline, + ).event() + ) + # Forward to server await self.write_event(wake_event) @@ -478,9 +595,9 @@ class SatelliteEventHandler(AsyncEventHandler): except Exception: _LOGGER.exception("Unexpected error in run_satellite") - def _process_audio(self, audio_event: Event) -> Event: - """Perform audio processing.""" - if not self.has_audio_processing: + 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 @@ -490,8 +607,27 @@ class SatelliteEventHandler(AsyncEventHandler): if self.process_webrtc is not None: audio_bytes = self.process_webrtc(audio_bytes) - if self.volume_multiplier != 1.0: - audio_bytes = multiply_volume(audio_bytes, self.volume_multiplier) + 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, @@ -516,6 +652,56 @@ class SatelliteEventHandler(AsyncEventHandler): async def disconnect(self) -> None: self.is_running = False + async def _play_wav(self, wav_path: Union[str, Path]) -> None: + 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() + # ----------------------------------------------------------------------------- @@ -532,18 +718,18 @@ def get_mac_address() -> str: def needs_webrtc(args: argparse.Namespace) -> bool: """Return True if webrtc must be used.""" - return ( - (args.noise_suppression > 0) or (args.auto_gain > 0) or (args.vad == "webrtc") - ) + return (args.noise_suppression > 0) or (args.auto_gain > 0) def needs_silero(args: argparse.Namespace) -> bool: """Return True if silero-vad must be used.""" - return args.vad == "silero" + return args.vad class WebRtcAudio: - _sub_chunk_samples: Final = 160 + """Audio processing using webrtc.""" + + _sub_chunk_samples: Final = 160 # 10ms @ 16Khz _sub_chunk_bytes: Final = _sub_chunk_samples * 2 # 16-bit def __init__(self, auto_gain: int, noise_suppression: int) -> None: @@ -553,6 +739,7 @@ class WebRtcAudio: self.audio_buffer = AudioBuffer(self._sub_chunk_bytes) def __call__(self, audio_bytes: bytes) -> bytes: + """Process in 10ms chunks.""" clean_chunk = bytes() for sub_chunk in chunk_samples( audio_bytes, self._sub_chunk_bytes, self.audio_buffer @@ -563,6 +750,34 @@ class WebRtcAudio: return clean_chunk +class SileroVad: + 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 + 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__":