feat(zeo): add MQTT push subscription and DPS cache with command trait

Subscribe to protocol-102 push messages on device connect, decode via decode_rpc_response, and feed _dps_cache. Add ZeoCommandTrait with start_program() (combine START+MODE+PROGRAM params, QoS 1) and resume(). _get_start_params reads from MQTT-push cache, falling back to ID_QUERY for MODE/PROGRAM on cache miss. Add ZeoFeatureTrait and ZeoFeatures for DP 237 capability discovery. DPS cache shared between ZeoApi and ZeoCommandTrait via reference.

Author: NOisi-X

roborock/data/zeo/zeo_containers.py

@@ -19,20 +19,31 @@
 
 @dataclass
 class ZeoStartParams(RoborockBase):
-    """Parameters that must be bundled with a START command.
+    """All parameters that may be bundled with a START command.
 
-    All Zeo devices require ``mode`` and ``program`` to be sent together
-    with the start signal. The remaining fields are optional and only
-    included when the device reports a non-None value.
+    ``mode`` and ``program`` are mandatory for every device.  Every other
+    field is optional — when ``None`` it is simply omitted from the MQTT
+    payload, so the same superset works for washers and dryers alike.
     """
 
     mode: ZeoMode
     program: ZeoProgram
+
+    # Washer
     temperature: ZeoTemperature | None = None
     rinse: ZeoRinse | None = None
     spin: ZeoSpin | None = None
     drying_mode: ZeoDryingMode | None = None
 
+    # Dryer
+    drying_method: ZeoDryingMethod | None = None
+    steam_volume: ZeoSteamVolume | None = None
+    total_time: int | None = None
+
+    # Optional across both device families
+    soak: ZeoSoak | None = None
+    dry_and_care: ZeoDryAndCare | None = None
+
 
 # ── DP 222 (LoadCloudProgram) bitfield decoder ──────────────────────────
 # The official app packs all custom-program parameters into a single

roborock/devices/traits/a01/__init__.py

@@ -38,16 +38,24 @@
     RoborockDyadStateCode,
 )
 from roborock.data.zeo.zeo_code_mappings import (
+    ZeoDetergentExpansionType,
     ZeoDetergentType,
+    ZeoDirtDetectionStatus,
+    ZeoDryAndCare,
+    ZeoDryerStartError,
+    ZeoDryingMethod,
     ZeoDryingMode,
     ZeoError,
     ZeoFeatureBits,
     ZeoMode,
     ZeoProgram,
     ZeoRinse,
+    ZeoSoak,
+    ZeoSoftenerExpansionType,
     ZeoSoftenerType,
     ZeoSpin,
     ZeoState,
+    ZeoSteamVolume,
     ZeoTemperature,
 )
 from roborock.devices.rpc.a01_channel import send_decoded_command
@@ -63,11 +71,16 @@
     RoborockZeoProtocol,
 )
 
+from .command import ZeoCommandTrait  # noqa: F401 — re‑export
+from .device_features import ZeoFeatures, ZeoFeatureTrait  # noqa: F401 — re‑export
+
 _LOGGER = logging.getLogger(__name__)
 
 __init__ = [
     "DyadApi",
     "ZeoApi",
+    "ZeoCommandTrait",
+    "ZeoFeatureTrait",
 ]
 
 
@@ -113,6 +126,27 @@
     RoborockZeoProtocol.TIMES_AFTER_CLEAN: lambda val: int(val),
     RoborockZeoProtocol.DETERGENT_EMPTY: lambda val: bool(val),
     RoborockZeoProtocol.SOFTENER_EMPTY: lambda val: bool(val),
+    RoborockZeoProtocol.DIRT_DETECTION_STATUS: lambda val: ZeoDirtDetectionStatus(val).name,
+    RoborockZeoProtocol.TOTAL_TIME: lambda val: int(val),
+    RoborockZeoProtocol.FEATURE_BITS: lambda val: int(val),
+    RoborockZeoProtocol.SMART_HOSTING_WAITED_TIME: lambda val: int(val),
+    RoborockZeoProtocol.IS_NEED_FLUFF_CLEAN: lambda val: bool(val),
+    RoborockZeoProtocol.PANEL_PROGRAM_PARAMS_SET_RESULT: lambda val: int(val),
+    RoborockZeoProtocol.DEVICE_BOUND: lambda val: bool(val),
+    RoborockZeoProtocol.CLOTH_PUT_IN: lambda val: bool(val),
+    RoborockZeoProtocol.CLOTH_READY_TO_DRY_COUNT_DOWN: lambda val: int(val),
+    RoborockZeoProtocol.START_DRYER_ERROR: lambda val: ZeoDryerStartError(val).name,
+    RoborockZeoProtocol.DOORLOCK_STATE: lambda val: bool(val),
+    RoborockZeoProtocol.APP_AUTHORIZATION: lambda val: bool(val),
+    RoborockZeoProtocol.SMART_HOSTING_TIME: lambda val: int(val),
+    RoborockZeoProtocol.CUSTOM_PROGRAM_CLEANING_TIME: lambda val: int(val),
+    RoborockZeoProtocol.PANEL_TIMING_PROGRAM_PARAMS: lambda val: int(val),
+    RoborockZeoProtocol.STEAM_CARE_TIME: lambda val: int(val),
+    # meta — read-only (JSON)
+    RoborockZeoProtocol.PRODUCT_INFO: lambda val: _try_json(val),
+    RoborockZeoProtocol.WASHING_LOG: lambda val: _try_json(val),
+    RoborockZeoProtocol.VOICE_RECORD_INFO: lambda val: _try_json(val),
+    RoborockZeoProtocol.VOICE_RECORD: lambda val: _try_json(val),
     # read-write
     RoborockZeoProtocol.MODE: lambda val: ZeoMode(val).name,
     RoborockZeoProtocol.PROGRAM: lambda val: ZeoProgram(val).name,
@@ -123,6 +157,41 @@
     RoborockZeoProtocol.DETERGENT_TYPE: lambda val: ZeoDetergentType(val).name,
     RoborockZeoProtocol.SOFTENER_TYPE: lambda val: ZeoSoftenerType(val).name,
     RoborockZeoProtocol.SOUND_SET: lambda val: bool(val),
+    RoborockZeoProtocol.DIRT_DETECTION_SWITCH: lambda val: bool(val),
+    RoborockZeoProtocol.SOAK: lambda val: ZeoSoak(val).name,
+    RoborockZeoProtocol.SILENT_MODE_ON: lambda val: bool(val),
+    RoborockZeoProtocol.SILENT_MODE_START_TIME: lambda val: int(val),
+    RoborockZeoProtocol.SILENT_MODE_END_TIME: lambda val: int(val),
+    RoborockZeoProtocol.DRY_CARE_MODE: lambda val: ZeoDryAndCare(val).name,
+    RoborockZeoProtocol.WASH_DRY_LINKED: lambda val: bool(val),
+    RoborockZeoProtocol.DRYING_METHOD: lambda val: ZeoDryingMethod(val).name,
+    RoborockZeoProtocol.STEAM_VOLUME: lambda val: ZeoSteamVolume(val).name,
+    RoborockZeoProtocol.ION_DEODORIZATION: lambda val: bool(val),
+    RoborockZeoProtocol.UV_LIGHT: lambda val: bool(val),
+    RoborockZeoProtocol.SMART_HOSTING: lambda val: bool(val),
+    RoborockZeoProtocol.SOFTENER_EXPANSION_TYPE: lambda val: ZeoSoftenerExpansionType(val).name,
+    RoborockZeoProtocol.DETERGENT_EXPANSION_TYPE: lambda val: ZeoDetergentExpansionType(val).name,
+    RoborockZeoProtocol.SMILE_LIGHT_STATUS: lambda val: bool(val),
+    RoborockZeoProtocol.POWER_LIGHT: lambda val: bool(val),
+    RoborockZeoProtocol.PANEL_PROGRAM_PARAMS_SET: lambda val: int(val),
+    RoborockZeoProtocol.WIFI_LINKAGE_RESET: lambda val: int(val),
+    RoborockZeoProtocol.SAVE_ADAPTED_CLOUD_PROGRAM: lambda val: int(val),
+    RoborockZeoProtocol.CHILD_LOCK: lambda val: bool(val),
+    RoborockZeoProtocol.DETERGENT_SET: lambda val: bool(val),
+    RoborockZeoProtocol.SOFTENER_SET: lambda val: bool(val),
+    RoborockZeoProtocol.FLUFF_CLEANED: lambda val: bool(val),
+    # read-write (int-valued)
+    RoborockZeoProtocol.CUSTOM_PARAM_SAVE: lambda val: int(val),
+    RoborockZeoProtocol.CUSTOM_PARAM_GET: lambda val: int(val),
+    RoborockZeoProtocol.DEFAULT_SETTING: lambda val: int(val),
+    RoborockZeoProtocol.LIGHT_SETTING: lambda val: bool(val),
+    RoborockZeoProtocol.DETERGENT_VOLUME: lambda val: int(val),
+    RoborockZeoProtocol.SOFTENER_VOLUME: lambda val: int(val),
+    # meta — read-write
+    RoborockZeoProtocol.SET_SOUND_PACKAGE: lambda val: val,
+    RoborockZeoProtocol.VOICE_VOLUME: lambda val: val,
+    RoborockZeoProtocol.VOICE_SWITCH: lambda val: bool(val),
+    RoborockZeoProtocol.VOICE_RECORD_DELETE: lambda val: int(val),
 }
 
 
@@ -173,20 +242,30 @@ class ZeoApi(Trait, TraitUpdateListener):
 
     name = "zeo"
 
-    def __init__(self, channel: MqttChannel) -> None:
+    def __init__(self, channel: MqttChannel, product_id: 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._feature_trait = ZeoFeatureTrait(channel, product_id)
+        self._command: ZeoCommandTrait | None = None
+
+    @property
+    def command(self) -> ZeoCommandTrait:
+        """Lazily-built trait for wash-programme commands."""
+        if self._command is None:
+            self._command = ZeoCommandTrait(
+                channel=self._channel,
+                dps_cache=self._dps_cache,
+                feature_trait=self._feature_trait,
+                proto_entries=ZEO_PROTOCOL_ENTRIES,
+            )
+        return self._command
 
     async def start(self) -> None:
-        """Subscribe to MQTT push and discover device features.
-
-        Subscribes to the DPS MQTT topic, then queries FEATURE_BITS
-        (DP 237) to wake the device and cache supported capabilities.
-        """
+        """Subscribe to MQTT push and discover device capabilities."""
         await self._ensure_subscribed()
         await self._discover_features()
 
@@ -203,13 +282,7 @@ async def _ensure_subscribed(self) -> None:
         self._dps_unsub = await self._channel.subscribe(self._on_dps_message)
 
     async def _discover_features(self) -> None:
-        """Query FEATURE_BITS to wake the device and cache capabilities.
-
-        Sending an RPC query after subscribing triggers the device to
-        start pushing its full state — equivalent to how V1's
-        ``discover_features()`` uses ``device_features.refresh()`` to
-        initiate the push cycle.
-        """
+        """Query FEATURE_BITS (DP 237) and cache device capabilities."""
         try:
             result = await self.query_values([RoborockZeoProtocol.FEATURE_BITS])
             self._feature_bits = result.get(RoborockZeoProtocol.FEATURE_BITS, 0)
@@ -261,6 +334,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, product_id=product.id)
         case _:
             raise NotImplementedError(f"Unsupported category {product.category}")

roborock/devices/traits/a01/command.py

@@ -0,0 +1,146 @@
+"""Zeo command trait"""
+
+import json
+import logging
+from collections.abc import Callable
+from typing import Any
+
+from roborock.data.zeo.zeo_containers import ZeoStartParams
+from roborock.devices.rpc.a01_channel import send_decoded_command
+from roborock.devices.traits.a01.device_features import ZeoFeatureTrait
+from roborock.devices.transport.mqtt_channel import MqttChannel
+from roborock.roborock_message import RoborockZeoProtocol
+
+_LOGGER = logging.getLogger(__name__)
+
+_START_PARAM_DPS: list[RoborockZeoProtocol] = [
+    RoborockZeoProtocol.MODE,
+    RoborockZeoProtocol.PROGRAM,
+    RoborockZeoProtocol.TEMP,
+    RoborockZeoProtocol.RINSE_TIMES,
+    RoborockZeoProtocol.SPIN_LEVEL,
+    RoborockZeoProtocol.DRYING_MODE,
+    RoborockZeoProtocol.DETERGENT_SET,
+    RoborockZeoProtocol.SOFTENER_SET,
+    RoborockZeoProtocol.COUNTDOWN,
+    RoborockZeoProtocol.SOAK,
+    RoborockZeoProtocol.TOTAL_TIME,
+    RoborockZeoProtocol.DRYING_METHOD,
+    RoborockZeoProtocol.STEAM_VOLUME,
+]
+
+_FIELD_TO_DP: dict[str, RoborockZeoProtocol] = {
+    "mode": RoborockZeoProtocol.MODE,
+    "program": RoborockZeoProtocol.PROGRAM,
+    "temperature": RoborockZeoProtocol.TEMP,
+    "rinse": RoborockZeoProtocol.RINSE_TIMES,
+    "spin": RoborockZeoProtocol.SPIN_LEVEL,
+    "drying_mode": RoborockZeoProtocol.DRYING_MODE,
+    "drying_method": RoborockZeoProtocol.DRYING_METHOD,
+    "steam_volume": RoborockZeoProtocol.STEAM_VOLUME,
+    "total_time": RoborockZeoProtocol.TOTAL_TIME,
+    "soak": RoborockZeoProtocol.SOAK,
+    "dry_and_care": RoborockZeoProtocol.DRY_CARE_MODE,
+}
+
+_FEATURE_GATED_DPS: dict[RoborockZeoProtocol, str] = {
+    RoborockZeoProtocol.ION_DEODORIZATION: "ion_deodorization",
+    RoborockZeoProtocol.WASH_DRY_LINKED: "wash_dry_linkage",
+    RoborockZeoProtocol.SMART_HOSTING: "smart_hosting",
+}
+
+class ZeoCommandTrait:
+    """Trait for sending wash-programme commands to Zeo devices."""
+
+    def __init__(
+        self,
+        *,
+        channel: MqttChannel,
+        dps_cache: dict[int, Any],
+        feature_trait: ZeoFeatureTrait,
+        proto_entries: dict[RoborockZeoProtocol, Callable],
+    ) -> None:
+        """Initialize the command trait."""
+        
+        self._channel = channel
+        self._dps_cache = dps_cache
+        self._feature_trait = feature_trait
+        self._proto_entries = proto_entries
+
+    def _convert_value(self, protocol: RoborockZeoProtocol, value: Any) -> Any:
+        """Convert a protocol value using the injected entries table."""
+        if (converter := self._proto_entries.get(protocol)) is not None:
+            try:
+                return converter(value)
+            except (ValueError, TypeError):
+                return None
+        return None
+
+
+    async def start_program(self) -> dict[RoborockZeoProtocol, Any]:
+        """Start the device, bundling the current programme parameters."""
+        _LOGGER.debug("Start command: discovering features and building payload")
+        features = self._feature_trait.features
+        p = await self._get_start_params()
+        dps: dict[RoborockZeoProtocol, Any] = {RoborockZeoProtocol.START: "True"}
+        for field_name, dp in _FIELD_TO_DP.items():
+            val = getattr(p, field_name)
+            if val is not None:
+                dps[dp] = val
+        for dp, attr_name in _FEATURE_GATED_DPS.items():
+            if features is not None and getattr(features, attr_name, False):
+                val = self._dps_cache.get(int(dp))
+                if val is not None:
+                    dps[dp] = val
+        result = await send_decoded_command(
+            self._channel,
+            dps,
+            qos=1,
+            value_encoder=lambda x: x,
+        )
+        for dp, v in dps.items():
+            self._dps_cache[int(dp)] = v
+        return {proto: self._convert_value(proto, dps.get(proto)) for proto in dps}
+
+    async def resume(self) -> dict[RoborockZeoProtocol, Any]:
+        """Resume a paused programme or start without rebundling).
+        Only works while the device is powered on."""
+        _LOGGER.debug("Resume command")
+        dps = {RoborockZeoProtocol.START: "True"}
+        result = await send_decoded_command(
+            self._channel,
+            dps,
+        )
+        self._dps_cache[int(RoborockZeoProtocol.START)] = 1
+        return result
+
+
+    async def _get_start_params(self) -> ZeoStartParams:
+        """Read programme settings, querying the device on cache miss."""
+        cache = self._dps_cache
+        need_refresh = (
+            int(RoborockZeoProtocol.MODE) not in cache
+            or int(RoborockZeoProtocol.PROGRAM) not in cache
+        )
+        if need_refresh:
+            raw = await send_decoded_command(
+                self._channel,
+                {RoborockZeoProtocol.ID_QUERY: [RoborockZeoProtocol.MODE, RoborockZeoProtocol.PROGRAM]},
+                value_encoder=json.dumps,
+            )
+            for dp in (RoborockZeoProtocol.MODE, RoborockZeoProtocol.PROGRAM):
+                if (val := raw.get(dp)) is not None:
+                    cache[int(dp)] = val
+        return ZeoStartParams(
+            mode=cache[int(RoborockZeoProtocol.MODE)],
+            program=cache[int(RoborockZeoProtocol.PROGRAM)],
+            temperature=cache.get(int(RoborockZeoProtocol.TEMP)),
+            rinse=cache.get(int(RoborockZeoProtocol.RINSE_TIMES)),
+            spin=cache.get(int(RoborockZeoProtocol.SPIN_LEVEL)),
+            drying_mode=cache.get(int(RoborockZeoProtocol.DRYING_MODE)),
+            drying_method=cache.get(int(RoborockZeoProtocol.DRYING_METHOD)),
+            steam_volume=cache.get(int(RoborockZeoProtocol.STEAM_VOLUME)),
+            total_time=cache.get(int(RoborockZeoProtocol.TOTAL_TIME)),
+            soak=cache.get(int(RoborockZeoProtocol.SOAK)),
+            dry_and_care=cache.get(int(RoborockZeoProtocol.DRY_CARE_MODE)),
+        )