Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-07-07 06:46:17 +01:00
Code Activity

Improve Wyoming satellite reconnect and tolerance of other satellites (#174460)

Browse Source 
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Michael Hansen
2026-06-24 02:22:03 -05:00
committed by GitHub
parent 687064d5cc
commit c2e780dfd2
4 changed files with 205 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/wyoming/assist_satellite.py
+34 -6
View File
@@ -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)
homeassistant/components/wyoming/data.py
+3 -1
View File
@@ -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
tests/components/wyoming/test_data.py
+52 -1
View File
@@ -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
tests/components/wyoming/test_satellite.py
+116 -1
View File
@@ -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()