Add fan_speed_presets() for querying available fan speeds

miio/vacuum.py

@@ -6,7 +6,7 @@
 import os
 import pathlib
 import time
-from typing import List, Optional, Union
+from typing import Dict, List, Optional, Union
 
 import click
 import pytz
@@ -46,6 +46,24 @@ class Consumable(enum.Enum):
     SensorDirty = "sensor_dirty_time"
 
 
+class FanspeedV1(enum.Enum):
+    Silent = 38
+    Standard = 60
+    Medium = 77
+    Turbo = 90
+
+
+class FanspeedV2(enum.Enum):
+    Silent = 101
+    Standard = 102
+    Medium = 103
+    Turbo = 104
+    Gentle = 105
+
+
+ROCKROBO_V1 = "rockrobo.vacuum.v1"
+
+
 class Vacuum(Device):
     """Main class representing the vacuum."""
 
@@ -54,6 +72,8 @@ def __init__(
     ) -> None:
         super().__init__(ip, token, start_id, debug)
         self.manual_seqnum = -1
+        self.model = None
+        self._fanspeeds = FanspeedV1
 
     @command()
     def start(self):
@@ -416,6 +436,47 @@ def fan_speed(self):
         """Return fan speed."""
         return self.send("get_custom_mode")[0]
 
+    def _autodetect_model(self):
+        """Detect the model of the vacuum.
+
+        For the moment this is used only for the fanspeeds,
+        but that could be extended to cover other supported features."""
+        try:
+            info = self.info()
+            self.model = info.model
+        except TypeError:
+            # cloud-blocked vacuums will not return proper payloads
+            self._fanspeeds = FanspeedV1
+            self.model = ROCKROBO_V1
+            _LOGGER.debug("Unable to query model, falling back to %s", self._fanspeeds)
+            return
+
+        _LOGGER.info("model: %s", self.model)
+
+        if info.model == ROCKROBO_V1:
+            _LOGGER.debug("Got robov1, checking for firmware version")
+            fw_version = info.firmware_version
+            version, build = fw_version.split("_")
+            version = tuple(map(int, version.split(".")))
+            if version >= (3, 5, 7):
+                self._fanspeeds = FanspeedV2
+            else:
+                self._fanspeeds = FanspeedV1
+        else:
+            self._fanspeeds = FanspeedV2
+
+        _LOGGER.debug(
+            "Using new fanspeed mapping %s for %s", self._fanspeeds, info.model
+        )
+
+    @command()
+    def fan_speed_presets(self) -> Dict[str, int]:
+        """Return dictionary containing supported fan speeds."""
+        if self.model is None:
+            self._autodetect_model()
+
+        return {x.name: x.value for x in list(self._fanspeeds)}
+
     @command()
     def sound_info(self):
         """Get voice settings."""
@@ -619,10 +680,7 @@ def cleanup(vac: Vacuum, *args, **kwargs):
             _LOGGER.debug("Writing %s to %s", seqs, id_file)
             path_obj = pathlib.Path(id_file)
             cache_dir = path_obj.parents[0]
-            try:
-                cache_dir.mkdir(parents=True)
-            except FileExistsError:
-                pass  # after dropping py3.4 support, use exist_ok for mkdir
+            cache_dir.mkdir(parents=True, exist_ok=True)
             with open(id_file, "w") as f:
                 json.dump(seqs, f)
 

miio/viomivacuum.py

@@ -3,6 +3,7 @@
 from collections import defaultdict
 from datetime import timedelta
 from enum import Enum
+from typing import Dict
 
 import click
 
@@ -338,3 +339,8 @@ def led(self, state: ViomiLedState):
     def carpet_mode(self, mode: ViomiCarpetTurbo):
         """Set the carpet mode."""
         return self.send("set_carpetturbo", [mode.value])
+
+    @command()
+    def fan_speed_presets(self) -> Dict[str, int]:
+        """Return dictionary containing supported fanspeeds."""
+        return {x.name: x.value for x in list(ViomiVacuumSpeed)}