diff --git a/README.md b/README.md index 1c8a13c..95ed1dd 100644 --- a/README.md +++ b/README.md @@ -117,8 +117,8 @@ Run the satellite with automatic gain control and noise suppression: ``` sh script/run \ ... \ - --auto-gain 5 \ - --noise-suppression 2 + --mic-auto-gain 5 \ + --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) diff --git a/wyoming_satellite/__main__.py b/wyoming_satellite/__main__.py index 0f99403..3093191 100644 --- a/wyoming_satellite/__main__.py +++ b/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__": diff --git a/wyoming_satellite/satellite.py b/wyoming_satellite/satellite.py index 93f9d01..a170fab 100644 --- a/wyoming_satellite/satellite.py +++ b/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: - pass + async def trigger_streaming_stop(self) -> None: + 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() diff --git a/wyoming_satellite/settings.py b/wyoming_satellite/settings.py index 2e795cb..e6a50e2 100644 --- a/wyoming_satellite/settings.py +++ b/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) diff --git a/wyoming_satellite/utils.py b/wyoming_satellite/utils.py index 4c67717..d080cd0 100644 --- a/wyoming_satellite/utils.py +++ b/wyoming_satellite/utils.py @@ -2,8 +2,12 @@ import array import asyncio import logging -import shlex -from typing import Iterable, List, Optional +import wave +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() diff --git a/wyoming_satellite/webrtc.py b/wyoming_satellite/webrtc.py new file mode 100644 index 0000000..03185c9 --- /dev/null +++ b/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