Update satellite.py

Implements non-blocking downloads.
This commit is contained in:
imonlinux
2025-11-19 18:47:37 -08:00
committed by GitHub
parent cf82a3158d
commit 6f5444237c
+44 -6
View File
@@ -1,5 +1,7 @@
"""Voice satellite protocol.""" """Voice satellite protocol."""
import asyncio
import functools
import hashlib import hashlib
import logging import logging
import posixpath import posixpath
@@ -7,7 +9,7 @@ import shutil
import time import time
from collections.abc import Iterable from collections.abc import Iterable
from pathlib import Path from pathlib import Path
from typing import Dict, Optional, Set, Union from typing import Dict, Optional, Set, Union, List
from urllib.parse import urlparse, urlunparse from urllib.parse import urlparse, urlunparse
from urllib.request import urlopen from urllib.request import urlopen
@@ -51,7 +53,7 @@ class VoiceSatelliteProtocol(APIServer):
def __init__(self, state: ServerState) -> None: def __init__(self, state: ServerState) -> None:
super().__init__(state.name) super().__init__(state.name)
self.state = state self.state = state
self.state.satellite = self # <-- ADDED BACK self.state.satellite = self
self.media_player_entity = MediaPlayerEntity( self.media_player_entity = MediaPlayerEntity(
server=self, server=self,
@@ -159,11 +161,16 @@ class VoiceSatelliteProtocol(APIServer):
self._play_timer_finished() self._play_timer_finished()
def handle_message(self, msg: message.Message) -> Iterable[message.Message]: def handle_message(self, msg: message.Message) -> Iterable[message.Message]:
"""
Handles incoming messages from Home Assistant.
Note: This method is synchronous. Long-running operations must be offloaded to Tasks.
"""
if isinstance(msg, VoiceAssistantEventResponse): if isinstance(msg, VoiceAssistantEventResponse):
data: Dict[str, str] = {} data: Dict[str, str] = {}
for arg in msg.data: for arg in msg.data:
data[arg.name] = arg.value data[arg.name] = arg.value
self.handle_voice_event(VoiceAssistantEventType(msg.event_type), data) self.handle_voice_event(VoiceAssistantEventType(msg.event_type), data)
elif isinstance(msg, VoiceAssistantAnnounceRequest): elif isinstance(msg, VoiceAssistantAnnounceRequest):
_LOGGER.debug("Announcing: %s", msg.text) _LOGGER.debug("Announcing: %s", msg.text)
urls = [] urls = []
@@ -177,8 +184,10 @@ class VoiceSatelliteProtocol(APIServer):
yield from self.media_player_entity.play( yield from self.media_player_entity.play(
urls, announcement=True, done_callback=self._tts_finished urls, announcement=True, done_callback=self._tts_finished
) )
elif isinstance(msg, VoiceAssistantTimerEventResponse): elif isinstance(msg, VoiceAssistantTimerEventResponse):
self.handle_timer_event(VoiceAssistantTimerEventType(msg.event_type), msg) self.handle_timer_event(VoiceAssistantTimerEventType(msg.event_type), msg)
elif isinstance(msg, DeviceInfoRequest): elif isinstance(msg, DeviceInfoRequest):
yield DeviceInfoResponse( yield DeviceInfoResponse(
uses_password=False, uses_password=False,
@@ -192,6 +201,7 @@ class VoiceSatelliteProtocol(APIServer):
| VoiceAssistantFeature.TIMERS | VoiceAssistantFeature.TIMERS
), ),
) )
elif isinstance( elif isinstance(
msg, msg,
( (
@@ -205,6 +215,7 @@ class VoiceSatelliteProtocol(APIServer):
if isinstance(msg, ListEntitiesRequest): if isinstance(msg, ListEntitiesRequest):
yield ListEntitiesDoneResponse() yield ListEntitiesDoneResponse()
elif isinstance(msg, VoiceAssistantConfigurationRequest): elif isinstance(msg, VoiceAssistantConfigurationRequest):
available_wake_words = [ available_wake_words = [
VoiceAssistantWakeWord( VoiceAssistantWakeWord(
@@ -242,8 +253,16 @@ class VoiceSatelliteProtocol(APIServer):
_LOGGER.info("Connected to Home Assistant") _LOGGER.info("Connected to Home Assistant")
self._set_state(SatelliteState.IDLE) self._set_state(SatelliteState.IDLE)
self.state.event_bus.publish("ha_connected") self.state.event_bus.publish("ha_connected")
elif isinstance(msg, VoiceAssistantSetConfiguration): elif isinstance(msg, VoiceAssistantSetConfiguration):
# FIX: Offload to background task to prevent blocking the loop during download
self.state.loop.create_task(self._handle_set_configuration_task(msg))
# Yield nothing immediately; response will be sent asynchronously
async def _handle_set_configuration_task(self, msg: VoiceAssistantSetConfiguration) -> None:
"""Asynchronous handler for SetConfiguration to avoid blocking I/O."""
active_wake_words: Set[str] = set() active_wake_words: Set[str] = set()
for wake_word_id in msg.active_wake_words: for wake_word_id in msg.active_wake_words:
if wake_word_id in self.state.wake_words: if wake_word_id in self.state.wake_words:
active_wake_words.add(wake_word_id) active_wake_words.add(wake_word_id)
@@ -257,7 +276,8 @@ class VoiceSatelliteProtocol(APIServer):
_LOGGER.warning("Unknown wake word: %s", wake_word_id) _LOGGER.warning("Unknown wake word: %s", wake_word_id)
continue continue
model_info = self._download_external_wake_word(external_wake_word) # Await the non-blocking download
model_info = await self._download_external_wake_word(external_wake_word)
if not model_info: if not model_info:
continue continue
@@ -361,9 +381,19 @@ class VoiceSatelliteProtocol(APIServer):
), ),
) )
def _download_external_wake_word( async def _download_external_wake_word(
self, external_wake_word: VoiceAssistantExternalWakeWord self, external_wake_word: VoiceAssistantExternalWakeWord
) -> Optional[AvailableWakeWord]: ) -> Optional[AvailableWakeWord]:
"""Wrapper to run the blocking download in a thread executor."""
return await self.state.loop.run_in_executor(
None,
functools.partial(self._download_external_wake_word_sync, external_wake_word)
)
def _download_external_wake_word_sync(
self, external_wake_word: VoiceAssistantExternalWakeWord
) -> Optional[AvailableWakeWord]:
"""Blocking download logic, intended to run in an executor."""
eww_dir = self.state.download_dir / "external_wake_words" eww_dir = self.state.download_dir / "external_wake_words"
eww_dir.mkdir(parents=True, exist_ok=True) eww_dir.mkdir(parents=True, exist_ok=True)
@@ -387,7 +417,8 @@ class VoiceSatelliteProtocol(APIServer):
if should_download_config or should_download_model: if should_download_config or should_download_model:
_LOGGER.debug("Downloading %s to %s", external_wake_word.url, config_path) _LOGGER.debug("Downloading %s to %s", external_wake_word.url, config_path)
with urlopen(external_wake_word.url) as request: try:
with urlopen(external_wake_word.url, timeout=10) as request:
if request.status != 200: if request.status != 200:
_LOGGER.warning( _LOGGER.warning(
"Failed to download: %s, status=%s", "Failed to download: %s, status=%s",
@@ -398,6 +429,9 @@ class VoiceSatelliteProtocol(APIServer):
with open(config_path, "wb") as model_file: with open(config_path, "wb") as model_file:
shutil.copyfileobj(request, model_file) shutil.copyfileobj(request, model_file)
except Exception as e:
_LOGGER.error("Exception downloading config: %s", e)
return None
if should_download_model: if should_download_model:
parsed_url = urlparse(external_wake_word.url) parsed_url = urlparse(external_wake_word.url)
@@ -407,7 +441,8 @@ class VoiceSatelliteProtocol(APIServer):
model_url = urlunparse(parsed_url) model_url = urlunparse(parsed_url)
_LOGGER.debug("Downloading %s to %s", model_url, model_path) _LOGGER.debug("Downloading %s to %s", model_url, model_path)
with urlopen(model_url) as request: try:
with urlopen(model_url, timeout=10) as request:
if request.status != 200: if request.status != 200:
_LOGGER.warning( _LOGGER.warning(
"Failed to download: %s, status=%s", model_url, request.status "Failed to download: %s, status=%s", model_url, request.status
@@ -416,6 +451,9 @@ class VoiceSatelliteProtocol(APIServer):
with open(model_path, "wb") as model_file: with open(model_path, "wb") as model_file:
shutil.copyfileobj(request, model_file) shutil.copyfileobj(request, model_file)
except Exception as e:
_LOGGER.error("Exception downloading model: %s", e)
return None
return AvailableWakeWord( return AvailableWakeWord(
id=external_wake_word.id, id=external_wake_word.id,