feat: add MQTT push subscription and real-time state tracking for Zeo devices

roborock/data/zeo/zeo_code_mappings.py

@@ -1,8 +1,5 @@
 """Zeo (washing machine) device enums.
 
-Member-level comments show the manufacturer's official identifiers
-extracted from the React Native app plugin bundle.
-
 For enums that map between protocol-level positions and physical units
 (e.g. spin speed, temperature), the comment format is
 ``# L<N>, <physical-value>`` where ``L<N>`` is the protocol position
@@ -18,12 +15,7 @@
 
 
 class ZeoFeatureBits(RoborockEnum):
- """Bit positions in DP 237 (FEATURE_BITS).
-
- Extracted from the official Roborock Washer app plugin bundle
- (index.ios.bundle, module 726). Each member's integer value
- is the exact bit offset the device reports at DP 237.
- """
+ """Bit positions in DP 237 (FEATURE_BITS)."""
 
  smart_hosting = 0
  silent_mode = 1
@@ -52,14 +44,15 @@ class ZeoFeatureBits(RoborockEnum):
 
 
 class ZeoMode(RoborockEnum):
- null = 0 # (not found in app bundle)
+ null = 0
  wash = 1
  wash_and_dry = 2
  dry = 3
  treatment = 4
 
 
 class ZeoState(RoborockEnum):
+ null = 0
  standby = 1
  weighing = 2 # Checking
  soaking = 3
@@ -83,7 +76,7 @@ class ZeoState(RoborockEnum):
 
 
 class ZeoProgram(RoborockEnum):
- null = 0 # (not found in app bundle)
+ null = 0
  standard = 1 # Mixed
  quick = 2
  sanitize = 3 # Sterilization
@@ -152,6 +145,7 @@ class ZeoSoak(RoborockEnum):
 
 
 class ZeoTemperature(RoborockEnum):
+ null = 0
  normal = 1 # L1, 0 C
  low = 2 # L2, 30 C
  medium = 3 # L3, 40 C
@@ -203,11 +197,13 @@ class ZeoSoftenerType(RoborockEnum):
 
 
 class ZeoDetergentExpansionType(RoborockEnum):
+ null = 0
  concentrated_detergent = 1
  detergent = 2
 
 
 class ZeoSoftenerExpansionType(RoborockEnum):
+ null = 0
  softener = 1
  softener_expansion = 2
  wool_detergent = 3
@@ -241,6 +237,7 @@ class ZeoError(RoborockEnum):
 
 
 class ZeoDryingMethod(RoborockEnum):
+ null = 0
  l1 = 1 # L1, Saving
  l2 = 2 # L2, Standard
  l3 = 3 # L3, SuperFast
@@ -255,11 +252,13 @@ class ZeoSteamVolume(RoborockEnum):
 
 
 class ZeoDryAndCare(RoborockEnum):
+ null = 0
  soft = 1
  normal = 2
 
 
 class ZeoDryerStartError(RoborockEnum):
+ null = 0
  dryer_running = 1 # Washer-Dryer Pairing cannot start: Dryer is running.
  dryer_error = 2 # Washer-Dryer Pairing cannot start: Dryer has an error.
  dryer_done = 3 # Dryer drying is complete, please remove the clothes first.
@@ -269,7 +268,7 @@ class ZeoDryerStartError(RoborockEnum):
  dryer_network_fail = 7 # Please check the dryer network connection.
 
 
-# The following are App-internal lookup tables extracted from the bundle.
+# The following are App-internal lookup tables.
 # They are NOT DP protocol enums — the device never receives these values.
 # They control how the official app adjusts recommended dosages or UI visibility.
 #

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()
+ elif 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.
@@ -230,6 +232,8 @@ async def close(self) -> None:
  self.v1_properties.close()
  if self.b01_q10_properties is not None:
  await self.b01_q10_properties.close()
+ if self.zeo is not None:
+ self.zeo.close()
  if self._unsub:
  self._unsub()
  self._unsub = None

roborock/devices/traits/a01/__init__.py

@@ -22,6 +22,7 @@
 """
 
 import json
+import logging
 from collections.abc import Callable
 from datetime import time
 from typing import Any
@@ -42,6 +43,7 @@
  ZeoDetergentType,
  ZeoDryingMode,
  ZeoError,
+ ZeoFeatureBits,
  ZeoMode,
  ZeoProgram,
  ZeoRinse,
@@ -52,10 +54,23 @@
 )
 from roborock.devices.rpc.a01_channel import send_decoded_command
 from roborock.devices.traits import Trait
+from roborock.devices.traits.a01.device_feature import (
+ build_feature_dp_list,
+ build_force_load_dp_list,
+ supports_uv_light,
+)
+from roborock.devices.traits.common import TraitUpdateListener
 from roborock.devices.transport.mqtt_channel import MqttChannel
 from roborock.exceptions import RoborockException
 from roborock.protocols.a01_protocol import decode_rpc_response
