Almost finished

README.md

@@ -117,8 +117,8 @@ Run the satellite with automatic gain control and noise suppression:
 ``` sh
 script/run \
   ... \
-  --auto-gain 5 \
+  --mic-auto-gain 5 \
-  --noise-suppression 2
+  --mic-noise-suppression 2
 ```
 
 Automatic gain control is between 0-31, which 31 being the loudest.
@@ -133,6 +133,7 @@ Satellites can respond to events from the server by running commands:
 * `--startup-command` - run when satellite starts
 * `--detect-command` - wake word detection has started, but not detected yet (no stdin)
 * `--streaming-start-command` - audio has started streaming to server
+* `--streaming-stop-command` - audio has stopped streaming to server
 * `--detection-command` - wake word is detected (wake word name on stdin)
 * `--transcript-command` - speech-to-text transcript is returned (text on stdin)
 * `--stt-start-command` - user started speaking (no stdin)

wyoming_satellite/__main__.py

@@ -11,7 +11,7 @@ import wave
 from enum import Enum
 from functools import partial
 from pathlib import Path
-from typing import Callable, Final, Iterator, Optional, Union
+from typing import Callable, Final, Iterator, List, Optional, Union
 
 from pyring_buffer import RingBuffer
 from wyoming.asr import Transcribe, Transcript
@@ -48,6 +48,8 @@ from .utils import AudioBuffer, chunk_samples, multiply_volume, run_event_comman
 _LOGGER = logging.getLogger()
 _DIR = Path(__file__).parent
 
+# TODO: events service
+
 
 async def main() -> None:
     """Main entry point."""
@@ -81,6 +83,10 @@ async def main() -> None:
         help="Sample per chunk for mic-command (default: 1024)",
     )
     parser.add_argument("--mic-volume-multiplier", type=float, default=1.0)
+    parser.add_argument(
+        "--mic-noise-suppression", type=int, default=0, choices=(0, 1, 2, 3, 4)
+    )
+    parser.add_argument("--mic-auto-gain", type=int, default=0, choices=list(range(32)))
 
     # Sound output
     parser.add_argument("--snd-uri", help="URI of Wyoming sound service")
@@ -147,12 +153,6 @@ async def main() -> None:
         help="Seconds before going back to waiting for speech when wake word isn't detected",
     )
 
-    # 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)))
-
     # External event handlers
     parser.add_argument(
         "--event-uri", help="URI of Wyoming service to forward events to"
@@ -194,6 +194,10 @@ async def main() -> None:
         "--streaming-start-command",
         help="Command to run when audio streaming starts",
     )
+    parser.add_argument(
+        "--streaming-stop-command",
+        help="Command to run when audio streaming stops",
+    )
     parser.add_argument(
         "--error-command",
         help="Command to run when an error occurs",
@@ -272,7 +276,17 @@ async def main() -> None:
     )
 
     settings = SatelliteSettings(
-        mic=MicSettings(uri=args.mic_uri, command=_split(args.mic_command)),
+        mic=MicSettings(
+            uri=args.mic_uri,
+            command=_split(args.mic_command),
+            rate=args.mic_command_rate,
+            width=args.mic_command_width,
+            channels=args.mic_command_channels,
+            samples_per_chunk=args.mic_command_samples_per_chunk,
+            volume_multiplier=args.mic_volume_multiplier,
+            auto_gain=args.mic_auto_gain,
+            noise_suppression=args.mic_noise_suppression,
+        ),
         vad=VadSettings(
             enabled=args.vad,
             threshold=args.vad_threshold,
@@ -285,19 +299,43 @@ async def main() -> None:
             command=_split(args.wake_command),
             names=args.wake_word_name,
         ),
-        snd=SndSettings(uri=args.snd_uri, command=_split(args.snd_command)),
+        snd=SndSettings(
+            uri=args.snd_uri,
+            command=_split(args.snd_command),
+            rate=args.snd_command_rate,
+            width=args.snd_command_width,
+            channels=args.snd_command_channels,
+            volume_multiplier=args.snd_volume_multiplier,
+            awake_wav=args.awake_wav,
+            done_wav=args.done_wav,
+        ),
         event=EventSettings(
-            uri=args.event_uri, streaming_start=_split(args.streaming_start_command)
+            uri=args.event_uri,
+            startup=_split(args.startup_command),
+            streaming_start=_split(args.streaming_start_command),
+            streaming_stop=_split(args.streaming_stop_command),
+            detect=_split(args.detect_command),
+            detection=_split(args.detection_command),
+            transcript=_split(args.transcript_command),
+            stt_start=_split(args.stt_start_command),
+            stt_stop=_split(args.stt_stop_command),
+            synthesize=_split(args.synthesize_command),
+            tts_start=_split(args.tts_start_command),
+            tts_stop=_split(args.tts_stop_command),
+            error=_split(args.error_command),
         ),
     )
 
     satellite: SatelliteBase
 
     if settings.wake.enabled:
+        # Local wake word detection
         satellite = WakeStreamingSatellite(settings)
     elif settings.vad.enabled:
+        # Stream after speech
         satellite = VadStreamingSatellite(settings)
     else:
+        # Stream all the time
         satellite = AlwaysStreamingSatellite(settings)
 
     if args.startup_command:
@@ -949,7 +987,7 @@ 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)
+    return (args.mic_noise_suppression > 0) or (args.mic_auto_gain > 0)
 
 
 def needs_silero(args: argparse.Namespace) -> bool:
@@ -964,30 +1002,6 @@ def _split(command: Optional[str]) -> Optional[List[str]]:
     return shlex.split(command)
 
 
-class WebRtcAudio:
-    """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:
-        from webrtc_noise_gain import AudioProcessor
-
-        self.audio_processor = AudioProcessor(auto_gain, noise_suppression)
-        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
-        ):
-            result = self.audio_processor.Process10ms(sub_chunk)
-            clean_chunk += result.audio
-
-        return clean_chunk
-
-
 # -----------------------------------------------------------------------------
 
 if __name__ == "__main__":

wyoming_satellite/satellite.py

@@ -5,7 +5,8 @@ import math
 import time
 from dataclasses import dataclass
 from enum import Enum, auto
-from typing import List, Optional, Set
+from pathlib import Path
+from typing import Callable, List, Optional, Set, Union
 
 from pyring_buffer import RingBuffer
 from wyoming.asr import Transcript
@@ -19,8 +20,9 @@ from wyoming.snd import SndProcessAsyncClient
 from wyoming.wake import Detect, Detection, WakeProcessAsyncClient
 
 from .settings import SatelliteSettings
-from .utils import run_event_command
+from .utils import multiply_volume, run_event_command, wav_to_events
 from .vad import SileroVad
+from .webrtc import WebRtcAudio
 
 _LOGGER = logging.getLogger()
 
@@ -46,6 +48,7 @@ class SatelliteBase:
         self._writer: Optional[asyncio.StreamWriter] = None
 
         self._mic_task: Optional[asyncio.Task] = None
+        self._mic_webrtc: Optional[Callable[[bytes], bytes]] = None
         self._snd_task: Optional[asyncio.Task] = None
         self._snd_queue: "Optional[asyncio.Queue[Event]]" = None
         self._wake_task: Optional[asyncio.Task] = None
@@ -229,6 +232,13 @@ class SatelliteBase:
 
     async def _mic_task_proc(self) -> None:
         mic_client: Optional[AsyncClient] = None
+        audio_bytes: Optional[bytes] = None
+
+        if self.settings.mic.needs_webrtc and (self._mic_webrtc is None):
+            _LOGGER.debug("Using webrtc audio enhancements")
+            self._mic_webrtc = WebRtcAudio(
+                self.settings.mic.auto_gain, self.settings.mic.noise_suppression
+            )
 
         async def _disconnect() -> None:
             try:
@@ -253,9 +263,22 @@ class SatelliteBase:
                     await asyncio.sleep(self.settings.mic.reconnect_seconds)
                     continue
 
-                # TODO: audio processing
+                # Audio processing
+                if self.settings.mic.needs_processing and AudioChunk.is_type(
+                    event.type
+                ):
+                    chunk = AudioChunk.from_event(event)
+                    audio_bytes = self._process_mic_audio(chunk.audio)
+                    event = AudioChunk(
+                        rate=chunk.rate,
+                        width=chunk.width,
+                        channels=chunk.channels,
+                        audio=audio_bytes,
+                    ).event()
+                else:
+                    audio_bytes = None
 
-                await self.event_from_mic(event)
+                await self.event_from_mic(event, audio_bytes)
             except asyncio.CancelledError:
                 break
             except Exception:
@@ -266,6 +289,17 @@ class SatelliteBase:
 
         await _disconnect()
 
+    def _process_mic_audio(self, audio_bytes: bytes) -> bytes:
+        if self.settings.mic.volume_multiplier != 1.0:
+            audio_bytes = multiply_volume(
+                audio_bytes, self.settings.mic.volume_multiplier
+            )
+
+        if self._mic_webrtc is not None:
+            audio_bytes = self._mic_webrtc(audio_bytes)
+
+        return audio_bytes
+
     # -------------------------------------------------------------------------
     # Sound
     # -------------------------------------------------------------------------
@@ -323,6 +357,18 @@ class SatelliteBase:
 
         await _disconnect()
 
+    async def _play_wav(self, wav_path: Optional[Union[str, Path]]) -> None:
+        """Send WAV as events to sound service."""
+        if (not wav_path) or (not self.settings.snd.enabled):
+            return
+
+        for event in wav_to_events(
+            wav_path,
+            samples_per_chunk=self.settings.snd.samples_per_chunk,
+            volume_multiplier=self.settings.snd.volume_multiplier,
+        ):
+            await self.event_to_snd(event)
+
     # -------------------------------------------------------------------------
     # Wake
     # -------------------------------------------------------------------------
@@ -430,16 +476,28 @@ class SatelliteBase:
     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())
+        await self.trigger_detect()
 
     # -------------------------------------------------------------------------
     # Events
     # -------------------------------------------------------------------------
 
-    async def event_streaming_start(self) -> None:
+    async def trigger_streaming_start(self) -> None:
         await run_event_command(self.settings.event.streaming_start)
 
-    async def event_streaming_stop(self) -> None:
+    async def trigger_streaming_stop(self) -> None:
-        pass
+        await run_event_command(self.settings.event.streaming_stop)
+
+    async def trigger_detect(self) -> None:
+        await run_event_command(self.settings.event.detect)
+
+    async def trigger_detection(self, detection: Detection) -> None:
+        await run_event_command(self.settings.event.detection, detection.name)
+        await self._play_wav(self.settings.snd.awake_wav)
+
+    async def trigger_transcript(self, transcript: Transcript) -> None:
+        await run_event_command(self.settings.event.transcript, transcript.text)
+        await self._play_wav(self.settings.snd.done_wav)
 
 
 # -----------------------------------------------------------------------------
@@ -458,7 +516,15 @@ class AlwaysStreamingSatellite(SatelliteBase):
         if RunSatellite.is_type(event.type):
             self.is_streaming = True
             _LOGGER.info("Streaming audio")
-            await self.event_streaming_start()
+            await self.trigger_streaming_start()
+        elif Detect.is_type(event.type):
+            await self.trigger_detect()
+        elif Detection.is_type(event.type):
+            _LOGGER.debug("Wake word detected")
+            await self.trigger_detection(Detection.from_event(event))
+        elif Transcript.is_type(event.type):
+            _LOGGER.debug(event)
+            await self.trigger_transcript(Transcript.from_event(event))
 
     async def event_from_mic(
         self, event: Event, audio_bytes: Optional[bytes] = None
@@ -499,6 +565,14 @@ class VadStreamingSatellite(SatelliteBase):
 
         if RunSatellite.is_type(event.type):
             _LOGGER.info("Waiting for speech")
+        elif Detect.is_type(event.type):
+            await self.trigger_detect()
+        elif Detection.is_type(event.type):
+            _LOGGER.debug("Wake word detected")
+            await self.trigger_detection(Detection.from_event(event))
+        elif Transcript.is_type(event.type):
+            _LOGGER.debug(event)
+            await self.trigger_transcript(Transcript.from_event(event))
 
     async def event_from_mic(
         self, event: Event, audio_bytes: Optional[bytes] = None
@@ -519,7 +593,7 @@ class VadStreamingSatellite(SatelliteBase):
             await self.event_to_server(AudioStop().event())
 
             _LOGGER.info("Waiting for speech")
-            await self.event_streaming_stop()
+            await self.trigger_streaming_stop()
 
         if not self.is_streaming:
             # Check VAD
@@ -538,7 +612,7 @@ class VadStreamingSatellite(SatelliteBase):
             # Speech detected
             self.is_streaming = True
             _LOGGER.info("Streaming audio")
-            await self.event_streaming_start()
+            await self.trigger_streaming_start()
 
             if self.settings.vad.wake_word_timeout is not None:
                 self.timeout_seconds = (
@@ -592,7 +666,9 @@ class WakeStreamingSatellite(SatelliteBase):
             _LOGGER.info("Waiting for wake word")
 
             if Transcript.is_type(event.type):
-                await self.event_streaming_stop()
+                _LOGGER.debug(event)
+                await self.trigger_streaming_stop()
+                await self.trigger_transcript(Transcript.from_event(event))
 
     async def event_from_mic(
         self, event: Event, audio_bytes: Optional[bytes] = None
@@ -614,4 +690,5 @@ class WakeStreamingSatellite(SatelliteBase):
         if Detection.is_type(event.type):
             self.is_streaming = True
             _LOGGER.debug("Streaming audio")
-            await self.event_streaming_start()
+            await self.trigger_detection(Detection.from_event(event))
+            await self.trigger_streaming_start()

wyoming_satellite/settings.py

@@ -42,6 +42,7 @@ class SndSettings(ServiceSettings):
     rate: int = 22050
     width: int = 2
     channels: int = 1
+    samples_per_chunk: int = 1024
 
     @property
     def needs_processing(self) -> bool:
@@ -67,7 +68,18 @@ class VadSettings:
 
 @dataclass(frozen=True)
 class EventSettings(ServiceSettings):
+    startup: Optional[List[str]] = None
     streaming_start: Optional[List[str]] = None
+    streaming_stop: Optional[List[str]] = None
+    detect: Optional[List[str]] = None
+    detection: Optional[List[str]] = None
+    transcript: Optional[List[str]] = None
+    stt_start: Optional[List[str]] = None
+    stt_stop: Optional[List[str]] = None
+    synthesize: Optional[List[str]] = None
+    tts_start: Optional[List[str]] = None
+    tts_stop: Optional[List[str]] = None
+    error: Optional[List[str]] = None
 
 
 @dataclass(frozen=True)

wyoming_satellite/utils.py

@@ -2,8 +2,12 @@
 import array
 import asyncio
 import logging
-import shlex
+import wave
-from typing import Iterable, List, Optional
+from pathlib import Path
+from typing import Iterable, Iterator, List, Optional, Union
+
+from wyoming.audio import AudioChunk, AudioStart, AudioStop
+from wyoming.event import Event
 
 _LOGGER = logging.getLogger()
 
@@ -113,3 +117,38 @@ async def run_event_command(
     else:
         proc.stdin.close()
         await proc.wait()
+
+
+def wav_to_events(
+    wav_path: Union[str, Path],
+    samples_per_chunk: int = 1024,
+    volume_multiplier: float = 1.0,
+) -> 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 volume_multiplier != 1.0:
+                audio_bytes = multiply_volume(audio_bytes, 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()

wyoming_satellite/webrtc.py

@@ -0,0 +1,28 @@
+"""webrtc audio processing."""
+from typing import Final
+
+from .utils import AudioBuffer, chunk_samples
+
+
+class WebRtcAudio:
+    """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:
+        from webrtc_noise_gain import AudioProcessor
+
+        self.audio_processor = AudioProcessor(auto_gain, noise_suppression)
+        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
+        ):
+            result = self.audio_processor.Process10ms(sub_chunk)
+            clean_chunk += result.audio
+
+        return clean_chunk