OZMO 905 Support

.pre-commit-config.yaml

@@ -32,7 +32,7 @@ repos:
     hooks:
       - id: codespell
         args:
-          - --ignore-words-list=deebot
+          - --ignore-words-list=deebot,MapP
           - --skip="./.*,*.csv,*.json"
           - --quiet-level=2
           - --exclude-file=deebot_client/util/continents.py

deebot_client/capabilities.py

@@ -175,9 +175,9 @@ class CapabilityMap:
     changed: CapabilityEvent[MapChangedEvent]
     clear: CapabilityExecute[[]] | None = None
     major: CapabilityEvent[MajorMapEvent]
-    multi_state: CapabilitySetEnable[MultimapStateEvent]
+    multi_state: CapabilitySetEnable[MultimapStateEvent] | None = None
     position: CapabilityEvent[PositionsEvent]
-    relocation: CapabilityExecute[[]]
+    relocation: CapabilityExecute[[]] | None = None
     rooms: CapabilityEvent[RoomsEvent]
     trace: CapabilityEvent[MapTraceEvent]
 
@@ -213,7 +213,7 @@ class CapabilitySettings:
     sweep_mode: CapabilitySetEnable[SweepModeEvent] | None = None
     true_detect: CapabilitySetEnable[TrueDetectEvent] | None = None
     voice_assistant: CapabilitySetEnable[VoiceAssistantStateEvent] | None = None
-    volume: CapabilitySet[VolumeEvent, [int]]
+    volume: CapabilitySet[VolumeEvent, [int]] | None = None
 
 
 @dataclass(frozen=True, kw_only=True)

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/json/common.py

@@ -23,6 +23,7 @@
     HandlingState,
     MessageBody,
     MessageBodyDataDict,
+    MessageDict,
 )
 from deebot_client.util import verify_required_class_variables_exists
 
@@ -62,6 +63,12 @@ class JsonCommandWithMessageHandling(
     """Command, which handle response by itself."""
 
 
+class JsonCommandWithRawMessageHandling(
+    JsonCommand, CommandWithMessageHandling, MessageDict, ABC
+):
+    """Command, which handle raw response by itself."""
+
+
 class ExecuteCommand(JsonCommandWithMessageHandling, ABC):
     """Command, which is executing something (ex. Charge)."""
 

deebot_client/commands/xml/__init__.py

@@ -9,35 +9,71 @@
 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 .error import GetError
-from .fan_speed import GetFanSpeed
+from .fan_speed import GetCleanSpeed, SetCleanSpeed
 from .life_span import GetLifeSpan
+from .map import GetMapM, GetMapSet, GetMapSt, GetTrM, PullM, PullMP
 from .play_sound import PlaySound
 from .pos import GetPos
 from .stats import GetCleanSum
+from .water_info import GetWaterBoxInfo, GetWaterPermeability
 
 if TYPE_CHECKING:
     from .common import XmlCommand
 
 __all__ = [
     "Charge",
+    "Clean",
+    "CleanArea",
     "GetBatteryInfo",
     "GetChargeState",
+    "GetCleanLogs",
+    "GetCleanSpeed",
+    "GetCleanState",
     "GetCleanSum",
     "GetError",
-    "GetFanSpeed",
     "GetLifeSpan",
+    "GetMapM",
+    "GetMapSet",
+    "GetMapSt",
     "GetPos",
+    "GetTrM",
+    "GetWaterBoxInfo",
+    "GetWaterPermeability",
     "PlaySound",
+    "PullM",
+    "PullMP",
+    "SetCleanSpeed",
 ]
 
 # fmt: off
 # ordered by file asc
 _COMMANDS: list[type[XmlCommand]] = [
+    Charge,
+    Clean,
+    CleanArea,
+    GetError,
     GetBatteryInfo,
+    GetChargeState,
+    GetCleanLogs,
+    GetCleanSpeed,
+    GetCleanState,
+    GetCleanSum,
     GetError,
     GetLifeSpan,
+    GetMapM,
+    GetMapSet,
+    GetMapSt,
+    GetPos,
+    GetTrM,
+    GetWaterBoxInfo,
+    GetWaterPermeability,
     PlaySound,
+    PullM,
+    PullMP,
+    SetCleanSpeed,
 ]
 # fmt: on
 

deebot_client/commands/xml/charge_state.py

@@ -38,7 +38,7 @@ def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
                 case "slotcharging" | "slot_charging" | "wirecharging":
                     status = State.DOCKED
                 case "idle":
-                    status = State.IDLE
+                    pass
                 case "going":
                     status = State.RETURNING
                 case _:

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()

deebot_client/commands/xml/clean_logs.py

@@ -0,0 +1,75 @@
+"""Clean Logs commands."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.command import CommandResult
+from deebot_client.events import (
+    CleanLogEntry,
+    CleanLogEvent,
+)
+from deebot_client.logging_filter import get_logger
+from deebot_client.message import HandlingResult
+from deebot_client.util import get_enum
+
+from .common import XmlCommandWithMessageHandling
+from .enum import XmlStopReason
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+
+_LOGGER = get_logger(__name__)
+
+
+class GetCleanLogs(XmlCommandWithMessageHandling):
+    """GetCleanLogs command."""
+
+    NAME = "GetCleanLogs"
+
+    def __init__(self, count: int = 0) -> None:
+        super().__init__({"count": str(count)})
+
+    @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 (resp_logs := xml.findall("CleanSt")) is None
+        ):
+            return HandlingResult.analyse()
+
+        if len(resp_logs) >= 0:
+            logs: list[CleanLogEntry] = []
+            for log in resp_logs:
+                xml_stop_reason_attrib = str(log.attrib["f"])
+                stop_reason = XmlStopReason.FINISHED
+                try:
+                    stop_reason = get_enum(XmlStopReason, xml_stop_reason_attrib)
+                except Exception as e:
+                    _LOGGER.error(
+                        "Could not decode stop reason: %s",
+                        xml_stop_reason_attrib,
+                        exc_info=e,
+                    )
+                try:
+                    logs.append(
+                        CleanLogEntry(
+                            timestamp=int(log.attrib["s"]),
+                            image_url="",  # Missing
+                            type=log.attrib["t"],
+                            area=int(log.attrib["a"]),
+                            stop_reason=stop_reason.to_clean_job_status(),  # To be extracted
+                            duration=int(log.attrib["l"]),
+                        )
+                    )
+                except Exception:  # pylint: disable=broad-except
+                    _LOGGER.warning("Skipping log entry: %s", log, exc_info=True)
+            event_bus.notify(CleanLogEvent(logs))
+            return CommandResult.success()
+        return HandlingResult.analyse()