Add support for set/getWashInfo commands and onWashInfo message

deebot_client/capabilities.py

@@ -43,14 +43,17 @@
     WorkMode,
     WorkModeEvent,
 )
+from deebot_client.events.wash_info import WashInfoEvent
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 
     from _typeshed import DataclassInstance
 
     from deebot_client.command import Command, SetCommand
+    from deebot_client.commands.json.wash_info import SetWashInfo
     from deebot_client.events.efficiency_mode import EfficiencyMode, EfficiencyModeEvent
+    from deebot_client.events.wash_info import WashMode
     from deebot_client.models import CleanAction, CleanMode
 
 
@@ -121,6 +124,14 @@ class CapabilityCleanAction:
     area: Callable[[CleanMode, str, int], Command]
 
 
+@dataclass(frozen=True, kw_only=True)
+class CapabilityWashInfo(CapabilityEvent[WashInfoEvent]):
+    """Capabilities for wash handling."""
+
+    set: Callable[[WashMode | None, int | None], SetWashInfo]
+    wash_modes: tuple[WashMode, ...]
+
+
 @dataclass(frozen=True, kw_only=True)
 class CapabilityClean:
     """Capabilities for clean."""
@@ -131,6 +142,7 @@ class CapabilityClean:
     log: CapabilityEvent[CleanLogEvent] | None = None
     preference: CapabilitySetEnable[CleanPreferenceEvent] | None = None
     work_mode: CapabilitySetTypes[WorkModeEvent, WorkMode] | None = None
+    wash_info: CapabilityWashInfo | None = None
 
 
 @dataclass(frozen=True)

deebot_client/command.py

@@ -9,7 +9,13 @@
 from deebot_client.events import AvailabilityEvent
 from deebot_client.exceptions import ApiTimeoutError, DeebotError
 
-from .const import PATH_API_IOT_DEVMANAGER, REQUEST_HEADERS, DataType
+from .const import (
+    PATH_API_IOT_DEVMANAGER,
+    REQUEST_HEADERS,
+    UNDEFINED,
+    DataType,
+    UndefinedType,
+)
 from .logging_filter import get_logger
 from .message import HandlingResult, HandlingState, Message
 
@@ -252,7 +258,8 @@ class InitParam:
     """Init param."""
 
     type_: type
-    name: str | None = None
+    name: str | None = field(kw_only=True, default=None)
+    default: UndefinedType | Any = field(kw_only=True, default=UNDEFINED)
 
 
 class CommandMqttP2P(Command, ABC):
@@ -276,24 +283,25 @@ def create_from_mqtt(cls, data: dict[str, Any]) -> CommandMqttP2P:
                 # Remove field
                 data.pop(name, None)
             else:
-                values[param.name or name] = _pop_or_raise(name, param.type_, data)
+                values[param.name or name] = _pop_or_raise(name, param, 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:
+def _pop_or_raise(name: str, param: InitParam, data: dict[str, Any]) -> Any:
+    if name not in data:
+        if param.default is not UNDEFINED:
+            return param.default
         msg = f'"{name}" is missing in {data}'
-        raise DeebotError(msg) from err
+        raise DeebotError(msg)
+    value = data.pop(name)
     try:
-        return type_(value)
+        return param.type_(value)
     except ValueError as err:
-        msg = f'Could not convert "{value}" of {name} into {type_}'
+        msg = f'Could not convert "{value}" of {name} into {param.type_}'
         raise DeebotError(msg) from err
 
 

deebot_client/commands/json/__init__.py

@@ -40,6 +40,7 @@
 from .true_detect import GetTrueDetect, SetTrueDetect
 from .voice_assistant_state import GetVoiceAssistantState, SetVoiceAssistantState
 from .volume import GetVolume, SetVolume
+from .wash_info import GetWashInfo, SetWashInfo
 from .water_info import GetWaterInfo, SetWaterInfo
 from .work_mode import GetWorkMode, SetWorkMode
 
@@ -97,6 +98,8 @@
     "SetVoiceAssistantState",
     "GetVolume",
     "SetVolume",
+    "GetWashInfo",
+    "SetWashInfo",
     "GetWaterInfo",
     "SetWaterInfo",
     "GetWorkMode",
@@ -183,6 +186,9 @@
     GetVolume,
     SetVolume,
 
+    GetWashInfo,
+    SetWashInfo,
+
     GetWaterInfo,
     SetWaterInfo,
 

deebot_client/commands/json/common.py

@@ -128,7 +128,9 @@ class SetEnableCommand(JsonSetCommand, ABC):
     _field_name = _ENABLE
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
-        cls._mqtt_params = MappingProxyType({cls._field_name: InitParam(bool, _ENABLE)})
+        cls._mqtt_params = MappingProxyType(
+            {cls._field_name: InitParam(bool, name=_ENABLE)}
+        )
         super().__init_subclass__(**kwargs)
 
     def __init__(self, enable: bool) -> None:  # noqa: FBT001

deebot_client/commands/json/life_span.py

@@ -50,7 +50,7 @@ class ResetLifeSpan(ExecuteCommand, CommandMqttP2P):
     """Reset life span command."""
 
     name = "resetLifeSpan"
-    _mqtt_params = MappingProxyType({"type": InitParam(LifeSpan, "life_span")})
+    _mqtt_params = MappingProxyType({"type": InitParam(LifeSpan, name="life_span")})
 
     def __init__(self, life_span: LifeSpan) -> None:
         super().__init__({"type": life_span.value})

deebot_client/commands/json/ota.py

@@ -52,7 +52,9 @@ class SetOta(JsonSetCommand):
     name = "setOta"
     get_command = GetOta
 
-    _mqtt_params = MappingProxyType({"autoSwitch": InitParam(bool, "auto_enabled")})
+    _mqtt_params = MappingProxyType(
+        {"autoSwitch": InitParam(bool, name="auto_enabled")}
+    )
 
     def __init__(self, auto_enabled: bool) -> None:  # noqa: FBT001
         super().__init__({"autoSwitch": 1 if auto_enabled else 0})

deebot_client/commands/json/wash_info.py

@@ -0,0 +1,49 @@
+"""WashInfo command module."""
+from __future__ import annotations
+
+from types import MappingProxyType
+from typing import Any
+
+from deebot_client.command import InitParam
+from deebot_client.events import WashMode
+from deebot_client.messages.json.wash_info import OnWashInfo
+
+from .common import JsonGetCommand, JsonSetCommand
+
+
+class GetWashInfo(OnWashInfo, JsonGetCommand):
+    """Get wash info command."""
+
+    name = "getWashInfo"
+
+
+class SetWashInfo(JsonSetCommand):
+    """Set wash info command."""
+
+    name = "setWashInfo"
+    get_command = GetWashInfo
+    _mqtt_params = MappingProxyType(
+        {
+            "mode": InitParam(int, default=None),
+            "hot_wash_amount": InitParam(int, default=None),
+        }
+    )
+
+    def __init__(
+        self,
+        mode: WashMode | str | int | None = None,
+        hot_wash_amount: int | None = None,
+    ) -> None:
+        args: dict[str, Any] = {}
+
+        if isinstance(mode, str):
+            mode = WashMode.get(mode)
+        if isinstance(mode, WashMode):
+            mode = mode.value
+
+        if mode is not None:
+            args["mode"] = mode
+
+        if hot_wash_amount is not None:
+            args["hot_wash_amount"] = hot_wash_amount
+        super().__init__(args)

deebot_client/events/__init__.py

@@ -24,6 +24,7 @@
     PositionType,
 )
 from .network import NetworkInfoEvent
+from .wash_info import WashInfoEvent, WashMode
 from .water_info import WaterAmount, WaterInfoEvent
 from .work_mode import WorkMode, WorkModeEvent
 
@@ -52,6 +53,8 @@
     "PositionType",
     "PositionsEvent",
     "SweepModeEvent",
