diff --git a/src/airtouch2/at2plus/At2PlusClient.py b/src/airtouch2/at2plus/At2PlusClient.py index e074d7b..c5126c3 100644 --- a/src/airtouch2/at2plus/At2PlusClient.py +++ b/src/airtouch2/at2plus/At2PlusClient.py @@ -5,7 +5,7 @@ from airtouch2.at2plus.At2PlusAircon import At2PlusAircon from airtouch2.at2plus.At2PlusGroup import At2PlusGroup from airtouch2.common.NetClient import NetClient -from airtouch2.protocol.at2plus.control_status_common import ControlStatusSubHeader, ControlStatusSubType +from airtouch2.protocol.at2plus.control_status_common import ControlStatusSubHeader, ControlStatusSubType, need_ack from airtouch2.protocol.at2plus.extended_common import ExtendedMessageSubType, ExtendedSubHeader from airtouch2.protocol.at2plus.message_common import HEADER_LENGTH, HEADER_MAGIC, Header, Message, MessageType from airtouch2.protocol.at2plus.messages.AcAbilityMessage import AcAbility, AcAbilityMessage, RequestAcAbilityMessage @@ -13,11 +13,17 @@ from airtouch2.common.Buffer import Buffer from airtouch2.protocol.at2plus.crc16_modbus import crc16 from airtouch2.common.interfaces import Callback, Serializable, TaskCreator +from airtouch2.protocol.at2plus.messages.Ack import Ack from airtouch2.protocol.at2plus.messages.GroupNames import RequestGroupNamesMessage, group_names_from_subdata from airtouch2.protocol.at2plus.messages.GroupStatus import GroupStatusMessage _LOGGER = logging.getLogger(__name__) +# The controller drops the TCP session after ~16 min of silence (it stops +# broadcasting when nothing changes, e.g. overnight). A lightweight status +# request every 4 min keeps the session alive and prevents that idle reset. +POLL_INTERVAL_SECONDS = 240 + class At2PlusClient: def __init__(self, host: str, dump_responses: bool = False, task_creator: TaskCreator = asyncio.create_task): @@ -29,6 +35,7 @@ def __init__(self, host: str, dump_responses: bool = False, task_creator: TaskCr self._client = NetClient(host, 9200, self._on_connect, self.handle_one_message, task_creator) self._dump_responses = dump_responses self._task_creator = task_creator + self._poll_task: asyncio.Task[None] | None = None self._new_ac_callbacks: list[Callback] = [] self._ability_message_queue: asyncio.Queue[AcAbilityMessage] = asyncio.Queue() self._found_ac = asyncio.Event() @@ -41,11 +48,15 @@ async def connect(self) -> bool: def run(self) -> None: self._client.run() + self._poll_task = self._task_creator(self._poll_loop()) async def wait_for_ac(self, timeout: int = 5) -> None: await asyncio.wait_for(self._found_ac.wait(), timeout) async def stop(self) -> None: + if self._poll_task is not None: + self._poll_task.cancel() + self._poll_task = None await self._client.stop() def add_new_ac_callback(self, callback: Callback): @@ -69,6 +80,25 @@ def remove_callback() -> None: async def send(self, msg: Serializable): await self._client.send(msg) + async def _send_keepalive_poll(self) -> None: + """Send a lightweight status request to keep the TCP session active. + + The controller resets the socket after ~16 min of silence (it stops + broadcasting when nothing changes, e.g. overnight), so a periodic + request keeps traffic flowing. Best-effort: transient send errors + during reconnection are ignored. + """ + try: + await self._client.send(AcStatusMessage([])) + _LOGGER.debug("Sent keep-alive status poll") + except Exception as e: + _LOGGER.debug(f"Keep-alive poll failed (likely reconnecting): {e}") + + async def _poll_loop(self) -> None: + while True: + await asyncio.sleep(POLL_INTERVAL_SECONDS) + await self._send_keepalive_poll() + async def handle_one_message(self) -> None: message = await self._read_message() if not message: @@ -86,9 +116,20 @@ async def handle_one_message(self) -> None: group_status_message = GroupStatusMessage.from_bytes( message.data_buffer.read_bytes(subheader.subdata_length.total())) self._task_creator(self._handle_group_status_message(group_status_message)) + # Currently unhandled + elif subheader.sub_type == ControlStatusSubType.AC_STATUS2: + pass + elif subheader.sub_type == ControlStatusSubType.SYSTEM_STATUS: + pass + elif subheader.sub_type == ControlStatusSubType.ZONE_STATUS: + pass + elif subheader.sub_type == ControlStatusSubType.SYSTEM_ID: + pass else: _LOGGER.warning( f"Unknown status message type: subtype={subheader.sub_type}, data={message.data_buffer.to_bytes().hex(':')}") + if subheader.sub_type in need_ack: + self._task_creator(self.send(Ack(subheader.sub_type))) elif message.header.type == MessageType.EXTENDED: subheader = ExtendedSubHeader.from_buffer(message.data_buffer) if subheader.sub_type == ExtendedMessageSubType.ABILITY: diff --git a/src/airtouch2/common/NetClient.py b/src/airtouch2/common/NetClient.py index 53b5037..c6ab196 100644 --- a/src/airtouch2/common/NetClient.py +++ b/src/airtouch2/common/NetClient.py @@ -101,7 +101,9 @@ async def send(self, message: Serializable) -> None: try: await self._writer.drain() drained = True - except (ConnectionResetError, asyncio.IncompleteReadError, TimeoutError) as e: + except (asyncio.IncompleteReadError, OSError) as e: + # OSError covers reset / timeout / host-unreachable — any + # of them means the socket is dead; reconnect and retry. await self._try_reconnect() async def read_bytes(self, size: int) -> bytes | None: @@ -116,8 +118,14 @@ async def read_bytes(self, size: int) -> bytes | None: except asyncio.IncompleteReadError as e: _LOGGER.debug(f"IncompleteReadError - partial bytes: {e.partial.hex(':')}") data = None - except (ConnectionResetError, TimeoutError) as e: - _LOGGER.debug("ConnectionResetError") + except OSError as e: + # Any socket-level failure means the connection is gone: reset, + # timeout, or host-unreachable (the controller dropping off WiFi + # mid-connection raises [Errno 113], which is NOT a + # ConnectionResetError). An uncaught error here kills the read + # loop and leaves the client permanently dead until restart, so + # treat every OSError as a lost connection and reconnect. + _LOGGER.debug(f"Socket error while reading: {e!r}") data = None if data is None: diff --git a/src/airtouch2/protocol/at2plus/control_status_common.py b/src/airtouch2/protocol/at2plus/control_status_common.py index 09fd222..24f3d3f 100644 --- a/src/airtouch2/protocol/at2plus/control_status_common.py +++ b/src/airtouch2/protocol/at2plus/control_status_common.py @@ -36,6 +36,19 @@ class ControlStatusSubType(IntEnum): AC_CONTROL = 0x22 AC_STATUS = 0x23 + AC_STATUS2 = 0x10 + SYSTEM_STATUS = 0x2B + ZONE_STATUS = 0x40 + SYSTEM_ID = 0x45 + + +need_ack: list[ControlStatusSubType] = [ + ControlStatusSubType.AC_STATUS2, + ControlStatusSubType.SYSTEM_STATUS, + ControlStatusSubType.ZONE_STATUS, + ControlStatusSubType.SYSTEM_ID, +] + @dataclass class SubDataLength(Serializable): diff --git a/src/airtouch2/protocol/at2plus/messages/Ack.py b/src/airtouch2/protocol/at2plus/messages/Ack.py new file mode 100644 index 0000000..f354540 --- /dev/null +++ b/src/airtouch2/protocol/at2plus/messages/Ack.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from airtouch2.common.interfaces import Serializable +from airtouch2.protocol.at2plus.control_status_common import ( + CONTROL_STATUS_SUBHEADER_LENGTH, + ControlStatusSubType, + ControlStatusSubHeader, + SubDataLength, +) +from airtouch2.protocol.at2plus.message_common import ( + Header, + MessageType, + add_checksum_message_buffer, + prime_message_buffer, +) + +# The controller requires the ACK's top-level address byte to carry the +# control/status identifier (0xC0), NOT the usual client value (0x80). Verified +# on a live AirTouch 2+: acking the status broadcasts with 0x80 triggers a +# continuous broadcast spam storm (~1.7/sec, ~100x normal) until the session is +# unusable; 0xC0 is accepted cleanly. It looks redundant with the message-type +# byte, but the controller genuinely validates it. +CONTROL_STATUS_REPLY_ADDRESS = 0xC0 + + +class Ack(Serializable): + message_type: ControlStatusSubType + + def __init__(self, message_type: ControlStatusSubType): + self.message_type = message_type + + def to_bytes(self) -> bytes: + subheader = ControlStatusSubHeader(self.message_type, SubDataLength(1, 0, 0)) + buffer = prime_message_buffer( + Header( + CONTROL_STATUS_REPLY_ADDRESS, + MessageType.CONTROL_STATUS, + CONTROL_STATUS_SUBHEADER_LENGTH + subheader.subdata_length.total(), + ) + ) + buffer.append(subheader) + buffer.append_bytes(bytes([0])) # zero payload byte + add_checksum_message_buffer(buffer) + return buffer.to_bytes() diff --git a/tests/protocol/at2plus/messages/test_ack.py b/tests/protocol/at2plus/messages/test_ack.py new file mode 100644 index 0000000..2bbed11 --- /dev/null +++ b/tests/protocol/at2plus/messages/test_ack.py @@ -0,0 +1,31 @@ +import unittest +from airtouch2.protocol.at2plus.control_status_common import ControlStatusSubType +from airtouch2.protocol.at2plus.messages.Ack import Ack + + +class TestAck(unittest.TestCase): + def _test_common(self, message_type: ControlStatusSubType): + ack = Ack(message_type) + serialised = ack.to_bytes() + expected = bytes([ + # Address byte must be 0xC0 (control/status), not 0x80 — verified on a + # live controller: 0x80 triggers a continuous broadcast spam storm. + 0x55, 0x55, 0xc0, 0xb0, 0x01, 0xc0, 0x00, 0x09, + # ControlStatusSubheader with matching sub message type + # Static, 1-byte, zero payload. + int(message_type), 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00 + ]) + self.assertEqual(serialised[:-2].hex(':'), expected.hex(':')) + + def test_ac_status2(self): + # aramshaw says this requires an extra payload byte + self._test_common(ControlStatusSubType.AC_STATUS2) + + def test_system_status(self): + self._test_common(ControlStatusSubType.SYSTEM_STATUS) + + def test_system_id(self): + self._test_common(ControlStatusSubType.SYSTEM_ID) + + def test_zone_status(self): + self._test_common(ControlStatusSubType.ZONE_STATUS) diff --git a/tests/protocol/at2plus/test_keepalive_poll.py b/tests/protocol/at2plus/test_keepalive_poll.py new file mode 100644 index 0000000..08f48d0 --- /dev/null +++ b/tests/protocol/at2plus/test_keepalive_poll.py @@ -0,0 +1,23 @@ +import unittest + +from airtouch2.at2plus.At2PlusClient import At2PlusClient +from airtouch2.protocol.at2plus.messages.AcStatus import AcStatusMessage + + +class TestKeepAlivePoll(unittest.IsolatedAsyncioTestCase): + """The controller drops the session after ~16 min of silence; a periodic + lightweight status request keeps it alive.""" + + async def test_poll_sends_status_request(self): + client = At2PlusClient("localhost") + sent = [] + + async def fake_send(msg): + sent.append(msg) + + client._client.send = fake_send + + await client._send_keepalive_poll() + + self.assertEqual(len(sent), 1) + self.assertIsInstance(sent[0], AcStatusMessage) diff --git a/tests/protocol/at2plus/test_netclient_socket_errors.py b/tests/protocol/at2plus/test_netclient_socket_errors.py new file mode 100644 index 0000000..095bd05 --- /dev/null +++ b/tests/protocol/at2plus/test_netclient_socket_errors.py @@ -0,0 +1,54 @@ +import unittest +from unittest.mock import AsyncMock, MagicMock + +from airtouch2.common.NetClient import NetClient + + +class _StubMessage: + def to_bytes(self) -> bytes: + return b"\x00" + + +class TestNetClientSocketErrors(unittest.IsolatedAsyncioTestCase): + """Regression tests for the zombie-client bug. + + When the controller drops off WiFi mid-connection, reads/writes fail with + OSError [Errno 113] Host is unreachable - NOT ConnectionResetError. The + old code only caught reset/timeout, so the read loop died un-handled and + the client never reconnected (dead until restart). Any socket-level + OSError must be treated as connection loss. + """ + + def _client(self) -> NetClient: + return NetClient("localhost", 9200, AsyncMock(), AsyncMock()) + + async def test_read_bytes_treats_host_unreachable_as_connection_loss(self): + client = self._client() + client._reader = MagicMock() + client._reader.readexactly = AsyncMock( + side_effect=OSError(113, "Host is unreachable")) + client._try_reconnect = AsyncMock() + + result = await client.read_bytes(1) + + self.assertIsNone(result) # signalled as a failed read... + client._try_reconnect.assert_awaited() # ...and reconnect kicked off + + async def test_send_reconnects_on_host_unreachable(self): + client = self._client() + dead_writer = MagicMock() + dead_writer.drain = AsyncMock( + side_effect=OSError(113, "Host is unreachable")) + good_writer = MagicMock() + good_writer.drain = AsyncMock() + + async def fake_reconnect(): + client._writer = good_writer + + client._writer = dead_writer + client._try_reconnect = AsyncMock(side_effect=fake_reconnect) + + await client.send(_StubMessage()) # must not raise + + client._try_reconnect.assert_awaited() + good_writer.drain.assert_awaited()