Add create_from_mqtt and make strict types for command init (#319)

deebot_client/command.py

@@ -4,6 +4,8 @@
 from dataclasses import dataclass, field
 from typing import Any, final
 
+from deebot_client.exceptions import DeebotError
+
 from .authentication import Authenticator
 from .const import PATH_API_IOT_DEVMANAGER, REQUEST_HEADERS, DataType
 from .events.event_bus import EventBus
@@ -186,9 +188,51 @@ def __hash__(self) -> int:
         return hash(self.name) + hash(self._args)
 
 
+@dataclass
+class InitParam:
+    """Init param."""
+
+    type_: type
+    name: str | None = None
+
+
 class CommandMqttP2P(Command, ABC):
     """Command which can handle mqtt p2p messages."""
 
+    _mqtt_params: dict[str, InitParam | None]
+
     @abstractmethod
     def handle_mqtt_p2p(self, event_bus: EventBus, response: dict[str, Any]) -> None:
         """Handle response received over the mqtt channel "p2p"."""
+
+    @classmethod
+    def create_from_mqtt(cls, data: dict[str, Any]) -> "CommandMqttP2P":
+        """Create a command from the mqtt data."""
+        values: dict[str, Any] = {}
+        if not hasattr(cls, "_mqtt_params"):
+            raise DeebotError("_mqtt_params not set")
+
+        for name, param in cls._mqtt_params.items():
+            if param is None:
+                # Remove field
+                data.pop(name, None)
+            else:
+                values[param.name or name] = _pop_or_raise(name, param.type_, data)
+
+        if data:
+            _LOGGER.debug("Following data will be ignored: %s", data)
+
+        return cls(**values)
+
+
+def _pop_or_raise(name: str, type_: type, data: dict[str, Any]) -> Any:
+    try:
+        value = data.pop(name)
+    except KeyError as err:
+        raise DeebotError(f'"{name}" is missing in {data}') from err
+    try:
+        return type_(value)
+    except ValueError as err:
+        raise DeebotError(
+            f'Could not convert "{value}" of {name} into {type_}'
+        ) from err

deebot_client/commands/json/clean_count.py

@@ -1,8 +1,8 @@
 """Clean count command module."""
 
-from collections.abc import Mapping
 from typing import Any
 
+from deebot_client.command import InitParam
 from deebot_client.events import CleanCountEvent
 from deebot_client.message import HandlingResult, MessageBodyDataDict
 
@@ -32,6 +32,7 @@ class SetCleanCount(SetCommand):
 
     name = "setCleanCount"
     get_command = GetCleanCount
+    _mqtt_params = {"count": InitParam(int)}
 
-    def __init__(self, count: int, **kwargs: Mapping[str, Any]) -> None:
-        super().__init__({"count": count}, **kwargs)
+    def __init__(self, count: int) -> None:
+        super().__init__({"count": count})

deebot_client/commands/json/common.py

@@ -1,10 +1,9 @@
 """Base commands."""
 from abc import ABC, abstractmethod
-from collections.abc import Mapping
 from datetime import datetime
 from typing import Any
 
-from deebot_client.command import Command, CommandMqttP2P, CommandResult
+from deebot_client.command import Command, CommandMqttP2P, CommandResult, InitParam
 from deebot_client.const import DataType
 from deebot_client.events import AvailabilityEvent, EnableEvent
 from deebot_client.events.event_bus import EventBus
@@ -108,16 +107,6 @@ class SetCommand(ExecuteCommand, CommandMqttP2P, ABC):
     Command needs to be linked to the "get" command, for handling (updating) the sensors.
     """
 
-    def __init__(
-        self,
-        args: dict | list | None,
-        **kwargs: Mapping[str, Any],
-    ) -> None:
-        if kwargs:
-            _LOGGER.debug("Following passed parameters will be ignored: %s", kwargs)
-
-        super().__init__(args)
-
     @property
     @abstractmethod
     def get_command(self) -> type[CommandWithMessageHandling]:
@@ -156,7 +145,7 @@ def _handle_body_data_dict(
 class SetEnableCommand(SetCommand, ABC):
     """Abstract set enable command."""
 
-    def __init__(self, enable: int | bool, **kwargs: Mapping[str, Any]) -> None:
-        if isinstance(enable, bool):
-            enable = 1 if enable else 0
-        super().__init__({"enable": enable}, **kwargs)
+    _mqtt_params = {"enable": InitParam(bool)}
+
+    def __init__(self, enable: bool) -> None:
+        super().__init__({"enable": 1 if enable else 0})

deebot_client/commands/json/fan_speed.py

@@ -1,7 +1,7 @@
 """(fan) speed commands."""
-from collections.abc import Mapping
 from typing import Any
 
+from deebot_client.command import InitParam
 from deebot_client.events import FanSpeedEvent, FanSpeedLevel
 from deebot_client.message import HandlingResult, MessageBodyDataDict
 
@@ -30,13 +30,7 @@ class SetFanSpeed(SetCommand):
 
     name = "setSpeed"
     get_command = GetFanSpeed
+    _mqtt_params = {"speed": InitParam(FanSpeedLevel)}
 
-    def __init__(
-        self, speed: str | int | FanSpeedLevel, **kwargs: Mapping[str, Any]
-    ) -> None:
-        if isinstance(speed, str):
-            speed = FanSpeedLevel.get(speed)
-        if isinstance(speed, FanSpeedLevel):
-            speed = speed.value
-
-        super().__init__({"speed": speed}, **kwargs)
+    def __init__(self, speed: FanSpeedLevel) -> None:
+        super().__init__({"speed": speed.value})

deebot_client/commands/json/life_span.py

@@ -1,33 +1,21 @@
 """Life span commands."""
 from typing import Any
 
-from deebot_client.command import CommandMqttP2P
+from deebot_client.command import CommandMqttP2P, InitParam
 from deebot_client.events import LifeSpan, LifeSpanEvent
 from deebot_client.message import HandlingResult, HandlingState, MessageBodyDataList
 from deebot_client.util import LST
 
 from .common import CommandWithMessageHandling, EventBus, ExecuteCommand
 
-LifeSpanType = LifeSpan | str
-
-
-def _get_str(_type: LifeSpanType) -> str:
-    if isinstance(_type, LifeSpan):
-        return _type.value
-
-    return _type
-
 
 class GetLifeSpan(CommandWithMessageHandling, MessageBodyDataList):
     """Get life span command."""
 
     name = "getLifeSpan"
 
-    def __init__(self, _types: LifeSpanType | LST[LifeSpanType]) -> None:
-        if isinstance(_types, LifeSpanType):  # type: ignore[misc, arg-type]
-            _types = set(_types)
-
-        args = [_get_str(life_span) for life_span in _types]
+    def __init__(self, life_spans: LST[LifeSpan]) -> None:
+        args = [life_span.value for life_span in life_spans]
         super().__init__(args)
 
     @classmethod
@@ -54,9 +42,10 @@ class ResetLifeSpan(ExecuteCommand, CommandMqttP2P):
     """Reset life span command."""
 
     name = "resetLifeSpan"
+    _mqtt_params = {"type": InitParam(LifeSpan, "life_span")}
 
-    def __init__(self, type: LifeSpanType) -> None:  # pylint: disable=redefined-builtin
-        super().__init__({"type": _get_str(type)})
+    def __init__(self, life_span: LifeSpan) -> None:
+        super().__init__({"type": life_span.value})
 
     def handle_mqtt_p2p(self, event_bus: EventBus, response: dict[str, Any]) -> None:
         """Handle response received over the mqtt channel "p2p"."""

deebot_client/commands/json/volume.py

@@ -1,8 +1,8 @@
 """Volume command module."""
 
-from collections.abc import Mapping
 from typing import Any
 
+from deebot_client.command import InitParam
 from deebot_client.events import VolumeEvent
 from deebot_client.message import HandlingResult, MessageBodyDataDict
 
@@ -34,8 +34,10 @@ class SetVolume(SetCommand):
 
     name = "setVolume"
     get_command = GetVolume
+    _mqtt_params = {
+        "volume": InitParam(int),
+        "total": None,  # Remove it as we don't can set it (App includes it)
+    }
 
-    def __init__(self, volume: int, **kwargs: Mapping[str, Any]) -> None:
-        # removing "total" as we don't can set it (App includes it)
-        kwargs.pop("total", None)
-        super().__init__({"volume": volume}, **kwargs)
+    def __init__(self, volume: int) -> None:
+        super().__init__({"volume": volume})

deebot_client/commands/json/water_info.py

@@ -1,7 +1,7 @@
 """Water info commands."""
-from collections.abc import Mapping
 from typing import Any
 
+from deebot_client.command import InitParam
 from deebot_client.events import WaterAmount, WaterInfoEvent
 from deebot_client.message import HandlingResult, MessageBodyDataDict
 
@@ -34,16 +34,10 @@ class SetWaterInfo(SetCommand):
 
     name = "setWaterInfo"
     get_command = GetWaterInfo
+    _mqtt_params = {
+        "amount": InitParam(WaterAmount),
+        "enable": None,  # Remove it as we don't can set it (App includes it)
+    }
 
-    def __init__(
-        self, amount: str | int | WaterAmount, **kwargs: Mapping[str, Any]
-    ) -> None:
-        # removing "enable" as we don't can set it
-        kwargs.pop("enable", None)
-
-        if isinstance(amount, str):
-            amount = WaterAmount.get(amount)
-        if isinstance(amount, WaterAmount):
-            amount = amount.value
-
-        super().__init__({"amount": amount}, **kwargs)
+    def __init__(self, amount: WaterAmount) -> None:
+        super().__init__({"amount": amount.value})

deebot_client/mqtt_client.py

@@ -289,7 +289,9 @@ def _handle_p2p(
                     )
                     return
 
-                self._received_p2p_commands[request_id] = command_type(**data)
+                self._received_p2p_commands[request_id] = command_type.create_from_mqtt(
+                    data
+                )
             else:
                 if command := self._received_p2p_commands.pop(request_id, None):
                     if sub_info := self._subscribtions.get(topic_split[3]):

tests/commands/json/__init__.py

@@ -6,8 +6,12 @@
 
 from deebot_client.authentication import Authenticator
 from deebot_client.command import Command
-from deebot_client.commands.json.common import ExecuteCommand, SetCommand
-from deebot_client.events import Event
+from deebot_client.commands.json.common import (
+    ExecuteCommand,
+    SetCommand,
+    SetEnableCommand,
+)
+from deebot_client.events import EnableEvent, Event
 from deebot_client.events.event_bus import EventBus
 from deebot_client.models import Credentials, DeviceInfo
 from tests.helpers import get_message_json, get_request_json, get_success_body
@@ -78,7 +82,7 @@ async def assert_execute_command(
 
 async def assert_set_command(
     command: SetCommand,
-    args: dict | list | None,
+    args: dict,
     expected_get_command_event: Event,
 ) -> None:
     await assert_execute_command(command, args)
@@ -98,3 +102,15 @@ async def assert_set_command(
     # Success
     command.handle_mqtt_p2p(event_bus, get_message_json(get_success_body()))
     event_bus.notify.assert_called_once_with(expected_get_command_event)
+
+    mqtt_command = command.create_from_mqtt(args)
+    assert mqtt_command == command
+
+
+async def assert_set_enable_command(
+    command: SetEnableCommand,
+    enabled: bool,
+    expected_get_command_event: type[EnableEvent],
+) -> None:
+    args = {"enable": 1 if enabled else 0}
+    await assert_set_command(command, args, expected_get_command_event(enabled))

tests/commands/json/test_advanced_mode.py

@@ -4,7 +4,7 @@
 from deebot_client.events import AdvancedModeEvent
 from tests.helpers import get_request_json, get_success_body
 
-from . import assert_command, assert_set_command
+from . import assert_command, assert_set_enable_command
 
 
 @pytest.mark.parametrize("value", [False, True])
@@ -15,5 +15,4 @@ async def test_GetAdvancedMode(value: bool) -> None:
 
 @pytest.mark.parametrize("value", [False, True])
 async def test_SetAdvancedMode(value: bool) -> None:
-    args = {"enable": 1 if value else 0}
-    await assert_set_command(SetAdvancedMode(value), args, AdvancedModeEvent(value))
+    await assert_set_enable_command(SetAdvancedMode(value), value, AdvancedModeEvent)

tests/commands/json/test_carpet.py

@@ -4,7 +4,7 @@
 from deebot_client.events import CarpetAutoFanBoostEvent
 from tests.helpers import get_request_json, get_success_body
 
-from . import assert_command, assert_set_command
+from . import assert_command, assert_set_enable_command
 
 
 @pytest.mark.parametrize("value", [False, True])
@@ -15,7 +15,6 @@ async def test_GetCarpetAutoFanBoost(value: bool) -> None:
 
 @pytest.mark.parametrize("value", [False, True])
 async def test_SetCarpetAutoFanBoost(value: bool) -> None:
-    args = {"enable": 1 if value else 0}
-    await assert_set_command(
-        SetCarpetAutoFanBoost(value), args, CarpetAutoFanBoostEvent(value)
+    await assert_set_enable_command(
+        SetCarpetAutoFanBoost(value), value, CarpetAutoFanBoostEvent
     )

tests/commands/json/test_clean_preference.py

@@ -4,7 +4,7 @@
 from deebot_client.events import CleanPreferenceEvent
 from tests.helpers import get_request_json, get_success_body
 
-from . import assert_command, assert_set_command
+from . import assert_command, assert_set_enable_command
 
 
 @pytest.mark.parametrize("value", [False, True])
@@ -15,7 +15,6 @@ async def test_GetCleanPreference(value: bool) -> None:
 
 @pytest.mark.parametrize("value", [False, True])
 async def test_SetCleanPreference(value: bool) -> None:
-    args = {"enable": 1 if value else 0}
-    await assert_set_command(
-        SetCleanPreference(value), args, CleanPreferenceEvent(value)
+    await assert_set_enable_command(
+        SetCleanPreference(value), value, CleanPreferenceEvent
     )

tests/commands/json/test_continuous_cleaning.py

@@ -4,7 +4,7 @@
 from deebot_client.events import ContinuousCleaningEvent
 from tests.helpers import get_request_json, get_success_body
 
-from . import assert_command, assert_set_command
+from . import assert_command, assert_set_enable_command
 
 
 @pytest.mark.parametrize("value", [False, True])
@@ -15,7 +15,6 @@ async def test_GetContinuousCleaning(value: bool) -> None:
 
 @pytest.mark.parametrize("value", [False, True])
 async def test_SetContinuousCleaning(value: bool) -> None:
-    args = {"enable": 1 if value else 0}
-    await assert_set_command(
-        SetContinuousCleaning(value), args, ContinuousCleaningEvent(value)
+    await assert_set_enable_command(
+        SetContinuousCleaning(value), value, ContinuousCleaningEvent
     )