Add XML Clean Commands

deebot_client/commands/xml/__init__.py

@@ -9,6 +9,7 @@
 from .battery import GetBatteryInfo
 from .charge import Charge
 from .charge_state import GetChargeState
+from .clean import Clean, CleanArea, GetCleanState
 from .clean_logs import GetCleanLogs
 from .clean_speed import GetCleanSpeed, SetCleanSpeed
 from .error import GetError
@@ -23,11 +24,14 @@
 
 __all__ = [
     "Charge",
+    "Clean",
+    "CleanArea",
     "GetBatteryInfo",
     "GetChargeState",
     "GetChargerPos",
     "GetCleanLogs",
     "GetCleanSpeed",
+    "GetCleanState",
     "GetCleanSum",
     "GetError",
     "GetLifeSpan",
@@ -51,6 +55,10 @@
 
     Charge,
 
+    Clean,
+    CleanArea,
+    GetCleanState,
+
     GetCleanLogs,
 
     GetCleanSpeed,

deebot_client/commands/xml/clean.py

@@ -0,0 +1,90 @@
+"""Clean commands."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.events import FanSpeedEvent, FanSpeedLevel, StateEvent
+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
+
+
+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:
+        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:
+        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))
+            elif clean_action in (CleanAction.RESUME, CleanAction.STOP):
+                event_bus.notify(StateEvent(State.IDLE))
+
+        return HandlingResult.success()

tests/commands/xml/test_clean.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from deebot_client.command import CommandResult
+from deebot_client.commands.xml import Clean, CleanArea, GetCleanState
+from deebot_client.events import FanSpeedEvent, FanSpeedLevel, StateEvent
+from deebot_client.message import HandlingState
+from deebot_client.models import CleanAction, CleanMode, State
+from tests.commands import assert_command
+
+from . import get_request_xml
+
+if TYPE_CHECKING:
+    from deebot_client.events.base import Event
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        Clean(CleanAction.START, speed=FanSpeedLevel.MAX),
+        Clean(CleanAction.PAUSE),
+    ],
+)
+async def test_Clean(command: Clean) -> None:
+    json = get_request_xml("<ctl ret='ok'/>")
+    await assert_command(
+        command, json, None, command_result=CommandResult(HandlingState.SUCCESS)
+    )
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        CleanArea(CleanMode.SPOT_AREA, "4", 1),
+    ],
+)
+async def test_CleanArea(command: CleanArea) -> None:
+    json = get_request_xml("<ctl ret='ok'/>")
+    await assert_command(
+        command, json, None, command_result=CommandResult(HandlingState.SUCCESS)
+    )
+
+
+@pytest.mark.parametrize(
+    ("params", "expected_events"),
+    [
+        (
+            "speed='standard' st='s'",
+            [FanSpeedEvent(FanSpeedLevel.NORMAL), StateEvent(State.CLEANING)],
+        ),
+        (
+            "speed='strong' st='s'",
+            [FanSpeedEvent(FanSpeedLevel.MAX), StateEvent(State.CLEANING)],
+        ),
+        (
+            "speed='standard' st='p'",
+            [FanSpeedEvent(FanSpeedLevel.NORMAL), StateEvent(State.PAUSED)],
+        ),
+        (
+            "st='r'",
+            [StateEvent(State.IDLE)],
+        ),
+        (
+            "st='h'",
+            [StateEvent(State.IDLE)],
+        ),
+        (
+            "speed='strong'",
+            [FanSpeedEvent(FanSpeedLevel.MAX)],
+        ),
+    ],
+    ids=[
+        "standard_cleaning",
+        "strong_cleaning",
+        "paused",
+        "resume/idle",
+        "stop/idle",
+        "fanspeed_only",
+    ],
+)
+async def test_get_clean_state(
+    params: str,
+    expected_events: list[Event],
+) -> None:
+    json = get_request_xml(
+        f"<ctl ret='ok'><clean type='auto' {params} t='0' a='0' s='0' tr=''/></ctl>"
+    )
+    await assert_command(
+        GetCleanState(),
+        json,
+        expected_events,
+    )
+
+
+@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),
+    )