Add more XML commands and messages

Author: nanomad

deebot_client/commands/xml/__init__.py

@@ -44,18 +44,22 @@
 # fmt: off
 # ordered by file asc
 _COMMANDS: list[type[XmlCommand]] = [
+    Charge,
     Clean,
     CleanArea,
-    GetError,
     GetBatteryInfo,
+    GetChargeState,
     GetCleanLogs,
     GetCleanSpeed,
     GetCleanState,
+    GetCleanSum,
+    GetError,
     GetLifeSpan,
+    GetPos,
     GetWaterBoxInfo,
     GetWaterPermeability,
-    SetCleanSpeed,
     PlaySound,
+    SetCleanSpeed,
 ]
 # fmt: on
 

deebot_client/commands/xml/clean_logs.py

@@ -6,14 +6,15 @@
 
 from deebot_client.command import CommandResult
 from deebot_client.events import (
-    CleanJobStatus,
     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
@@ -46,14 +47,24 @@ def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
         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=CleanJobStatus.FINISHED,  # To be extracted
+                            stop_reason=stop_reason.to_clean_job_status(),  # To be extracted
                             duration=int(log.attrib["l"]),
                         )
                     )

deebot_client/commands/xml/enum.py

@@ -0,0 +1,35 @@
+"""Enums used only by XML messages."""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+from deebot_client.events import CleanJobStatus
+
+
+class XmlStopReason(StrEnum):
+    """Reasons why cleaning has been stopped."""
+
+    FINISHED = "s"
+    BATTERY_LOW = "r"
+    STOPPED_BY_APP = "a"
+    STOPPED_BY_REMOTE_CONTROL = "i"
+    STOPPED_BY_BUTTON = "b"
+    STOPPED_BY_WARNING = "w"
+    STOPPED_BY_NO_DISTURB = "f"
+    STOPPED_BY_CLEARMAP = "m"
+    STOPPED_BY_NO_PATH = "n"
+    STOPPED_BY_NOT_IN_MAP = "u"
+    STOPPED_BY_VIRTUAL_WALL = "v"
+
+    def to_clean_job_status(self) -> CleanJobStatus:
+        """Convert this value to a CleanJobStatus for compatibility proposes."""
+        if self == XmlStopReason.FINISHED:
+            return CleanJobStatus.FINISHED
+        if self in (
+            XmlStopReason.STOPPED_BY_APP,
+            XmlStopReason.STOPPED_BY_REMOTE_CONTROL,
+            XmlStopReason.STOPPED_BY_BUTTON,
+        ):
+            return CleanJobStatus.MANUALLY_STOPPED
+        return CleanJobStatus.FINISHED_WITH_WARNINGS

deebot_client/messages/xml/__init__.py

@@ -6,9 +6,15 @@
 
 from deebot_client.messages.xml.battery import BatteryInfo
 from deebot_client.messages.xml.charge import ChargeState
-from deebot_client.messages.xml.clean import CleanReport, CleanSt
+from deebot_client.messages.xml.clean import (
+    CleanedPos,
+    CleanReport,
+    CleanReportServer,
+    CleanSt,
+)
 from deebot_client.messages.xml.map import MapP
 from deebot_client.messages.xml.pos import Pos
+from deebot_client.messages.xml.sleep import SleepStatus
 from deebot_client.messages.xml.water_info import WaterBoxInfo
 
 if TYPE_CHECKING:
@@ -18,9 +24,12 @@
     "BatteryInfo",
     "ChargeState",
     "CleanReport",
+    "CleanReportServer",
     "CleanSt",
+    "CleanedPos",
     "MapP",
     "Pos",
+    "SleepStatus",
     "WaterBoxInfo",
 ]
 # fmt: off
@@ -29,10 +38,13 @@
     BatteryInfo,
     ChargeState,
     CleanReport,
-    WaterBoxInfo,
-    Pos,
+    CleanReportServer,
+    CleanSt,
+    CleanedPos,
     MapP,
-    CleanSt
+    Pos,
+    SleepStatus,
+    WaterBoxInfo,
 ]
 # fmt: on
 

deebot_client/messages/xml/clean.py

@@ -4,11 +4,18 @@
 
 from typing import TYPE_CHECKING
 
-from deebot_client.events import FanSpeedEvent, FanSpeedLevel, StateEvent
+from deebot_client.events import (
+    FanSpeedEvent,
+    FanSpeedLevel,
+    Position,
+    PositionsEvent,
+    StateEvent,
+)
 from deebot_client.logging_filter import get_logger
 from deebot_client.message import HandlingResult
 from deebot_client.messages.xml.common import XmlMessage
 from deebot_client.models import CleanAction, State
+from deebot_client.rs.map import PositionType
 
 if TYPE_CHECKING:
     from xml.etree.ElementTree import Element
@@ -43,6 +50,8 @@ class CleanReport(XmlMessage):
     def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
         """Handle xml message and notify the correct event subscribers.
 
+        b"<ctl ts='1744467249311' td='CleanReport'><clean type='auto' speed='standard' st='s' rsn='a' a='' l='' sts=''/></ctl>"
+
         :return: A message response
         """
         if (clean := xml.find("clean")) is None:
@@ -64,3 +73,45 @@ def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
                 _LOGGER.debug("Ignored CleanState %s", clean_action)
 
         return HandlingResult.success()
+
+
+class CleanReportServer(XmlMessage):
+    """CleanReportServer message."""
+
+    NAME = "CleanReportServer"
+
+    @classmethod
+    def _handle_xml(cls, _event_bus: EventBus, _xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        b"<ctl ts='1744467262312' td='CleanReportServer' act='s' type='auto' cs='1134230540'/>"
+        b"<ctl ts='1744467393682' td='CleanReportServer' act='h' type='auto' sts='1744467262' cs='1134230540' area='1' last='76' mapCount='6'/>"
+
+        :return: A message response
+        """
+        return HandlingResult.analyse()
+
+
+class CleanedPos(XmlMessage):
+    """CleanedPos message."""
+
+    NAME = "CleanedPos"
+
+    @classmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        b"<ctl ts='1744467393682' td='CleanedPos' t='p' p='-2450,-996' a='-88' csid='1134230540'/>"
+
+        :return: A message response
+        """
+        if p := xml.attrib.get("p"):
+            p_x, p_y = p.split(",", 2)
+            p_a = xml.attrib.get("a", 0)
+            position = Position(
+                type=PositionType.DEEBOT, x=int(p_x), y=int(p_y), a=int(p_a)
+            )
+            event_bus.notify(PositionsEvent(positions=[position]))
+            return HandlingResult.success()
+
+        return HandlingResult.analyse()

deebot_client/messages/xml/sleep.py

@@ -0,0 +1,29 @@
+"""Sleep messages."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.message import HandlingResult
+from deebot_client.messages.xml.common import XmlMessage
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+
+
+class SleepStatus(XmlMessage):
+    """SleepStatus message."""
+
+    NAME = "SleepStatus"
+
+    @classmethod
+    def _handle_xml(cls, _event_bus: EventBus, _xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        b"<ctl ts='1744467249545' td='SleepStatus' st='0'/>"
+
+        :return: A message response
+        """
+        return HandlingResult.analyse()