Initial Ozmo 900 support

Author: nanomad

deebot_client/command.py

@@ -324,8 +324,15 @@ def _pop_or_raise(name: str, type_: type, data: dict[str, Any]) -> Any:
     try:
         return type_(value)
     except ValueError as err:
-        msg = f'Could not convert "{value}" of {name} into {type_}'
-        raise DeebotError(msg) from err
+        if hasattr(type_, "from_xml"):
+            try:
+                return type_.from_xml(value)
+            except ValueError as err2:
+                msg = f'Could not convert "{value}" of {name} into {type_}'
+                raise DeebotError(msg) from err2
+        else:
+            msg = f'Could not convert "{value}" of {name} into {type_}'
+            raise DeebotError(msg) from err
 
 
 class GetCommand(CommandWithMessageHandling, ABC):

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

@@ -6,10 +6,12 @@
 
 from deebot_client.command import Command, CommandMqttP2P
 
+from .battery import GetBatteryInfo
 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 .fan_speed import GetCleanSpeed, SetCleanSpeed
 from .life_span import GetLifeSpan
 from .play_sound import PlaySound
 from .pos import GetPos
@@ -20,20 +22,31 @@
 
 __all__ = [
     "Charge",
+    "Clean",
+    "CleanArea",
+    "GetBatteryInfo",
     "GetChargeState",
+    "GetCleanSpeed",
+    "GetCleanState",
     "GetCleanSum",
     "GetError",
-    "GetFanSpeed",
     "GetLifeSpan",
     "GetPos",
     "PlaySound",
+    "SetCleanSpeed",
 ]
 
 # fmt: off
 # ordered by file asc
 _COMMANDS: list[type[XmlCommand]] = [
+    Clean,
+    CleanArea,
     GetError,
+    GetBatteryInfo,
+    GetCleanSpeed,
+    GetCleanState,
     GetLifeSpan,
+    SetCleanSpeed,
     PlaySound,
 ]
 # fmt: on

deebot_client/commands/xml/battery.py

@@ -0,0 +1,40 @@
+"""Battery Info command."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.events import BatteryEvent
+from deebot_client.message import HandlingResult
+
+from .common import XmlCommandWithMessageHandling
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+
+
+class GetBatteryInfo(XmlCommandWithMessageHandling):
+    """GetBatteryInfo command."""
+
+    NAME = "GetBatteryInfo"
+
+    @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 (battery := xml.find("battery")) is None
+            or (power := battery.attrib.get("power")) is None
+        ):
+            return HandlingResult.analyse()
+
+        if power:
+            event_bus.notify(BatteryEvent(int(power)))
+            return HandlingResult.success()
+
+        return HandlingResult.analyse()

deebot_client/commands/xml/clean.py

@@ -0,0 +1,89 @@
+"""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:
+        # <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))
+        return HandlingResult.success()

deebot_client/commands/xml/common.py

@@ -3,12 +3,18 @@
 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,
+    CommandMqttP2P,
+    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
@@ -81,8 +87,41 @@ def _handle_xml(cls, _: EventBus, xml: Element) -> HandlingResult:
         return HandlingResult(HandlingState.FAILED)
 
 
-class XmlSetCommand(ExecuteCommand, SetCommand, ABC):
+class XmlCommandMqttP2P(XmlCommand, CommandMqttP2P, ABC):
+    """Json base command for mqtt p2p channel."""
+
+    @classmethod
+    def create_from_mqtt(cls, payload: str | bytes | bytearray) -> CommandMqttP2P:
+        """Create a command from the mqtt data."""
+        xml = ElementTree.fromstring(payload)
+        return cls._create_from_mqtt(xml.attrib)
+
+    def handle_mqtt_p2p(
+        self, event_bus: EventBus, response_payload: str | bytes | bytearray
+    ) -> None:
+        """Handle response received over the mqtt channel "p2p"."""
+        self._handle_mqtt_p2p(event_bus, str(response_payload))
+
+    @abstractmethod
+    def _handle_mqtt_p2p(
+        self, event_bus: EventBus, response: dict[str, Any] | str
+    ) -> None:
+        """Handle response received over the mqtt channel "p2p"."""
+
+
+class XmlSetCommand(ExecuteCommand, SetCommand, XmlCommandMqttP2P, ABC):
     """Xml base set command.
 
     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.from_xml(str(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.
@@ -29,16 +42,18 @@ def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
         if xml.attrib.get("ret") != "ok" or not (speed := xml.attrib.get("speed")):
             return HandlingResult.analyse()
 
-        event: FanSpeedEvent | None = None
+        event_bus.notify(FanSpeedEvent(FanSpeedLevel.from_xml(speed)))
+        return HandlingResult.success()
+
 
-        match speed.lower():
-            case "standard":
-                event = FanSpeedEvent(FanSpeedLevel.NORMAL)
-            case "strong":
-                event = FanSpeedEvent(FanSpeedLevel.MAX)
+class SetCleanSpeed(XmlSetCommand):
+    """SetCleanSpeed command."""
 
-        if event:
-            event_bus.notify(event)
-            return HandlingResult.success()
+    NAME = "SetCleanSpeed"
+    get_command = GetCleanSpeed
+    _mqtt_params = MappingProxyType({"speed": InitParam(FanSpeedLevel)})
 
-        return HandlingResult.analyse()
+    def __init__(self, speed: FanSpeedLevel | str) -> None:
+        if isinstance(speed, FanSpeedLevel):
+            speed = speed.xml_value
+        super().__init__({"speed": speed})

deebot_client/events/fan_speed.py

@@ -3,20 +3,22 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from enum import IntEnum, unique
+from enum import unique
+
+from deebot_client.util.enum import IntEnumWithXml
 
 from .base import Event
 
 
 @unique
-class FanSpeedLevel(IntEnum):
+class FanSpeedLevel(IntEnumWithXml):
     """Enum class for all possible fan speed levels."""
 
     # Values should be sort from low to high on their meanings
-    QUIET = 1000
-    NORMAL = 0
-    MAX = 1
-    MAX_PLUS = 2
+    QUIET = 1000, ""
+    NORMAL = 0, "standard"
+    MAX = 1, "strong"
+    MAX_PLUS = 2, ""
 
 
 @dataclass(frozen=True)