Fix the AirTouch 2+ connection dropout: ACK address (0xC0) + keep-alive

src/airtouch2/at2plus/At2PlusClient.py

@@ -5,19 +5,25 @@
 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
 from airtouch2.protocol.at2plus.messages.AcStatus import AcStatusMessage
 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:

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:

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):

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()

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)

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)

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()