Add VAD

requirements.txt

@@ -1,3 +1,3 @@
-wyoming==1.3.0
+wyoming==1.4.0
 zeroconf==0.88.0
 pyring-buffer==1.0.0

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-trigger-level", type=int, default=1)
-    parser.add_argument("--vad-buffer-chunks", type=int, default=40)
+    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(
+                self.state = SatelliteState.VAD_RESET
-                    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.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,36 +330,30 @@ class SatelliteEventHandler(AsyncEventHandler):
             or AudioChunk.is_type(event.type)
             or AudioStop.is_type(event.type)
         ):
-            if self.snd_client is None:
+            if AudioStart.is_type(event.type):
-                if self.cli_args.snd_command:
+                # TTS start
-                    _LOGGER.debug("Running %s", self.cli_args.snd_command)
+                await self._ensure_snd_client()
-                    program, *program_args = shlex.split(self.cli_args.snd_command)
+                if self.snd_client is not None:
-                    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)
-
                     await self.snd_client.connect()
 
-            assert self.snd_client is not None
+                await self.forward_event(event)
-
+                _LOGGER.debug(event)
-            try:
+            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)
-            except Exception:
+            elif AudioStop.is_type(event.type):
-                _LOGGER.exception("Unexpected error while sending snd audio")
+                # TTS stop
+                await self.forward_event(event)
+                _LOGGER.debug(event)
 
-                # Reconnect
+                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)
 
-            if AudioStart.is_type(event.type) or AudioStop.is_type(event.type):
-                # TTS start/stop
             await self.forward_event(event)
             _LOGGER.debug(event)
         elif (
@@ -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):
+                    if not AudioChunk.is_type(mic_event.type):
-                        mic_event = self._process_audio(mic_event)
+                        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:
+    def _process_mic_audio(self, audio_event: Event) -> Event:
-        """Perform audio processing."""
+        """Perform microphone audio processing, if necessary."""
-        if not self.has_audio_processing:
+        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:
+        if self.mic_volume_multiplier != 1.0:
-            audio_bytes = multiply_volume(audio_bytes, self.volume_multiplier)
+            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 (
+    return (args.noise_suppression > 0) or (args.auto_gain > 0)
-        (args.noise_suppression > 0) or (args.auto_gain > 0) or (args.vad == "webrtc")
-    )
 
 
 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__":