From c2e780dfd2b20d3b6b45e5ba0110198a9d22116f Mon Sep 17 00:00:00 2001 From: Michael Hansen Date: Wed, 24 Jun 2026 02:22:03 -0500 Subject: [PATCH] Improve Wyoming satellite reconnect and tolerance of other satellites (#174460) Co-authored-by: Claude Opus 4.8 (1M context) --- .../components/wyoming/assist_satellite.py | 40 +++++- homeassistant/components/wyoming/data.py | 4 +- tests/components/wyoming/test_data.py | 53 +++++++- tests/components/wyoming/test_satellite.py | 117 +++++++++++++++++- 4 files changed, 205 insertions(+), 9 deletions(-) diff --git a/homeassistant/components/wyoming/assist_satellite.py b/homeassistant/components/wyoming/assist_satellite.py index 41ec62d8a290..2176f89e2fe5 100644 --- a/homeassistant/components/wyoming/assist_satellite.py +++ b/homeassistant/components/wyoming/assist_satellite.py @@ -2,6 +2,7 @@ import asyncio from collections.abc import AsyncGenerator +import contextlib import io import logging import time @@ -416,10 +417,8 @@ class WyomingAssistSatellite(WyomingSatelliteEntity, AssistSatelliteEntity): # Stop existing pipeline self._audio_queue.put_nowait(None) - # Tell satellite to stop running - self._send_pause() - - # Stop task loop + # Stop task loop. The satellite is paused and disconnected in run()'s + # teardown so the pause is reliably sent before the socket is closed. self.is_running = False # Unblock waiting for unmuted @@ -475,6 +474,17 @@ class WyomingAssistSatellite(WyomingSatelliteEntity, AssistSatelliteEntity): # Ensure sensor is off (before stop) self.device.set_is_active(False) + # Pause the satellite, then close the connection. The pause is sent + # and flushed before disconnecting so the satellite reliably sees + # it. Without an explicit disconnect the socket is only released + # when the client is garbage collected, which leaves satellites + # that allow a single connection unable to reconnect when + # re-enabled. + if self._client is not None: + with contextlib.suppress(ConnectionError, OSError): + await self._client.write_event(PauseSatellite().event()) + await self._disconnect() + await self.on_stopped() async def on_restart(self) -> None: @@ -637,9 +647,27 @@ class WyomingAssistSatellite(WyomingSatelliteEntity, AssistSatelliteEntity): # Satellite requested pipeline run run_pipeline = RunPipeline.from_event(client_event) self._run_pipeline_once(run_pipeline, wake_word_phrase) - elif ( - AudioChunk.is_type(client_event.type) and self._is_pipeline_running + elif AudioChunk.is_type(client_event.type) and ( + self._is_pipeline_running or (wake_word_phrase is not None) ): + if not self._is_pipeline_running: + # Some satellites report a local wake word detection and + # then start streaming audio without sending a + # RunPipeline event. Start a pipeline so the audio isn't + # silently dropped. Begin at ASR since the wake word was + # already detected on the satellite. + _LOGGER.debug( + "Received audio after detection without RunPipeline; " + "starting a pipeline automatically" + ) + self._run_pipeline_once( + RunPipeline( + start_stage=PipelineStage.ASR, + end_stage=PipelineStage.TTS, + ), + wake_word_phrase, + ) + # Microphone audio chunk = AudioChunk.from_event(client_event) chunk = self._chunk_converter.convert(chunk) diff --git a/homeassistant/components/wyoming/data.py b/homeassistant/components/wyoming/data.py index d314e14d0e7a..92d138f39376 100644 --- a/homeassistant/components/wyoming/data.py +++ b/homeassistant/components/wyoming/data.py @@ -9,7 +9,9 @@ from homeassistant.const import Platform from .error import WyomingError -_INFO_TIMEOUT = 1 +# Allow time for satellites that are briefly busy on connect (e.g. playing a +# startup sound) to answer the Describe before we time out and retry. +_INFO_TIMEOUT = 5 _INFO_RETRY_WAIT = 2 _INFO_RETRIES = 3 diff --git a/tests/components/wyoming/test_data.py b/tests/components/wyoming/test_data.py index bec27db057ed..cc120ce51c03 100644 --- a/tests/components/wyoming/test_data.py +++ b/tests/components/wyoming/test_data.py @@ -1,11 +1,18 @@ """Test tts.""" +import asyncio from unittest.mock import patch +import pytest from syrupy.assertion import SnapshotAssertion +from wyoming.event import Event from wyoming.info import Info -from homeassistant.components.wyoming.data import WyomingService, load_wyoming_info +from homeassistant.components.wyoming.data import ( + _INFO_TIMEOUT, + WyomingService, + load_wyoming_info, +) from homeassistant.core import HomeAssistant from . import SATELLITE_INFO, STT_INFO, TTS_INFO, WAKE_WORD_INFO, MockAsyncTcpClient @@ -96,3 +103,47 @@ async def test_satellite_with_wake_word(hass: HomeAssistant) -> None: assert service is not None assert service.get_name() == satellite_info.satellite.name assert not service.platforms + + +class SlowMockAsyncTcpClient(MockAsyncTcpClient): + """Mock client that delays its response to simulate a busy satellite.""" + + def __init__(self, responses: list[Event | None], delay: float) -> None: + """Initialize.""" + super().__init__(responses) + self._delay = delay + + async def read_event(self) -> Event | None: + """Receive after a delay.""" + await asyncio.sleep(self._delay) + return await super().read_event() + + +def test_info_timeout_is_relaxed() -> None: + """Test the default info timeout tolerates satellites busy on connect. + + Some satellites do not answer the initial Describe immediately (e.g. while + playing a startup sound), so the default must be generous enough to avoid + spurious setup failures. + """ + assert _INFO_TIMEOUT >= 5 + + +@pytest.mark.parametrize( + ("timeout", "expect_loaded"), + [(0.05, False), (0.5, True)], + ids=["too_short", "long_enough"], +) +async def test_load_info_timeout_governs_slow_satellite( + hass: HomeAssistant, timeout: float, expect_loaded: bool +) -> None: + """Test a slow satellite loads only when the timeout is generous enough.""" + with patch( + "homeassistant.components.wyoming.data.AsyncTcpClient", + SlowMockAsyncTcpClient([STT_INFO.event()], delay=0.2), + ): + info = await load_wyoming_info( + "localhost", 1234, retries=0, retry_wait=0, timeout=timeout + ) + + assert (info is not None) == expect_loaded diff --git a/tests/components/wyoming/test_satellite.py b/tests/components/wyoming/test_satellite.py index 8e6dc84b2e46..d621471916b9 100644 --- a/tests/components/wyoming/test_satellite.py +++ b/tests/components/wyoming/test_satellite.py @@ -15,7 +15,7 @@ from wyoming.event import Event from wyoming.info import Info from wyoming.ping import Ping, Pong from wyoming.pipeline import PipelineStage, RunPipeline -from wyoming.satellite import RunSatellite +from wyoming.satellite import PauseSatellite, RunSatellite from wyoming.snd import Played from wyoming.timer import TimerCancelled, TimerFinished, TimerStarted, TimerUpdated from wyoming.tts import Synthesize @@ -88,6 +88,10 @@ class SatelliteAsyncTcpClient(MockAsyncTcpClient): self.run_satellite_event = asyncio.Event() self.detect_event = asyncio.Event() + self.pause_satellite_event = asyncio.Event() + self.disconnect_event = asyncio.Event() + self.paused_before_disconnect = False + self.detection_event = asyncio.Event() self.detection: Detection | None = None @@ -141,10 +145,18 @@ class SatelliteAsyncTcpClient(MockAsyncTcpClient): """Connect.""" self.connect_event.set() + async def disconnect(self) -> None: + """Disconnect.""" + self.paused_before_disconnect = self.pause_satellite_event.is_set() + self.disconnect_event.set() + await super().disconnect() + async def write_event(self, event: Event): """Send.""" if RunSatellite.is_type(event.type): self.run_satellite_event.set() + elif PauseSatellite.is_type(event.type): + self.pause_satellite_event.set() elif Detect.is_type(event.type): self.detect_event.set() elif Detection.is_type(event.type): @@ -2043,3 +2055,106 @@ async def test_satellite_tts_streaming(hass: HomeAssistant) -> None: # Stop the satellite await hass.config_entries.async_unload(entry.entry_id) await hass.async_block_till_done() + + +async def test_satellite_pauses_and_disconnects_on_unload( + hass: HomeAssistant, +) -> None: + """Test the satellite is paused and disconnected when the entry unloads. + + Without an explicit disconnect the socket is only released on garbage + collection, which prevents satellites that allow a single connection from + reconnecting when re-enabled. + """ + with ( + patch( + "homeassistant.components.wyoming.data.load_wyoming_info", + return_value=SATELLITE_INFO, + ), + patch( + "homeassistant.components.wyoming.assist_satellite.AsyncTcpClient", + SatelliteAsyncTcpClient([], block_until_inject=True), + ) as mock_client, + ): + entry = await setup_config_entry(hass) + + async with asyncio.timeout(1): + await mock_client.connect_event.wait() + await mock_client.run_satellite_event.wait() + + await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + async with asyncio.timeout(1): + await mock_client.disconnect_event.wait() + + # The satellite is paused, and the pause is sent before the disconnect. + assert mock_client.pause_satellite_event.is_set() + assert mock_client.paused_before_disconnect + assert mock_client.is_connected is False + + +async def test_satellite_audio_without_run_pipeline(hass: HomeAssistant) -> None: + """Test audio after a wake detection without RunPipeline starts a pipeline. + + Some satellites report a local wake word detection and then start streaming + audio without sending a RunPipeline event. The audio must not be silently + dropped. + """ + assert await async_setup_component(hass, assist_pipeline.DOMAIN, {}) + + events = [ + # Satellite info (with local wake word) followed by a wake detection, + # but no RunPipeline event before audio is streamed. + Info(satellite=SATELLITE_INFO.satellite, wake=WAKE_WORD_INFO.wake).event(), + Detection(name="Test Model").event(), + ] + + pipeline_kwargs: dict[str, Any] = {} + run_pipeline_called = asyncio.Event() + audio_chunk_received = asyncio.Event() + + async def async_pipeline_from_audio_stream( + hass: HomeAssistant, + context, + event_callback, + stt_metadata, + stt_stream, + **kwargs, + ) -> None: + nonlocal pipeline_kwargs + pipeline_kwargs = kwargs + run_pipeline_called.set() + async for chunk in stt_stream: + if chunk: + audio_chunk_received.set() + break + + with ( + patch( + "homeassistant.components.wyoming.data.load_wyoming_info", + return_value=SATELLITE_INFO, + ), + patch( + "homeassistant.components.wyoming.assist_satellite.AsyncTcpClient", + SatelliteAsyncTcpClient(events), + ), + patch( + "homeassistant.components.assist_satellite.entity.async_pipeline_from_audio_stream", + async_pipeline_from_audio_stream, + ), + patch("homeassistant.components.wyoming.assist_satellite._PING_SEND_DELAY", 0), + ): + entry = await setup_config_entry(hass) + + async with asyncio.timeout(1): + await run_pipeline_called.wait() + await audio_chunk_received.wait() + + # Pipeline starts at STT (wake word already detected on the satellite) + # and is given the resolved wake word phrase. + assert pipeline_kwargs.get("start_stage") == assist_pipeline.PipelineStage.STT + assert pipeline_kwargs.get("wake_word_phrase") == "Test Phrase" + + await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done()