Add initial test

script/format

@@ -7,7 +7,12 @@ _DIR = Path(__file__).parent
 _PROGRAM_DIR = _DIR.parent
 _VENV_DIR = _PROGRAM_DIR / ".venv"
 _MODULE_DIR = _PROGRAM_DIR / "wyoming_satellite"
+_TESTS_DIR = _PROGRAM_DIR / "tests"
 
 context = venv.EnvBuilder().ensure_directories(_VENV_DIR)
-subprocess.check_call([context.env_exe, "-m", "black", str(_MODULE_DIR)])
+subprocess.check_call(
-subprocess.check_call([context.env_exe, "-m", "isort", str(_MODULE_DIR)])
+    [context.env_exe, "-m", "black", str(_MODULE_DIR), str(_TESTS_DIR)]
+)
+subprocess.check_call(
+    [context.env_exe, "-m", "isort", str(_MODULE_DIR), str(_TESTS_DIR)]
+)

script/lint

@@ -7,10 +7,21 @@ _DIR = Path(__file__).parent
 _PROGRAM_DIR = _DIR.parent
 _VENV_DIR = _PROGRAM_DIR / ".venv"
 _MODULE_DIR = _PROGRAM_DIR / "wyoming_satellite"
+_TESTS_DIR = _PROGRAM_DIR / "tests"
 
 context = venv.EnvBuilder().ensure_directories(_VENV_DIR)
-subprocess.check_call([context.env_exe, "-m", "black", str(_MODULE_DIR), "--check"])
+subprocess.check_call(
-subprocess.check_call([context.env_exe, "-m", "isort", str(_MODULE_DIR), "--check"])
+    [context.env_exe, "-m", "black", str(_MODULE_DIR), str(_TESTS_DIR), "--check"]
-subprocess.check_call([context.env_exe, "-m", "flake8", str(_MODULE_DIR)])
+)
-subprocess.check_call([context.env_exe, "-m", "pylint", str(_MODULE_DIR)])
+subprocess.check_call(
-subprocess.check_call([context.env_exe, "-m", "mypy", str(_MODULE_DIR)])
+    [context.env_exe, "-m", "isort", str(_MODULE_DIR), str(_TESTS_DIR), "--check"]
+)
+subprocess.check_call(
+    [context.env_exe, "-m", "flake8", str(_MODULE_DIR), str(_TESTS_DIR)]
+)
+subprocess.check_call(
+    [context.env_exe, "-m", "pylint", str(_MODULE_DIR), str(_TESTS_DIR)]
+)
+subprocess.check_call(
+    [context.env_exe, "-m", "mypy", str(_MODULE_DIR), str(_TESTS_DIR)]
+)

tests/shared.py

@@ -1,17 +1,6 @@
 """Shared code for Wyoming satellite tests."""
-from wyoming.info import Info, Attribution, Satellite
 from wyoming.audio import AudioChunk
 
-WYOMING_INFO = Info(
-    satellite=Satellite(
-        name="test satellite",
-        area="test area",
-        description="test description",
-        attribution=Attribution(name="", url=""),
-        installed=True,
-    )
-)
-
 AUDIO_CHUNK = AudioChunk(
     rate=16000, width=2, channels=1, audio=bytes([255] * 960)  # 30ms
 )

tests/test_always_streaming.py

