diff --git a/script/format b/script/format index 615bd8e..3f8a9b2 100755 --- a/script/format +++ b/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([context.env_exe, "-m", "isort", str(_MODULE_DIR)]) +subprocess.check_call( + [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)] +) diff --git a/script/lint b/script/lint index d535816..46ba277 100755 --- a/script/lint +++ b/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([context.env_exe, "-m", "isort", str(_MODULE_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([context.env_exe, "-m", "mypy", str(_MODULE_DIR)]) +subprocess.check_call( + [context.env_exe, "-m", "black", str(_MODULE_DIR), str(_TESTS_DIR), "--check"] +) +subprocess.check_call( + [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)] +) diff --git a/tests/shared.py b/tests/shared.py index aa626f3..8d6e01d 100644 --- a/tests/shared.py +++ b/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 ) diff --git a/tests/test_always_streaming.py b/tests/test_always_streaming.py deleted file mode 100644 index 332c87e..0000000 --- a/tests/test_always_streaming.py +++ /dev/null @@ -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) diff --git a/tests/test_wake_streaming.py b/tests/test_wake_streaming.py new file mode 100644 index 0000000..c2c6077 --- /dev/null +++ b/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 diff --git a/wyoming_satellite/__main__.py b/wyoming_satellite/__main__.py index 4805b77..7178f0a 100644 --- a/wyoming_satellite/__main__.py +++ b/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, diff --git a/wyoming_satellite/satellite.py b/wyoming_satellite/satellite.py index 096fd6e..cfc037d 100644 --- a/wyoming_satellite/satellite.py +++ b/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)) diff --git a/wyoming_satellite/settings.py b/wyoming_satellite/settings.py index 34c99ff..442b70b 100644 --- a/wyoming_satellite/settings.py +++ b/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: