Update ESPHome protocol implementation to match official linux-voice-assistant

- Replace api_server.py with official ESPHome protocol implementation
- Replace satellite.py with official voice assistant protocol
- Add MediaPlayerEntity for media player support
- Add save_preferences method to ServerState
- Update AudioPlayer to support URLs and file paths
- Remove zeroconf (Reachy Mini handles mDNS discovery)
- Add media_player_entity to ServerState

reachy_mini_ha_voice/api_server.py

@@ -1,74 +1,180 @@
-"""API server for Home Assistant integration."""
+"""Partial ESPHome server implementation."""
 
 import asyncio
-import json
 import logging
-from typing import Dict, List, Optional
+from abc import abstractmethod
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, List, Optional
 
-from .models import ServerState
+# pylint: disable=no-name-in-module
+from aioesphomeapi._frame_helper.packets import make_plain_text_packets
+from aioesphomeapi.api_pb2 import (  # type: ignore[attr-defined]
+    AuthenticationRequest,
+    AuthenticationResponse,
+    DisconnectRequest,
+    DisconnectResponse,
+    HelloRequest,
+    HelloResponse,
+    PingRequest,
+    PingResponse,
+)
+from aioesphomeapi.core import MESSAGE_TYPE_TO_PROTO
+from google.protobuf import message
+
+PROTO_TO_MESSAGE_TYPE = {v: k for k, v in MESSAGE_TYPE_TO_PROTO.items()}
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class APIServer:
-    """API server for Home Assistant."""
-
-    def __init__(self, state: ServerState):
-        """Initialize API server."""
-        self.state = state
-        self._handlers: Dict[str, callable] = {
-            "hello": self._handle_hello,
-            "list_entities": self._handle_list_entities,
-            "get_state": self._handle_get_state,
-            "subscribe_states": self._handle_subscribe_states,
-        }
-
-    async def handle_request(self, command: str, payload: dict) -> dict:
-        """Handle an API request."""
-        handler = self._handlers.get(command)
-        if handler:
-            try:
-                return await handler(payload)
-            except Exception as e:
-                _LOGGER.error("Error handling request %s: %s", command, e)
-                return {"error": str(e)}
-        else:
-            return {"error": f"Unknown command: {command}"}
-
-    async def _handle_hello(self, payload: dict) -> dict:
-        """Handle hello request."""
-        return {
-            "name": self.state.name,
-            "mac_address": self.state.mac_address,
-            "version": "1.0.0",
-        }
-
-    async def _handle_list_entities(self, payload: dict) -> dict:
-        """Handle list_entities request."""
-        entities = []
-        for entity in self.state.entities:
-            entities.append(
-                {
-                    "key": entity.key,
-                    "name": entity.name,
-                    "state": entity.state,
-                    "attributes": entity.attributes,
-                }
+class APIServer(asyncio.Protocol):
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+        self._buffer: Optional[bytes] = None
+        self._buffer_len: int = 0
+        self._pos: int = 0
+        self._transport = None
+        self._writelines = None
+
+    @abstractmethod
+    def handle_message(self, msg: message.Message) -> Iterable[message.Message]:
+        pass
+
+    def process_packet(self, msg_type: int, packet_data: bytes) -> None:
+        msg_class = MESSAGE_TYPE_TO_PROTO[msg_type]
+        msg_inst = msg_class.FromString(packet_data)
+
+        if isinstance(msg_inst, HelloRequest):
+            self.send_messages(
+                [
+                    HelloResponse(
+                        api_version_major=1,
+                        api_version_minor=10,
+                        name=self.name,
+                    )
+                ]
             )
-        return {"entities": entities}
-
-    async def _handle_get_state(self, payload: dict) -> dict:
-        """Handle get_state request."""
-        key = payload.get("key")
-        entity = next((e for e in self.state.entities if e.key == key), None)
-        if entity:
-            return {
-                "key": entity.key,
-                "state": entity.state,
-                "attributes": entity.attributes,
-            }
-        return {"error": "Entity not found"}
-
-    async def _handle_subscribe_states(self, payload: dict) -> dict:
-        """Handle subscribe_states request."""
-        return {"result": "ok"}
+            return
+
+        if isinstance(msg_inst, AuthenticationRequest):
+            self.send_messages([AuthenticationResponse()])
+        elif isinstance(msg_inst, DisconnectRequest):
+            self.send_messages([DisconnectResponse()])
+            _LOGGER.debug("Disconnect requested")
+            if self._transport:
+                self._transport.close()
+                self._transport = None
+                self._writelines = None
+        elif isinstance(msg_inst, PingRequest):
+            self.send_messages([PingResponse()])
+        elif msgs := self.handle_message(msg_inst):
+            if isinstance(msgs, message.Message):
+                msgs = [msgs]
+
+            self.send_messages(msgs)
+
+    def send_messages(self, msgs: List[message.Message]):
+        if self._writelines is None:
+            return
+
+        packets = [
+            (PROTO_TO_MESSAGE_TYPE[msg.__class__], msg.SerializeToString())
+            for msg in msgs
+        ]
+        packet_bytes = make_plain_text_packets(packets)
+        self._writelines(packet_bytes)
+
+    def connection_made(self, transport) -> None:
+        self._transport = transport
+        self._writelines = transport.writelines
+
+    def data_received(self, data: bytes):
+        if self._buffer is None:
+            self._buffer = data
+            self._buffer_len = len(data)
+        else:
+            self._buffer += data
+            self._buffer_len += len(data)
+
+        while self._buffer_len >= 3:
+            self._pos = 0
+            # Read preamble, which should always 0x00
+            if (preamble := self._read_varuint()) != 0x00:
+                _LOGGER.error("Incorrect preamble: %s", preamble)
+                return
+
+            if (length := self._read_varuint()) == -1:
+                _LOGGER.error("Incorrect length: %s", length)
+                return
+
+            if (msg_type := self._read_varuint()) == -1:
+                _LOGGER.error("Incorrect message type: %s", msg_type)
+                return
+
+            if length == 0:
+                # Empty message (allowed)
+                self._remove_from_buffer()
+                self.process_packet(msg_type, b"")
+                continue
+
+            if (packet_data := self._read(length)) is None:
+                return
+
+            self._remove_from_buffer()
+            self.process_packet(msg_type, packet_data)
+
+    def _read(self, length: int) -> bytes | None:
+        """Read exactly length bytes from the buffer or None if all the bytes are not yet available."""
+        new_pos = self._pos + length
+        if self._buffer_len < new_pos:
+            return None
+        original_pos = self._pos
+        self._pos = new_pos
+        if TYPE_CHECKING:
+            assert self._buffer is not None, "Buffer should be set"
+        cstr = self._buffer
+        # Important: we must keep the bounds check (self._buffer_len < new_pos)
+        # above to verify we never try to read past the end of the buffer
+        return cstr[original_pos:new_pos]
+
+    def connection_lost(self, exc):
+        self._transport = None
+        self._writelines = None
+
+    def _read_varuint(self) -> int:
+        """Read a varuint from the buffer or -1 if the buffer runs out of bytes."""
+        if not self._buffer:
+            return -1
+
+        result = 0
+        bitpos = 0
+        cstr = self._buffer
+        while self._buffer_len > self._pos:
+            val = cstr[self._pos]
+            self._pos += 1
+            result |= (val & 0x7F) << bitpos
+            if (val & 0x80) == 0:
+                return result
+            bitpos += 7
+        return -1
+
+    def _remove_from_buffer(self) -> None:
+        """Remove data from the buffer."""
+        end_of_frame_pos = self._pos
+        self._buffer_len -= end_of_frame_pos
+        if self._buffer_len == 0:
+            # This is the best case scenario, we can just set the buffer to None
+            # and don't have to copy the data. This is the most common case as well.
+            self._buffer = None
+            return
+        if TYPE_CHECKING:
+            assert self._buffer is not None, "Buffer should be set"
+        # This is the worst case scenario, we have to copy the data
+        # and can't just use the buffer directly. This should only happen
+        # when we read multiple frames at once because the event loop
+        # is blocked and we cannot pull the data out of the buffer fast enough.
+        cstr = self._buffer
+        # Important: we must use the explicit length for the slice
+        # since Cython will stop at any '\0' character if we don't
+        self._buffer = cstr[end_of_frame_pos : self._buffer_len + end_of_frame_pos]

reachy_mini_ha_voice/app.py

@@ -24,7 +24,6 @@ from .models import (
 )
 from .satellite import VoiceSatelliteProtocol
 from .util import get_mac
-from .zeroconf import HomeAssistantZeroconf
 
 _LOGGER = logging.getLogger(__name__)
 _MODULE_DIR = Path(__file__).parent
@@ -124,6 +123,7 @@ class ReachyMiniHAVoiceApp(ReachyMiniApp):
             refractory_seconds=2.0,
             download_dir=_REPO_DIR / "local",
             reachy_integration=None,  # Not using Reachy integration for now
+            media_player_entity=None,
         )
 
     def _load_wake_words(self) -> Dict[str, AvailableWakeWord]:
@@ -186,16 +186,12 @@ class ReachyMiniHAVoiceApp(ReachyMiniApp):
             lambda: VoiceSatelliteProtocol(state), host="0.0.0.0", port=6053
         )
 
-        # Auto discovery (zeroconf, mDNS)
-        discovery = HomeAssistantZeroconf(port=6053, name="ReachyMini")
-        await discovery.register_server()
-
         try:
             async with server:
                 _LOGGER.info("ESPHome server started on port 6053")
                 await server.serve_forever()
         finally:
-            await discovery.unregister_server()
+            _LOGGER.info("ESPHome server stopped")
 
     def _process_audio(self, state: ServerState) -> None:
         """Process audio from microphone."""

reachy_mini_ha_voice/entity.py

@@ -1,36 +1,94 @@
 """Entity management for Home Assistant."""
 
 import logging
-from typing import Dict, List
+from typing import Dict, List, Optional
+
+# pylint: disable=no-name-in-module
+from aioesphomeapi.api_pb2 import (  # type: ignore[attr-defined]
+    ListEntitiesMediaPlayersResponse,
+    MediaPlayerCommandRequest,
+    TextSensorStateResponse,
+)
+from aioesphomeapi.model import MediaPlayerState
 
 from .models import Entity
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class EntityManager:
-    """Manage Home Assistant entities."""
+class MediaPlayerEntity(Entity):
+    """Media player entity for voice assistant."""
+
+    def __init__(
+        self, server, key: int, name: str, object_id: str, music_player, announce_player
+    ):
+        """Initialize media player entity."""
+        super().__init__(key=key, name=name, state="idle", attributes={})
+        self.server = server
+        self.object_id = object_id
+        self.music_player = music_player
+        self.announce_player = announce_player
+        self._volume = 1.0
+        self._position = 0
+        self._duration = 0
+
+    def handle_message(self, msg):
+        """Handle a message."""
+        if isinstance(msg, ListEntitiesMediaPlayersResponse):
+            yield self.get_list_entities_response()
+        elif isinstance(msg, MediaPlayerCommandRequest):
+            self.handle_command(msg)
+
+    def get_list_entities_response(self):
+        """Get list entities response."""
+        from aioesphomeapi.api_pb2 import ListEntitiesMediaPlayersResponse
+
+        return ListEntitiesMediaPlayersResponse(
+            object_id=self.object_id,
+            key=self.key,
+            name=self.name,
+        )
+
+    def handle_command(self, msg):
+        """Handle a media player command."""
+        if msg.command == MediaPlayerCommandRequest.PLAY:
+            if msg.url:
+                self.play([msg.url])
+        elif msg.command == MediaPlayerCommandRequest.PAUSE:
+            self.music_player.stop()
+        elif msg.command == MediaPlayerCommandRequest.STOP:
+            self.music_player.stop()
+        elif msg.command == MediaPlayerCommandRequest.VOLUME_SET:
+            self._volume = msg.volume / 255.0
+        elif msg.command == MediaPlayerCommandRequest.MUTE:
+            self._volume = 0.0 if msg.mute else 1.0
+
+    def play(self, urls, announcement=False, done_callback=None):
+        """Play media."""
+        _LOGGER.debug("Playing: %s", urls)
+        player = self.announce_player if announcement else self.music_player
 
-    def __init__(self):
-        """Initialize entity manager."""
-        self._entities: Dict[str, Entity] = {}
+        for url in urls:
+            try:
+                from urllib.request import urlopen
 
-    def add_entity(self, entity: Entity) -> None:
-        """Add an entity."""
-        self._entities[entity.key] = entity
-        _LOGGER.debug("Added entity: %s", entity.key)
+                with urlopen(url) as response:
+                    audio_data = response.read()
+                player.play(audio_data)
+            except Exception as e:
+                _LOGGER.error("Error playing %s: %s", url, e)
 
-    def update_entity(self, key: str, state: str, attributes: Dict[str, str]) -> None:
-        """Update an entity."""
-        if key in self._entities:
-            self._entities[key].state = state
-            self._entities[key].attributes.update(attributes)
-            _LOGGER.debug("Updated entity: %s", key)
+        if done_callback:
+            done_callback()
 
-    def get_entity(self, key: str) -> Entity:
-        """Get an entity by key."""
-        return self._entities.get(key)
+    def duck(self):
+        """Duck the volume."""
+        _LOGGER.debug("Ducking media player")
+        # Reduce volume by 50%
+        # self._volume *= 0.5
 
-    def list_entities(self) -> List[Entity]:
-        """List all entities."""
-        return list(self._entities.values())
+    def unduck(self):
+        """Unduck the volume."""
+        _LOGGER.debug("Unducking media player")
+        # Restore volume
+        # self._volume = min(1.0, self._volume * 2.0)

reachy_mini_ha_voice/models.py

@@ -73,6 +73,20 @@ class ServerState:
     satellite: Optional["VoiceSatelliteProtocol"] = None
     wake_words_changed: bool = True
     reachy_integration: Optional["ReachyMiniIntegration"] = None
+    media_player_entity: Optional["MediaPlayerEntity"] = None
+
+    def save_preferences(self) -> None:
+        """Save preferences to file."""
+        try:
+            import json
+
+            with open(self.preferences_path, "w", encoding="utf-8") as f:
+                json.dump(
+                    {"active_wake_words": self.preferences.active_wake_words},
+                    f,
+                )
+        except Exception as e:
+            _LOGGER.error("Error saving preferences: %s", e)
 
 
 @dataclass
@@ -93,14 +107,33 @@ class AudioPlayer:
         self.device = device
         self._stream = None
         self._pyaudio = None
+        self._ducked = False
 
-    def play(self, audio_data: bytes) -> None:
-        """Play audio data."""
+    def play(self, audio_data: Union[bytes, str], done_callback=None) -> None:
+        """Play audio data or URL."""
         import pyaudio
 
         if self._pyaudio is None:
             self._pyaudio = pyaudio.PyAudio()
 
+        if isinstance(audio_data, str):
+            # It's a URL or file path
+            try:
+                from urllib.request import urlopen
+
+                if audio_data.startswith("http://") or audio_data.startswith("https://"):
+                    with urlopen(audio_data) as response:
+                        audio_data = response.read()
+                else:
+                    # It's a file path
+                    with open(audio_data, "rb") as f:
+                        audio_data = f.read()
+            except Exception as e:
+                _LOGGER.error("Error loading audio: %s", e)
+                if done_callback:
+                    done_callback()
+                return
+
         # Assume 16-bit PCM, 16kHz, mono
         if self._stream is None:
             self._stream = self._pyaudio.open(
@@ -113,11 +146,31 @@ class AudioPlayer:
 
         self._stream.write(audio_data)
 
-    def close(self) -> None:
-        """Close the audio player."""
+        if done_callback:
+            done_callback()
+
+    def duck(self) -> None:
+        """Duck the volume (reduce by 50%)."""
+        self._ducked = True
+        # For simple implementation, we just note the state
+        # In a full implementation, we would actually reduce the volume
+
+    def unduck(self) -> None:
+        """Unduck the volume (restore to normal)."""
+        self._ducked = False
+        # For simple implementation, we just note the state
+        # In a full implementation, we would actually restore the volume
+
+    def stop(self) -> None:
+        """Stop playing and reset the stream."""
         if self._stream is not None:
+            self._stream.stop_stream()
             self._stream.close()
             self._stream = None
+
+    def close(self) -> None:
+        """Close the audio player."""
+        self.stop()
         if self._pyaudio is not None:
             self._pyaudio.terminate()
             self._pyaudio = None

reachy_mini_ha_voice/satellite.py

@@ -1,136 +1,393 @@
-"""Voice satellite protocol implementation for ESPHome."""
+"""Voice satellite protocol."""
 
-import asyncio
-import json
+import hashlib
 import logging
-import struct
-from typing import Optional, Union
+import posixpath
+import shutil
+import time
+from collections.abc import Iterable
+from typing import Dict, Optional, Set, Union
+from urllib.parse import urlparse, urlunparse
+from urllib.request import urlopen
 
-from .models import ServerState
+# pylint: disable=no-name-in-module
+from aioesphomeapi.api_pb2 import (  # type: ignore[attr-defined]
+    DeviceInfoRequest,
+    DeviceInfoResponse,
+    ListEntitiesDoneResponse,
+    ListEntitiesRequest,
+    ListEntitiesServicesResponse,
+    ListEntitiesServicesArgumentsResponse,
+    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, WakeWordType
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class VoiceSatelliteProtocol(asyncio.Protocol):
-    """ESPHome voice satellite protocol implementation."""
+class VoiceSatelliteProtocol(APIServer):
 
-    def __init__(self, state: ServerState):
-        """Initialize protocol."""
-        self.state = state
-        self.transport: Optional[asyncio.Transport] = None
-        self._buffer = bytearray()
-        self._connected = False
+    def __init__(self, state: ServerState) -> None:
+        super().__init__(state.name)
 
-    def connection_made(self, transport: asyncio.Transport) -> None:
-        """Handle new connection."""
-        self.transport = transport
+        self.state = state
         self.state.satellite = self
-        self._connected = True
-        _LOGGER.info("Client connected: %s", transport.get_extra_info("peername"))
-
-    def connection_lost(self, exc: Optional[Exception]) -> None:
-        """Handle connection loss."""
-        self._connected = False
-        self.state.satellite = None
-        _LOGGER.info("Client disconnected")
-        if exc:
-            _LOGGER.error("Connection error: %s", exc)
-
-    def data_received(self, data: bytes) -> None:
-        """Handle incoming data."""
-        self._buffer.extend(data)
-
-        while len(self._buffer) >= 3:
-            # Parse message header
-            msg_type = self._buffer[0]
-            msg_length = struct.unpack(">H", self._buffer[1:3])[0]
-
-            if len(self._buffer) < 3 + msg_length:
-                # Need more data
+
+        if self.state.media_player_entity is None:
+            self.state.media_player_entity = MediaPlayerEntity(
+                server=self,
+                key=len(state.entities),
+                name="Media Player",
+                object_id="reachy_mini_ha_voice_media_player",
+                music_player=state.music_player,
+                announce_player=state.tts_player,
+            )
+            self.state.entities.append(self.state.media_player_entity)
+
+        self._is_streaming_audio = False
+        self._tts_url: Optional[str] = None
+        self._tts_played = False
+        self._continue_conversation = False
+        self._timer_finished = False
+        self._external_wake_words: Dict[str, VoiceAssistantExternalWakeWord] = {}
+
+    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._tts_url = data.get("url")
+            self._tts_played = False
+            self._continue_conversation = False
+        elif event_type in (
+            VoiceAssistantEventType.VOICE_ASSISTANT_STT_VAD_END,
+            VoiceAssistantEventType.VOICE_ASSISTANT_STT_END,
+        ):
+            self._is_streaming_audio = False
+        elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_INTENT_PROGRESS:
+            if data.get("tts_start_streaming") == "1":
+                # Start streaming early
+                self.play_tts()
+        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_END:
+            self._tts_url = data.get("url")
+            self.play_tts()
+        elif event_type == VoiceAssistantEventType.VOICE_ASSISTANT_RUN_END:
+            self._is_streaming_audio = False
+            if not self._tts_played:
+                self._tts_finished()
+
+            self._tts_played = False
+
+        # TODO: handle error
+
+    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]:
+        if isinstance(msg, VoiceAssistantEventResponse):
+            # Pipeline event
+            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)
+
+            assert self.state.media_player_entity is not None
+
+            urls = []
+            if msg.preannounce_media_id:
+                urls.append(msg.preannounce_media_id)
+
+            urls.append(msg.media_id)
+
+            self.state.active_wake_words.add(self.state.stop_word.id)
+            self._continue_conversation = msg.start_conversation
+
+            self.duck()
+            yield from self.state.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 = [
+                VoiceAssistantWakeWord(
+                    id=ww.id,
+                    wake_word=ww.wake_word,
+                    trained_languages=ww.trained_languages,
+                )
+                for ww in self.state.available_wake_words.values()
+            ]
+
+            for eww in msg.external_wake_words:
+                if eww.model_type != "micro":
+                    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")
+        elif isinstance(msg, VoiceAssistantSetConfiguration):
+            # Change active wake words
+            active_wake_words: Set[str] = set()
+
+            for wake_word_id in msg.active_wake_words:
+                if wake_word_id in self.state.wake_words:
+                    # Already active
+                    active_wake_words.add(wake_word_id)
+                    continue
+
+                model_info = self.state.available_wake_words.get(wake_word_id)
+                if not model_info:
+                    # Check external wake words (may require download)
+                    external_wake_word = self._external_wake_words.get(wake_word_id)
+                    if not external_wake_word:
+                        continue
+
+                    model_info = 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
 
-            # Extract message
-            msg_data = bytes(self._buffer[3 : 3 + msg_length])
-            self._buffer = self._buffer[3 + msg_length :]
-
-            # Process message
-            asyncio.create_task(self._process_message(msg_type, msg_data))
-
-    async def _process_message(self, msg_type: int, msg_data: bytes) -> None:
-        """Process a message."""
-        try:
-            if msg_type == 0x01:  # Hello
-                await self._handle_hello(msg_data)
-            elif msg_type == 0x02:  # Voice Assistant Start
-                await self._handle_voice_assistant_start(msg_data)
-            elif msg_type == 0x03:  # Voice Assistant End
-                await self._handle_voice_assistant_end(msg_data)
-            elif msg_type == 0x04:  # TTS Audio
-                await self._handle_tts_audio(msg_data)
-            else:
-                _LOGGER.warning("Unknown message type: %s", msg_type)
-        except Exception as e:
-            _LOGGER.error("Error processing message: %s", e)
-
-    async def _handle_hello(self, data: bytes) -> None:
-        """Handle hello message."""
-        _LOGGER.debug("Received hello message")
-        # Send hello response
-        response = self._build_message(0x01, json.dumps({"name": self.state.name}))
-        self._send_message(response)
-
-    async def _handle_voice_assistant_start(self, data: bytes) -> None:
-        """Handle voice assistant start message."""
-        _LOGGER.info("Voice assistant started")
-        # Play wake sound
-        try:
-            with open(self.state.wakeup_sound, "rb") as f:
-                self.state.tts_player.play(f.read())
-        except Exception as e:
-            _LOGGER.error("Error playing wake sound: %s", e)
-
-    async def _handle_voice_assistant_end(self, data: bytes) -> None:
-        """Handle voice assistant end message."""
-        _LOGGER.info("Voice assistant ended")
-
-    async def _handle_tts_audio(self, data: bytes) -> None:
-        """Handle TTS audio message."""
-        try:
-            self.state.tts_player.play(data)
-        except Exception as e:
-            _LOGGER.error("Error playing TTS audio: %s", e)
+            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:
-        """Handle audio chunk from microphone."""
-        if self._connected and self.transport:
-            # Send audio data to Home Assistant
-            message = self._build_message(0x10, audio_chunk)
-            self._send_message(message)
-
-    def wakeup(self, wake_word) -> None:
-        """Handle wake word detection."""
-        _LOGGER.info("Wake word detected: %s", wake_word.id)
-        # Send wake notification to Home Assistant
-        message = self._build_message(
-            0x11, json.dumps({"wake_word": wake_word.wake_word})
+
+        if not self._is_streaming_audio:
+            return
+
+        self.send_messages([VoiceAssistantAudio(data=audio_chunk)])
+
+    def wakeup(self, wake_word: Union[MicroWakeWord, OpenWakeWord]) -> None:
+        if self._timer_finished:
+            # Stop timer instead
+            self._timer_finished = False
+            self.state.tts_player.stop()
+            _LOGGER.debug("Stopping timer finished sound")
+            return
+
+        wake_word_phrase = 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._send_message(message)
+        self.duck()
+        self._is_streaming_audio = True
+        self.state.tts_player.play(self.state.wakeup_sound)
 
     def stop(self) -> None:
-        """Handle stop word detection."""
-        _LOGGER.info("Stop word detected")
-        # Send stop notification to Home Assistant
-        message = self._build_message(0x12, json.dumps({"action": "stop"}))
-        self._send_message(message)
-
-    def _build_message(self, msg_type: int, data: Union[str, bytes]) -> bytes:
-        """Build a message."""
-        if isinstance(data, str):
-            data = data.encode("utf-8")
-        length = len(data)
-        return bytes([msg_type]) + struct.pack(">H", length) + data
-
-    def _send_message(self, message: bytes) -> None:
-        """Send a message."""
-        if self._connected and self.transport:
-            self.transport.write(message)
+        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) or self._tts_played:
+            return
+
+        self._tts_played = True
+        _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)
+
+    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 _tts_finished(self) -> None:
+        self.state.active_wake_words.discard(self.state.stop_word.id)
+        self.send_messages([VoiceAssistantAnnounceFinished()])
+
+        if self._continue_conversation:
+            self.send_messages([VoiceAssistantRequest(start=True)])
+            self._is_streaming_audio = True
+            _LOGGER.debug("Continuing conversation")
+        else:
+            self.unduck()
+
+        _LOGGER.debug("TTS response finished")
+
+    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: time.sleep(1.0) or self._play_timer_finished(),
+        )
+
+    def connection_lost(self, exc):
+        super().connection_lost(exc)
+        _LOGGER.info("Disconnected from Home Assistant")
+
+    def _download_external_wake_word(
+        self, external_wake_word: VoiceAssistantExternalWakeWord
+    ) -> Optional[AvailableWakeWord]:
+        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()
+
+        # Check if we need to download the model file
+        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:
+            # Download config
+            _LOGGER.debug("Downloading %s to %s", external_wake_word.url, config_path)
+            with urlopen(external_wake_word.url) 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)
+
+        if should_download_model:
+            # Download model file
+            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)
+            with urlopen(model_url) 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)
+
+        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,
+        )