Add ping
This commit is contained in:
Michael Hansen
@@ -5,7 +5,7 @@ import math
|
||||
import time
|
||||
from enum import Enum, auto
|
||||
from pathlib import Path
|
||||
from typing import Callable, Dict, Optional, Set, Union
|
||||
from typing import Callable, Dict, Optional, Set, Union, Final
|
||||
|
||||
from pyring_buffer import RingBuffer
|
||||
from wyoming.asr import Transcript
|
||||
@@ -15,11 +15,18 @@ 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, StreamingStarted, StreamingStopped
|
||||
from wyoming.satellite import (
|
||||
RunSatellite,
|
||||
StreamingStarted,
|
||||
StreamingStopped,
|
||||
SatelliteConnected,
|
||||
SatelliteDisconnected,
|
||||
)
|
||||
from wyoming.snd import SndProcessAsyncClient
|
||||
from wyoming.tts import Synthesize
|
||||
from wyoming.vad import VoiceStarted, VoiceStopped
|
||||
from wyoming.wake import Detect, Detection, WakeProcessAsyncClient
|
||||
from wyoming.ping import Ping, Pong
|
||||
|
||||
from .settings import SatelliteSettings
|
||||
from .utils import DebugAudioWriter, multiply_volume, run_event_command, wav_to_events
|
||||
@@ -28,6 +35,9 @@ from .webrtc import WebRtcAudio
|
||||
|
||||
_LOGGER = logging.getLogger()
|
||||
|
||||
_PONG_TIMEOUT: Final = 5
|
||||
_PING_SEND_DELAY: Final = 2
|
||||
|
||||
|
||||
class State(Enum):
|
||||
NOT_STARTED = auto()
|
||||
@@ -60,6 +70,9 @@ class SatelliteBase:
|
||||
self._event_task: Optional[asyncio.Task] = None
|
||||
self._event_queue: "Optional[asyncio.Queue[Event]]" = None
|
||||
|
||||
self._pong_received_event = asyncio.Event()
|
||||
self._ping_server_task = asyncio.create_task(self._ping_server(), name="ping")
|
||||
|
||||
# Debug audio recording
|
||||
self.wake_audio_writer: Optional[DebugAudioWriter] = None
|
||||
self.stt_audio_writer: Optional[DebugAudioWriter] = None
|
||||
@@ -120,17 +133,19 @@ class SatelliteBase:
|
||||
while self.state != State.STOPPED:
|
||||
await self._state_changed.wait()
|
||||
|
||||
def set_server(self, server_id: str, writer: asyncio.StreamWriter) -> None:
|
||||
async def set_server(self, server_id: str, writer: asyncio.StreamWriter) -> None:
|
||||
"""Set event writer."""
|
||||
self.server_id = server_id
|
||||
self._writer = writer
|
||||
_LOGGER.debug("Server set: %s", server_id)
|
||||
await self.trigger_server_connected()
|
||||
|
||||
def clear_server(self) -> None:
|
||||
async def clear_server(self) -> None:
|
||||
"""Remove writer."""
|
||||
self.server_id = None
|
||||
self._writer = None
|
||||
_LOGGER.debug("Server disconnected")
|
||||
await self.trigger_server_disonnected()
|
||||
|
||||
async def event_to_server(self, event: Event) -> None:
|
||||
"""Send an event to the server."""
|
||||
@@ -147,6 +162,31 @@ class SatelliteBase:
|
||||
else:
|
||||
_LOGGER.exception("Unexpected error sending event to server")
|
||||
|
||||
async def _ping_server(self) -> None:
|
||||
try:
|
||||
while self.is_running:
|
||||
await asyncio.sleep(_PING_SEND_DELAY)
|
||||
if self.server_id is None:
|
||||
# No server connected
|
||||
continue
|
||||
|
||||
# Send ping and wait for pong
|
||||
self._pong_received_event.clear()
|
||||
await self.event_to_server(Ping().event())
|
||||
try:
|
||||
await asyncio.wait_for(
|
||||
self._pong_received_event.wait(), timeout=_PONG_TIMEOUT
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
if self.server_id is None:
|
||||
# No server connected
|
||||
continue
|
||||
|
||||
_LOGGER.warning("Did not receive ping response within timeout")
|
||||
await self.clear_server()
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error in ping server task")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
async def _start(self) -> None:
|
||||
@@ -172,7 +212,14 @@ class SatelliteBase:
|
||||
|
||||
async def event_from_server(self, event: Event) -> None:
|
||||
"""Called when an event is received from the server."""
|
||||
if AudioChunk.is_type(event.type):
|
||||
if Ping.is_type(event.type):
|
||||
# Respond with pong
|
||||
ping = Ping.from_event(event)
|
||||
await self.event_to_server(Pong(text=ping.text).event())
|
||||
elif Pong.is_type(event.type):
|
||||
# Response from our ping
|
||||
self._pong_received_event.set()
|
||||
elif AudioChunk.is_type(event.type):
|
||||
# TTS audio
|
||||
await self.event_to_snd(event)
|
||||
elif AudioStart.is_type(event.type):
|
||||
@@ -217,9 +264,11 @@ class SatelliteBase:
|
||||
if self.settings.wake.enabled:
|
||||
# Local wake word detection
|
||||
start_stage = PipelineStage.ASR
|
||||
restart_on_end = False
|
||||
else:
|
||||
# Remote wake word detection
|
||||
start_stage = PipelineStage.WAKE
|
||||
restart_on_end = not self.settings.vad.enabled
|
||||
|
||||
if self.settings.snd.enabled:
|
||||
# Play TTS response
|
||||
@@ -228,7 +277,9 @@ class SatelliteBase:
|
||||
# No audio output
|
||||
end_stage = PipelineStage.HANDLE
|
||||
|
||||
run_pipeline = RunPipeline(start_stage=start_stage, end_stage=end_stage).event()
|
||||
run_pipeline = RunPipeline(
|
||||
start_stage=start_stage, end_stage=end_stage, restart_on_end=restart_on_end
|
||||
).event()
|
||||
await self.event_to_server(run_pipeline)
|
||||
await self.forward_event(run_pipeline)
|
||||
|
||||
@@ -639,6 +690,18 @@ class SatelliteBase:
|
||||
# Events
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
async def trigger_server_connected(self) -> None:
|
||||
"""Called when connected to server."""
|
||||
_LOGGER.info("Connected to server")
|
||||
await run_event_command(self.settings.event.connected)
|
||||
await self.forward_event(SatelliteConnected().event())
|
||||
|
||||
async def trigger_server_disonnected(self) -> None:
|
||||
"""Called when disconnected from server."""
|
||||
_LOGGER.info("Disconnected from server")
|
||||
await run_event_command(self.settings.event.disconnected)
|
||||
await self.forward_event(SatelliteDisconnected().event())
|
||||
|
||||
async def trigger_streaming_start(self) -> None:
|
||||
"""Called when audio streaming starts."""
|
||||
await run_event_command(self.settings.event.streaming_start)
|
||||
@@ -997,6 +1060,17 @@ class WakeStreamingSatellite(SatelliteBase):
|
||||
if self.wake_audio_writer is not None:
|
||||
self.wake_audio_writer.start(timestamp=self._debug_recording_timestamp)
|
||||
|
||||
async def trigger_server_disonnected(self) -> None:
|
||||
await super().trigger_server_disonnected()
|
||||
|
||||
self.is_streaming = False
|
||||
|
||||
# Stop debug recording (stt)
|
||||
if self.stt_audio_writer is not None:
|
||||
self.stt_audio_writer.stop()
|
||||
|
||||
await self.trigger_streaming_stop()
|
||||
|
||||
async def event_from_mic(
|
||||
self, event: Event, audio_bytes: Optional[bytes] = None
|
||||
) -> None:
|
||||
@@ -1023,7 +1097,8 @@ class WakeStreamingSatellite(SatelliteBase):
|
||||
await self.event_to_wake(event)
|
||||
|
||||
async def event_from_wake(self, event: Event) -> None:
|
||||
if self.is_streaming:
|
||||
if self.is_streaming or (self.server_id is None):
|
||||
# Not streaming or no server connected
|
||||
return
|
||||
|
||||
if Detection.is_type(event.type):
|
||||
|
||||
Reference in New Issue
Block a user