diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index bad7ffa..dc27a08 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -1,5 +1,7 @@ +--- name: test +# yamllint disable-line rule:truthy on: workflow_dispatch: pull_request: diff --git a/CHANGELOG.md b/CHANGELOG.md index eb974cc..9d811a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## 1.2.0 + +- Add `--tts-played-command` +- Add `--mic-seconds-to-mute-after-awake-wav` and `--mic-no-mute-during-awake-wav` +- Send preferred sound format to server + ## 1.1.1 - Bump to wyoming 1.5.2 (package fix) diff --git a/requirements.txt b/requirements.txt index 66d5da8..1eb7bdf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ -wyoming==1.5.2 +wyoming==1.5.3 zeroconf==0.88.0 pyring-buffer==1.0.0 diff --git a/setup.py b/setup.py index d8422f3..8aba9bd 100644 --- a/setup.py +++ b/setup.py @@ -52,8 +52,6 @@ setup( "Development Status :: 3 - Alpha", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", - "Programming Language :: Python :: 3.7", - "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", diff --git a/tests/test_satellite.py b/tests/test_satellite.py new file mode 100644 index 0000000..528dd5f --- /dev/null +++ b/tests/test_satellite.py @@ -0,0 +1,198 @@ +import asyncio +import io +import logging +from collections.abc import Iterable +from pathlib import Path +from typing import Final, Optional +from unittest.mock import patch + +import pytest +from wyoming.asr import Transcript +from wyoming.audio import AudioChunk +from wyoming.client import AsyncClient +from wyoming.event import Event, async_read_event +from wyoming.pipeline import PipelineStage, RunPipeline +from wyoming.satellite import RunSatellite, StreamingStarted, StreamingStopped +from wyoming.wake import Detection + +from wyoming_satellite import ( + EventSettings, + MicSettings, + SatelliteSettings, + WakeSettings, + WakeStreamingSatellite, +) + +from .shared import AUDIO_CHUNK + +_LOGGER = logging.getLogger() + +TIMEOUT: Final = 1 + + +class MicClient(AsyncClient): + def __init__(self) -> None: + super().__init__() + + async def read_event(self) -> Optional[Event]: + await asyncio.sleep(AUDIO_CHUNK.seconds) + return AUDIO_CHUNK.event() + + async def write_event(self, event: Event) -> None: + # Output only + pass + + +class WakeClient(AsyncClient): + def __init__(self) -> None: + super().__init__() + self._event_ready = asyncio.Event() + self._event: Optional[Event] = None + self._detected: bool = False + + async def read_event(self) -> Optional[Event]: + await self._event_ready.wait() + self._event_ready.clear() + return self._event + + async def write_event(self, event: Event) -> None: + if AudioChunk.is_type(event.type): + if not self._detected: + self._detected = True + self._event = Detection().event() + self._event_ready.set() + + +class EventClient(AsyncClient): + def __init__(self) -> None: + super().__init__() + self.detection = asyncio.Event() + self.streaming_started = asyncio.Event() + self.streaming_stopped = asyncio.Event() + + async def read_event(self) -> Optional[Event]: + # Input only + return None + + async def write_event(self, event: Event) -> None: + if Detection.is_type(event.type): + self.detection.set() + elif StreamingStarted.is_type(event.type): + self.streaming_started.set() + elif StreamingStopped.is_type(event.type): + self.streaming_stopped.set() + + +class FakeStreamReaderWriter: + def __init__(self) -> None: + self._undrained_data = bytes() + self._value = bytes() + self._data_ready = asyncio.Event() + + def write(self, data: bytes) -> None: + self._undrained_data += data + + def writelines(self, data: Iterable[bytes]) -> None: + for line in data: + self.write(line) + + async def drain(self) -> None: + self._value += self._undrained_data + self._undrained_data = bytes() + self._data_ready.set() + self._data_ready.clear() + + async def readline(self) -> bytes: + while b"\n" not in self._value: + await self._data_ready.wait() + + with io.BytesIO(self._value) as value_io: + data = value_io.readline() + self._value = self._value[len(data) :] + return data + + async def readexactly(self, n: int) -> bytes: + while len(self._value) < n: + await self._data_ready.wait() + + data = self._value[:n] + self._value = self._value[n:] + return data + + +@pytest.mark.asyncio +async def test_satellite_and_server(tmp_path: Path) -> None: + mic_client = MicClient() + wake_client = WakeClient() + event_client = EventClient() + + with patch( + "wyoming_satellite.satellite.SatelliteBase._make_mic_client", + return_value=mic_client, + ), patch( + "wyoming_satellite.satellite.SatelliteBase._make_wake_client", + return_value=wake_client, + ), patch( + "wyoming_satellite.satellite.SatelliteBase._make_event_client", + return_value=event_client, + ): + satellite = WakeStreamingSatellite( + SatelliteSettings( + mic=MicSettings(uri="test"), + wake=WakeSettings(uri="test"), + event=EventSettings(uri="test"), + ) + ) + + # Fake server connection + server_io = FakeStreamReaderWriter() + await satellite.set_server("test", server_io) # type: ignore + + async def event_from_satellite() -> Optional[Event]: + return await async_read_event(server_io) + + satellite_task = asyncio.create_task(satellite.run(), name="satellite") + await satellite.event_from_server(RunSatellite().event()) + + # Trigger detection + event = await asyncio.wait_for(event_from_satellite(), timeout=TIMEOUT) + assert event is not None + assert Detection.is_type(event.type), event + + # Pipeline should start + event = await asyncio.wait_for(event_from_satellite(), timeout=TIMEOUT) + assert event is not None + assert RunPipeline.is_type(event.type), event + run_pipeline = RunPipeline.from_event(event) + assert run_pipeline.start_stage == PipelineStage.ASR + + # No TTS + assert run_pipeline.end_stage == PipelineStage.HANDLE + + # Event service should have received detection + await asyncio.wait_for(event_client.detection.wait(), timeout=TIMEOUT) + + # Server should be receiving audio now + assert satellite.is_streaming, "Not streaming" + for _ in range(5): + event = await asyncio.wait_for(event_from_satellite(), timeout=TIMEOUT) + assert event is not None + assert AudioChunk.is_type(event.type) + + # Event service should have received streaming start + await asyncio.wait_for(event_client.streaming_started.wait(), timeout=TIMEOUT) + + # Send transcript + await satellite.event_from_server(Transcript(text="test").event()) + + # Wait for streaming to stop + while satellite.is_streaming: + event = await asyncio.wait_for(event_from_satellite(), timeout=TIMEOUT) + assert event is not None + assert AudioChunk.is_type(event.type) + + # Event service should have received streaming stop + await asyncio.wait_for(event_client.streaming_stopped.wait(), timeout=TIMEOUT) + + await satellite.stop() + await satellite_task diff --git a/tests/test_wake_streaming.py b/tests/test_wake_streaming.py index ec98ab9..0977509 100644 --- a/tests/test_wake_streaming.py +++ b/tests/test_wake_streaming.py @@ -101,8 +101,8 @@ async def test_multiple_wakeups(tmp_path: Path) -> None: await satellite.event_from_server(Transcript("test").event()) # Should not trigger again within refractory period (default: 5 sec) - # with pytest.raises(asyncio.TimeoutError): - # await asyncio.wait_for(event_client.wake_event.wait(), timeout=0.15) + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(event_client.wake_event.wait(), timeout=0.15) await satellite.stop() await satellite_task diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..80dd236 --- /dev/null +++ b/tox.ini @@ -0,0 +1,14 @@ +[tox] +env_list = + py{39,310,311} +minversion = 4.12.1 + +[testenv] +description = run the tests with pytest +package = wheel +wheel_build_env = .pkg +deps = + pytest>=7,<8 + pytest-asyncio<1 +commands = + pytest {tty:--color=yes} {posargs} diff --git a/wyoming_satellite/__main__.py b/wyoming_satellite/__main__.py index 941a9c4..9bcb284 100644 --- a/wyoming_satellite/__main__.py +++ b/wyoming_satellite/__main__.py @@ -24,6 +24,7 @@ from .settings import ( SndSettings, VadSettings, WakeSettings, + WakeWordAndPipeline, ) from .utils import ( get_mac_address, @@ -84,6 +85,11 @@ async def main() -> None: action="store_true", help="Don't mute the microphone while awake wav is being played", ) + parser.add_argument( + "--mic-channel-index", + type=int, + help="Take microphone input from a specific channel (first channel is 0)", + ) # Sound output parser.add_argument("--snd-uri", help="URI of Wyoming sound service") @@ -114,7 +120,9 @@ async def main() -> None: "--wake-word-name", action="append", default=[], - help="Name of wake word to listen for (requires --wake-uri)", + nargs="+", + metavar=("name", "pipeline"), + help="Name of wake word to listen for and optional pipeline name to run (requires --wake-uri)", ) parser.add_argument("--wake-command", help="Program to run for wake word detection") parser.add_argument( @@ -318,6 +326,7 @@ async def main() -> None: noise_suppression=args.mic_noise_suppression, seconds_to_mute_after_awake_wav=args.mic_seconds_to_mute_after_awake_wav, mute_during_awake_wav=(not args.mic_no_mute_during_awake_wav), + channel_index=args.mic_channel_index, ), vad=VadSettings( enabled=args.vad, @@ -329,7 +338,9 @@ async def main() -> None: wake=WakeSettings( uri=args.wake_uri, command=split_command(args.wake_command), - names=args.wake_word_name, + names=[ + WakeWordAndPipeline(*wake_name) for wake_name in args.wake_word_name + ], refractory_seconds=args.wake_refractory_seconds if args.wake_refractory_seconds > 0 else None, diff --git a/wyoming_satellite/event_handler.py b/wyoming_satellite/event_handler.py index f68179b..58f046b 100644 --- a/wyoming_satellite/event_handler.py +++ b/wyoming_satellite/event_handler.py @@ -26,7 +26,7 @@ class SatelliteEventHandler(AsyncEventHandler): super().__init__(*args, **kwargs) self.cli_args = cli_args - self.wyoming_info_event = wyoming_info.event() + self.wyoming_info = wyoming_info self.client_id = str(time.monotonic_ns()) self.satellite = satellite @@ -35,7 +35,8 @@ class SatelliteEventHandler(AsyncEventHandler): 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) + await self.satellite.update_info(self.wyoming_info) + await self.write_event(self.wyoming_info.event()) return True if self.satellite.server_id is None: diff --git a/wyoming_satellite/satellite.py b/wyoming_satellite/satellite.py index 81d9307..c708405 100644 --- a/wyoming_satellite/satellite.py +++ b/wyoming_satellite/satellite.py @@ -1,4 +1,5 @@ """Satellite code.""" +import array import asyncio import logging import math @@ -6,14 +7,15 @@ import time import wave from enum import Enum, auto from pathlib import Path -from typing import Callable, Dict, Final, Optional, Set, Union +from typing import Callable, Dict, Final, List, Optional, Set, Union from pyring_buffer import RingBuffer from wyoming.asr import Transcript -from wyoming.audio import AudioChunk, AudioStart, AudioStop +from wyoming.audio import AudioChunk, AudioFormat, AudioStart, AudioStop from wyoming.client import AsyncClient from wyoming.error import Error from wyoming.event import Event, async_write_event +from wyoming.info import Describe, Info from wyoming.mic import MicProcessAsyncClient from wyoming.ping import Ping, Pong from wyoming.pipeline import PipelineStage, RunPipeline @@ -31,7 +33,13 @@ from wyoming.vad import VoiceStarted, VoiceStopped from wyoming.wake import Detect, Detection, WakeProcessAsyncClient from .settings import SatelliteSettings -from .utils import DebugAudioWriter, multiply_volume, run_event_command, wav_to_events +from .utils import ( + DebugAudioWriter, + multiply_volume, + normalize_wake_word, + run_event_command, + wav_to_events, +) from .vad import SileroVad from .webrtc import WebRtcAudio @@ -39,6 +47,7 @@ _LOGGER = logging.getLogger() _PONG_TIMEOUT: Final = 5 _PING_SEND_DELAY: Final = 2 +_WAKE_INFO_TIMEOUT: Final = 2 class State(Enum): @@ -285,7 +294,7 @@ class SatelliteBase: if not AudioChunk.is_type(event.type): await self.forward_event(event) - async def _send_run_pipeline(self) -> None: + async def _send_run_pipeline(self, pipeline_name: Optional[str] = None) -> None: """Sends a RunPipeline event with the correct stages.""" if self.settings.wake.enabled: # Local wake word detection @@ -304,8 +313,17 @@ class SatelliteBase: end_stage = PipelineStage.HANDLE run_pipeline = RunPipeline( - start_stage=start_stage, end_stage=end_stage, restart_on_end=restart_on_end + start_stage=start_stage, + end_stage=end_stage, + name=pipeline_name, + restart_on_end=restart_on_end, + snd_format=AudioFormat( + rate=self.settings.snd.rate, + width=self.settings.snd.width, + channels=self.settings.snd.channels, + ), ).event() + _LOGGER.debug(run_pipeline) await self.event_to_server(run_pipeline) await self.forward_event(run_pipeline) @@ -443,15 +461,36 @@ class SatelliteBase: continue # Audio processing - if self.settings.mic.needs_processing and AudioChunk.is_type( - event.type + if ( + self.settings.mic.needs_processing + or (self.settings.mic.channel_index is not None) + and AudioChunk.is_type(event.type) ): chunk = AudioChunk.from_event(event) - audio_bytes = self._process_mic_audio(chunk.audio) + if self.settings.mic.channel_index is not None: + if chunk.width != 2: + raise ValueError( + "Mic channel index selection requires 16-bit samples" + ) + + # Convert to unsigned 16-bit array to make channel extraction easier + audio_array = array.array("H", chunk.audio) + audio_bytes = audio_array[ + self.settings.mic.channel_index :: chunk.channels + ].tobytes() + + if self.settings.mic.needs_processing: + if audio_bytes is None: + audio_bytes = chunk.audio + + audio_bytes = self._process_mic_audio(chunk.audio) + event = AudioChunk( rate=chunk.rate, width=chunk.width, - channels=chunk.channels, + channels=chunk.channels + if (self.settings.mic.channel_index is None) + else 1, audio=audio_bytes, ).event() else: @@ -733,7 +772,11 @@ 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()) + wake_names: Optional[List[str]] = None + if self.settings.wake.names: + wake_names = [w.name for w in self.settings.wake.names] + + await self.event_to_wake(Detect(names=wake_names).event()) await self.trigger_detect() # ------------------------------------------------------------------------- @@ -856,6 +899,13 @@ class SatelliteBase: await _disconnect() + # ------------------------------------------------------------------------- + # Info + # ------------------------------------------------------------------------- + + async def update_info(self, info: Info) -> None: + pass + # ----------------------------------------------------------------------------- @@ -1108,6 +1158,9 @@ class WakeStreamingSatellite(SatelliteBase): self._is_paused = False + self._wake_info: Optional[Info] = None + self._wake_info_ready = asyncio.Event() + async def event_from_server(self, event: Event) -> None: # Only check event types once is_run_satellite = False @@ -1201,8 +1254,13 @@ class WakeStreamingSatellite(SatelliteBase): await self.event_to_wake(event) async def event_from_wake(self, event: Event) -> None: + if Info.is_type(event.type): + self._wake_info = Info.from_event(event) + self._wake_info_ready.set() + return + if self.is_streaming or (self.server_id is None): - # Not streaming or no server connected + # Not detecting or no server connected return if Detection.is_type(event.type): @@ -1224,6 +1282,8 @@ class WakeStreamingSatellite(SatelliteBase): if self.stt_audio_writer is not None: self.stt_audio_writer.start(timestamp=self._debug_recording_timestamp) + _LOGGER.debug(detection) + self.is_streaming = True _LOGGER.debug("Streaming audio") @@ -1237,7 +1297,35 @@ class WakeStreamingSatellite(SatelliteBase): # No refractory period self.refractory_timestamp.pop(detection.name, None) - await self._send_run_pipeline() + # Forward to the server + await self.event_to_server(event) + + # Match detected wake word name with pipeline name + pipeline_name: Optional[str] = None + if self.settings.wake.names: + detection_name = normalize_wake_word(detection.name) + for wake_name in self.settings.wake.names: + if normalize_wake_word(wake_name.name) == detection_name: + pipeline_name = wake_name.pipeline + break + + await self._send_run_pipeline(pipeline_name=pipeline_name) await self.forward_event(event) # forward to event service await self.trigger_detection(Detection.from_event(event)) await self.trigger_streaming_start() + + async def update_info(self, info: Info) -> None: + self._wake_info = None + self._wake_info_ready.clear() + await self.event_to_wake(Describe().event()) + + try: + await asyncio.wait_for( + self._wake_info_ready.wait(), timeout=_WAKE_INFO_TIMEOUT + ) + + if self._wake_info is not None: + # Update wake info only + info.wake = self._wake_info.wake + except asyncio.TimeoutError: + _LOGGER.warning("Failed to get info from wake service") diff --git a/wyoming_satellite/settings.py b/wyoming_satellite/settings.py index 138f884..903ff27 100644 --- a/wyoming_satellite/settings.py +++ b/wyoming_satellite/settings.py @@ -55,6 +55,9 @@ class MicSettings(ServiceSettings): seconds_to_mute_after_awake_wav: float = 0.5 """Extra second(s) of microphone audio to mute after awake WAV has finished playing.""" + channel_index: Optional[int] = None + """Index of microphone channel to use. When not set, all input channels are used as-is.""" + @property def needs_webrtc(self) -> bool: """True if webrtc audio enhancements are needed.""" @@ -100,11 +103,19 @@ class SndSettings(ServiceSettings): return self.enabled and (self.volume_multiplier != 1.0) +@dataclass(frozen=True) +class WakeWordAndPipeline: + """Wake word name + optional pipeline name.""" + + name: str + pipeline: Optional[str] = None + + @dataclass(frozen=True) class WakeSettings(ServiceSettings): """Wake word service settings.""" - names: Optional[List[str]] = None + names: Optional[List[WakeWordAndPipeline]] = None """List of wake word names to listen for.""" rate: int = 16000 diff --git a/wyoming_satellite/utils/__init__.py b/wyoming_satellite/utils/__init__.py index c91a7b7..e47c60b 100644 --- a/wyoming_satellite/utils/__init__.py +++ b/wyoming_satellite/utils/__init__.py @@ -10,6 +10,7 @@ from .misc import ( get_mac_address, needs_silero, needs_webrtc, + normalize_wake_word, run_event_command, split_command, ) @@ -22,6 +23,7 @@ __all__ = [ "multiply_volume", "needs_silero", "needs_webrtc", + "normalize_wake_word", "run_event_command", "split_command", "wav_to_events", diff --git a/wyoming_satellite/utils/misc.py b/wyoming_satellite/utils/misc.py index a5d59ff..5d39d14 100644 --- a/wyoming_satellite/utils/misc.py +++ b/wyoming_satellite/utils/misc.py @@ -2,8 +2,11 @@ import argparse import asyncio import logging +import re import shlex +import unicodedata import uuid +from functools import lru_cache from typing import List, Optional _LOGGER = logging.getLogger() @@ -56,3 +59,24 @@ def split_command(command: Optional[str]) -> Optional[List[str]]: return None return shlex.split(command) + + +@lru_cache +def normalize_wake_word(wake_word: str) -> str: + """Normalizes a wake word name for comparison.""" + + # Lower case + wake_word = wake_word.strip().casefold() + + # Remove version numbers like v1.0 + wake_word = re.sub(r"v[0-9]+(\.[0-9])+", "", wake_word) + + # Replace anything besides letters/numbers with whitespace + wake_word = "".join( + c if unicodedata.category(c) in ("Ll", "Nd") else " " for c in wake_word + ) + + # Normalize whitespace + wake_word = " ".join(wake_word.strip().split()) + + return wake_word