Merge branch 'main' into supported_features_markdown

Author: Lash-L

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "python-roborock"
-version = "2.33.0"
+version = "2.35.0"
 description = "A package to control Roborock vacuums."
 authors = ["humbertogontijo <humbertogontijo@users.noreply.github.com>"]
 license = "GPL-3.0-only"

pyproject.toml

@@ -3,9 +3,7 @@
 from __future__ import annotations
 
 import asyncio
-import base64
 import logging
-import secrets
 import time
 from abc import ABC, abstractmethod
 from typing import Any
@@ -37,14 +35,11 @@ class RoborockClient(ABC):
     def __init__(self, device_info: DeviceData) -> None:
         """Initialize RoborockClient."""
         self.device_info = device_info
-        self._nonce = secrets.token_bytes(16)
         self._waiting_queue: dict[int, RoborockFuture] = {}
         self._last_device_msg_in = time.monotonic()
         self._last_disconnection = time.monotonic()
         self.keep_alive = KEEPALIVE
-        self._diagnostic_data: dict[str, dict[str, Any]] = {
-            "misc_info": {"Nonce": base64.b64encode(self._nonce).decode("utf-8")}
-        }
+        self._diagnostic_data: dict[str, dict[str, Any]] = {}
         self.is_available: bool = True
 
     async def async_release(self) -> None:
@@ -116,7 +111,3 @@ def _async_response(self, request_id: int, protocol_id: int = 0) -> Any:
             request_id = new_id
         self._waiting_queue[request_id] = queue
         return asyncio.ensure_future(self._wait_response(request_id, queue))
-
-    @abstractmethod
-    async def send_message(self, roborock_message: RoborockMessage):
-        """Send a message to the Roborock device."""

roborock/api.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from enum import StrEnum
+
+from roborock import DeviceFeatures
+
+
+class RoborockModeEnum(StrEnum):
+    """A custom StrEnum that also stores an integer code for each member."""
+
+    code: int
+
+    def __new__(cls, value: str, code: int) -> RoborockModeEnum:
+        """Creates a new enum member."""
+        member = str.__new__(cls, value)
+        member._value_ = value
+        member.code = code
+        return member
+
+
+class CleanModes(RoborockModeEnum):
+    GENTLE = ("gentle", 105)
+    OFF = ("off", 105)
+    QUIET = ("quiet", 101)
+    BALANCED = ("balanced", 102)
+    TURBO = ("turbo", 103)
+    MAX = ("max", 104)
+    MAX_PLUS = ("max_plus", 108)
+    CUSTOMIZED = ("custom", 106)
+    SMART_MODE = ("smart_mode", 110)
+
+
+class CleanRoutes(RoborockModeEnum):
+    STANDARD = ("standard", 300)
+    DEEP = ("deep", 301)
+    DEEP_PLUS = ("deep_plus", 303)
+    FAST = ("fast", 304)
+    DEEP_PLUS_CN = ("deep_plus", 305)
+    SMART_MODE = ("smart_mode", 306)
+    CUSTOMIZED = ("custom", 302)
+
+
+class CleanModesOld(RoborockModeEnum):
+    QUIET = ("quiet", 38)
+    BALANCED = ("balanced", 60)
+    TURBO = ("turbo", 75)
+    MAX = ("max", 100)
+
+
+class WaterModes(RoborockModeEnum):
+    OFF = ("off", 200)
+    LOW = ("low", 201)
+    MILD = ("mild", 201)
+    MEDIUM = ("medium", 202)
+    STANDARD = ("standard", 202)
+    HIGH = ("high", 203)
+    INTENSE = ("intense", 203)
+    CUSTOMIZED = ("custom", 204)
+    CUSTOM = ("custom_water_flow", 207)
+    EXTREME = ("extreme", 208)
+    SMART_MODE = ("smart_mode", 209)
+
+
+def get_clean_modes(features: DeviceFeatures) -> list[CleanModes]:
+    """Get the valid clean modes for the device - also known as 'fan power' or 'suction mode'"""
+    modes = [CleanModes.QUIET, CleanModes.BALANCED, CleanModes.TURBO, CleanModes.MAX]
+    if features.is_max_plus_mode_supported or features.is_none_pure_clean_mop_with_max_plus:
+        # If the vacuum has max plus mode supported
+        modes.append(CleanModes.MAX_PLUS)
+    if features.is_pure_clean_mop_supported:
+        # If the vacuum is capable of 'pure mop clean' aka no vacuum
+        modes.append(CleanModes.OFF)
+    else:
+        # If not, we can add gentle
+        modes.append(CleanModes.GENTLE)
+    return modes
+
+
+def get_clean_routes(features: DeviceFeatures, region: str) -> list[CleanRoutes]:
+    """The routes that the vacuum will take while mopping"""
+    if features.is_none_pure_clean_mop_with_max_plus:
+        return [CleanRoutes.FAST, CleanRoutes.STANDARD]
+    supported = [CleanRoutes.STANDARD, CleanRoutes.DEEP]
+    if features.is_careful_slow_mop_supported:
+        if not (
+            features.is_corner_clean_mode_supported
+            and features.is_clean_route_deep_slow_plus_supported
+            and region == "CN"
+        ):
+            # for some reason there is a china specific deep plus mode
+            supported.append(CleanRoutes.DEEP_PLUS_CN)
+        else:
+            supported.append(CleanRoutes.DEEP_PLUS)
+
+    if features.is_clean_route_fast_mode_supported:
+        supported.append(CleanRoutes.FAST)
+    return supported
+
+
+def get_water_modes(features: DeviceFeatures) -> list[WaterModes]:
+    """Get the valid water modes for the device - also known as 'water flow' or 'water level'"""
+    supported_modes = [WaterModes.OFF]
+    if features.is_mop_shake_module_supported:
+        # For mops that have the vibrating mop pad, they do mild standard intense
+        supported_modes.extend([WaterModes.MILD, WaterModes.STANDARD, WaterModes.INTENSE])
+    else:
+        supported_modes.extend([WaterModes.LOW, WaterModes.MEDIUM, WaterModes.HIGH])
+    if features.is_custom_water_box_distance_supported:
+        # This is for devices that allow you to set a custom water flow from 0-100
+        supported_modes.append(WaterModes.CUSTOM)
+    if features.is_mop_shake_module_supported and features.is_mop_shake_water_max_supported:
+        supported_modes.append(WaterModes.EXTREME)
+    return supported_modes