@@ -1,102 +0,0 @@
-import asyncio
-import logging
-import os
-from pathlib import Path
-
-import pytest
-from wyoming.audio import AudioChunk
-from wyoming.wake import Detection
-from wyoming.client import AsyncClient
-from wyoming.event import Event, async_write_event
-from wyoming.server import AsyncServer, AsyncEventHandler
-from wyoming_satellite import (
-    AlwaysStreamingSatellite,
-    EventSettings,
-    SatelliteSettings,
-    MicSettings,
-    WakeSettings,
-)
-
-from .shared import WYOMING_INFO, AUDIO_CHUNK
-
-_LOGGER = logging.getLogger()
-
-class MicEventHandler(AsyncEventHandler):
-    def __init__(self, *args, **kwargs) -> None:
-        super().__init__(*args, **kwargs)
-        self._is_connected = True
-        self._task = asyncio.create_task(self.stream_audio())
-
-    async def handle_event(self, event: Event) -> bool:
-        """Output only."""
-        return True
-
-    async def stream_audio(self) -> None:
-        while self._is_connected:
-            await self.write_event(AUDIO_CHUNK.event())
-            await asyncio.sleep(AUDIO_CHUNK.seconds)
-
-    async def disconnect(self) -> None:
-        self._is_connected = False
-        await self._task
-
-
-class WakeEventHandler(AsyncEventHandler):
-    async def handle_event(self, event: Event) -> bool:
-        _LOGGER.error(event)
-        if AudioChunk.is_type(event.type):
-            await self.write_event(Detection().event())
-
-        return True
-
-
-@pytest.mark.asyncio
-async def test_multiple_wakeups(tmp_path: Path) -> None:
-    mic_socket = tmp_path / "mic.socket"
-    os.mkfifo(mic_socket)
-    mic_uri = f"unix://{mic_socket}"
-    mic_server = AsyncServer.from_uri(mic_uri)
-    mic_task = asyncio.create_task(mic_server.run(MicEventHandler))
-
-    wake_socket = tmp_path / "wake.socket"
-    os.mkfifo(wake_socket)
-    wake_uri = f"unix://{wake_socket}"
-    wake_server = AsyncServer.from_uri(wake_uri)
-    wake_task = asyncio.create_task(wake_server.run(WakeEventHandler))
-
-    wake_event = asyncio.Event()
-
-    class EventEventHandler(AsyncEventHandler):
-        async def handle_event(self, event: Event) -> bool:
-            if Detection.is_type(event.type):
-                wake_event.set()
-
-            return True
-
-    event_socket = tmp_path / "event.socket"
-    os.mkfifo(event_socket)
-    event_uri = f"unix://{event_socket}"
-    event_server = AsyncServer.from_uri(event_uri)
-    event_task = asyncio.create_task(event_server.run(EventEventHandler))
-
-    satellite = AlwaysStreamingSatellite(
-        SatelliteSettings(
-            mic=MicSettings(uri=mic_uri),
-            wake=WakeSettings(uri=wake_uri),
-            event=EventSettings(uri=event_uri),
-        )
-    )
-
-    satellite_task = asyncio.create_task(satellite.run())
-
-    await asyncio.sleep(0.5)
-
-    # async with asyncio.timeout(1):
-    #     await wake_event.wait()
-
-    # await satellite.stop()
-
-    tasks = [mic_task, wake_task, event_task]
-    for task in tasks:
-        task.cancel()
-    await asyncio.wait(tasks)

tests/test_wake_streaming.py

@@ -0,0 +1,108 @@
+import asyncio
+import logging
+from pathlib import Path
+from typing import 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
+from wyoming.satellite import RunSatellite
+from wyoming.wake import Detection
+
+from wyoming_satellite import (
+    EventSettings,
+    MicSettings,
+    SatelliteSettings,
+    WakeSettings,
+    WakeStreamingSatellite,
+)
+
+from .shared import AUDIO_CHUNK
+
+_LOGGER = logging.getLogger()
+
+
+class MicClient(AsyncClient):
+    async def read_event(self) -> Optional[Event]:
+        # Send 30ms of audio every 30ms
+        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._detection_event = asyncio.Event()
+
+    async def read_event(self) -> Optional[Event]:
+        await self._detection_event.wait()
+        return Detection().event()
+
+    async def write_event(self, event: Event) -> None:
+        if AudioChunk.is_type(event.type):
+            self._detection_event.set()
+
+
+class EventClient(AsyncClient):
+    def __init__(self) -> None:
+        super().__init__()
+        self.wake_event = asyncio.Event()
+
+    async def read_event(self) -> Optional[Event]:
+        # Input only
+        return None
+
+    async def write_event(self, event: Event) -> None:
+        _LOGGER.error(event.type)
+        if Detection.is_type(event.type):
+            self.wake_event.set()
+
+
+@pytest.mark.asyncio
+async def test_multiple_wakeups(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"),
+            )
+        )
+
+        satellite_task = asyncio.create_task(satellite.run(), name="satellite")
+        await satellite.event_from_server(RunSatellite().event())
+
+        async with asyncio.timeout(1):
+            await event_client.wake_event.wait()
+            event_client.wake_event.clear()
+
+        # Stop streaming
+        await satellite.event_from_server(Transcript("test").event())
+
+        # Should not trigger again within refractory period (default: 5 sec)
+        with pytest.raises(asyncio.TimeoutError):
+            async with asyncio.timeout(0.15):
+                await event_client.wake_event.wait()
+
+        await satellite.stop()
+        await satellite_task

