Add xml GetError

deebot_client/commands/json/error.py

@@ -1,6 +1,7 @@
 """Error commands."""
 from typing import Any
 
+from deebot_client.const import ERROR_CODES
 from deebot_client.event_bus import EventBus
 from deebot_client.events import ErrorEvent, StateEvent
 from deebot_client.message import HandlingResult, MessageBodyDataDict
@@ -28,54 +29,10 @@ def _handle_body_data_dict(
             error = codes[-1]
 
             if error is not None:
-                description = _ERROR_CODES.get(error)
+                description = ERROR_CODES.get(error)
                 if error != 0:
                     event_bus.notify(StateEvent(State.ERROR))
                 event_bus.notify(ErrorEvent(error, description))
                 return HandlingResult.success()
 
         return HandlingResult.analyse()
-
-
-# from https://github.com/mrbungle64/ecovacs-deebot.js/blob/master/library/errorCodes.json
-_ERROR_CODES = {
-    -3: "Error parsing response data",
-    -2: "Internal error",
-    -1: "Host not reachable or communication malfunction",
-    0: "NoError: Robot is operational",
-    3: "RequestOAuthError: Authentication error",
-    7: "log data is not found",
-    100: "NoError: Robot is operational",
-    101: "BatteryLow: Low battery",
-    102: "HostHang: Robot is off the floor",
-    103: "WheelAbnormal: Driving Wheel malfunction",
-    104: "DownSensorAbnormal: Excess dust on the Anti-Drop Sensors",
-    105: "Stuck: Robot is stuck",
-    106: "SideBrushExhausted: Side Brushes have expired",
-    107: "DustCaseHeapExhausted: Dust case filter expired",
-    108: "SideAbnormal: Side Brushes are tangled",
-    109: "RollAbnormal: Main Brush is tangled",
-    110: "NoDustBox: Dust Bin Not installed",
-    111: "BumpAbnormal: Bump sensor stuck",
-    112: 'LDS: LDS "Laser Distance Sensor" malfunction',
-    113: "MainBrushExhausted: Main brush has expired",
-    114: "DustCaseFilled: Dust bin full",
-    115: "BatteryError:",
-    116: "ForwardLookingError:",
-    117: "GyroscopeError:",
-    118: "StrainerBlock:",
-    119: "FanError:",
-    120: "WaterBoxError:",
-    201: "AirFilterUninstall:",
-    202: "UltrasonicComponentAbnormal",
-    203: "SmallWheelError",
-    204: "WheelHang",
-    205: "IonSterilizeExhausted",
-    206: "IonSterilizeAbnormal",
-    207: "IonSterilizeFault",
-    312: "Please replace the Dust Bag.",
-    404: "Recipient unavailable",
-    500: "Request Timeout",
-    601: "ERROR_ClosedAIVISideAbnormal",
-    602: "ClosedAIVIRollAbnormal",
-}

deebot_client/commands/xml/__init__.py

@@ -1 +1,27 @@
-"""Xml commands."""
+"""Xml commands module."""
+from deebot_client.command import Command, CommandMqttP2P
+
+from .common import XmlCommand
+from .error import GetError
+
+__all__ = [
+    "GetError",
+]
+
+# fmt: off
+# ordered by file asc
+_COMMANDS: list[type[XmlCommand]] = [
+    GetError,
+]
+# fmt: on
+
+COMMANDS: dict[str, type[Command]] = {
+    cmd.name: cmd  # type: ignore[misc]
+    for cmd in _COMMANDS
+}
+
+COMMANDS_WITH_MQTT_P2P_HANDLING: dict[str, type[CommandMqttP2P]] = {
+    cmd_name: cmd
+    for (cmd_name, cmd) in COMMANDS.items()
+    if issubclass(cmd, CommandMqttP2P)
+}

deebot_client/commands/xml/common.py

@@ -1,31 +1,62 @@
 """Common xml based commands."""
+from abc import ABC, abstractmethod
+from typing import cast
+from xml.etree.ElementTree import Element, SubElement
 
+from defusedxml import ElementTree  # type: ignore[import-untyped]
 
-from xml.etree import ElementTree
-
-from deebot_client.command import Command
+from deebot_client.command import Command, CommandWithMessageHandling
 from deebot_client.const import DataType
+from deebot_client.event_bus import EventBus
+from deebot_client.logging_filter import get_logger
+from deebot_client.message import HandlingResult, MessageStr
+
+_LOGGER = get_logger(__name__)
 
 
 class XmlCommand(Command):
     """Xml command."""
 
     data_type: DataType = DataType.XML
 
-    @property
-    def has_sub_element(self) -> bool:
+    @property  # type: ignore[misc]
+    @classmethod
+    def has_sub_element(cls) -> bool:
         """Return True if command has inner element."""
         return False
 
     def _get_payload(self) -> str:
-        element = ctl_element = ElementTree.Element("ctl")
+        element = ctl_element = Element("ctl")
 
         if len(self._args) > 0:
             if self.has_sub_element:
-                element = ElementTree.SubElement(element, self.name.lower())
+                element = SubElement(element, self.name.lower())
 
             if isinstance(self._args, dict):
                 for key, value in self._args.items():
                     element.set(key, value)
 
-        return ElementTree.tostring(ctl_element, "unicode")
+        return cast(str, ElementTree.tostring(ctl_element, "unicode"))
+
+
+class XmlCommandWithMessageHandling(
+    XmlCommand, CommandWithMessageHandling, MessageStr, ABC
+):
+    """Xml command, which handle response by itself."""
+
+    @classmethod
+    @abstractmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+
+    @classmethod
+    def _handle_str(cls, event_bus: EventBus, message: str) -> HandlingResult:
+        """Handle string message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        xml = ElementTree.fromstring(message)
+        return cls._handle_xml(event_bus, xml)

deebot_client/commands/xml/error.py

@@ -0,0 +1,31 @@
+"""Error commands."""
+from xml.etree.ElementTree import Element
+
+from deebot_client.const import ERROR_CODES
+from deebot_client.event_bus import EventBus
+from deebot_client.events import ErrorEvent, StateEvent
+from deebot_client.message import HandlingResult
+from deebot_client.models import State
+
+from .common import XmlCommandWithMessageHandling
+
+
+class GetError(XmlCommandWithMessageHandling):
+    """Get error command."""
+
+    name = "GetError"
+
+    @classmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        error_code = int(errs) if (errs := xml.attrib["errs"]) else 0
+
+        if error_code != 0:
+            event_bus.notify(StateEvent(State.ERROR))
+
+        description = ERROR_CODES.get(error_code)
+        event_bus.notify(ErrorEvent(error_code, description))
+        return HandlingResult.success()

deebot_client/const.py

@@ -26,3 +26,47 @@ def get(cls, value: str) -> Self | None:
             return cls(value.lower())
         except ValueError:
             return None
+
+
+# from https://github.com/mrbungle64/ecovacs-deebot.js/blob/master/library/errorCodes.json
+ERROR_CODES = {
+    -3: "Error parsing response data",
+    -2: "Internal error",
+    -1: "Host not reachable or communication malfunction",
+    0: "NoError: Robot is operational",
+    3: "RequestOAuthError: Authentication error",
+    7: "log data is not found",
+    100: "NoError: Robot is operational",
+    101: "BatteryLow: Low battery",
+    102: "HostHang: Robot is off the floor",
+    103: "WheelAbnormal: Driving Wheel malfunction",
+    104: "DownSensorAbnormal: Excess dust on the Anti-Drop Sensors",
+    105: "Stuck: Robot is stuck",
+    106: "SideBrushExhausted: Side Brushes have expired",
+    107: "DustCaseHeapExhausted: Dust case filter expired",
+    108: "SideAbnormal: Side Brushes are tangled",
+    109: "RollAbnormal: Main Brush is tangled",
+    110: "NoDustBox: Dust Bin Not installed",
+    111: "BumpAbnormal: Bump sensor stuck",
+    112: 'LDS: LDS "Laser Distance Sensor" malfunction',
+    113: "MainBrushExhausted: Main brush has expired",
+    114: "DustCaseFilled: Dust bin full",
+    115: "BatteryError:",
+    116: "ForwardLookingError:",
+    117: "GyroscopeError:",
+    118: "StrainerBlock:",
+    119: "FanError:",
+    120: "WaterBoxError:",
+    201: "AirFilterUninstall:",
+    202: "UltrasonicComponentAbnormal",
+    203: "SmallWheelError",
+    204: "WheelHang",
+    205: "IonSterilizeExhausted",
+    206: "IonSterilizeAbnormal",
+    207: "IonSterilizeFault",
+    312: "Please replace the Dust Bag.",
+    404: "Recipient unavailable",
+    500: "Request Timeout",
+    601: "ERROR_ClosedAIVISideAbnormal",
+    602: "ClosedAIVIRollAbnormal",
+}

deebot_client/message.py

@@ -99,6 +99,38 @@
         return cls._handle(event_bus, message)
 
 
+class MessageStr(Message):
+    """String message."""
+
+    @classmethod
+    @abstractmethod
+    def _handle_str(cls, event_bus: EventBus, message: str) -> HandlingResult:
+        """Handle string message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+
+    @classmethod
+    # @_handle_error_or_analyse @edenhaus will make the decorator to work again
+    @final
+    def __handle_str(cls, event_bus: EventBus, message: str) -> HandlingResult:
+        return cls._handle_str(event_bus, message)
+
+    @classmethod
+    def _handle(
+        cls, event_bus: EventBus, message: dict[str, Any] | str
+    ) -> HandlingResult:
+        """Handle message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        # This basically means an XML message
+        if isinstance(message, str):
+            return cls.__handle_str(event_bus, message)
+
+        return super()._handle(event_bus, message)
+
+
 class MessageBody(Message):
     """Dict message with body attribute."""
 

tests/commands/json/test_error.py

@@ -0,0 +1,23 @@
+from collections.abc import Sequence
+
+import pytest
+
+from deebot_client.commands.json import GetError
+from deebot_client.events import ErrorEvent, StateEvent
+from deebot_client.events.base import Event
+from deebot_client.models import State
+from tests.helpers import get_request_json, get_success_body
+
+from . import assert_command
+
+
+@pytest.mark.parametrize(
+    ("code", "expected_events"),
+    [
+        (0, ErrorEvent(0, "NoError: Robot is operational")),
+        (105, [StateEvent(State.ERROR), ErrorEvent(105, "Stuck: Robot is stuck")]),
+    ],
+)
+async def test_getErrors(code: int, expected_events: Event | Sequence[Event]) -> None:
+    json = get_request_json(get_success_body({"code": [code]}))
+    await assert_command(GetError(), json, expected_events)

tests/commands/xml/__init__.py

@@ -0,0 +1,13 @@
+from functools import partial
+from typing import Any
+
+from deebot_client.hardware.deebot import get_static_device_info
+from tests.commands import assert_command as assert_command_base
+
+assert_command = partial(
+    assert_command_base, static_device_info=get_static_device_info("ls1ok3")
+)
+
+
+def get_request_xml(data: str | None) -> dict[str, Any]:
+    return {"id": "ALZf", "ret": "ok", "resp": data, "payloadType": "x"}

tests/commands/xml/test_error.py

@@ -0,0 +1,22 @@
+from collections.abc import Sequence
+
+import pytest
+
+from deebot_client.commands.xml import GetError
+from deebot_client.events import ErrorEvent, StateEvent
+from deebot_client.events.base import Event
+from deebot_client.models import State
+
+from . import assert_command, get_request_xml
+
+
+@pytest.mark.parametrize(
+    ("errs", "expected_events"),
+    [
+        ("", ErrorEvent(0, "NoError: Robot is operational")),
+        ("105", [StateEvent(State.ERROR), ErrorEvent(105, "Stuck: Robot is stuck")]),
+    ],
+)
+async def test_getErrors(errs: str, expected_events: Event | Sequence[Event]) -> None:
+    json = get_request_xml(f"<ctl ret='ok' errs='{errs}'/>")
+    await assert_command(GetError(), json, expected_events)