Implement vacuum dustbin module (dust_bucket) (#1423)

Initial implementation for dustbin auto-emptying.

New features:
- `dustbin_empty` action to empty the dustbin immediately
- `dustbin_autocollection_enabled` to toggle the auto collection
- `dustbin_mode` to choose how often the auto collection is performed

devtools/helpers/smartrequests.py

@@ -455,7 +455,10 @@ COMPONENT_REQUESTS = {
         SmartRequest.get_raw_request("getMapData"),
     ],
     "auto_change_map": [SmartRequest.get_raw_request("getAutoChangeMap")],
-    "dust_bucket": [SmartRequest.get_raw_request("getAutoDustCollection")],
+    "dust_bucket": [
+        SmartRequest.get_raw_request("getAutoDustCollection"),
+        SmartRequest.get_raw_request("getDustCollectionInfo"),
+    ],
     "mop": [SmartRequest.get_raw_request("getMopState")],
     "do_not_disturb": [SmartRequest.get_raw_request("getDoNotDisturb")],
     "charge_pose_clean": [],

kasa/module.py

@@ -163,6 +163,7 @@ class Module(ABC):
 
     # Vacuum modules
     Clean: Final[ModuleName[smart.Clean]] = ModuleName("Clean")
+    Dustbin: Final[ModuleName[smart.Dustbin]] = ModuleName("Dustbin")
 
     def __init__(self, device: Device, module: str) -> None:
         self._device = device

kasa/smart/modules/__init__.py

@@ -13,6 +13,7 @@ from .color import Color
 from .colortemperature import ColorTemperature
 from .contactsensor import ContactSensor
 from .devicemodule import DeviceModule
+from .dustbin import Dustbin
 from .energy import Energy
 from .fan import Fan
 from .firmware import Firmware
@@ -72,4 +73,5 @@ __all__ = [
     "OverheatProtection",
     "HomeKit",
     "Matter",
+    "Dustbin",
 ]

kasa/smart/modules/dustbin.py

@@ -0,0 +1,117 @@
+"""Implementation of vacuum dustbin."""
+
+from __future__ import annotations
+
+import logging
+from enum import IntEnum
+
+from ...feature import Feature
+from ..smartmodule import SmartModule
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class Mode(IntEnum):
+    """Dust collection modes."""
+
+    Smart = 0
+    Light = 1
+    Balanced = 2
+    Max = 3
+
+
+class Dustbin(SmartModule):
+    """Implementation of vacuum dustbin."""
+
+    REQUIRED_COMPONENT = "dust_bucket"
+
+    def _initialize_features(self) -> None:
+        """Initialize features."""
+        self._add_feature(
+            Feature(
+                self._device,
+                id="dustbin_empty",
+                name="Empty dustbin",
+                container=self,
+                attribute_setter="start_emptying",
+                category=Feature.Category.Primary,
+                type=Feature.Action,
+            )
+        )
+
+        self._add_feature(
+            Feature(
+                self._device,
+                id="dustbin_autocollection_enabled",
+                name="Automatic emptying enabled",
+                container=self,
+                attribute_getter="auto_collection",
+                attribute_setter="set_auto_collection",
+                category=Feature.Category.Config,
+                type=Feature.Switch,
+            )
+        )
+
+        self._add_feature(
+            Feature(
+                self._device,
+                id="dustbin_mode",
+                name="Automatic emptying mode",
+                container=self,
+                attribute_getter="mode",
+                attribute_setter="set_mode",
+                icon="mdi:fan",
+                choices_getter=lambda: list(Mode.__members__),
+                category=Feature.Category.Config,
+                type=Feature.Type.Choice,
+            )
+        )
+
+    def query(self) -> dict:
+        """Query to execute during the update cycle."""
+        return {
+            "getAutoDustCollection": {},
+            "getDustCollectionInfo": {},
+        }
+
+    async def start_emptying(self) -> dict:
+        """Start emptying the bin."""
+        return await self.call(
+            "setSwitchDustCollection",
+            {
+                "switch_dust_collection": True,
+            },
+        )
+
+    @property
+    def _settings(self) -> dict:
+        """Return auto-empty settings."""
+        return self.data["getDustCollectionInfo"]
+
+    @property
+    def mode(self) -> str:
+        """Return auto-emptying mode."""
+        return Mode(self._settings["dust_collection_mode"]).name
+
+    async def set_mode(self, mode: str) -> dict:
+        """Set auto-emptying mode."""
+        name_to_value = {x.name: x.value for x in Mode}
+        if mode not in name_to_value:
+            raise ValueError(
+                "Invalid auto/emptying mode speed %s, available %s", mode, name_to_value
+            )
+
+        settings = self._settings.copy()
+        settings["dust_collection_mode"] = name_to_value[mode]
+        return await self.call("setDustCollectionInfo", settings)
+
+    @property
+    def auto_collection(self) -> dict:
+        """Return auto-emptying config."""
+        return self._settings["auto_dust_collection"]
+
+    async def set_auto_collection(self, on: bool) -> dict:
+        """Toggle auto-emptying."""
+        settings = self._settings.copy()
+        settings["auto_dust_collection"] = on
+        return await self.call("setDustCollectionInfo", settings)

pyproject.toml

@@ -112,7 +112,7 @@ markers = [
 ]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
-timeout = 10
+#timeout = 10
 # dist=loadgroup enables grouping of tests into single worker.
 # required as caplog doesn't play nicely with multiple workers.
 addopts = "--disable-socket --allow-unix-socket --dist=loadgroup"

tests/fakeprotocol_smart.py

@@ -640,7 +640,12 @@ class FakeSmartTransport(BaseTransport):
         elif method[:3] == "set":
             target_method = f"get{method[3:]}"
             # Some vacuum commands do not have a getter
-            if method in ["setRobotPause", "setSwitchClean", "setSwitchCharge"]:
+            if method in [
+                "setRobotPause",
+                "setSwitchClean",
+                "setSwitchCharge",
+                "setSwitchDustCollection",
+            ]:
                 return {"error_code": 0}
 
             info[target_method].update(params)

tests/fixtures/smart/RV20 Max Plus(EU)_1.0_1.0.7.json

@@ -202,6 +202,10 @@
     "getMopState": {
         "mop_state": false
     },
+    "getDustCollectionInfo": {
+        "auto_dust_collection": true,
+        "dust_collection_mode": 0
+    },
     "getVacStatus": {
         "err_status": [
             0

tests/smart/modules/test_dustbin.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import pytest
+from pytest_mock import MockerFixture
+
+from kasa import Module
+from kasa.smart import SmartDevice
+from kasa.smart.modules.dustbin import Mode
+
+from ...device_fixtures import get_parent_and_child_modules, parametrize
+
+dustbin = parametrize(
+    "has dustbin", component_filter="dust_bucket", protocol_filter={"SMART"}
+)
+
+
+@dustbin
+@pytest.mark.parametrize(
+    ("feature", "prop_name", "type"),
+    [
+        ("dustbin_autocollection_enabled", "auto_collection", bool),
+        ("dustbin_mode", "mode", str),
+    ],
+)
+async def test_features(dev: SmartDevice, feature: str, prop_name: str, type: type):
+    """Test that features are registered and work as expected."""
+    dustbin = next(get_parent_and_child_modules(dev, Module.Dustbin))
+    assert dustbin is not None
+
+    prop = getattr(dustbin, prop_name)
+    assert isinstance(prop, type)
+
+    feat = dustbin._device.features[feature]
+    assert feat.value == prop
+    assert isinstance(feat.value, type)
+
+
+@dustbin
+async def test_dustbin_mode(dev: SmartDevice, mocker: MockerFixture):
+    """Test dust mode."""
+    dustbin = next(get_parent_and_child_modules(dev, Module.Dustbin))
+    call = mocker.spy(dustbin, "call")
+
+    mode_feature = dustbin._device.features["dustbin_mode"]
+    assert dustbin.mode == mode_feature.value
+
+    new_mode = Mode.Max
+    await dustbin.set_mode(new_mode.name)
+
+    params = dustbin._settings.copy()
+    params["dust_collection_mode"] = new_mode.value
+
+    call.assert_called_with("setDustCollectionInfo", params)
+
+    await dev.update()
+
+    assert dustbin.mode == new_mode.name
+
+    with pytest.raises(ValueError, match="Invalid auto/emptying mode speed"):
+        await dustbin.set_mode("invalid")
+
+
+@dustbin
+async def test_autocollection(dev: SmartDevice, mocker: MockerFixture):
+    """Test autocollection switch."""
+    dustbin = next(get_parent_and_child_modules(dev, Module.Dustbin))
+    call = mocker.spy(dustbin, "call")
+
+    auto_collection = dustbin._device.features["dustbin_autocollection_enabled"]
+    assert dustbin.auto_collection == auto_collection.value
+
+    await auto_collection.set_value(True)
+
+    params = dustbin._settings.copy()
+    params["auto_dust_collection"] = True
+
+    call.assert_called_with("setDustCollectionInfo", params)
+
+    await dev.update()
+
+    assert dustbin.auto_collection is True
+
+
+@dustbin
+async def test_empty_dustbin(dev: SmartDevice, mocker: MockerFixture):
+    """Test the empty dustbin feature."""
+    dustbin = next(get_parent_and_child_modules(dev, Module.Dustbin))
+    call = mocker.spy(dustbin, "call")
+
+    await dustbin.start_emptying()
+
+    call.assert_called_with("setSwitchDustCollection", {"switch_dust_collection": True})