Split WaterInfoEvent into multiple (#917)

Author: edenhaus

deebot_client/capabilities.py

@@ -48,11 +48,10 @@
     TrueDetectEvent,
     VoiceAssistantStateEvent,
     VolumeEvent,
-    WaterAmount,
-    WaterInfoEvent,
     WorkMode,
     WorkModeEvent,
     auto_empty,
+    water_info,
 )
 
 if TYPE_CHECKING:
@@ -229,6 +228,18 @@ class CapabilityStation:
     state: CapabilityEvent[StationEvent]
 
 
+@dataclass(frozen=True, kw_only=True)
+class CapabilityWater:
+    """Capabilities for water."""
+
+    amount: CapabilitySetTypes[
+        water_info.WaterAmountEvent,
+        [water_info.WaterAmount | str],
+        water_info.WaterAmount,
+    ]
+    mop_attached: CapabilityEvent[water_info.MopAttachedEvent]
+
+
 @dataclass(frozen=True, kw_only=True)
 class Capabilities(ABC):
     """Capabilities."""
@@ -252,9 +263,7 @@ class Capabilities(ABC):
     state: CapabilityEvent[StateEvent]
     station: CapabilityStation | None = None
     stats: CapabilityStats
-    water: (
-        CapabilitySetTypes[WaterInfoEvent, [WaterAmount | str], WaterAmount] | None
-    ) = None
+    water: CapabilityWater | None = None
 
     _events: MappingProxyType[type[Event], list[Command]] = field(init=False)
 

deebot_client/commands/json/water_info.py

@@ -6,7 +6,13 @@
 from typing import TYPE_CHECKING, Any
 
 from deebot_client.command import InitParam
-from deebot_client.events import SweepType, WaterAmount, WaterInfoEvent
+from deebot_client.events.water_info import (
+    MopAttachedEvent,
+    SweepType,
+    WaterAmount,
+    WaterAmountEvent,
+    WaterSweepTypeEvent,
+)
 from deebot_client.message import HandlingResult
 from deebot_client.util import get_enum
 
