Merge branch 'rhasspy:master' into bugfix-wav-volume

.github/workflows/test.yml

@@ -1,5 +1,7 @@
+---
 name: test
 
+# yamllint disable-line rule:truthy
 on:
   workflow_dispatch:
   pull_request:

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)

requirements.txt

@@ -1,3 +1,3 @@
-wyoming==1.5.2
+wyoming==1.5.3
 zeroconf==0.88.0
 pyring-buffer==1.0.0

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",

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

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):
+        with pytest.raises(asyncio.TimeoutError):
-        #     await asyncio.wait_for(event_client.wake_event.wait(), timeout=0.15)
+            await asyncio.wait_for(event_client.wake_event.wait(), timeout=0.15)
 
         await satellite.stop()
         await satellite_task

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}

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,

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:

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(
+                if (
-                    event.type
+                    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)
+                    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")

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

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",

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