diff --git a/README.md b/README.md index 95ed1dd..71d94c9 100644 --- a/README.md +++ b/README.md @@ -130,10 +130,10 @@ You can also use `--mic-volume-multiplier X` to multiply all audio samples by `X Satellites can respond to events from the server by running commands: -* `--startup-command` - run when satellite starts +* `--startup-command` - run when satellite starts (no stdin) * `--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 +* `--streaming-start-command` - audio has started streaming to server (no stdin) +* `--streaming-stop-command` - audio has stopped streaming to server (no stdin) * `--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 3093191..1e5370a 100644 --- a/wyoming_satellite/__main__.py +++ b/wyoming_satellite/__main__.py @@ -2,32 +2,17 @@ import argparse import asyncio import logging -import math import shlex import sys import time import uuid -import wave -from enum import Enum from functools import partial from pathlib import Path -from typing import Callable, Final, Iterator, List, Optional, Union +from typing import List, Optional -from pyring_buffer import RingBuffer -from wyoming.asr import Transcribe, Transcript -from wyoming.audio import AudioChunk, AudioStart, AudioStop -from wyoming.client import AsyncClient -from wyoming.error import Error from wyoming.event import Event from wyoming.info import Attribution, Describe, Info, Satellite -from wyoming.mic import MicProcessAsyncClient -from wyoming.pipeline import PipelineStage, RunPipeline -from wyoming.satellite import RunSatellite from wyoming.server import AsyncEventHandler, AsyncServer, AsyncTcpServer -from wyoming.snd import SndProcessAsyncClient -from wyoming.tts import Synthesize -from wyoming.vad import VoiceStarted, VoiceStopped -from wyoming.wake import Detect, Detection, WakeProcessAsyncClient from .satellite import ( AlwaysStreamingSatellite, @@ -43,13 +28,11 @@ from .settings import ( VadSettings, WakeSettings, ) -from .utils import AudioBuffer, chunk_samples, multiply_volume, run_event_command +from .utils import run_event_command _LOGGER = logging.getLogger() _DIR = Path(__file__).parent -# TODO: events service - async def main() -> None: """Main entry point.""" @@ -396,59 +379,9 @@ class SatelliteEventHandler(AsyncEventHandler): self.client_id = str(time.monotonic_ns()) self.satellite = satellite - # self.is_running = True - # self.state: Optional[Satellite] = None - # self.after_tts_state: Optional[Satellite] = None - # self.satellite_task: Optional[asyncio.Task] = None - - # self.mic_volume_multiplier = self.cli_args.mic_volume_multiplier - # self.has_mic_audio_processing = needs_webrtc(self.cli_args) or ( - # self.mic_volume_multiplier != 1.0 - # ) - # self.process_webrtc: Optional[Callable[[bytes], bytes]] = None - # if needs_webrtc(self.cli_args): - # self.process_webrtc = WebRtcAudio( - # self.cli_args.auto_gain, self.cli_args.noise_suppression - # ) - - # self.has_wake = bool(self.cli_args.wake_uri or self.cli_args.wake_command) - - # self.has_snd = bool(self.cli_args.snd_uri or self.cli_args.snd_command) - # self.snd_client: Optional[AsyncClient] = None - # self.snd_volume_multiplier = self.cli_args.snd_volume_multiplier - # self.has_snd_audio_processing = self.snd_volume_multiplier != 1.0 - - # self.process_vad: Optional[Callable[[Optional[bytes]], bool]] = None - # self.vad_buffer: Optional[RingBuffer] = None - # if needs_silero(self.cli_args): - # self.process_vad = SileroVad( - # self.cli_args.vad_threshold, self.cli_args.vad_trigger_level - # ) - # if self.cli_args.vad_buffer_seconds > 0: - # # Assume 16Khz, 16-bit mono samples - # vad_buffer_bytes = int( - # math.ceil(self.cli_args.vad_buffer_seconds * 16000 * 2) - # ) - # self.vad_buffer = RingBuffer(vad_buffer_bytes) - - # self.events_client: Optional[AsyncClient] = None - - # self.server_state = server_state - # self.server_state.current_client_id = self.client_id _LOGGER.debug("Client connected: %s", self.client_id) self.satellite.set_server(self.client_id, self.writer) - # self._stop_if_new_client_task = asyncio.create_task(self._stop_if_new_client()) - - # async def _stop_if_new_client(self) -> None: - # """Stop this event handler if a new client connects.""" - # while self.is_running: - # if self.server_state.current_client_id != self.client_id: - # self.is_running = False - # _LOGGER.debug("Detected new client. Stopping old client.") - # else: - # await asyncio.sleep(1) - # ------------------------------------------------------------------------- async def handle_event(self, event: Event) -> bool: @@ -466,511 +399,6 @@ class SatelliteEventHandler(AsyncEventHandler): return True - # 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) - # _LOGGER.debug("Sent info to client: %s", self.client_id) - # return True - - # if RunSatellite.is_type(event.type) and (self.satellite_task is None): - # if self.has_wake: - # # Local wake word detection - # self.state = SatelliteState.WAKE_START - # self.satellite_task = asyncio.create_task(self.run_satellite_wake()) - # else: - # # Remote wake word detection - # self.state = SatelliteState.VAD_RESET - # self.satellite_task = asyncio.create_task(self.run_satellite()) - # elif Transcript.is_type(event.type): - # if self.cli_args.done_wav: - # self.state = SatelliteState.PLAYING_AUDIO - # await self._play_wav(self.cli_args.done_wav) - - # # STT transcript - # await self._forward_event(event) - # _LOGGER.debug("-> %s", event) - - # if self.has_wake: - # # Ready for next wake word detection - # self.state = SatelliteState.WAKE_START - # else: - # self.state = SatelliteState.VAD_RESET - # elif self.has_snd and ( - # AudioStart.is_type(event.type) - # or AudioChunk.is_type(event.type) - # or AudioStop.is_type(event.type) - # ): - # if AudioStart.is_type(event.type): - # # TTS start - # await self._ensure_snd_client() - # if self.snd_client is not None: - # await self.snd_client.connect() - - # # State will be restored when audio ends - # self.after_tts_state = self.state - # self.state = SatelliteState.PLAYING_AUDIO - - # await self._forward_event(event) - # _LOGGER.debug("-> %s", event) - # elif AudioChunk.is_type(event.type) and (self.snd_client is not None): - # # Forward to sound service - # event = self._process_snd_audio(event) - # await self.snd_client.write_event(event) - # elif AudioStop.is_type(event.type): - # # TTS stop - # await self._forward_event(event) - # _LOGGER.debug("-> %s", event) - - # if self.snd_client is not None: - # await self.snd_client.disconnect() - # self.snd_client = None - - # if self.after_tts_state is not None: - # # Restore state after playing audio - # self.state = self.after_tts_state - # self.after_tts_state = None - # elif Detection.is_type(event.type): - # if self.cli_args.awake_wav: - # last_state = self.state - # self.state = SatelliteState.PLAYING_AUDIO - # await self._play_wav(self.cli_args.awake_wav) - # self.state = last_state - - # await self._forward_event(event) - # _LOGGER.debug("-> %s", event) - # elif ( - # Detect.is_type(event.type) - # or Transcribe.is_type(event.type) - # or VoiceStarted.is_type(event.type) - # or VoiceStopped.is_type(event.type) - # or Synthesize.is_type(event.type) - # ): - # # Other client events: - # # - Detect for start of wake word detection - # # - Transcribe for when STT starts - # # - VoiceStarted/VoiceStopped for when user starts/stops speaking - # # - Synthesize for TTS text - # await self._forward_event(event) - # _LOGGER.debug("-> %s", event) - # elif Error.is_type(event.type): - # await self._forward_event(event) - # _LOGGER.warning(event) - # else: - # _LOGGER.debug("Unexpected event: type=%s, data=%s", event.type, event.data) - - # return True - - # async def disconnect(self) -> None: - # self.is_running = False - - # # ------------------------------------------------------------------------- - # # Tasks - # # ------------------------------------------------------------------------- - - # async def run_satellite(self) -> None: - # """Task to read mic input and do remote wake word detection.""" - # vad_timeout_sec: Optional[float] = None - - # try: - # mic_client = self._make_mic_client() - # async with mic_client: - # while self.is_running: - # mic_event = await mic_client.read_event() - # if mic_event is None: - # _LOGGER.warning("Microphone service disconnected") - # break - - # if not AudioChunk.is_type(mic_event.type): - # continue - - # mic_event = self._process_mic_audio(mic_event) - - # if self.state == SatelliteState.VAD_RESET: - # # Reset vad state - # if self.process_vad is not None: - # self.process_vad(None) - - # if self.vad_buffer is not None: - # self.vad_buffer.put(bytes(self.vad_buffer.maxlen)) - - # self.state = SatelliteState.VAD_WAIT - # if self.process_vad is not None: - # _LOGGER.info("Waiting for speech") - - # if self.state == SatelliteState.VAD_WAIT: - # if self.process_vad is None: - # # No VAD - # self.state = SatelliteState.ASR - # _LOGGER.info("Streaming audio") - - # # Start pipeline - # await self._run_pipeline(PipelineStage.WAKE) - # else: - # # Wait for speech - # chunk = AudioChunk.from_event(mic_event) - # if self.process_vad(chunk.audio): - # # Ready to stream - # self.state = SatelliteState.WAKE - - # # Start pipeline - # await self._run_pipeline(PipelineStage.WAKE) - - # if self.vad_buffer: - # # Send contents of VAD buffer first - # await self.write_event( - # AudioChunk( - # rate=chunk.rate, - # width=chunk.width, - # channels=chunk.channels, - # audio=self.vad_buffer.getvalue(), - # ).event() - # ) - - # if self.cli_args.vad_wake_word_timeout is not None: - # # Time in the future when timeout will occur - # vad_timeout_sec = ( - # time.monotonic() - # + self.cli_args.vad_wake_word_timeout - # ) - # else: - # # No timeout - # vad_timeout_sec = None - - # _LOGGER.info("Streaming audio") - # elif self.vad_buffer is not None: - # # Save audio right before speech - # self.vad_buffer.put(chunk.audio) - - # if self.state == SatelliteState.WAKE: - # if (vad_timeout_sec is not None) and ( - # time.monotonic() >= vad_timeout_sec - # ): - # # Wake word was not detected within timeout - # self.state = SatelliteState.VAD_RESET - - # # Cancel current pipeline with audio stop - # chunk = AudioChunk.from_event(mic_event) - # await self.write_event(AudioStop().event()) - # continue - - # # Remote wake word detection - # await self.write_event(mic_event) - # elif self.state == SatelliteState.ASR: - # # Forward all audio to server - # await self.write_event(mic_event) - - # except ConnectionResetError: - # _LOGGER.info("Server disconnected") - # except Exception: - # _LOGGER.exception("Unexpected error in run_satellite") - - # async def run_satellite_wake(self) -> None: - # """Task to read mic input and do local wake word detection.""" - # try: - # mic_client = self._make_mic_client() - # wake_client = await self._connect_to_wake() - - # async with mic_client, wake_client: - # # Read events in parallel - # mic_task = asyncio.create_task(mic_client.read_event()) - # wake_task = asyncio.create_task(wake_client.read_event()) - # pending = {mic_task, wake_task} - - # while self.is_running: - # if self.state == SatelliteState.WAKE_START: - # # Send detect event - # _LOGGER.info("Detecting wake word") - - # if self.cli_args.wake_word_name: - # # Request specific wake word(s) - # detect_event = Detect( - # names=self.cli_args.wake_word_name - # ).event() - # else: - # detect_event = Detect().event() - - # _LOGGER.debug("<- %s", detect_event) - # await wake_client.write_event(detect_event) - # await self._forward_event(detect_event) - - # self.state = SatelliteState.WAKE - - # # Wait for mic/wake events - # done, pending = await asyncio.wait( - # pending, return_when=asyncio.FIRST_COMPLETED - # ) - - # if mic_task in done: - # mic_event = mic_task.result() - # if mic_event is None: - # _LOGGER.warning("Microphone service disconnected") - # break - - # if AudioChunk.is_type(mic_event.type): - # mic_event = self._process_mic_audio(mic_event) - - # # Forward to wake service - # await wake_client.write_event(mic_event) - - # if self.state == SatelliteState.ASR: - # # Forward to server - # await self.write_event(mic_event) - - # # Next audio chunk - # mic_task = asyncio.create_task(mic_client.read_event()) - # pending.add(mic_task) - # elif wake_task in done: - # wake_event = wake_task.result() - # if wake_event is None: - # _LOGGER.warning("Wake word service disconnected") - # break - - # if Detection.is_type(wake_event.type) and ( - # self.state == SatelliteState.WAKE - # ): - # _LOGGER.debug("Wake word detected: %s", wake_event) - - # if self.cli_args.awake_wav: - # self.state = SatelliteState.PLAYING_AUDIO - # await self._play_wav(self.cli_args.awake_wav) - - # self.state = SatelliteState.ASR - - # # Start pipeline - # await self._run_pipeline(PipelineStage.ASR) - - # # Forward to server - # await self.write_event(wake_event) - - # # Forward to client - # await self._forward_event(wake_event) - - # _LOGGER.info("Streaming audio") - - # # Next wake event - # wake_task = asyncio.create_task(wake_client.read_event()) - # pending.add(wake_task) - # except ConnectionResetError: - # _LOGGER.info("Server disconnected") - # except Exception: - # _LOGGER.exception("Unexpected error in run_satellite") - - # async def _connect_to_wake(self) -> AsyncClient: - # assert self.cli_args.wake_uri or self.cli_args.wake_command - - # if self.cli_args.wake_command: - # _LOGGER.debug("Running %s", self.cli_args.wake_command) - # program, *program_args = shlex.split(self.cli_args.wake_command) - # return WakeProcessAsyncClient( - # rate=self.cli_args.wake_command_rate, - # width=self.cli_args.wake_command_width, - # channels=self.cli_args.wake_command_channels, - # program=program, - # program_args=program_args, - # ) - - # while True: - # try: - # _LOGGER.debug("Connecting to %s", self.cli_args.wake_uri) - # return AsyncClient.from_uri(self.cli_args.wake_uri) - # except (ConnectionError, OSError): - # _LOGGER.exception( - # "Unexpected error connecting to wake service. " - # "Reconnecting in %s second(s)", - # _WAKE_RECONNECT_SECONDS, - # ) - # await asyncio.sleep(_WAKE_RECONNECT_SECONDS) - - # # ------------------------------------------------------------------------- - - # def _make_mic_client(self) -> AsyncClient: - # """Create mic client.""" - # if self.cli_args.mic_command: - # _LOGGER.debug("Running %s", self.cli_args.mic_command) - # program, *program_args = shlex.split(self.cli_args.mic_command) - # return MicProcessAsyncClient( - # rate=self.cli_args.mic_command_rate, - # width=self.cli_args.mic_command_width, - # channels=self.cli_args.mic_command_channels, - # samples_per_chunk=self.cli_args.mic_command_samples_per_chunk, - # program=program, - # program_args=program_args, - # ) - - # assert self.cli_args.mic_uri - # _LOGGER.debug("Connecting to %s", self.cli_args.mic_uri) - # return AsyncClient.from_uri(self.cli_args.mic_uri) - - # async def _ensure_snd_client(self) -> None: - # """Create snd client if necessary.""" - # if (not self.has_snd) or (self.snd_client is not None): - # return - - # if self.cli_args.snd_command: - # _LOGGER.debug("Running %s", self.cli_args.snd_command) - # program, *program_args = shlex.split(self.cli_args.snd_command) - # self.snd_client = SndProcessAsyncClient( - # rate=self.cli_args.snd_command_rate, - # width=self.cli_args.snd_command_width, - # channels=self.cli_args.snd_command_channels, - # program=program, - # program_args=program_args, - # ) - # else: - # _LOGGER.debug("Connecting to %s", self.cli_args.snd_uri) - # self.snd_client = AsyncClient.from_uri(self.cli_args.snd_uri) - - # async def _run_pipeline(self, start_stage: PipelineStage) -> None: - # """Tell server to start running the pipeline.""" - # await self.write_event( - # RunPipeline( - # start_stage=start_stage, - # end_stage=PipelineStage.TTS if self.has_snd else PipelineStage.HANDLE, - # ).event() - # ) - - # if self.cli_args.streaming_start_command: - # await run_event_command(self.cli_args.streaming_start_command) - - # def _process_mic_audio(self, audio_event: Event) -> Event: - # """Perform microphone audio processing, if necessary.""" - # if not self.has_mic_audio_processing: - # # No processing needed - # return audio_event - - # chunk = AudioChunk.from_event(audio_event) - # audio_bytes = chunk.audio - - # if self.process_webrtc is not None: - # audio_bytes = self.process_webrtc(audio_bytes) - - # if self.mic_volume_multiplier != 1.0: - # audio_bytes = multiply_volume(audio_bytes, self.mic_volume_multiplier) - - # return AudioChunk( - # rate=chunk.rate, - # width=chunk.width, - # channels=chunk.channels, - # audio=audio_bytes, - # ).event() - - # def _process_snd_audio(self, audio_event: Event) -> Event: - # """Perform output audio processing, if necessary.""" - # if not self.has_snd_audio_processing: - # # No processing needed - # return audio_event - - # chunk = AudioChunk.from_event(audio_event) - # audio_bytes = chunk.audio - - # if self.snd_volume_multiplier != 1.0: - # audio_bytes = multiply_volume(audio_bytes, self.snd_volume_multiplier) - - # return AudioChunk( - # rate=chunk.rate, - # width=chunk.width, - # channels=chunk.channels, - # audio=audio_bytes, - # ).event() - - # async def _forward_event(self, event: Event) -> None: - # """Forward a Wyoming event to a client and run event commands.""" - # if self.cli_args.detect_command and Detect.is_type(event.type): - # # Wake word detection started - # await run_event_command(self.cli_args.detect_command) - # elif self.cli_args.detection_command and Detection.is_type(event.type): - # # Wake word is detected - # detection = Detection.from_event(event) - # await run_event_command(self.cli_args.detection_command, detection.name) - # elif self.cli_args.transcript_command and Transcript.is_type(event.type): - # # STT text is available - # transcript = Transcript.from_event(event) - # await run_event_command(self.cli_args.transcript_command, transcript.text) - # elif self.cli_args.stt_start_command and VoiceStarted.is_type(event.type): - # # User starts speaking - # await run_event_command(self.cli_args.stt_start_command) - # elif self.cli_args.stt_stop_command and VoiceStopped.is_type(event.type): - # # User stops speaking - # await run_event_command(self.cli_args.stt_stop_command) - # elif self.cli_args.synthesize_command and Synthesize.is_type(event.type): - # # TTS text is available - # synthesize = Synthesize.from_event(event) - # await run_event_command(self.cli_args.synthesize_command, synthesize.text) - # elif self.cli_args.tts_start_command and AudioStart.is_type(event.type): - # # TTS audio start - # await run_event_command(self.cli_args.tts_start_command) - # elif self.cli_args.tts_stop_command and AudioStop.is_type(event.type): - # # TTS audio stop - # await run_event_command(self.cli_args.tts_stop_command) - # elif self.cli_args.error_command and Error.is_type(event.type): - # # Error occurred - # error = Error.from_event(event) - # await run_event_command(self.cli_args.error_command, error.text) - - # if not self.cli_args.event_uri: - # # No external service - # return - - # if self.events_client is None: - # self.events_client = AsyncClient.from_uri(self.cli_args.event_uri) - # _LOGGER.debug("Connecting to %s", self.cli_args.event_uri) - # await self.events_client.connect() - - # assert self.events_client is not None - # await self.events_client.write_event(event) - - # async def _play_wav(self, wav_path: Union[str, Path]) -> None: - # """Send WAV file to snd client as audio chunks.""" - # await self._ensure_snd_client() - - # if self.snd_client is None: - # _LOGGER.debug("Cannot play WAV, no snd service") - # return - - # try: - # async with self.snd_client: - # for event in self._wav_to_events(wav_path): - # await self.snd_client.write_event(event) - # except Exception: - # _LOGGER.exception("Unexpected error while playing WAV: %s", wav_path) - # finally: - # self.snd_client = None - - # def _wav_to_events( - # self, wav_path: Union[str, Path], samples_per_chunk: int = 1024 - # ) -> 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 self.snd_volume_multiplier != 1.0: - # audio_bytes = multiply_volume( - # audio_bytes, self.snd_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/satellite.py b/wyoming_satellite/satellite.py index a170fab..6d8539e 100644 --- a/wyoming_satellite/satellite.py +++ b/wyoming_satellite/satellite.py @@ -3,20 +3,22 @@ import asyncio import logging import math import time -from dataclasses import dataclass from enum import Enum, auto from pathlib import Path -from typing import Callable, List, Optional, Set, Union +from typing import Callable, Optional, Set, Union from pyring_buffer import RingBuffer from wyoming.asr import Transcript from wyoming.audio import AudioChunk, AudioStart, AudioStop -from wyoming.client import AsyncClient, AsyncTcpClient -from wyoming.event import Event, async_read_event, async_write_event +from wyoming.client import AsyncClient +from wyoming.error import Error +from wyoming.event import Event, async_write_event from wyoming.mic import MicProcessAsyncClient from wyoming.pipeline import PipelineStage, RunPipeline from wyoming.satellite import RunSatellite from wyoming.snd import SndProcessAsyncClient +from wyoming.tts import Synthesize +from wyoming.vad import VoiceStarted, VoiceStopped from wyoming.wake import Detect, Detection, WakeProcessAsyncClient from .settings import SatelliteSettings @@ -53,6 +55,8 @@ class SatelliteBase: self._snd_queue: "Optional[asyncio.Queue[Event]]" = None self._wake_task: Optional[asyncio.Task] = None self._wake_queue: "Optional[asyncio.Queue[Event]]" = None + self._event_task: Optional[asyncio.Task] = None + self._event_queue: "Optional[asyncio.Queue[Event]]" = None @property def is_running(self) -> bool: @@ -121,31 +125,45 @@ class SatelliteBase: async def event_from_server(self, event: Event) -> None: """Called when an event is received from the server.""" - if RunSatellite.is_type(event.type): - if self.settings.wake.enabled: - # Local wake word detection - start_stage = PipelineStage.ASR - else: - # Remote wake word detection - start_stage = PipelineStage.WAKE - - if self.settings.snd.enabled: - # Play TTS response - end_stage = PipelineStage.TTS - else: - # No audio output - end_stage = PipelineStage.INTENT - - await self.event_to_server( - RunPipeline(start_stage=start_stage, end_stage=end_stage).event() - ) - elif ( - AudioChunk.is_type(event.type) - or AudioStart.is_type(event.type) - or AudioStop.is_type(event.type) - ): - # Forward to sound service + if AudioChunk.is_type(event.type): + # TTS audio await self.event_to_snd(event) + elif AudioStart.is_type(event.type): + # TTS started + await self.event_to_snd(event) + await self.trigger_tts_start() + elif AudioStop.is_type(event.type): + # TTS stopped + await self.event_to_snd(event) + await self.trigger_tts_stop() + elif Detect.is_type(event.type): + # Wake word detection started + await self.trigger_detect() + elif Detection.is_type(event.type): + # Wake word detected + _LOGGER.debug("Wake word detected") + await self.trigger_detection(Detection.from_event(event)) + elif VoiceStarted.is_type(event.type): + # STT start + await self.trigger_stt_start() + elif VoiceStopped.is_type(event.type): + # STT stop + await self.trigger_stt_stop() + elif Transcript.is_type(event.type): + # STT text + _LOGGER.debug(event) + await self.trigger_transcript(Transcript.from_event(event)) + elif Synthesize.is_type(event.type): + # TTS request + _LOGGER.debug(event) + await self.trigger_synthesize(Synthesize.from_event(event)) + elif Error.is_type(event.type): + _LOGGER.warning(event) + await self.trigger_error(Error.from_event(event)) + + # Forward everything except audio to event service + if not AudioChunk.is_type(event.type): + await self.forward_event(event) async def event_to_server(self, event: Event) -> None: """Send an event to the server.""" @@ -154,10 +172,24 @@ class SatelliteBase: await async_write_event(event, self._writer) - async def event_to_snd(self, event: Event) -> None: - """Send an event to the sound service.""" - if self._snd_queue is not None: - self._snd_queue.put_nowait(event) + async def _send_run_pipeline(self) -> None: + if self.settings.wake.enabled: + # Local wake word detection + start_stage = PipelineStage.ASR + else: + # Remote wake word detection + start_stage = PipelineStage.WAKE + + if self.settings.snd.enabled: + # Play TTS response + end_stage = PipelineStage.TTS + else: + # No audio output + end_stage = PipelineStage.INTENT + + run_pipeline = RunPipeline(start_stage=start_stage, end_stage=end_stage).event() + await self.event_to_server(run_pipeline) + await self.forward_event(run_pipeline) async def _restart(self) -> None: self.state = State.RESTARTING @@ -189,19 +221,36 @@ class SatelliteBase: ) self._wake_task = asyncio.create_task(self._wake_task_proc()) + 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()) + _LOGGER.info("Connected to services") async def _disconnect_from_services(self) -> None: if self._mic_task is not None: - _LOGGER.debug("Stopping microphone") + _LOGGER.debug("Stopping microphone service") self._mic_task.cancel() self._mic_task = None if self._snd_task is not None: - _LOGGER.debug("Stopping sound") + _LOGGER.debug("Stopping sound service") self._snd_task.cancel() self._snd_task = None + if self._wake_task is not None: + _LOGGER.debug("Stopping wake service") + self._wake_task.cancel() + self._wake_task = None + + if self._event_task is not None: + _LOGGER.debug("Stopping event service") + self._event_task.cancel() + self._event_task = None + _LOGGER.debug("Disconnected from services") # ------------------------------------------------------------------------- @@ -304,6 +353,11 @@ class SatelliteBase: # Sound # ------------------------------------------------------------------------- + async def event_to_snd(self, event: Event) -> None: + """Send an event to the sound service.""" + if self._snd_queue is not None: + self._snd_queue.put_nowait(event) + def _make_snd_client(self) -> Optional[AsyncClient]: if self.settings.snd.command: program, *program_args = self.settings.snd.command @@ -499,6 +553,70 @@ class SatelliteBase: await run_event_command(self.settings.event.transcript, transcript.text) await self._play_wav(self.settings.snd.done_wav) + async def trigger_stt_start(self) -> None: + await run_event_command(self.settings.event.stt_start) + + async def trigger_stt_stop(self) -> None: + await run_event_command(self.settings.event.stt_stop) + + async def trigger_synthesize(self, synthesize: Synthesize) -> None: + await run_event_command(self.settings.event.synthesize, synthesize.text) + + async def trigger_tts_start(self) -> None: + await run_event_command(self.settings.event.tts_start) + + async def trigger_tts_stop(self) -> None: + await run_event_command(self.settings.event.tts_stop) + + async def trigger_error(self, error: Error) -> None: + await run_event_command(self.settings.event.error, error.text) + + async def forward_event(self, event: Event) -> None: + """Forward an event to the event service.""" + if self._event_queue is not None: + self._event_queue.put_nowait(event) + + def _make_event_client(self) -> Optional[AsyncClient]: + if self.settings.event.uri: + return AsyncClient.from_uri(self.settings.event.uri) + + return None + + async def _event_task_proc(self) -> None: + event_client: Optional[AsyncClient] = None + + async def _disconnect() -> None: + try: + if event_client is not None: + await event_client.disconnect() + except Exception: + pass # ignore disconnect errors + + while self.is_running: + try: + if self._event_queue is None: + self._event_queue = asyncio.Queue() + + event = await self._event_queue.get() + + if event_client is None: + event_client = self._make_event_client() + assert event_client is not None + await event_client.connect() + _LOGGER.debug("Connected to event service") + + await event_client.write_event(event) + except asyncio.CancelledError: + break + except Exception: + _LOGGER.exception("Unexpected error in event read task") + await _disconnect() + event_client = None # reconnect + self._event_queue = None + await asyncio.sleep(self.settings.event.reconnect_seconds) + + await _disconnect() + # ----------------------------------------------------------------------------- @@ -516,15 +634,8 @@ class AlwaysStreamingSatellite(SatelliteBase): if RunSatellite.is_type(event.type): self.is_streaming = True _LOGGER.info("Streaming audio") + await self._send_run_pipeline() 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 @@ -565,14 +676,6 @@ 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 @@ -612,6 +715,7 @@ class VadStreamingSatellite(SatelliteBase): # Speech detected self.is_streaming = True _LOGGER.info("Streaming audio") + await self._send_run_pipeline() await self.trigger_streaming_start() if self.settings.vad.wake_word_timeout is not None: @@ -666,9 +770,7 @@ class WakeStreamingSatellite(SatelliteBase): _LOGGER.info("Waiting for wake word") if Transcript.is_type(event.type): - _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 @@ -690,5 +792,6 @@ class WakeStreamingSatellite(SatelliteBase): if Detection.is_type(event.type): self.is_streaming = True _LOGGER.debug("Streaming audio") + await self._send_run_pipeline() await self.trigger_detection(Detection.from_event(event)) await self.trigger_streaming_start()