roborock/clean_modes.py

@@ -164,7 +164,7 @@ async def _discover(ctx):
     if not cache_data:
         raise Exception("You need to login first")
     client = RoborockApiClient(cache_data.email)
-    home_data = await client.get_home_data(cache_data.user_data)
+    home_data = await client.get_home_data_v3(cache_data.user_data)
     cache_data.home_data = home_data
     context.update(cache_data)
     click.echo(f"Discovered devices {', '.join([device.name for device in home_data.get_all_devices()])}")

roborock/cli.py

@@ -725,6 +725,29 @@ class NetworkInfo(RoborockBase):
     rssi: int | None = None
 
 
+@dataclass
+class AppInitStatusLocalInfo(RoborockBase):
+    location: str
+    bom: str | None = None
+    featureset: int | None = None
+    language: str | None = None
+    logserver: str | None = None
+    wifiplan: str | None = None
+    timezone: str | None = None
+    name: str | None = None
+
+
+@dataclass
+class AppInitStatus(RoborockBase):
+    local_info: AppInitStatusLocalInfo
+    feature_info: list[int]
+    new_feature_info: int
+    new_feature_info_str: str
+    new_feature_info_2: int | None = None
+    carriage_type: int | None = None
+    dsp_version: int | None = None
+
+
 @dataclass
 class DeviceData(RoborockBase):
     device: HomeDataDevice

roborock/containers.py

@@ -423,6 +423,7 @@ class DeviceFeatures:
     is_clean_route_setting_supported: bool = field(
         metadata={"product_features": [ProductFeatures.MOP_SHAKE_MODULE, ProductFeatures.MOP_SPIN_MODULE]}
     )
+    is_mop_shake_module_supported: bool = field(metadata={"product_features": [ProductFeatures.MOP_SHAKE_MODULE]})
 
     @classmethod
     def from_feature_flags(

roborock/device_features.py

@@ -0,0 +1,30 @@
+"""Thin wrapper around the MQTT channel for Roborock B01 devices."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from roborock.protocols.b01_protocol import (
+    CommandType,
+    ParamsType,
+    decode_rpc_response,
+    encode_mqtt_payload,
+)
+
+from .mqtt_channel import MqttChannel
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def send_decoded_command(
+    mqtt_channel: MqttChannel,
+    dps: int,
+    command: CommandType,
+    params: ParamsType,
+) -> dict[int, Any]:
+    """Send a command on the MQTT channel and get a decoded response."""
+    _LOGGER.debug("Sending MQTT command: %s", params)
+    roborock_message = encode_mqtt_payload(dps, command, params)
+    response = await mqtt_channel.send_message(roborock_message)
+    return decode_rpc_response(response)  # type: ignore[return-value]

roborock/devices/b01_channel.py

@@ -21,6 +21,7 @@
 from .cache import Cache, NoCache
 from .channel import Channel
 from .mqtt_channel import create_mqtt_channel
+from .traits.b01.props import B01PropsApi
 from .traits.dyad import DyadApi
 from .traits.status import StatusTrait
 from .traits.trait import Trait
@@ -45,6 +46,7 @@ class DeviceVersion(enum.StrEnum):
 
     V1 = "1.0"
     A01 = "A01"
+    B01 = "B01"
     UNKNOWN = "unknown"
 
 
@@ -120,7 +122,7 @@ def create_home_data_api(email: str, user_data: UserData) -> HomeDataApi:
     client = RoborockApiClient(email)
 
     async def home_data_api() -> HomeData:
-        return await client.get_home_data(user_data)
+        return await client.get_home_data_v3(user_data)
 
     return home_data_api
 
@@ -159,6 +161,9 @@ def device_creator(device: HomeDataDevice, product: HomeDataProduct) -> Roborock
                         traits.append(ZeoApi(mqtt_channel))
                     case _:
                         raise NotImplementedError(f"Device {device.name} has unsupported category {product.category}")
+            case DeviceVersion.B01:
+                mqtt_channel = create_mqtt_channel(user_data, mqtt_params, mqtt_session, device)
+                traits.append(B01PropsApi(mqtt_channel))
             case _:
                 raise NotImplementedError(f"Device {device.name} has unsupported version {device.pv}")
         return RoborockDevice(device, channel, traits)