Add wash interval commands

deebot_client/capabilities.py

@@ -37,6 +37,7 @@
     TrueDetectEvent,
     VoiceAssistantStateEvent,
     VolumeEvent,
+    WashIntervalEvent,
     WaterAmount,
     WaterInfoEvent,
     WorkMode,
@@ -129,6 +130,7 @@ class CapabilityClean:
     count: CapabilitySet[CleanCountEvent, int] | None = None
     log: CapabilityEvent[CleanLogEvent] | None = None
     preference: CapabilitySetEnable[CleanPreferenceEvent] | None = None
+    wash_interval: CapabilitySet[WashIntervalEvent, int] | None = None
     work_mode: CapabilitySetTypes[WorkModeEvent, WorkMode] | None = None
 
 

deebot_client/commands/json/__init__.py

@@ -37,6 +37,7 @@
 from .true_detect import GetTrueDetect, SetTrueDetect
 from .voice_assistant_state import GetVoiceAssistantState, SetVoiceAssistantState
 from .volume import GetVolume, SetVolume
+from .wash_interval import GetWashInterval, SetWashInterval
 from .water_info import GetWaterInfo, SetWaterInfo
 from .work_mode import GetWorkMode, SetWorkMode
 
@@ -90,6 +91,8 @@
     "SetVoiceAssistantState",
     "GetVolume",
     "SetVolume",
+    "GetWashInterval",
+    "SetWashInterval",
     "GetWaterInfo",
     "SetWaterInfo",
     "GetWorkMode",
@@ -170,6 +173,9 @@
     GetVolume,
     SetVolume,
 
+    GetWashInterval,
+    SetWashInterval,
+
     GetWaterInfo,
     SetWaterInfo,
 

deebot_client/commands/json/wash_interval.py

@@ -0,0 +1,40 @@
+"""Pads cleaning interval commands."""
+from types import MappingProxyType
+from typing import Any
+
+from deebot_client.command import InitParam
+from deebot_client.event_bus import EventBus
+from deebot_client.events.wash_interval import WashIntervalEvent
+from deebot_client.message import HandlingResult
+
+from .common import JsonGetCommand, JsonSetCommand
+
+
+class GetWashInterval(JsonGetCommand):
+    """Get wash interval command."""
+
+    name = "getWashInterval"
+
+    @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
+        """
+        event_bus.notify(WashIntervalEvent(int(data["interval"])))
+        return HandlingResult.success()
+
+
+class SetWashInterval(JsonSetCommand):
+    """Set wash interval command."""
+
+    name = "setWashInterval"
+    get_command = GetWashInterval
+    _mqtt_params = MappingProxyType({"interval": InitParam(int)})
+
+    def __init__(self, interval: int) -> None:
+        if interval <= 0:
+            raise ValueError("'interval' must be positive")
+        super().__init__({"interval": interval})

deebot_client/events/__init__.py

@@ -24,6 +24,7 @@
     PositionType,
 )
 from .network import NetworkInfoEvent
+from .wash_interval import WashIntervalEvent
 from .water_info import WaterAmount, WaterInfoEvent
 from .work_mode import WorkMode, WorkModeEvent
 
@@ -51,6 +52,7 @@
     "Position",
     "PositionType",
     "PositionsEvent",
+    "WashIntervalEvent",
     "SweepModeEvent",
     "WaterAmount",
     "WaterInfoEvent",

deebot_client/events/wash_interval.py

@@ -0,0 +1,11 @@
+"""Cleaning pads interval event module."""
+from dataclasses import dataclass
+
+from .base import Event
+
+
+@dataclass(frozen=True)
+class WashIntervalEvent(Event):
+    """Wash interval event representation."""
+
+    interval: int

deebot_client/hardware/deebot/p1jij8.py

@@ -51,6 +51,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_interval import (
+    GetWashInterval,
+    SetWashInterval,
+)
 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
@@ -83,6 +87,7 @@
     TotalStatsEvent,
     TrueDetectEvent,
     VolumeEvent,
+    WashIntervalEvent,
     WaterAmount,
     WaterInfoEvent,
     WorkMode,
@@ -113,6 +118,11 @@
             preference=CapabilitySetEnable(
                 CleanPreferenceEvent, [GetCleanPreference()], SetCleanPreference
             ),
+            wash_interval=CapabilitySet(
+                event=WashIntervalEvent,
+                get=[GetWashInterval()],
+                set=SetWashInterval,
+            ),
             work_mode=CapabilitySetTypes(
                 event=WorkModeEvent,
                 get=[GetWorkMode()],

tests/commands/json/test_wash_interval.py

@@ -0,0 +1,37 @@
+from typing import Any
+
+import pytest
+
+from deebot_client.commands.json import GetWashInterval, SetWashInterval
+from deebot_client.events import WashIntervalEvent
+from tests.helpers import (
+    get_request_json,
+    get_success_body,
+)
+
+from . import assert_command, assert_set_command
+
+
+@pytest.mark.parametrize(
+    ("json", "expected"),
+    [
+        ({"interval": 6}, WashIntervalEvent(6)),
+        ({"interval": 10}, WashIntervalEvent(10)),
+    ],
+)
+async def test_GetPadsCleaningInterval(
+    json: dict[str, Any], expected: WashIntervalEvent
+) -> None:
+    json = get_request_json(get_success_body(json))
+    await assert_command(GetWashInterval(), json, expected)
+
+
+async def test_SetWashInterval() -> None:
+    command = SetWashInterval(60)
+    args = {"interval": 60}
+    await assert_set_command(command, args, WashIntervalEvent(60))
+
+
+def test_SetWashInterval_invalid_value() -> None:
+    with pytest.raises(ValueError, match="'interval' must be positive"):
+        SetWashInterval(0)