fix(zeo): call TraitUpdateListener.__init__ in ZeoApi and fix connect() to use zeo attribute

Author: NOisi-X

roborock/devices/device.py

@@ -202,6 +202,8 @@ async def connect(self) -> None:
                 await self.v1_properties.start()
             elif self.b01_q10_properties is not None:
                 await self.b01_q10_properties.start()
+            if self.zeo is not None:
+                await self.zeo.start()
         except RoborockException:
             # Expected: start() can fail transiently. Unsubscribe before propagating
             # so the retry by connect_loop() gets a clean channel.

roborock/devices/traits/a01/__init__.py

@@ -20,10 +20,14 @@
 """
 
 import json
+import logging
 from collections.abc import Callable
 from datetime import time
 from typing import Any
 
+from roborock.exceptions import RoborockException
+from roborock.protocols.a01_protocol import decode_rpc_response
+
 from roborock.data import DyadProductInfo, DyadSndState, HomeDataProduct, RoborockCategory
 from roborock.data.dyad.dyad_code_mappings import (
     DyadBrushSpeed,
@@ -50,8 +54,16 @@
 )
 from roborock.devices.rpc.a01_channel import send_decoded_command
 from roborock.devices.traits import Trait
+from roborock.devices.traits.common import TraitUpdateListener
 from roborock.devices.transport.mqtt_channel import MqttChannel
-from roborock.roborock_message import RoborockDyadDataProtocol, RoborockZeoProtocol
+from roborock.roborock_message import (
+    RoborockDyadDataProtocol,
+    RoborockMessage,
+    RoborockMessageProtocol,
+    RoborockZeoProtocol,
+)
+
+_LOGGER = logging.getLogger(__name__)
 
 __init__ = [
     "DyadApi",
@@ -156,14 +168,50 @@ async def set_value(self, protocol: RoborockDyadDataProtocol, value: Any) -> dic
         return await send_decoded_command(self._channel, params)
 
 
-class ZeoApi(Trait):
+class ZeoApi(Trait, TraitUpdateListener):
     """API for interacting with Zeo devices."""
 
     name = "zeo"
 
     def __init__(self, channel: MqttChannel) -> None:
         """Initialize the Zeo API."""
+        TraitUpdateListener.__init__(self, _LOGGER)
         self._channel = channel
+        self._dps_cache: dict[int, Any] = {}
+        self._dps_unsub: Callable[[], None] | None = None
+
+    async def start(self) -> None:
+        """Subscribe to MQTT push (called once after device connects)."""
+        await self._ensure_subscribed()
+
+    def close(self) -> None:
+        """Unsubscribe from MQTT push and release resources."""
+        if self._dps_unsub is not None:
+            self._dps_unsub()
+            self._dps_unsub = None
+
+    async def _ensure_subscribed(self) -> None:
+        """Subscribe to MQTT DPS push (idempotent)."""
+        if self._dps_unsub is not None:
+            return
+        self._dps_unsub = await self._channel.subscribe(self._on_dps_message)
+
+    def _on_dps_message(self, message: RoborockMessage) -> None:
+        """Handle unsolicited MQTT push (protocol 102 — RPC_RESPONSE).
+
+        Zeo devices broadcast status changes as ``{"dps": {...}}`` JSON
+        payloads.  This callback decodes them and feeds the cache so
+        that ``query_values`` can skip the device round-trip when the
+        requested DPs are already up to date.
+        """
+        if message.protocol != RoborockMessageProtocol.RPC_RESPONSE:
+            return
+        try:
+            decoded = decode_rpc_response(message)
+            self._dps_cache.update(decoded)
+            self._notify_update()
+        except RoborockException:
+            _LOGGER.debug("Failed to decode push message, skipping: %s", message, exc_info=True)
 
     async def query_values(self, protocols: list[RoborockZeoProtocol]) -> dict[RoborockZeoProtocol, Any]:
         """Query the device for the values of the given protocols."""
@@ -172,6 +220,9 @@ async def query_values(self, protocols: list[RoborockZeoProtocol]) -> dict[Robor
             {RoborockZeoProtocol.ID_QUERY: protocols},
             value_encoder=json.dumps,
         )
+        for protocol, value in response.items():
+            if value is not None:
+                self._dps_cache[int(protocol)] = value
         return {protocol: convert_zeo_value(protocol, response.get(protocol)) for protocol in protocols}
 
     async def set_value(self, protocol: RoborockZeoProtocol, value: Any) -> dict[RoborockZeoProtocol, Any]: