Add work mode

deebot_client/capabilities.py

@@ -37,6 +37,8 @@
     VolumeEvent,
     WaterAmount,
     WaterInfoEvent,
+    WorkMode,
+    WorkModeEvent,
 )
 from deebot_client.models import CleanAction, CleanMode
 
@@ -120,6 +122,7 @@ class CapabilityClean:
     count: CapabilitySet[CleanCountEvent, int] | None = None
     log: CapabilityEvent[CleanLogEvent]
     preference: CapabilitySetEnable[CleanPreferenceEvent] | None = None
+    work_mode: CapabilitySetTypes[WorkModeEvent, WorkMode] | None = None
 
 
 @dataclass(frozen=True)

deebot_client/commands/json/__init__.py

@@ -31,6 +31,7 @@
 from .true_detect import GetTrueDetect, SetTrueDetect
 from .volume import GetVolume, SetVolume
 from .water_info import GetWaterInfo, SetWaterInfo
+from .work_mode import GetWorkMode, SetWorkMode
 
 __all__ = [
     "GetAdvancedMode",
@@ -74,6 +75,8 @@
     "SetVolume",
     "GetWaterInfo",
     "SetWaterInfo",
+    "GetWorkMode",
+    "SetWorkMode",
 ]
 
 # fmt: off
@@ -141,6 +144,9 @@
 
     GetWaterInfo,
     SetWaterInfo,
+
+    GetWorkMode,
+    SetWorkMode
 ]
 # fmt: on
 

deebot_client/commands/json/work_mode.py

@@ -0,0 +1,39 @@
+"""Work mode commands."""
+from typing import Any
+
+from deebot_client.command import InitParam
+from deebot_client.event_bus import EventBus
+from deebot_client.events import WorkMode, WorkModeEvent
+from deebot_client.message import HandlingResult, MessageBodyDataDict
+
+from .common import CommandWithMessageHandling, SetCommand
+
+
+class GetWorkMode(CommandWithMessageHandling, MessageBodyDataDict):
+    """Get work mode command."""
+
+    name = "getWorkMode"
+
+    @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(WorkModeEvent(WorkMode(int(data["mode"]))))
+        return HandlingResult.success()
+
+
+class SetWorkMode(SetCommand):
+    """Set work mode command."""
+
+    name = "setWorkMode"
+    get_command = GetWorkMode
+    _mqtt_params = {"mode": InitParam(WorkMode)}
+
+    def __init__(self, mode: WorkMode | str) -> None:
+        if isinstance(mode, str):
+            mode = WorkMode.get(mode)
+        super().__init__({"mode": mode.value})

deebot_client/events/__init__.py

@@ -23,6 +23,7 @@
     PositionType,
 )
 from .water_info import WaterAmount, WaterInfoEvent
+from .work_mode import WorkMode, WorkModeEvent
 
 __all__ = [
     "BatteryEvent",
@@ -44,6 +45,8 @@
     "PositionsEvent",
     "WaterAmount",
     "WaterInfoEvent",
+    "WorkMode",
+    "WorkModeEvent",
 ]
 
 

deebot_client/events/work_mode.py

@@ -0,0 +1,22 @@
+"""Work mode event module."""
+from dataclasses import dataclass
+
+from deebot_client.util import DisplayNameIntEnum
+
+from .base import Event
+
+
+class WorkMode(DisplayNameIntEnum):
+    """Enum class for all possible work modes."""
+
+    VACUUM_AND_MOP = 0
+    VACUUM = 1
+    MOP = 2
+    MOP_AFTER_VACUUM = 3
+
+
+@dataclass(frozen=True)
+class WorkModeEvent(Event):
+    """Work mode event representation."""
+
+    mode: WorkMode

deebot_client/hardware/deebot/fallback.py

@@ -1,4 +1,4 @@
-"""Deebot ozmo 950 Capabilities."""
+"""Fallback Capabilities."""
 from deebot_client.capabilities import (
     Capabilities,
     CapabilityClean,
@@ -82,10 +82,11 @@
     WaterInfoEvent,
 )
 from deebot_client.models import StaticDeviceInfo
+from deebot_client.util import short_name
 
-from . import DEVICES, FALLBACK
+from . import DEVICES
 
