Add XML Clean Commands

Author: nanomad

deebot_client/commands/xml/__init__.py

@@ -8,6 +8,7 @@
 
 from .charge import Charge
 from .charge_state import GetChargeState
+from .clean import Clean, CleanArea, GetCleanState
 from .error import GetError
 from .fan_speed import GetFanSpeed
 from .life_span import GetLifeSpan
@@ -20,7 +21,10 @@
 
 __all__ = [
     "Charge",
+    "Clean",
+    "CleanArea",
     "GetChargeState",
+    "GetCleanState",
     "GetCleanSum",
     "GetError",
     "GetFanSpeed",
@@ -32,7 +36,10 @@
 # fmt: off
 # ordered by file asc
 _COMMANDS: list[type[XmlCommand]] = [
+    Clean,
+    CleanArea,
     GetError,
+    GetCleanState,
     GetLifeSpan,
     PlaySound,
 ]

deebot_client/commands/xml/clean.py

@@ -0,0 +1,97 @@
+"""Clean commands."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.events import FanSpeedEvent, FanSpeedLevel, StateEvent
+from deebot_client.logging_filter import get_logger
+from deebot_client.message import HandlingResult
+from deebot_client.models import CleanAction, CleanMode, State
+
+from .common import ExecuteCommand, XmlCommandWithMessageHandling
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+
+_LOGGER = get_logger(__name__)
+
+
+class Clean(ExecuteCommand):
+    """Generic start/pause/stop cleaning command."""
+
+    NAME = "Clean"
+    HAS_SUB_ELEMENT = True
+
+    def __init__(
+        self, action: CleanAction, speed: FanSpeedLevel = FanSpeedLevel.NORMAL
+    ) -> None:
+        # <ctl><clean type='SpotArea' act='s' speed='standard' deep='1' mid='4,5'/></ctl>
+
+        super().__init__(
+            {
+                "type": CleanMode.AUTO.xml_value,
+                "act": action.xml_value,
+                "speed": speed.xml_value,
+            }
+        )
+
+
+class CleanArea(ExecuteCommand):
+    """Clean area command."""
+
+    NAME = "Clean"
+    HAS_SUB_ELEMENT = True
+
+    def __init__(
+        self,
+        mode: CleanMode,
+        area: str,
+        cleanings: int = 1,
+        speed: FanSpeedLevel = FanSpeedLevel.NORMAL,
+    ) -> None:
+        # <ctl><clean type='SpotArea' act='s' speed='standard' deep='1' mid='4,5'/></ctl>
+
+        super().__init__(
+            {
+                "type": mode.xml_value,
+                "act": CleanAction.START.xml_value,
+                "speed": speed.xml_value,
+                "deep": str(cleanings),
+                "mid": area,
+            }
+        )
+
+
+class GetCleanState(XmlCommandWithMessageHandling):
+    """GetCleanState command."""
+
+    NAME = "GetCleanState"
+
+    @classmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        if xml.attrib.get("ret") != "ok" or (clean := xml.find("clean")) is None:
+            return HandlingResult.analyse()
+
+        speed_attrib = clean.attrib.get("speed")
+        if speed_attrib is not None:
+            fan_speed_level = FanSpeedLevel.from_xml(speed_attrib)
+            event_bus.notify(FanSpeedEvent(fan_speed_level))
+
+        clean_attrib = clean.attrib.get("st")
+        if clean_attrib is not None:
+            clean_action = CleanAction.from_xml(clean_attrib)
+            if clean_action == CleanAction.START:
+                event_bus.notify(StateEvent(State.CLEANING))
+            elif clean_action == CleanAction.PAUSE:
+                event_bus.notify(StateEvent(State.PAUSED))
+            else:
+                _LOGGER.debug("Ignored CleanState %s", clean_action)
+
+        return HandlingResult.success()

tests/commands/xml/test_clean.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import pytest
+
+from deebot_client.command import CommandResult
+from deebot_client.commands.xml import GetCleanState
+from deebot_client.commands.xml.clean import CleanArea
+from deebot_client.events import FanSpeedEvent, FanSpeedLevel, StateEvent
+from deebot_client.message import HandlingState
+from deebot_client.models import CleanMode, State
+from tests.commands import assert_command
+
+from . import get_request_xml
+
+
+@pytest.mark.parametrize(
+    ("command", "command_result"),
+    [
+        (CleanArea(CleanMode.SPOT_AREA, "4", 1), HandlingState.SUCCESS),
+    ],
+)
+async def test_CleanArea(command: CleanArea, command_result: HandlingState) -> None:
+    json = get_request_xml("<ctl ret='ok'/>")
+    await assert_command(
+        command, json, None, command_result=CommandResult(command_result)
+    )
+
+
+@pytest.mark.parametrize(
+    ("speed", "state", "expected_fan_speed_event", "expected_state_event"),
+    [
+        (
+            "standard",
+            "s",
+            FanSpeedEvent(FanSpeedLevel.NORMAL),
+            StateEvent(State.CLEANING),
+        ),
+        (
+            "strong",
+            "s",
+            FanSpeedEvent(FanSpeedLevel.MAX),
+            StateEvent(State.CLEANING),
+        ),
+        (
+            "standard",
+            "p",
+            FanSpeedEvent(FanSpeedLevel.NORMAL),
+            StateEvent(State.PAUSED),
+        ),
+    ],
+    ids=["standard_cleaning", "strong_cleaning", "paused"],
+)
+async def test_get_clean_state(
+    speed: str,
+    state: str,
+    expected_fan_speed_event: FanSpeedEvent,
+    expected_state_event: StateEvent,
+) -> None:
+    json = get_request_xml(
+        f"<ctl ret='ok'><clean type='auto' speed='{speed}' st='{state}' t='0' a='0' s='0' tr=''/></ctl>"
+    )
+    await assert_command(
+        GetCleanState(),
+        json,
+        [x for x in [expected_fan_speed_event, expected_state_event] if x is not None],
+    )
+
+
+@pytest.mark.parametrize(
+    "xml",
+    ["<ctl ret='error'/>", "<ctl ret='ok'></ctl>"],
+    ids=["error", "no_state"],
+)
+async def test_get_clean_state_error(xml: str) -> None:
+    json = get_request_xml(xml)
+    await assert_command(
+        GetCleanState(),
+        json,
+        None,
+        command_result=CommandResult(HandlingState.ANALYSE_LOGGED),
+    )