523 lines
20 KiB
Python
523 lines
20 KiB
Python
"""Voice satellite protocol."""
|
|
|
|
import asyncio
|
|
import functools
|
|
import hashlib
|
|
import logging
|
|
import posixpath
|
|
import shutil
|
|
import time
|
|
from collections.abc import Iterable
|
|
from pathlib import Path
|
|
from typing import Dict, Optional, Set, Union, List
|
|
from urllib.parse import urlparse, urlunparse
|
|
from urllib.request import urlopen
|
|
|
|
from aioesphomeapi.api_pb2 import (
|
|
DeviceInfoRequest,
|
|
DeviceInfoResponse,
|
|
ListEntitiesDoneResponse,
|
|
ListEntitiesRequest,
|
|
MediaPlayerCommandRequest,
|
|
SubscribeHomeAssistantStatesRequest,
|
|
VoiceAssistantAnnounceFinished,
|
|
VoiceAssistantAnnounceRequest,
|
|
VoiceAssistantAudio,
|
|
VoiceAssistantConfigurationRequest,
|
|
VoiceAssistantConfigurationResponse,
|
|
VoiceAssistantEventResponse,
|
|
VoiceAssistantExternalWakeWord,
|
|
VoiceAssistantRequest,
|
|
VoiceAssistantSetConfiguration,
|
|
VoiceAssistantTimerEventResponse,
|
|
VoiceAssistantWakeWord,
|
|
)
|
|
from aioesphomeapi.model import (
|
|
VoiceAssistantEventType,
|
|
VoiceAssistantFeature,
|
|
VoiceAssistantTimerEventType,
|
|
)
|
|
from google.protobuf import message
|
|
from pymicro_wakeword import MicroWakeWord
|
|
from pyopen_wakeword import OpenWakeWord
|
|
|
|
from .api_server import APIServer
|
|
from .entity import MediaPlayerEntity
|
|
from .models import AvailableWakeWord, ServerState, SatelliteState, WakeWordType
|
|
from .util import call_all
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class VoiceSatelliteProtocol(APIServer):
|
|
def __init__(self, state: ServerState) -> None:
|
|
super().__init__(state.name)
|
|
self.state = state
|
|
self.state.satellite = self
|
|
|
|
# --- Ensure exactly one MediaPlayerEntity with a stable key ---
|
|
if self.state.entities:
|
|
existing = self.state.entities[0]
|
|
if not isinstance(existing, MediaPlayerEntity):
|
|
_LOGGER.warning(
|
|
"First ESPHome entity is not MediaPlayerEntity (%r). "
|
|
"Replacing it with a new MediaPlayerEntity.",
|
|
type(existing),
|
|
)
|
|
self.media_player_entity = MediaPlayerEntity(
|
|
server=self,
|
|
state=state,
|
|
key=0,
|
|
name="Media Player",
|
|
object_id="linux_voice_assistant_media_player",
|
|
music_player=state.music_player,
|
|
announce_player=state.tts_player,
|
|
)
|
|
self.state.entities[0] = self.media_player_entity
|
|
else:
|
|
# Reuse existing entity but bind it to the current protocol
|
|
self.media_player_entity = existing
|
|
self.media_player_entity.server = self
|
|
self.media_player_entity.key = 0
|
|
else:
|
|
# First connection in this process: create the media player once
|
|
self.media_player_entity = MediaPlayerEntity(
|
|
server=self,
|
|
state=state,
|
|
key=0,
|
|
name="Media Player",
|
|
object_id="linux_voice_assistant_media_player",
|
|
music_player=state.music_player,
|
|
announce_player=state.tts_player,
|
|
)
|
|
self.state.entities.append(self.media_player_entity)
|
|
|
|
# If more entities somehow accumulated, prune them to avoid confusing HA
|
|
if len(self.state.entities) > 1:
|
|
_LOGGER.warning(
|
|
"Pruning %d extra ESPHome entities; keeping only the first.",
|
|
len(self.state.entities) - 1,
|
|
)
|
|
del self.state.entities[1:]
|
|
|
|
# State machine
|
|
self._state: SatelliteState = SatelliteState.STARTING
|
|
self._is_streaming_audio: bool = False
|
|
|
|
self._tts_url: Optional[str] = None
|
|
self._continue_conversation: bool = False
|
|
self._timer_finished: bool = False
|
|
|
|
self._run_end_received: bool = False
|
|
self._tts_end_received: bool = False
|
|
# Track if current audio is an announcement
|
|
self._is_announcement: bool = False
|
|
|
|
self._external_wake_words: Dict[str, VoiceAssistantExternalWakeWord] = {}
|
|
|
|
def _set_state(self, new_state: SatelliteState):
|
|
if self._state == new_state:
|
|
return
|
|
|
|
_LOGGER.debug("State transition: %s -> %s", self._state, new_state)
|
|
self._state = new_state
|
|
|
|
if new_state == SatelliteState.IDLE:
|
|
self.unduck()
|
|
self.state.active_wake_words.discard(self.state.stop_word.id)
|
|
self.state.event_bus.publish("voice_idle")
|
|
elif new_state == SatelliteState.LISTENING:
|
|
self.duck()
|
|
self.state.event_bus.publish("voice_listen")
|
|
elif new_state == SatelliteState.THINKING:
|
|
self.state.event_bus.publish("voice_thinking")
|
|
elif new_state == SatelliteState.RESPONDING:
|
|
self.state.active_wake_words.add(self.state.stop_word.id)
|
|
self.state.event_bus.publish("voice_responding")
|
|
elif new_state == SatelliteState.ERROR:
|
|
self.state.event_bus.publish("voice_error")
|
|
|
|
def handle_voice_event(
|
|
self, event_type: VoiceAssistantEventType, data: Dict[str, str]
|
|
) -> None:
|
|
_LOGGER.debug("Voice event: type=%s, data=%s", event_type.name, data)
|
|
|
|
if event_type == VoiceAssistantEventType.VOICE_ASSISTANT_RUN_START:
|
|
self._run_end_received = False
|
|
self._tts_end_received = False
|
|
self._is_announcement = False
|
|
self._tts_url = data.get("url")
|
|
self._continue_conversation = False
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_STT_START:
|
|
self._set_state(SatelliteState.LISTENING)
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_STT_VAD_START:
|
|
self.state.event_bus.publish("voice_vad_start", data)
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_STT_END:
|
|
self._is_streaming_audio = False
|
|
self._set_state(SatelliteState.THINKING)
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_STT_VAD_END:
|
|
# No-op for now; could be used for LED cues
|
|
pass
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_END:
|
|
if data.get("continue_conversation") == "1":
|
|
self._continue_conversation = True
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_TTS_START:
|
|
self._set_state(SatelliteState.RESPONDING)
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_TTS_END:
|
|
self._tts_url = data.get("url")
|
|
self._tts_end_received = True
|
|
self.play_tts()
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_RUN_END:
|
|
self._run_end_received = True
|
|
if self._state != SatelliteState.RESPONDING:
|
|
self._determine_final_state()
|
|
|
|
elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_ERROR:
|
|
self._set_state(SatelliteState.ERROR)
|
|
self.state.loop.call_later(5.0, self._set_state, SatelliteState.IDLE)
|
|
|
|
def handle_timer_event(
|
|
self,
|
|
event_type: VoiceAssistantTimerEventType,
|
|
msg: VoiceAssistantTimerEventResponse,
|
|
) -> None:
|
|
_LOGGER.debug("Timer event: type=%s", event_type.name)
|
|
if event_type == VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_FINISHED:
|
|
if not self._timer_finished:
|
|
self.state.active_wake_words.add(self.state.stop_word.id)
|
|
self._timer_finished = True
|
|
self.duck()
|
|
self._play_timer_finished()
|
|
|
|
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):
|
|
data: Dict[str, str] = {}
|
|
for arg in msg.data:
|
|
data[arg.name] = arg.value
|
|
self.handle_voice_event(VoiceAssistantEventType(msg.event_type), data)
|
|
|
|
elif isinstance(msg, VoiceAssistantAnnounceRequest):
|
|
_LOGGER.debug("Announcing: %s", msg.text)
|
|
urls: List[str] = []
|
|
if msg.preannounce_media_id:
|
|
urls.append(msg.preannounce_media_id)
|
|
urls.append(msg.media_id)
|
|
|
|
self._is_announcement = True
|
|
self.state.active_wake_words.add(self.state.stop_word.id)
|
|
self._continue_conversation = msg.start_conversation
|
|
self.duck()
|
|
self._set_state(SatelliteState.RESPONDING)
|
|
yield from self.media_player_entity.play(
|
|
urls, announcement=True, done_callback=self._tts_finished
|
|
)
|
|
|
|
elif isinstance(msg, VoiceAssistantTimerEventResponse):
|
|
self.handle_timer_event(VoiceAssistantTimerEventType(msg.event_type), msg)
|
|
|
|
elif isinstance(msg, DeviceInfoRequest):
|
|
yield DeviceInfoResponse(
|
|
uses_password=False,
|
|
name=self.state.name,
|
|
mac_address=self.state.mac_address,
|
|
voice_assistant_feature_flags=(
|
|
VoiceAssistantFeature.VOICE_ASSISTANT
|
|
| VoiceAssistantFeature.API_AUDIO
|
|
| VoiceAssistantFeature.ANNOUNCE
|
|
| VoiceAssistantFeature.START_CONVERSATION
|
|
| VoiceAssistantFeature.TIMERS
|
|
),
|
|
)
|
|
|
|
elif isinstance(
|
|
msg,
|
|
(
|
|
ListEntitiesRequest,
|
|
SubscribeHomeAssistantStatesRequest,
|
|
MediaPlayerCommandRequest,
|
|
),
|
|
):
|
|
for entity in self.state.entities:
|
|
yield from entity.handle_message(msg)
|
|
|
|
if isinstance(msg, ListEntitiesRequest):
|
|
yield ListEntitiesDoneResponse()
|
|
|
|
elif isinstance(msg, VoiceAssistantConfigurationRequest):
|
|
available_wake_words: List[VoiceAssistantWakeWord] = [
|
|
VoiceAssistantWakeWord(
|
|
id=ww.id,
|
|
wake_word=ww.wake_word,
|
|
trained_languages=ww.trained_languages,
|
|
)
|
|
for ww in self.state.available_wake_words.values()
|
|
]
|
|
|
|
# Reset external wake words cache and add new ones
|
|
self._external_wake_words.clear()
|
|
for eww in msg.external_wake_words:
|
|
if eww.model_type != "micro":
|
|
_LOGGER.warning(
|
|
"Skipping external wake word %s (type=%s)",
|
|
eww.id,
|
|
eww.model_type,
|
|
)
|
|
continue
|
|
|
|
available_wake_words.append(
|
|
VoiceAssistantWakeWord(
|
|
id=eww.id,
|
|
wake_word=eww.wake_word,
|
|
trained_languages=eww.trained_languages,
|
|
)
|
|
)
|
|
self._external_wake_words[eww.id] = eww
|
|
|
|
yield VoiceAssistantConfigurationResponse(
|
|
available_wake_words=available_wake_words,
|
|
active_wake_words=[
|
|
ww.id
|
|
for ww in self.state.wake_words.values()
|
|
if ww.id in self.state.active_wake_words
|
|
],
|
|
max_active_wake_words=2,
|
|
)
|
|
|
|
_LOGGER.info("Connected to Home Assistant")
|
|
self._set_state(SatelliteState.IDLE)
|
|
self.state.event_bus.publish("ha_connected")
|
|
|
|
elif isinstance(msg, VoiceAssistantSetConfiguration):
|
|
# 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()
|
|
|
|
for wake_word_id in msg.active_wake_words:
|
|
if wake_word_id in self.state.wake_words:
|
|
active_wake_words.add(wake_word_id)
|
|
continue
|
|
|
|
model_info = self.state.available_wake_words.get(wake_word_id)
|
|
|
|
if not model_info:
|
|
external_wake_word = self._external_wake_words.get(wake_word_id)
|
|
if not external_wake_word:
|
|
_LOGGER.warning("Unknown wake word: %s", wake_word_id)
|
|
continue
|
|
|
|
# Await the non-blocking download
|
|
model_info = await self._download_external_wake_word(external_wake_word)
|
|
if not model_info:
|
|
continue
|
|
|
|
self.state.available_wake_words[wake_word_id] = model_info
|
|
|
|
_LOGGER.debug("Loading wake word: %s", model_info.wake_word_path)
|
|
self.state.wake_words[wake_word_id] = model_info.load()
|
|
|
|
_LOGGER.info("Wake word set: %s", wake_word_id)
|
|
active_wake_words.add(wake_word_id)
|
|
break
|
|
|
|
self.state.active_wake_words = active_wake_words
|
|
_LOGGER.debug("Active wake words: %s", active_wake_words)
|
|
|
|
self.state.preferences.active_wake_words = list(active_wake_words)
|
|
self.state.save_preferences()
|
|
self.state.wake_words_changed = True
|
|
|
|
def handle_audio(self, audio_chunk: bytes) -> None:
|
|
if self._is_streaming_audio:
|
|
self.send_messages([VoiceAssistantAudio(data=audio_chunk)])
|
|
|
|
def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None:
|
|
if self._state not in (SatelliteState.IDLE, SatelliteState.STARTING):
|
|
return
|
|
|
|
if self._timer_finished:
|
|
self._timer_finished = False
|
|
self.state.tts_player.stop()
|
|
_LOGGER.debug("Stopping timer finished sound")
|
|
return
|
|
|
|
wake_word_phrase = getattr(wake_word, "wake_word", "wake word")
|
|
_LOGGER.debug("Detected wake word: %s", wake_word_phrase)
|
|
self.send_messages(
|
|
[VoiceAssistantRequest(start=True, wake_word_phrase=wake_word_phrase)]
|
|
)
|
|
self._set_state(SatelliteState.LISTENING)
|
|
self._is_streaming_audio = True
|
|
self.state.tts_player.play(self.state.wakeup_sound)
|
|
|
|
def stop(self) -> None:
|
|
self.state.active_wake_words.discard(self.state.stop_word.id)
|
|
self.state.tts_player.stop()
|
|
if self._timer_finished:
|
|
self._timer_finished = False
|
|
_LOGGER.debug("Stopping timer finished sound")
|
|
else:
|
|
_LOGGER.debug("TTS response stopped manually")
|
|
self._tts_finished()
|
|
|
|
def play_tts(self) -> None:
|
|
if not self._tts_url:
|
|
return
|
|
|
|
_LOGGER.debug("Playing TTS response: %s", self._tts_url)
|
|
self.state.active_wake_words.add(self.state.stop_word.id)
|
|
self.state.tts_player.play(self._tts_url, done_callback=self._tts_finished)
|
|
self._tts_url = None
|
|
|
|
def duck(self) -> None:
|
|
_LOGGER.debug("Ducking music")
|
|
self.state.music_player.duck()
|
|
|
|
def unduck(self) -> None:
|
|
_LOGGER.debug("Unducking music")
|
|
self.state.music_player.unduck()
|
|
|
|
def _determine_final_state(self) -> None:
|
|
if self._continue_conversation:
|
|
self.send_messages([VoiceAssistantRequest(start=True)])
|
|
self._is_streaming_audio = True
|
|
_LOGGER.debug("Continuing conversation")
|
|
self._set_state(SatelliteState.LISTENING)
|
|
else:
|
|
self._set_state(SatelliteState.IDLE)
|
|
|
|
_LOGGER.debug("Final state determined")
|
|
|
|
def _tts_finished(self) -> None:
|
|
self.state.active_wake_words.discard(self.state.stop_word.id)
|
|
self.send_messages([VoiceAssistantAnnounceFinished()])
|
|
_LOGGER.debug("TTS audio playback finished")
|
|
|
|
# If this was just an announcement, we are done,
|
|
# or if we received the official Run End.
|
|
if self._is_announcement or self._run_end_received:
|
|
self._determine_final_state()
|
|
self._is_announcement = False
|
|
else:
|
|
if self._state == SatelliteState.RESPONDING:
|
|
self._set_state(SatelliteState.THINKING)
|
|
|
|
def _play_timer_finished(self) -> None:
|
|
if not self._timer_finished:
|
|
self.unduck()
|
|
return
|
|
self.state.tts_player.play(
|
|
self.state.timer_finished_sound,
|
|
done_callback=lambda: call_all(
|
|
lambda: time.sleep(1.0), self._play_timer_finished
|
|
),
|
|
)
|
|
|
|
async def _download_external_wake_word(
|
|
self, external_wake_word: VoiceAssistantExternalWakeWord
|
|
) -> 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.mkdir(parents=True, exist_ok=True)
|
|
|
|
config_path = eww_dir / f"{external_wake_word.id}.json"
|
|
should_download_config = not config_path.exists()
|
|
|
|
model_path = eww_dir / f"{external_wake_word.id}.tflite"
|
|
should_download_model = True
|
|
if model_path.exists():
|
|
model_size = model_path.stat().st_size
|
|
if model_size == external_wake_word.model_size:
|
|
with open(model_path, "rb") as model_file:
|
|
model_hash = hashlib.sha256(model_file.read()).hexdigest()
|
|
|
|
if model_hash == external_wake_word.model_hash:
|
|
should_download_model = False
|
|
_LOGGER.debug(
|
|
"Model size and hash match for %s. Skipping download.",
|
|
external_wake_word.id,
|
|
)
|
|
|
|
if should_download_config or should_download_model:
|
|
_LOGGER.debug(
|
|
"Downloading %s to %s", external_wake_word.url, config_path
|
|
)
|
|
try:
|
|
with urlopen(external_wake_word.url, timeout=10) as request:
|
|
if request.status != 200:
|
|
_LOGGER.warning(
|
|
"Failed to download: %s, status=%s",
|
|
external_wake_word.url,
|
|
request.status,
|
|
)
|
|
return None
|
|
|
|
with open(config_path, "wb") as model_file:
|
|
shutil.copyfileobj(request, model_file)
|
|
except Exception as exc: # noqa: BLE001
|
|
_LOGGER.error("Exception downloading config: %s", exc)
|
|
return None
|
|
|
|
if should_download_model:
|
|
parsed_url = urlparse(external_wake_word.url)
|
|
parsed_url = parsed_url._replace(
|
|
path=posixpath.join(posixpath.dirname(parsed_url.path), model_path.name)
|
|
)
|
|
model_url = urlunparse(parsed_url)
|
|
|
|
_LOGGER.debug("Downloading %s to %s", model_url, model_path)
|
|
try:
|
|
with urlopen(model_url, timeout=10) as request:
|
|
if request.status != 200:
|
|
_LOGGER.warning(
|
|
"Failed to download: %s, status=%s",
|
|
model_url,
|
|
request.status,
|
|
)
|
|
return None
|
|
|
|
with open(model_path, "wb") as model_file:
|
|
shutil.copyfileobj(request, model_file)
|
|
except Exception as exc: # noqa: BLE001
|
|
_LOGGER.error("Exception downloading model: %s", exc)
|
|
return None
|
|
|
|
return AvailableWakeWord(
|
|
id=external_wake_word.id,
|
|
type=WakeWordType.MICRO_WAKE_WORD,
|
|
wake_word=external_wake_word.wake_word,
|
|
trained_languages=external_wake_word.trained_languages,
|
|
wake_word_path=config_path,
|
|
)
|
|
|
|
def connection_lost(self, exc):
|
|
super().connection_lost(exc)
|
|
self._set_state(SatelliteState.ERROR)
|
|
_LOGGER.info("Disconnected from Home Assistant")
|