-DEVICES[FALLBACK] = StaticDeviceInfo(
+DEVICES[short_name(__name__)] = StaticDeviceInfo(
     DataType.JSON,
     Capabilities(
         availability=CapabilityEvent(AvailabilityEvent, [GetBattery(True)]),

deebot_client/hardware/deebot/p1jij8.py

@@ -0,0 +1,188 @@
+"""Deebot T20 Omni Capabilities."""
+from deebot_client.capabilities import (
+    Capabilities,
+    CapabilityClean,
+    CapabilityCleanAction,
+    CapabilityCustomCommand,
+    CapabilityEvent,
+    CapabilityExecute,
+    CapabilityLifeSpan,
+    CapabilityMap,
+    CapabilitySet,
+    CapabilitySetEnable,
+    CapabilitySettings,
+    CapabilitySetTypes,
+    CapabilityStats,
+)
+from deebot_client.commands.json.advanced_mode import GetAdvancedMode, SetAdvancedMode
+from deebot_client.commands.json.battery import GetBattery
+from deebot_client.commands.json.carpet import (
+    GetCarpetAutoFanBoost,
+    SetCarpetAutoFanBoost,
+)
+from deebot_client.commands.json.charge import Charge
+from deebot_client.commands.json.charge_state import GetChargeState
+from deebot_client.commands.json.clean import Clean, CleanArea, GetCleanInfo
+from deebot_client.commands.json.clean_count import GetCleanCount, SetCleanCount
+from deebot_client.commands.json.clean_logs import GetCleanLogs
+from deebot_client.commands.json.clean_preference import (
+    GetCleanPreference,
+    SetCleanPreference,
+)
+from deebot_client.commands.json.continuous_cleaning import (
+    GetContinuousCleaning,
+    SetContinuousCleaning,
+)
+from deebot_client.commands.json.custom import CustomCommand
+from deebot_client.commands.json.error import GetError
+from deebot_client.commands.json.fan_speed import GetFanSpeed, SetFanSpeed
+from deebot_client.commands.json.life_span import GetLifeSpan, ResetLifeSpan
+from deebot_client.commands.json.map import GetCachedMapInfo, GetMajorMap, GetMapTrace
+from deebot_client.commands.json.multimap_state import (
+    GetMultimapState,
+    SetMultimapState,
+)
+from deebot_client.commands.json.play_sound import PlaySound
+from deebot_client.commands.json.pos import GetPos
+from deebot_client.commands.json.relocation import SetRelocationState
+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.water_info import GetWaterInfo, SetWaterInfo
+from deebot_client.commands.json.work_mode import GetWorkMode, SetWorkMode
+from deebot_client.const import DataType
+from deebot_client.events import (
+    AdvancedModeEvent,
+    AvailabilityEvent,
+    BatteryEvent,
+    CachedMapInfoEvent,
+    CarpetAutoFanBoostEvent,
+    CleanCountEvent,
+    CleanLogEvent,
+    CleanPreferenceEvent,
+    ContinuousCleaningEvent,
+    CustomCommandEvent,
+    ErrorEvent,
+    FanSpeedEvent,
+    FanSpeedLevel,
+    LifeSpan,
+    LifeSpanEvent,
+    MajorMapEvent,
+    MapChangedEvent,
+    MapTraceEvent,
+    MultimapStateEvent,
+    PositionsEvent,
+    ReportStatsEvent,
+    RoomsEvent,
+    StateEvent,
+    StatsEvent,
+    TotalStatsEvent,
+    TrueDetectEvent,
+    VolumeEvent,
+    WaterAmount,
+    WaterInfoEvent,
+    WorkMode,
+    WorkModeEvent,
+)
+from deebot_client.models import StaticDeviceInfo
+from deebot_client.util import short_name
+
+from . import DEVICES
+
+DEVICES[short_name(__name__)] = StaticDeviceInfo(
+    DataType.JSON,
+    Capabilities(
+        availability=CapabilityEvent(AvailabilityEvent, [GetBattery(True)]),
+        battery=CapabilityEvent(BatteryEvent, [GetBattery()]),
+        charge=CapabilityExecute(Charge),
+        clean=CapabilityClean(
+            action=CapabilityCleanAction(command=Clean, area=CleanArea),
+            continuous=CapabilitySetEnable(
+                ContinuousCleaningEvent,
+                [GetContinuousCleaning()],
+                SetContinuousCleaning,
+            ),
+            count=CapabilitySet(CleanCountEvent, [GetCleanCount()], SetCleanCount),
+            log=CapabilityEvent(CleanLogEvent, [GetCleanLogs()]),
+            preference=CapabilitySetEnable(
+                CleanPreferenceEvent, [GetCleanPreference()], SetCleanPreference
+            ),
+            work_mode=CapabilitySetTypes(
+                event=WorkModeEvent,
+                get=[GetWorkMode()],
+                set=SetWorkMode,
+                types=(
+                    WorkMode.MOP,
+                    WorkMode.MOP_AFTER_VACUUM,
+                    WorkMode.VACUUM,
+                    WorkMode.VACUUM_AND_MOP,
+                ),
+            ),
+        ),
+        custom=CapabilityCustomCommand(
+            event=CustomCommandEvent, get=[], set=CustomCommand
+        ),
+        error=CapabilityEvent(ErrorEvent, [GetError()]),
+        fan_speed=CapabilitySetTypes(
+            event=FanSpeedEvent,
+            get=[GetFanSpeed()],
+            set=SetFanSpeed,
+            types=(
+                FanSpeedLevel.QUIET,
+                FanSpeedLevel.NORMAL,
+                FanSpeedLevel.MAX,
+                FanSpeedLevel.MAX_PLUS,
+            ),
+        ),
+        life_span=CapabilityLifeSpan(
+            types=(LifeSpan.BRUSH, LifeSpan.FILTER, LifeSpan.SIDE_BRUSH),
+            event=LifeSpanEvent,
+            get=[GetLifeSpan([LifeSpan.BRUSH, LifeSpan.FILTER, LifeSpan.SIDE_BRUSH])],
+            reset=ResetLifeSpan,
+        ),
+        map=CapabilityMap(
+            chached_info=CapabilityEvent(CachedMapInfoEvent, [GetCachedMapInfo()]),
+            changed=CapabilityEvent(MapChangedEvent, []),
+            major=CapabilityEvent(MajorMapEvent, [GetMajorMap()]),
+            multi_state=CapabilitySetEnable(
+                MultimapStateEvent, [GetMultimapState()], SetMultimapState
+            ),
+            position=CapabilityEvent(PositionsEvent, [GetPos()]),
+            relocation=CapabilityExecute(SetRelocationState),
+            rooms=CapabilityEvent(RoomsEvent, [GetCachedMapInfo()]),
+            trace=CapabilityEvent(MapTraceEvent, [GetMapTrace()]),
+        ),
+        play_sound=CapabilityExecute(PlaySound),
+        settings=CapabilitySettings(
+            advanced_mode=CapabilitySetEnable(
+                AdvancedModeEvent, [GetAdvancedMode()], SetAdvancedMode
+            ),
+            carpet_auto_fan_boost=CapabilitySetEnable(
+                CarpetAutoFanBoostEvent,
+                [GetCarpetAutoFanBoost()],
+                SetCarpetAutoFanBoost,
+            ),
+            true_detect=CapabilitySetEnable(
+                TrueDetectEvent, [GetTrueDetect()], SetTrueDetect
+            ),
+            volume=CapabilitySet(VolumeEvent, [GetVolume()], SetVolume),
+        ),
+        state=CapabilityEvent(StateEvent, [GetChargeState(), GetCleanInfo()]),
+        stats=CapabilityStats(
+            clean=CapabilityEvent(StatsEvent, [GetStats()]),
+            report=CapabilityEvent(ReportStatsEvent, []),
+            total=CapabilityEvent(TotalStatsEvent, [GetTotalStats()]),
+        ),
+        water=CapabilitySetTypes(
+            event=WaterInfoEvent,
+            get=[GetWaterInfo()],
+            set=SetWaterInfo,
+            types=(
+                WaterAmount.LOW,
+                WaterAmount.MEDIUM,
+                WaterAmount.HIGH,
+                WaterAmount.ULTRAHIGH,
+            ),
+        ),
+    ),
+)

tests/commands/json/test_work_mode.py

@@ -0,0 +1,43 @@
+from typing import Any
+
+import pytest
+
+from deebot_client.commands.json import GetWorkMode, SetWorkMode
+from deebot_client.events import WorkMode, WorkModeEvent
+from tests.helpers import (
+    get_request_json,
+    get_success_body,
+    verify_DisplayNameEnum_unique,
+)
+
+from . import assert_command, assert_set_command
+
+
+def test_WorkMode_unique() -> None:
+    verify_DisplayNameEnum_unique(WorkMode)
+
+
+@pytest.mark.parametrize(
+    ("json", "expected"),
+    [
+        ({"mode": 0}, WorkModeEvent(WorkMode.VACUUM_AND_MOP)),
+        ({"mode": 1}, WorkModeEvent(WorkMode.VACUUM)),
+        ({"mode": 2}, WorkModeEvent(WorkMode.MOP)),
+        ({"mode": 3}, WorkModeEvent(WorkMode.MOP_AFTER_VACUUM)),
+    ],
+)
+async def test_GetWaterInfo(json: dict[str, Any], expected: WorkModeEvent) -> None:
+    json = get_request_json(get_success_body(json))
+    await assert_command(GetWorkMode(), json, expected)
+
+
+@pytest.mark.parametrize(("value"), [WorkMode.MOP_AFTER_VACUUM, "mop_after_vacuum"])
+async def test_SetWaterInfo(value: WorkMode | str) -> None:
+    command = SetWorkMode(value)
+    args = {"mode": 3}
+    await assert_set_command(command, args, WorkModeEvent(WorkMode.MOP_AFTER_VACUUM))
+
+
+def test_SetWaterInfo_inexisting_value() -> None:
+    with pytest.raises(ValueError, match="'INEXSTING' is not a valid WorkMode member"):
+        SetWorkMode("inexsting")