Add XML command "SetFanSpeed"

deebot_client/commands/__init__.py

@@ -11,14 +11,18 @@
     COMMANDS as JSON_COMMANDS,
     COMMANDS_WITH_MQTT_P2P_HANDLING as JSON_COMMANDS_WITH_MQTT_P2P_HANDLING,
 )
+from .xml import (
+    COMMANDS_WITH_MQTT_P2P_HANDLING as XML_COMMANDS_WITH_MQTT_P2P_HANDLING,
+)
 
 if TYPE_CHECKING:
     from deebot_client.command import Command, CommandMqttP2P
 
 COMMANDS: dict[DataType, dict[str, type[Command]]] = {DataType.JSON: JSON_COMMANDS}
 
 COMMANDS_WITH_MQTT_P2P_HANDLING: dict[DataType, dict[str, type[CommandMqttP2P]]] = {
-    DataType.JSON: JSON_COMMANDS_WITH_MQTT_P2P_HANDLING
+    DataType.JSON: JSON_COMMANDS_WITH_MQTT_P2P_HANDLING,
+    DataType.XML: XML_COMMANDS_WITH_MQTT_P2P_HANDLING,
 }
 
 

deebot_client/commands/xml/__init__.py

@@ -8,7 +8,7 @@
 
 from .charge_state import GetChargeState
 from .error import GetError
-from .fan_speed import GetFanSpeed
+from .fan_speed import GetCleanSpeed, SetCleanSpeed
 from .life_span import GetLifeSpan
 from .play_sound import PlaySound
 from .pos import GetPos
