Add features for newer vacuums (eg Roborock S7)

miio/tests/test_vacuum.py

@@ -5,6 +5,7 @@
 import pytest
 
 from miio import Vacuum, VacuumStatus
+from miio.vacuum import CarpetCleaningMode, MopMode
 
 from .dummies import DummyDevice
 
@@ -275,3 +276,27 @@ def test_history_empty(self):
             )
 
             assert len(self.device.clean_history().ids) == 0
+
+    def test_carpet_cleaning_mode(self):
+        with patch.object(self.device, "send", return_value=[{"carpet_clean_mode": 0}]):
+            assert self.device.carpet_cleaning_mode() == CarpetCleaningMode.Avoid
+
+        with patch.object(self.device, "send", return_value="unknown_method"):
+            assert self.device.carpet_cleaning_mode() is None
+
+        with patch.object(self.device, "send", return_value=["ok"]) as mock_method:
+            assert self.device.set_carpet_cleaning_mode(CarpetCleaningMode.Rise) is True
+            mock_method.assert_called_once_with(
+                "set_carpet_clean_mode", {"carpet_clean_mode": 1}
+            )
+
+    def test_mop_mode(self):
+        with patch.object(self.device, "send", return_value=["ok"]) as mock_method:
+            assert self.device.set_mop_mode(MopMode.Deep) is True
+            mock_method.assert_called_once_with("set_mop_mode", [301])
+
+        with patch.object(self.device, "send", return_value=[300]):
+            assert self.device.mop_mode() == MopMode.Standard
+
+        with patch.object(self.device, "send", return_value=[32453]):
+            assert self.device.mop_mode() is None

miio/vacuum.py

@@ -84,6 +84,13 @@ class FanspeedE2(enum.Enum):
     Turbo = 100
 
 
+class FanspeedS7(enum.Enum):
+    Silent = 101
+    Standard = 102
+    Medium = 103
+    Turbo = 104
+
+
 class WaterFlow(enum.Enum):
     """Water flow strength on s5 max."""
 
@@ -93,10 +100,26 @@ class WaterFlow(enum.Enum):
     Maximum = 203
 
 
+class MopMode(enum.Enum):
+    """Mop routing on S7."""
+
+    Standard = 300
+    Deep = 301
+
+
+class CarpetCleaningMode(enum.Enum):
+    """Type of carpet cleaning/avoidance."""
+
+    Avoid = 0
+    Rise = 1
+    Ignore = 2
+
+
 ROCKROBO_V1 = "rockrobo.vacuum.v1"
 ROCKROBO_S5 = "roborock.vacuum.s5"
 ROCKROBO_S6 = "roborock.vacuum.s6"
 ROCKROBO_S6_MAXV = "roborock.vacuum.a10"
+ROCKROBO_S7 = "roborock.vacuum.a15"
 
 
 class Vacuum(Device):
@@ -546,6 +569,8 @@ def _autodetect_model(self):
                 self._fanspeeds = FanspeedV1
         elif self.model == "roborock.vacuum.e2":
             self._fanspeeds = FanspeedE2
+        elif self.model == ROCKROBO_S7:
+            self._fanspeeds = FanspeedS7
         else:
             self._fanspeeds = FanspeedV2
 
@@ -681,6 +706,25 @@ def set_carpet_mode(
         }
         return self.send("set_carpet_mode", [data])[0] == "ok"
 
+    @command()
+    def carpet_cleaning_mode(self) -> CarpetCleaningMode:
+        """Get carpet cleaning mode/avoidance setting."""
+        try:
+            return CarpetCleaningMode(
+                self.send("get_carpet_clean_mode")[0]["carpet_clean_mode"]
+            )
+        except Exception as err:
+            _LOGGER.warning("Error while requesting carpet clean mode: %s", err)
+            return None
+
+    @command(click.argument("mode", type=EnumType(CarpetCleaningMode)))
+    def set_carpet_cleaning_mode(self, mode: CarpetCleaningMode):
+        """Set carpet cleaning mode/avoidance setting."""
+        return (
+            self.send("set_carpet_clean_mode", {"carpet_clean_mode": mode.value})[0]
+            == "ok"
+        )
+
     @command()
     def stop_zoned_clean(self):
         """Stop cleaning a zone."""
@@ -747,6 +791,30 @@ def set_waterflow(self, waterflow: WaterFlow):
         """Set water flow setting."""
         return self.send("set_water_box_custom_mode", [waterflow.value])
 
+    @command()
+    def mop_mode(self) -> Optional[MopMode]:
+        """Get mop mode setting."""
+        try:
+            return MopMode(self.send("get_mop_mode")[0])
+        except ValueError as err:
+            _LOGGER.warning("Device returned unknown MopMode: %s", err)
+            return None
+
+    @command(click.argument("mop_mode", type=EnumType(MopMode)))
+    def set_mop_mode(self, mop_mode: MopMode):
+        """Set mop mode setting."""
+        return self.send("set_mop_mode", [mop_mode.value])[0] == "ok"
+
+    @command()
+    def child_lock(self) -> bool:
+        """Get child lock setting."""
+        return self.send("get_child_lock_status")["lock_status"] == 1
+
+    @command(click.argument("lock", type=bool))
+    def set_child_lock(self, lock: bool) -> bool:
+        """Set child lock setting."""
+        return self.send("set_child_lock_status", {"lock_status": int(lock)})[0] == "ok"
+
     @classmethod
     def get_device_group(cls):
         @click.pass_context

miio/vacuum_cli.py

@@ -24,6 +24,7 @@
 from miio.exceptions import DeviceInfoUnavailableException
 from miio.miioprotocol import MiIOProtocol
 from miio.updater import OneShotServer
+from miio.vacuum import CarpetCleaningMode
 
 _LOGGER = logging.getLogger(__name__)
 pass_dev = click.make_pass_decorator(miio.Device, ensure=True)
@@ -115,6 +116,8 @@ def status(vac: miio.Vacuum):
 
     if res.error_code:
         click.echo(click.style("Error: %s !" % res.error, bold=True, fg="red"))
+    if res.is_water_shortage:
+        click.echo(click.style("Water is running low!", bold=True, fg="blue"))
     click.echo(click.style("State: %s" % res.state, bold=True))
     click.echo("Battery: %s %%" % res.battery)
     click.echo("Fanspeed: %s %%" % res.fanspeed)
@@ -124,6 +127,8 @@ def status(vac: miio.Vacuum):
     # click.echo("Map present: %s" % res.map)
     # click.echo("in_cleaning: %s" % res.in_cleaning)
     click.echo("Water box attached: %s" % res.is_water_box_attached)
+    if res.is_water_box_carriage_attached is not None:
+        click.echo("Mop attached: %s" % res.is_water_box_carriage_attached)
 
 
 @cli.command()
@@ -557,6 +562,24 @@ def carpet_mode(vac: miio.Vacuum, enabled=None):
         click.echo(vac.set_carpet_mode(enabled))
 
 
+@cli.command()
+@click.argument("mode", required=False, type=str)
+@pass_dev
+def carpet_cleaning_mode(vac: miio.Vacuum, mode=None):
+    """Query or set the carpet cleaning/avoidance mode.
+
+    Allowed values: Avoid, Rise, Ignore
+    """
+
+    if mode is None:
+        click.echo("Carpet cleaning mode: %s" % vac.carpet_cleaning_mode())
+    else:
+        click.echo(
+            "Setting carpet cleaning mode: %s"
+            % vac.set_carpet_cleaning_mode(CarpetCleaningMode[mode])
+        )
+
+
 @cli.command()
 @click.argument("ssid", required=True)
 @click.argument("password", required=True)

miio/vacuumcontainers.py

@@ -69,6 +69,21 @@ def __init__(self, data: Dict[str, Any]) -> None:
         # 'map_present': 1, 'in_cleaning': 3, 'in_returning': 0,
         # 'in_fresh_state': 0, 'lab_status': 1, 'water_box_status': 0,
         # 'fan_power': 102, 'dnd_enabled': 0, 'map_status': 3, 'lock_status': 0}]
+
+        # Example of S7 in charging mode
+        # new items: is_locating, water_box_mode, water_box_carriage_status,
+        # mop_forbidden_enable, adbumper_status, water_shortage_status,
+        # dock_type, dust_collection_status, auto_dust_collection, mop_mode, debug_mode
+        #
+        # [{'msg_ver': 2, 'msg_seq': 1839, 'state': 8, 'battery': 100,
+        # 'clean_time': 2311, 'clean_area': 35545000, 'error_code': 0,
+        # 'map_present': 1, 'in_cleaning': 0, 'in_returning': 0,
+        # 'in_fresh_state': 1, 'lab_status': 3, 'water_box_status': 1,
+        # 'fan_power': 102, 'dnd_enabled': 0, 'map_status': 3, 'is_locating': 0,
+        # 'lock_status': 0, 'water_box_mode': 202, 'water_box_carriage_status': 0,
+        # 'mop_forbidden_enable': 0, 'adbumper_status': [0, 0, 0],
+        # 'water_shortage_status': 0, 'dock_type': 0, 'dust_collection_status': 0,
+        # 'auto_dust_collection': 1,  'mop_mode': 300, 'debug_mode': 0}]
         self.data = data
 
     @property
@@ -175,6 +190,21 @@ def is_water_box_attached(self) -> bool:
         """Return True is water box is installed."""
         return "water_box_status" in self.data and self.data["water_box_status"] == 1
 
+    @property
+    def is_water_box_carriage_attached(self) -> Optional[bool]:
+        """Return True if water box carriage (mop) is installed, None if sensor not
+        present."""
+        if "water_box_carriage_status" in self.data:
+            return self.data["water_box_carriage_status"] == 1
+        return None
+
+    @property
+    def is_water_shortage(self) -> Optional[bool]:
+        """Returns True if water is low in the tank, None if sensor not present."""
+        if "water_shortage_status" in self.data:
+            return self.data["water_shortage_status"] == 1
+        return None
+
     @property
     def got_error(self) -> bool:
         """True if an error has occured."""