wyoming_satellite/__main__.py

@@ -123,6 +123,12 @@ async def main() -> None:
         default=1,
         help="Sample channels of wake-command (default: 1)",
     )
+    parser.add_argument(
+        "--wake-refractory-seconds",
+        type=float,
+        default=5.0,
+        help="Seconds after a wake word detection before another detection is handled (default: 5)",
+    )
 
     # Voice activity detector
     parser.add_argument(
@@ -285,6 +291,7 @@ async def main() -> None:
             uri=args.wake_uri,
             command=split_command(args.wake_command),
             names=args.wake_word_name,
+            refractory_seconds=args.wake_refractory_seconds,
         ),
         snd=SndSettings(
             uri=args.snd_uri,

wyoming_satellite/satellite.py

@@ -5,7 +5,7 @@ import math
 import time
 from enum import Enum, auto
 from pathlib import Path
-from typing import Callable, Optional, Set, Union
+from typing import Callable, Dict, Optional, Set, Union
 
 from pyring_buffer import RingBuffer
 from wyoming.asr import Transcript
@@ -63,7 +63,7 @@ class SatelliteBase:
     @property
     def is_running(self) -> bool:
         """True if not stopping/stopped."""
-        return self._state not in (State.STOPPING, State.STOPPED)
+        return self._state not in (State.STOPPED,)
 
     @property
     def state(self) -> State:
@@ -235,28 +235,30 @@ class SatelliteBase:
                 "Connecting to mic service: %s",
                 self.settings.mic.uri or self.settings.mic.command,
             )
-            self._mic_task = asyncio.create_task(self._mic_task_proc())
+            self._mic_task = asyncio.create_task(self._mic_task_proc(), name="mic")
 
         if self.settings.snd.enabled:
             _LOGGER.debug(
                 "Connecting to snd service: %s",
                 self.settings.snd.uri or self.settings.snd.command,
             )
-            self._snd_task = asyncio.create_task(self._snd_task_proc())
+            self._snd_task = asyncio.create_task(self._snd_task_proc(), name="snd")
 
         if self.settings.wake.enabled:
             _LOGGER.debug(
                 "Connecting to wake service: %s",
                 self.settings.wake.uri or self.settings.wake.command,
             )
-            self._wake_task = asyncio.create_task(self._wake_task_proc())
+            self._wake_task = asyncio.create_task(self._wake_task_proc(), name="wake")
 
         if self.settings.event.enabled:
             _LOGGER.debug(
                 "Connecting to event service: %s",
                 self.settings.event.uri or self.settings.event.command,
             )
-            self._event_task = asyncio.create_task(self._event_task_proc())
+            self._event_task = asyncio.create_task(
+                self._event_task_proc(), name="event"
+            )
 
         _LOGGER.info("Connected to services")
 
@@ -528,9 +530,19 @@ class SatelliteBase:
         pending: Set[asyncio.Task] = set()
 
         async def _disconnect() -> None:
+            nonlocal to_client_task, from_client_task
             try:
                 if wake_client is not None:
                     await wake_client.disconnect()