@@ -19,17 +19,20 @@
 
 __all__ = [
     "GetChargeState",
+    "GetCleanSpeed",
     "GetCleanSum",
     "GetError",
-    "GetFanSpeed",
     "GetLifeSpan",
     "GetPos",
     "PlaySound",
+    "SetCleanSpeed",
 ]
 
 # fmt: off
 # ordered by file asc
 _COMMANDS: list[type[XmlCommand]] = [
+    GetCleanSpeed,
+    SetCleanSpeed,
     GetError,
     GetLifeSpan,
     PlaySound,

deebot_client/commands/xml/common.py

@@ -3,15 +3,24 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, Any, cast
 from xml.etree.ElementTree import Element, SubElement
 
 from defusedxml import ElementTree  # type: ignore[import-untyped]
 
-from deebot_client.command import Command, CommandWithMessageHandling, SetCommand
+from deebot_client.command import (
+    Command,
+    CommandWithMessageHandling,
+    GetCommand,
+    SetCommand,
+)
 from deebot_client.const import DataType
 from deebot_client.logging_filter import get_logger
-from deebot_client.message import HandlingResult, HandlingState, MessageStr
+from deebot_client.message import (
+    HandlingResult,
+    HandlingState,
+    MessageStr,
+)
 
 if TYPE_CHECKING:
     from deebot_client.event_bus import EventBus
@@ -86,3 +95,14 @@ class XmlSetCommand(ExecuteCommand, SetCommand, ABC):
 
     Command needs to be linked to the "get" command, for handling (updating) the sensors.
     """
+
+
+class XmlGetCommand(XmlCommandWithMessageHandling, GetCommand, ABC):
+    """Xml get command."""
+
+    @classmethod
+    @abstractmethod
+    def handle_set_args(
+        cls, event_bus: EventBus, args: dict[str, Any]
+    ) -> HandlingResult:
+        """Handle arguments of set command."""

deebot_client/commands/xml/fan_speed.py

@@ -2,24 +2,37 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any
 
+from deebot_client.command import InitParam
 from deebot_client.events import FanSpeedEvent, FanSpeedLevel
 from deebot_client.message import HandlingResult
 
-from .common import XmlCommandWithMessageHandling
+from .common import XmlGetCommand, XmlSetCommand
 
 if TYPE_CHECKING:
     from xml.etree.ElementTree import Element
 
     from deebot_client.event_bus import EventBus
 
 
-class GetFanSpeed(XmlCommandWithMessageHandling):
-    """GetFanSpeed command."""
+class GetCleanSpeed(XmlGetCommand):
+    """GetCleanSpeed command."""
 
     NAME = "GetCleanSpeed"
 
+    @classmethod
+    def handle_set_args(
+        cls, event_bus: EventBus, args: dict[str, Any]
+    ) -> HandlingResult:
+        """Handle message->body->data and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        event_bus.notify(FanSpeedEvent(FanSpeedLevel(int(args["speed"]))))
+        return HandlingResult.success()
+
     @classmethod
     def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
         """Handle xml message and notify the correct event subscribers.
@@ -33,12 +46,25 @@
 
         match speed.lower():
             case "standard":
-                event = FanSpeedEvent(FanSpeedLevel.NORMAL)
+                event = FanSpeedEvent(FanSpeedLevel.STANDARD)
             case "strong":
-                event = FanSpeedEvent(FanSpeedLevel.MAX)
+                event = FanSpeedEvent(FanSpeedLevel.STRONG)
 
         if event:
             event_bus.notify(event)
             return HandlingResult.success()
 
         return HandlingResult.analyse()
+
+
+class SetCleanSpeed(XmlSetCommand):
+    """Set clean speed command."""
+
+    NAME = "SetCleanSpeed"
+    get_command = GetCleanSpeed
+    _mqtt_params = MappingProxyType({"speed": InitParam(FanSpeedLevel)})
+
+    def __init__(self, speed: FanSpeedLevel | str) -> None:
+        if isinstance(speed, FanSpeedLevel):
+            speed = speed.name.lower()
+        super().__init__({"speed": speed})

deebot_client/events/fan_speed.py

@@ -14,8 +14,10 @@ class FanSpeedLevel(IntEnum):
 
     # Values should be sort from low to high on their meanings
     QUIET = 1000
+    STANDARD = -1
     NORMAL = 0
     MAX = 1
+    STRONG = -2
     MAX_PLUS = 2
 
 

tests/commands/xml/test_fan_speed.py

@@ -4,8 +4,8 @@
 
 import pytest
 
-from deebot_client.command import CommandResult
-from deebot_client.commands.xml import GetFanSpeed
+from deebot_client.command import CommandResult, CommandWithMessageHandling
+from deebot_client.commands.xml import GetCleanSpeed, SetCleanSpeed
 from deebot_client.events import FanSpeedEvent, FanSpeedLevel
 from deebot_client.message import HandlingState
 from tests.commands import assert_command
@@ -19,26 +19,44 @@
 @pytest.mark.parametrize(
     ("speed", "expected_event"),
     [
-        ("standard", FanSpeedEvent(FanSpeedLevel.NORMAL)),
-        ("strong", FanSpeedEvent(FanSpeedLevel.MAX)),
+        ("standard", FanSpeedEvent(FanSpeedLevel.STANDARD)),
+        ("strong", FanSpeedEvent(FanSpeedLevel.STRONG)),
     ],
     ids=["standard", "strong"],
 )
-async def test_get_fan_speed(speed: str, expected_event: Event) -> None:
+async def test_get_clean_speed(speed: str, expected_event: Event) -> None:
     json = get_request_xml(f"<ctl ret='ok' speed='{speed}'/>")
-    await assert_command(GetFanSpeed(), json, expected_event)
+    await assert_command(GetCleanSpeed(), json, expected_event)
 
 
 @pytest.mark.parametrize(
     "xml",
     ["<ctl ret='error'/>", "<ctl ret='ok' speed='invalid'/>"],
     ids=["error", "no_state"],
 )
-async def test_get_fan_speed_error(xml: str) -> None:
+async def test_get_clean_speed_error(xml: str) -> None:
     json = get_request_xml(xml)
     await assert_command(
-        GetFanSpeed(),
+        GetCleanSpeed(),
         json,
         None,
         command_result=CommandResult(HandlingState.ANALYSE_LOGGED),
     )
+
+
+@pytest.mark.parametrize(
+    ("command", "xml", "result"),
+    [
+        (
+            SetCleanSpeed(FanSpeedLevel.STRONG),
+            "<ctl ret='ok' />",
+            HandlingState.SUCCESS,
+        ),
+        (SetCleanSpeed("invalid"), "<ctl ret='error' />", HandlingState.FAILED),
+    ],
+)
+async def test_set_clean_speed(
+    command: CommandWithMessageHandling, xml: str, result: HandlingState
+) -> None:
+    json = get_request_xml(xml)
+    await assert_command(command, json, None, command_result=CommandResult(result))