Add support for multiple data types

deebot_client/command.py

@@ -2,11 +2,10 @@
 import asyncio
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from datetime import datetime
 from typing import Any, final
 
 from .authentication import Authenticator
-from .const import PATH_API_IOT_DEVMANAGER, REQUEST_HEADERS
+from .const import PATH_API_IOT_DEVMANAGER, REQUEST_HEADERS, DataType
 from .events.event_bus import EventBus
 from .logging_filter import get_logger
 from .message import HandlingResult, HandlingState
@@ -48,6 +47,16 @@ def __init__(self, args: dict | list | None = None) -> None:
     def name(cls) -> str:
         """Command name."""
 
+    @property  # type: ignore[misc]
+    @classmethod
+    @abstractmethod
+    def data_type(cls) -> DataType:
+        """Data type."""  # noqa: D401
+
+    @abstractmethod
+    def _get_payload(self) -> dict[str, Any] | list | str:
+        """Get the payload for the rest call."""
+
     @final
     async def execute(
         self, authenticator: Authenticator, device_info: DeviceInfo, event_bus: EventBus
@@ -95,30 +104,17 @@ async def _execute(
                 result.args,
                 result.requested_commands,
             )
+        if result.state == HandlingState.ERROR:
+            _LOGGER.warning("Could not parse %s: %s", self.name, response)
         return result
 
-    def _get_payload(self) -> dict[str, Any] | list:
-        payload = {
-            "header": {
-                "pri": "1",
-                "ts": datetime.now().timestamp(),
-                "tzm": 480,
-                "ver": "0.0.50",
-            }
-        }
-
-        if len(self._args) > 0:
-            payload["body"] = {"data": self._args}
-
-        return payload
-
     async def _execute_api_request(
         self, authenticator: Authenticator, device_info: DeviceInfo
     ) -> dict[str, Any]:
-        json = {
+        payload = {
             "cmdName": self.name,
             "payload": self._get_payload(),
-            "payloadType": "j",
+            "payloadType": self.data_type.value,
             "td": "q",
             "toId": device_info.did,
             "toRes": device_info.resource,
@@ -127,9 +123,9 @@ async def _execute_api_request(
 
         credentials = await authenticator.authenticate()
         query_params = {
-            "mid": json["toType"],
-            "did": json["toId"],
-            "td": json["td"],
+            "mid": payload["toType"],
+            "did": payload["toId"],
+            "td": payload["td"],
             "u": credentials.user_id,
             "cv": "1.67.3",
             "t": "a",
@@ -138,7 +134,7 @@ async def _execute_api_request(
 
         return await authenticator.post_authenticated(
             PATH_API_IOT_DEVMANAGER,
-            json,
+            payload,
             query_params=query_params,
             headers=REQUEST_HEADERS,
         )
@@ -188,3 +184,11 @@ def __eq__(self, obj: object) -> bool:
 
     def __hash__(self) -> int:
         return hash(self.name) + hash(self._args)
+
+
+class CommandMqttP2P(Command, ABC):
+    """Command which can handle mqtt p2p messages."""
+
+    @abstractmethod
+    def handle_mqtt_p2p(self, event_bus: EventBus, response: dict[str, Any]) -> None:
+        """Handle response received over the mqtt channel "p2p"."""

deebot_client/commands/__init__.py

@@ -1,110 +1,14 @@
 """Commands module."""
-from ..command import Command
-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 CommandHandlingMqttP2P, SetCommand
-from .continuous_cleaning import GetContinuousCleaning, SetContinuousCleaning
-from .error import GetError
-from .fan_speed import FanSpeedLevel, 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
-
-# fmt: off
-# ordered by file asc
-_COMMANDS: list[type[Command]] = [
-    GetAdvancedMode,
-    SetAdvancedMode,
-
-    GetBattery,
-
-    GetCarpetAutoFanBoost,
-    SetCarpetAutoFanBoost,
-
-    GetCleanCount,
-    SetCleanCount,
-
-    GetCleanPreference,
-    SetCleanPreference,
-
-    Charge,
-
-    GetChargeState,
-
-    Clean,
-    CleanArea,
-    GetCleanInfo,
-
-    GetCleanLogs,
-
-    GetContinuousCleaning,
-    SetContinuousCleaning,
-
-    GetError,
+from deebot_client.command import Command, CommandMqttP2P
+from deebot_client.const import DataType
 
-    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
+from .json import COMMANDS as JSON_COMMANDS
+from .json import (
+    COMMANDS_WITH_MQTT_P2P_HANDLING as JSON_COMMANDS_WITH_MQTT_P2P_HANDLING,
+)
 
-COMMANDS: dict[str, type[Command]] = {
-    cmd.name: cmd for cmd in _COMMANDS  # type: ignore[misc]
-}
+COMMANDS: dict[DataType, dict[str, type[Command]]] = {DataType.JSON: JSON_COMMANDS}
 
-COMMANDS_WITH_MQTT_P2P_HANDLING: dict[str, type[CommandHandlingMqttP2P]] = {
-    cmd_name: cmd
-    for (cmd_name, cmd) in COMMANDS.items()
-    if issubclass(cmd, CommandHandlingMqttP2P)
+COMMANDS_WITH_MQTT_P2P_HANDLING: dict[DataType, dict[str, type[CommandMqttP2P]]] = {
+    DataType.JSON: JSON_COMMANDS_WITH_MQTT_P2P_HANDLING
 }

deebot_client/commands/json/__init__.py

@@ -0,0 +1,111 @@
+"""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 JsonCommand
+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
+
+# fmt: off
+# ordered by file asc
+_COMMANDS: list[type[JsonCommand]] = [
+    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/advanced_mode.py → deebot_client/commands/json/advanced_mode.py

@@ -1,6 +1,7 @@
 """Advanced mode command module."""
 
-from ..events import AdvancedModeEvent
+from deebot_client.events import AdvancedModeEvent
+
 from .common import GetEnableCommand, SetEnableCommand
 
 

deebot_client/commands/battery.py → deebot_client/commands/json/battery.py

@@ -1,5 +1,6 @@
 """Battery commands."""
-from ..messages import OnBattery
+from deebot_client.messages.json import OnBattery
+
 from .common import NoArgsCommand
 
 

deebot_client/commands/carpet.py → deebot_client/commands/json/carpet.py

@@ -1,6 +1,7 @@
 """Carpet pressure command module."""
 
-from ..events import CarpetAutoFanBoostEvent
+from deebot_client.events import CarpetAutoFanBoostEvent
+
 from .common import GetEnableCommand, SetEnableCommand
 
 

deebot_client/commands/charge.py → deebot_client/commands/json/charge.py

@@ -1,10 +1,11 @@
 """Charge commands."""
 from typing import Any
 
-from ..events import StateEvent
-from ..logging_filter import get_logger
-from ..message import HandlingResult
-from ..models import VacuumState
+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 VacuumState
+
 from .common import EventBus, ExecuteCommand
 from .const import CODE
 

deebot_client/commands/charge_state.py → deebot_client/commands/json/charge_state.py

@@ -1,9 +1,10 @@
 """Charge state commands."""
 from typing import Any
 
-from ..events import StateEvent
-from ..message import HandlingResult, MessageBodyDataDict
-from ..models import VacuumState
+from deebot_client.events import StateEvent
+from deebot_client.message import HandlingResult, MessageBodyDataDict
+from deebot_client.models import VacuumState
+
 from .common import EventBus, NoArgsCommand
 from .const import CODE