-from roborock.roborock_message import RoborockDyadDataProtocol, RoborockMessage, RoborockZeoProtocol
+from roborock.roborock_message import (
+ RoborockDyadDataProtocol,
+ RoborockMessage,
+ RoborockMessageProtocol,
+ RoborockZeoProtocol,
+)
+
+_LOGGER = logging.getLogger(__name__)
 
 __init__ = [
  "DyadApi",
@@ -115,6 +130,7 @@
  RoborockZeoProtocol.DETERGENT_TYPE: lambda val: ZeoDetergentType(val).name,
  RoborockZeoProtocol.SOFTENER_TYPE: lambda val: ZeoSoftenerType(val).name,
  RoborockZeoProtocol.SOUND_SET: lambda val: bool(val),
+ RoborockZeoProtocol.FEATURE_BITS: lambda val: int(val),
 }
 
 
@@ -187,14 +203,93 @@ def on_message(message: RoborockMessage) -> None:
  return await self._channel.subscribe(on_message)
 
 
-class ZeoApi(Trait):
+class ZeoApi(Trait, TraitUpdateListener):
  """API for interacting with Zeo devices."""
 
  name = "zeo"
 
- def __init__(self, channel: MqttChannel) -> None:
+ def __init__(self, channel: MqttChannel, model: str | None = None) -> 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
+ self._feature_bits: int = 0
+ self._model = model
+
+ async def start(self) -> None:
+ """Subscribe to MQTT push and trigger a full state sync.
+
+ Subscribes to the DPS MQTT topic, then performs a two-stage
+ force-load: first the base DP list (including FEATURE_BITS),
+ then a second query for the DPs gated behind each enabled feature.
+ The device responds with a complete state dump;
+ subsequent changes arrive via incremental MQTT push.
+ """
+ await self._ensure_subscribed()
+ await self._force_load()
+ await self._load_feature_dps()
+
+ 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)
+
+ async def _force_load(self) -> None:
+ """Send ID_QUERY with the base DP list to trigger a full state push.
+
+ For devices known to lack FEATURE_BITS, the DP is excluded
+ from the query list and ``_feature_bits`` stays at 0.
+ """
+ dp_list = build_force_load_dp_list(self._model)
+ result = await self.query_values(dp_list)
+ self._feature_bits = result.get(RoborockZeoProtocol.FEATURE_BITS, 0)
+
+ async def _load_feature_dps(self) -> None:
+ """Second-stage query for feature-gated DPs.
+
+ Called unconditionally after the first force-load; each DP is
+ independently gated:
+
+ - Feature-gated DPs are queried only when their feature bit is set
+ in FEATURE_BITS (DP 237).
+ - UV light (DP 228) is gated by :func:`supports_uv_light` (series
+ whitelist), independent of the feature bits.
+ """
+ feature_dps: list[RoborockZeoProtocol] = []
+ if self._feature_bits:
+ feature_dps.extend(build_feature_dp_list(self._feature_bits))
+ if supports_uv_light(self._model):
+ feature_dps.append(RoborockZeoProtocol.UV_LIGHT)
+ if not feature_dps:
+ return
+ try:
+ await self.query_values(feature_dps)
+ except RoborockException as exc:
+ _LOGGER.warning("Feature DPS load failed (non-fatal): %s", exc)
+
+ def supports(self, feature: ZeoFeatureBits) -> bool:
+ """Check whether the device supports a given feature bit."""
+ return bool(self._feature_bits & (1 << feature.value))
+
+ def _on_dps_message(self, message: RoborockMessage) -> None:
+ """Handle unsolicited MQTT push (protocol 102 — RPC_RESPONSE)."""
+ if message.protocol != RoborockMessageProtocol.RPC_RESPONSE:
+ return
+ try:
+ decoded = decode_rpc_response(message)
+ except RoborockException:
+ _LOGGER.debug("Dropped malformed push message", exc_info=True)
+ return
+ self._dps_cache.update(decoded)
+ self._notify_update()
 
  async def query_values(self, protocols: list[RoborockZeoProtocol]) -> dict[RoborockZeoProtocol, Any]:
  """Query the device for the values of the given protocols."""
@@ -217,6 +312,6 @@ def create(product: HomeDataProduct, mqtt_channel: MqttChannel) -> DyadApi | Zeo
  case RoborockCategory.WET_DRY_VAC:
  return DyadApi(mqtt_channel)
  case RoborockCategory.WASHING_MACHINE:
- return ZeoApi(mqtt_channel)
+ return ZeoApi(mqtt_channel, model=product.model)
  case _:
  raise NotImplementedError(f"Unsupported category {product.category}")