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