Add support for MQTT + XML based devices

deebot_client/command.py

@@ -67,7 +67,7 @@ async def execute(
 
         Returns
         -------
-            bot_reached (bool): True if the command was targeting the bot and it responded in time. False otherwise.
+            bot_reached (bool): True if the command was targeting the bot, and it responded in time. False otherwise.
                                 This value is not indicating if the command was executed successfully.
         """
         try:

deebot_client/commands/json/error.py

@@ -37,7 +37,7 @@ def _handle_body_data_dict(
         return HandlingResult.analyse()
 
 
-# from https://github.com/mrbungle64/ecovacs-deebot.js/blob/master/library/errorCodes.js
+# from https://github.com/mrbungle64/ecovacs-deebot.js/blob/master/library/errorCodes.json
 _ERROR_CODES = {
     -3: "Error parsing response data",
     -2: "Internal error",

deebot_client/commands/xml/__init__.py

@@ -0,0 +1,158 @@
+"""Commands module."""
+from deebot_client.command import Command, CommandMqttP2P
+
+# from .advanced_mode import GetAdvancedMode, SetAdvancedMode
+from .battery import GetBattery
+
+# from .carpet import GetCarpetAutoFanBoost, SetCarpetAutoFanBoost
+from .charge import Charge
+
+# from .charge_state import GetChargeState
+# from .clean import Clean, CleanArea, GetCleanInfo
+# from .clean_count import GetCleanCount, SetCleanCount
+from .clean_logs import GetCleanLogs
+
+# from .clean_preference import GetCleanPreference, SetCleanPreference
+from .common import XmlCommand
+
+# from .continuous_cleaning import GetContinuousCleaning, SetContinuousCleaning
+from .error import GetError
+
+# from .fan_speed import GetFanSpeed, SetFanSpeed
+# from .life_span import GetLifeSpan, ResetLifeSpan
+from .map import (
+    GetCachedMapInfo,
+    GetMajorMap,
+    GetMapSet,
+    GetMapSubSet,
+    GetMapTrace,
+    GetMinorMap,
+)
+
+# from .multimap_state import GetMultimapState, SetMultimapState
+from .play_sound import PlaySound
+
+# from .pos import GetPos
+from .relocation import SetRelocationState
+from .stats import GetStats, GetTotalStats
+
+# from .true_detect import GetTrueDetect, SetTrueDetect
+# from .volume import GetVolume, SetVolume
+# from .water_info import GetWaterInfo, SetWaterInfo
+
+__all__ = [
+    # "GetAdvancedMode",
+    # "SetAdvancedMode",
+    "GetBattery",
+    #     "GetCarpetAutoFanBoost",
+    #     "SetCarpetAutoFanBoost",
+    #     "GetCleanCount",
+    #     "SetCleanCount",
+    #     "GetCleanPreference",
+    #     "SetCleanPreference",
+    "Charge",
+    #     "GetChargeState",
+    #     "Clean",
+    #     "CleanArea",
+    #     "GetCleanInfo",
+    "GetCleanLogs",
+    # "GetContinuousCleaning",
+    # "SetContinuousCleaning",
+    "GetError",
+    "GetCachedMapInfo",
+    "GetMajorMap",
+    "GetMapSet",
+    "GetMapSubSet",
+    "GetMapTrace",
+    "GetMinorMap",
+    # "GetMultimapState",
+    # "SetMultimapState",
+    "PlaySound",
+    "SetRelocationState",
+    "GetStats",
+    "GetTotalStats",
+    # "GetTrueDetect",
+    # "SetTrueDetect",
+    # "GetVolume",
+    # "SetVolume",
+    # "GetWaterInfo",
+    # "SetWaterInfo",
+]
+
+# fmt: off
+# ordered by file asc
+_COMMANDS: list[type[XmlCommand]] = [
+    # GetAdvancedMode,
+    # SetAdvancedMode,
+
+    GetBattery,
+
+#     GetCarpetAutoFanBoost,
+#     SetCarpetAutoFanBoost,
+
+#     GetCleanCount,
+#     SetCleanCount,
+
+    # GetCleanPreference,
+    # SetCleanPreference,
+
+    Charge,
+
+#     GetChargeState,
+
+    # Clean,
+    # CleanArea,
+    # GetCleanInfo,
+
+    GetCleanLogs,
+
+    # GetContinuousCleaning,
+    # SetContinuousCleaning,
+
+    GetError,
+
+    # GetFanSpeed,
+    # SetFanSpeed,
+    #
+    # GetLifeSpan,
+    # ResetLifeSpan,
+
+    GetCachedMapInfo,
+    GetMajorMap,
+    GetMapSet,
+    GetMapSubSet,
+    GetMapTrace,
+    GetMinorMap,
+
+    # GetMultimapState,
+    # SetMultimapState,
+
+    PlaySound,
+
+    # GetPos,
+
+    SetRelocationState,
+
+    GetStats,
+    GetTotalStats,
+
+#     GetTrueDetect,
+#     SetTrueDetect,
+
+#     GetVolume,
+#     SetVolume,
+
+#     GetWaterInfo,
+#     SetWaterInfo,
+]
+# fmt: on
+
+COMMANDS: dict[str, type[Command]] = {
+    cmd.name: cmd for cmd in _COMMANDS  # type: ignore[misc]
+}
+
+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/battery.py

@@ -0,0 +1,14 @@
+"""Battery commands."""
+from deebot_client.messages.xml import OnBattery
+
+from .common import CommandWithMessageHandling
+
+
+class GetBattery(OnBattery, CommandWithMessageHandling):
+    """Get battery command."""
+
+    name = "GetBatteryInfo"
+
+    def __init__(self, is_available_check: bool = False) -> None:
+        super().__init__()
+        self._is_available_check = is_available_check

deebot_client/commands/xml/charge.py

@@ -0,0 +1,63 @@
+"""Charge commands."""
+from typing import Any
+
+from defusedxml import ElementTree
+
+from deebot_client.authentication import Authenticator
+from deebot_client.command import CommandResult
+from deebot_client.event_bus import EventBus
+from deebot_client.events import StateEvent
+from deebot_client.logging_filter import get_logger
+from deebot_client.message import HandlingResult
+from deebot_client.models import DeviceInfo, State
+
+from .common import ExecuteCommand
+
+_LOGGER = get_logger(__name__)
+
+
+class Charge(ExecuteCommand):
+    """Charge command."""
+
+    name = "Charge"
+
+    xml_has_own_element = True
+
+    @classmethod
+    def _handle_body(
+        cls, event_bus: EventBus, body: dict[str, Any] | str
+    ) -> HandlingResult:
+        """Handle message->body and notify the correct event subscribers.
+
+        :return: A message response
+        """
+
+        tree = ElementTree.fromstring(body)
+        attributes = tree.attrib.keys()
+
+        if len(attributes) == 0:
+            return HandlingResult.analyse()
+
+        # "resp": "<ctl ret='ok'/>", == returning
+        if "ret" in attributes and tree.attrib.get("ret") == "ok":
+            event_bus.notify(StateEvent(State.RETURNING))
+            return HandlingResult.success()
+
+        # "<ctl ret='fail' errno='8'/>", == already charging
+        is_already_charging = (
+            "errno" in attributes and int(tree.attrib.get("errno")) == 8
+        )
+        if is_already_charging:
+            # bot is already charging
+            event_bus.notify(StateEvent(State.DOCKED))
+            return HandlingResult.success()
+
+        return HandlingResult.success()
+
+    async def _execute(
+        self, authenticator: Authenticator, device_info: DeviceInfo, event_bus: EventBus
+    ) -> CommandResult:
+        if isinstance(self._args, dict):
+            self._args.update({"type": "go"})
+
+        return await super()._execute(authenticator, device_info, event_bus)

deebot_client/commands/xml/clean.py

@@ -0,0 +1,102 @@
+# from xml.etree import ElementTree
+#
+# from deebot_client.authentication import Authenticator
+# from deebot_client.command import CommandResult
+# from deebot_client.commands.xml.common import ExecuteCommand
+# from deebot_client.event_bus import EventBus
+# from deebot_client.events import StateEvent
+# from deebot_client.message import HandlingResult, MessageBodyDataDict
+# from deebot_client.models import CleanAction, CleanMode, DeviceInfo, VacuumState
+#
+#
+# @unique
+# class CleanSpeed(str, Enum):
+#     """Enum class for all possible clean speeds.
+#
+#     Currently only used for the Deebot 900.
+#     """
+#
+#     STANDARD = "standard"
+#     STRONG = "strong"
+#
+#
+# class GetCleanInfo(NoArgsCommand, MessageBodyDataDict):
+#     """Get clean info command."""
+#
+#     name = "GetCleanState"
+#
+#     @classmethod
+#     def _handle_body_data_xml(
+#         cls, event_bus: EventBus, xml_message: str
+#     ) -> HandlingResult:
+#         status: VacuumState | None = None
+#
+#         tree = ElementTree.fromstring(xml_message)
+#
+#         element = tree.find("clean")
+#         if element is None:
+#             return HandlingResult.analyse()
+#
+#         raw_state = element.attrib.get("st")
+#         a = element.attrib.get("a")  # Action ?
+#
+#         if raw_state == "h" and a == "0":
+#             status = VacuumState.IDLE
+#         elif raw_state == "p" and a == "0":
+#             status = VacuumState.PAUSED
+#         elif raw_state == "s" and a == "0":
+#             status = VacuumState.CLEANING
+#         elif raw_state == "h" and a == "1":
+#             status = VacuumState.RETURNING
+#
+#         if status:
+#             event_bus.notify(StateEvent(status))
+#             return HandlingResult.success()
+#
+#         return HandlingResult.analyse()
+#
+#
+# class CleanArea(Clean):
+#     """Clean area command."""
+#
+#     def __init__(self, mode: CleanMode, area: str, cleanings: int = 1) -> None:
+#         super().__init__(CleanAction.START)
+#         if not isinstance(self._args, dict):
+#             raise ValueError("args must be a dict!")
+#
+#         self._args["type"] = mode.value
+#         self._args["content"] = str(area)
+#         self._args["count"] = cleanings
+#
+#
+# class Clean(ExecuteCommand):
+#     name = "Clean"
+#
+#     has_sub_element = True
+#
+#     async def _execute(
+#         self, authenticator: Authenticator, device_info: DeviceInfo, event_bus: EventBus
+#     ) -> CommandResult:
+#         state = event_bus.get_last_event(StateEvent)
+#
+#         if state and isinstance(self._args, dict):
+#             if (
+#                 self._args["act"] == CleanAction.RESUME.value
+#                 and state.state != VacuumState.PAUSED
+#             ):
+#                 self._args = str(self.__get_args(CleanAction.START))[0]
+#             elif (
+#                 self._args["act"] == CleanAction.START.value
+#                 and state.state == VacuumState.PAUSED
+#             ):
+#                 self._args = str(self.__get_args(CleanAction.RESUME))[0]
+#             elif self._args["act"] == CleanAction.START.value:
+#                 self._args["act"] = str(CleanAction.START.value)[0]
+#
+#             elif self._args["act"] == CleanAction.STOP.value:
+#                 self._args["act"] = str(CleanAction.HALT.value)[0]
+#
+#             if "speed" not in self._args:
+#                 self._args["speed"] = CleanSpeed.STANDARD.value
+#
+#         return await super()._execute(authenticator, device_info, event_bus)