+
+                # Clean up tasks
+                if to_client_task is not None:
+                    to_client_task.cancel()
+                    to_client_task = None
+
+                if from_client_task is not None:
+                    from_client_task.cancel()
+                    from_client_task = None
             except Exception:
                 pass  # ignore disconnect errors
 
@@ -557,12 +569,16 @@ class SatelliteBase:
                 # Read/write in "parallel"
                 if to_client_task is None:
                     # From satellite to wake service
-                    to_client_task = asyncio.create_task(self._wake_queue.get())
+                    to_client_task = asyncio.create_task(
+                        self._wake_queue.get(), name="wake_to_client"
+                    )
                     pending.add(to_client_task)
 
                 if from_client_task is None:
                     # From wake service to satellite
-                    from_client_task = asyncio.create_task(wake_client.read_event())
+                    from_client_task = asyncio.create_task(
+                        wake_client.read_event(), name="wake_from_client"
+                    )
                     pending.add(from_client_task)
 
                 done, pending = await asyncio.wait(
@@ -717,6 +733,12 @@ class AlwaysStreamingSatellite(SatelliteBase):
         super().__init__(settings)
         self.is_streaming = False
 
+        if settings.vad.enabled:
+            _LOGGER.warning("VAD is enabled but will not be used")
+
+        if settings.wake.enabled:
+            _LOGGER.warning("Local wake word detection is enabled but will not be used")
+
     async def event_from_server(self, event: Event) -> None:
         await super().event_from_server(event)
 
@@ -771,6 +793,9 @@ class VadStreamingSatellite(SatelliteBase):
             vad_buffer_bytes = int(math.ceil(settings.vad.buffer_seconds * 16000 * 2))
             self.vad_buffer = RingBuffer(maxlen=vad_buffer_bytes)
 
+        if settings.wake.enabled:
+            _LOGGER.warning("Local wake word detection is enabled but will not be used")
+
     async def event_from_server(self, event: Event) -> None:
         await super().event_from_server(event)
 
@@ -872,6 +897,14 @@ class WakeStreamingSatellite(SatelliteBase):
         super().__init__(settings)
         self.is_streaming = False
 
+        # Timestamp in the future when the refractory period is over (set with
+        # time.monotonic()).
+        # wake word id -> seconds
+        self.refractory_timestamp: Dict[Optional[str], float] = {}
+
+        if settings.vad.enabled:
+            _LOGGER.warning("VAD is enabled but will not be used")
+
     async def event_from_server(self, event: Event) -> None:
         if Transcript.is_type(event.type):
             # Stop streaming before event_from_server is called because it will
@@ -904,8 +937,29 @@ class WakeStreamingSatellite(SatelliteBase):
             return
 
         if Detection.is_type(event.type):
+            detection = Detection.from_event(event)
+
+            # Check refractory period to avoid multiple back-to-back detections
+            refractory_timestamp = self.refractory_timestamp.get(detection.name)
+            if (refractory_timestamp is not None) and (
+                refractory_timestamp > time.monotonic()
+            ):
+                _LOGGER.debug("Wake word detection occurred during refractory period")
+                return
+
             self.is_streaming = True
             _LOGGER.debug("Streaming audio")
+
+            if self.settings.wake.refractory_seconds is not None:
+                # Another detection may not occur for this wake word until
+                # refractory period is over.
+                self.refractory_timestamp[detection.name] = (
+                    time.monotonic() + self.settings.wake.refractory_seconds
+                )
+            else:
+                # No refractory period
+                self.refractory_timestamp.pop(detection.name, None)
+
             await self._send_run_pipeline()
             await self.forward_event(event)  # forward to event service
             await self.trigger_detection(Detection.from_event(event))

wyoming_satellite/settings.py

@@ -109,6 +109,9 @@ class WakeSettings(ServiceSettings):
     channels: int = 1
     """Sample channels in wake word audio"""
 
+    refractory_seconds: Optional[float] = 5.0
+    """Seconds after a wake word detection before another detection is handled."""
+
 
 @dataclass(frozen=True)
 class VadSettings: