Implement MQTT for Advanced LVA Controls

Includes various compatibility fixes, reversions, and clean-up from the previous development cycle to create a stable base for the MQTT implementation. This ensures the application starts reliably with the pinned aioesphomeapi version.

linux_voice_assistant/api_server.py

@@ -9,8 +9,6 @@ from typing import TYPE_CHECKING, List, Optional
 # 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]
-    ConnectRequest,
-    ConnectResponse,
     DisconnectRequest,
     DisconnectResponse,
     HelloRequest,
@@ -57,9 +55,7 @@ class APIServer(asyncio.Protocol):
             )
             return
 
-        if isinstance(msg_inst, ConnectRequest):
-            self.send_messages([ConnectResponse()])
-        elif isinstance(msg_inst, DisconnectRequest):
+        if isinstance(msg_inst, DisconnectRequest):
             self.send_messages([DisconnectResponse()])
             _LOGGER.debug("Disconnect requested")
             if self._transport:
@@ -99,7 +95,6 @@ class APIServer(asyncio.Protocol):
 
         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
@@ -113,7 +108,6 @@ class APIServer(asyncio.Protocol):
                 return
 
             if length == 0:
-                # Empty message (allowed)
                 self._remove_from_buffer()
                 self.process_packet(msg_type, b"")
                 continue
@@ -125,7 +119,6 @@ class APIServer(asyncio.Protocol):
             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
@@ -134,8 +127,6 @@ class APIServer(asyncio.Protocol):
         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):
@@ -143,7 +134,6 @@ class APIServer(asyncio.Protocol):
         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
 
@@ -160,21 +150,12 @@ class APIServer(asyncio.Protocol):
         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]