@@ -29,20 +35,14 @@ def _handle_body_data_dict(
 
         :return: A message response
         """
-        mop_attached = data.get("enable")
-        if mop_attached is not None:
-            mop_attached = bool(mop_attached)
+        event_bus.notify(WaterAmountEvent(WaterAmount(int(data["amount"]))))
+
+        if (mop_attached := data.get("enable")) is not None:
+            event_bus.notify(MopAttachedEvent(bool(mop_attached)))
 
         if sweep_type := data.get("sweepType"):
-            sweep_type = SweepType(int(sweep_type))
-
-        event_bus.notify(
-            WaterInfoEvent(
-                WaterAmount(int(data["amount"])),
-                sweep_type,
-                mop_attached=mop_attached,
-            )
-        )
+            event_bus.notify(WaterSweepTypeEvent(SweepType(int(sweep_type))))
+
         return HandlingResult.success()
 
 

deebot_client/events/__init__.py

@@ -8,7 +8,7 @@
 
 from deebot_client.util.enum import StrEnumWithXml
 
-from . import auto_empty, station
+from . import auto_empty, station, water_info
 from .auto_empty import AutoEmptyEvent
 from .base import Event
 from .efficiency_mode import EfficiencyMode, EfficiencyModeEvent
@@ -27,7 +27,6 @@
 )
 from .network import NetworkInfoEvent
 from .station import StationEvent
-from .water_info import SweepType, WaterAmount, WaterInfoEvent
 from .work_mode import WorkMode, WorkModeEvent
 
 if TYPE_CHECKING:
@@ -56,13 +55,11 @@
     "PositionsEvent",
     "StationEvent",
     "SweepModeEvent",
-    "SweepType",
-    "WaterAmount",
-    "WaterInfoEvent",
     "WorkMode",
     "WorkModeEvent",
     "auto_empty",
     "station",
+    "water_info",
 ]
 
 

deebot_client/events/base.py

@@ -2,6 +2,15 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+
 
 class Event:
     """Event base class."""
+
+
+@dataclass(frozen=True)
+class ValueEvent[T](Event):
+    """Value event class."""
+
+    value: T

deebot_client/events/water_info.py

@@ -2,10 +2,17 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass, field
 from enum import IntEnum, unique
 
-from .base import Event
+from .base import ValueEvent
+
+__all__ = [
+    "MopAttachedEvent",
+    "SweepType",
+    "WaterAmount",
+    "WaterAmountEvent",
+    "WaterSweepTypeEvent",
+]
 
 
 @unique
@@ -26,11 +33,6 @@ class SweepType(IntEnum):
     DEEP = 2
 
 
-@dataclass(frozen=True)
-class WaterInfoEvent(Event):
-    """Water info event representation."""
-
-    amount: WaterAmount
-    # None means no data available
-    sweep_type: SweepType | None = None
-    mop_attached: bool | None = field(kw_only=True, default=None)
+WaterAmountEvent = ValueEvent[WaterAmount]
+WaterSweepTypeEvent = ValueEvent[SweepType]
+MopAttachedEvent = ValueEvent[bool]

deebot_client/hardware/deebot/2o4lnm.py

@@ -16,6 +16,7 @@
     CapabilitySettings,
     CapabilitySetTypes,
     CapabilityStats,
+    CapabilityWater,
     DeviceType,
 )
 from deebot_client.commands.json.advanced_mode import GetAdvancedMode, SetAdvancedMode
@@ -91,8 +92,7 @@
     TrueDetectEvent,
     VoiceAssistantStateEvent,
     VolumeEvent,
-    WaterAmount,
-    WaterInfoEvent,
+    water_info,
 )
 from deebot_client.models import StaticDeviceInfo
 from deebot_client.util import short_name
@@ -200,15 +200,18 @@
             report=CapabilityEvent(ReportStatsEvent, []),
             total=CapabilityEvent(TotalStatsEvent, [GetTotalStats()]),
         ),
-        water=CapabilitySetTypes(
-            event=WaterInfoEvent,
-            get=[GetWaterInfo()],
-            set=SetWaterInfo,
-            types=(
-                WaterAmount.LOW,
-                WaterAmount.MEDIUM,
-                WaterAmount.HIGH,
+        water=CapabilityWater(
+            amount=CapabilitySetTypes(
+                event=water_info.WaterAmountEvent,
+                get=[GetWaterInfo()],
+                set=SetWaterInfo,
+                types=(
+                    water_info.WaterAmount.LOW,
+                    water_info.WaterAmount.MEDIUM,
+                    water_info.WaterAmount.HIGH,
+                ),
             ),
+            mop_attached=CapabilityEvent(water_info.MopAttachedEvent, [GetWaterInfo()]),
         ),
     ),
 )

deebot_client/hardware/deebot/4bdkrs.py

@@ -18,6 +18,7 @@
     CapabilitySetTypes,
     CapabilityStation,
     CapabilityStats,
+    CapabilityWater,
     DeviceType,
 )
 from deebot_client.commands import StationAction
@@ -105,11 +106,10 @@
     TrueDetectEvent,
     VoiceAssistantStateEvent,
     VolumeEvent,
-    WaterAmount,
-    WaterInfoEvent,
     WorkMode,
     WorkModeEvent,
     auto_empty,
+    water_info,
 )
 from deebot_client.events.auto_empty import AutoEmptyEvent
 from deebot_client.events.efficiency_mode import EfficiencyMode
@@ -258,16 +258,19 @@
             report=CapabilityEvent(ReportStatsEvent, []),
             total=CapabilityEvent(TotalStatsEvent, [GetTotalStats()]),
         ),
-        water=CapabilitySetTypes(
-            event=WaterInfoEvent,
-            get=[GetWaterInfo()],
-            set=SetWaterInfo,
-            types=(
-                WaterAmount.LOW,
-                WaterAmount.MEDIUM,
-                WaterAmount.HIGH,
-                WaterAmount.ULTRAHIGH,
+        water=CapabilityWater(
+            amount=CapabilitySetTypes(
+                event=water_info.WaterAmountEvent,
+                get=[GetWaterInfo()],
+                set=SetWaterInfo,
+                types=(
+                    water_info.WaterAmount.LOW,
+                    water_info.WaterAmount.MEDIUM,
+                    water_info.WaterAmount.HIGH,
+                    water_info.WaterAmount.ULTRAHIGH,
+                ),
             ),
+            mop_attached=CapabilityEvent(water_info.MopAttachedEvent, [GetWaterInfo()]),
         ),
     ),
 )

deebot_client/hardware/deebot/7bryc5.py

@@ -16,6 +16,7 @@
     CapabilitySettings,
     CapabilitySetTypes,
     CapabilityStats,
+    CapabilityWater,
     DeviceType,
 )
 from deebot_client.commands.json.advanced_mode import GetAdvancedMode, SetAdvancedMode
@@ -84,8 +85,7 @@
     TotalStatsEvent,
     TrueDetectEvent,
     VolumeEvent,
-    WaterAmount,
-    WaterInfoEvent,
+    water_info,
 )
 from deebot_client.models import StaticDeviceInfo
 from deebot_client.util import short_name
@@ -183,16 +183,19 @@
             report=CapabilityEvent(ReportStatsEvent, []),
             total=CapabilityEvent(TotalStatsEvent, [GetTotalStats()]),
         ),
-        water=CapabilitySetTypes(
-            event=WaterInfoEvent,
-            get=[GetWaterInfo()],
-            set=SetWaterInfo,
-            types=(
-                WaterAmount.LOW,
-                WaterAmount.MEDIUM,
-                WaterAmount.HIGH,
-                WaterAmount.ULTRAHIGH,
+        water=CapabilityWater(
+            amount=CapabilitySetTypes(
+                event=water_info.WaterAmountEvent,
+                get=[GetWaterInfo()],
+                set=SetWaterInfo,
+                types=(
+                    water_info.WaterAmount.LOW,
+                    water_info.WaterAmount.MEDIUM,
+                    water_info.WaterAmount.HIGH,
+                    water_info.WaterAmount.ULTRAHIGH,
+                ),
             ),
+            mop_attached=CapabilityEvent(water_info.MopAttachedEvent, [GetWaterInfo()]),
         ),
     ),
 )

deebot_client/hardware/deebot/7j1tu6.py

@@ -14,6 +14,7 @@
     CapabilitySettings,
     CapabilitySetTypes,
     CapabilityStats,
+    CapabilityWater,
     DeviceType,
 )
 from deebot_client.commands.json.battery import GetBattery
@@ -47,8 +48,7 @@
     StatsEvent,
     TotalStatsEvent,
     VolumeEvent,
-    WaterAmount,
-    WaterInfoEvent,
+    water_info,
 )
 from deebot_client.models import StaticDeviceInfo
 from deebot_client.util import short_name
@@ -110,15 +110,18 @@
             report=CapabilityEvent(ReportStatsEvent, []),
             total=CapabilityEvent(TotalStatsEvent, [GetTotalStats()]),
         ),
-        water=CapabilitySetTypes(
-            event=WaterInfoEvent,
-            get=[GetWaterInfo()],
-            set=SetWaterInfo,
-            types=(
-                WaterAmount.LOW,
-                WaterAmount.MEDIUM,
-                WaterAmount.HIGH,
+        water=CapabilityWater(
+            amount=CapabilitySetTypes(
+                event=water_info.WaterAmountEvent,
+                get=[GetWaterInfo()],
+                set=SetWaterInfo,
+                types=(
+                    water_info.WaterAmount.LOW,
+                    water_info.WaterAmount.MEDIUM,
+                    water_info.WaterAmount.HIGH,
+                ),
             ),
+            mop_attached=CapabilityEvent(water_info.MopAttachedEvent, [GetWaterInfo()]),
         ),
     ),
 )