+    "WashMode",
+    "WashInfoEvent",
     "WaterAmount",
     "WaterInfoEvent",
     "WorkMode",

deebot_client/events/wash_info.py

@@ -0,0 +1,24 @@
+"""Wash info event module."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from deebot_client.util import DisplayNameIntEnum
+
+from .base import Event
+
+
+class WashMode(DisplayNameIntEnum):
+    """Enum class for all possible wash modes."""
+
+    STANDARD = 0
+    HOT = 1
+
+
+@dataclass(frozen=True)
+class WashInfoEvent(Event):
+    """Wash info event representation."""
+
+    mode: WashMode | None
+    interval: int | None
+    hot_wash_amount: int | None

deebot_client/hardware/deebot/p1jij8.py

@@ -15,6 +15,7 @@
     CapabilitySettings,
     CapabilitySetTypes,
     CapabilityStats,
+    CapabilityWashInfo,
 )
 from deebot_client.commands.json.advanced_mode import GetAdvancedMode, SetAdvancedMode
 from deebot_client.commands.json.battery import GetBattery
@@ -51,6 +52,10 @@
 from deebot_client.commands.json.stats import GetStats, GetTotalStats
 from deebot_client.commands.json.true_detect import GetTrueDetect, SetTrueDetect
 from deebot_client.commands.json.volume import GetVolume, SetVolume
+from deebot_client.commands.json.wash_info import (
+    GetWashInfo,
+    SetWashInfo,
+)
 from deebot_client.commands.json.water_info import GetWaterInfo, SetWaterInfo
 from deebot_client.commands.json.work_mode import GetWorkMode, SetWorkMode
 from deebot_client.const import DataType
@@ -88,6 +93,7 @@
     WorkMode,
     WorkModeEvent,
 )
+from deebot_client.events.wash_info import WashInfoEvent, WashMode
 from deebot_client.models import StaticDeviceInfo
 from deebot_client.util import short_name
 
@@ -124,6 +130,15 @@
                     WorkMode.VACUUM_AND_MOP,
                 ),
             ),
+            wash_info=CapabilityWashInfo(
+                event=WashInfoEvent,
+                get=[GetWashInfo()],
+                set=SetWashInfo,
+                wash_modes=(
+                    WashMode.STANDARD,
+                    WashMode.HOT,
+                ),
+            ),
         ),
         custom=CapabilityCustomCommand(
             event=CustomCommandEvent, get=[], set=CustomCommand

deebot_client/messages/json/__init__.py

@@ -6,13 +6,15 @@
 from .battery import OnBattery
 from .map import OnMapSetV2
 from .stats import ReportStats
+from .wash_info import OnWashInfo
 
 if TYPE_CHECKING:
     from deebot_client.message import Message
 
 __all__ = [
     "OnBattery",
     "OnMapSetV2",
+    "OnWashInfo",
     "ReportStats",
 ]
 

deebot_client/messages/json/wash_info.py

@@ -0,0 +1,38 @@
+"""WashInfo messages."""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from deebot_client.events import WashInfoEvent
+from deebot_client.events.wash_info import WashMode
+from deebot_client.message import HandlingResult, MessageBodyDataDict
+
+if TYPE_CHECKING:
+    from deebot_client.event_bus import EventBus
+
+
+class OnWashInfo(MessageBodyDataDict):
+    """On battery message."""
+
+    name = "onWashInfo"
+
+    @classmethod
+    def _handle_body_data_dict(
+        cls, event_bus: EventBus, data: dict[str, Any]
+    ) -> HandlingResult:
+        """Handle message->body->data and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        mode = data.get("mode")
+        if isinstance(mode, int):
+            mode = WashMode(mode)
+
+        event_bus.notify(
+            WashInfoEvent(
+                mode=mode,
+                hot_wash_amount=data.get("hot_wash_amount"),
+                interval=data.get("interval"),
+            )
+        )
+        return HandlingResult.success()