Add additional sensors and settings to Roborock vacuums

miio/device.py

@@ -65,6 +65,8 @@ def __init__(
         self.token: Optional[str] = token
         self._model: Optional[str] = model
         self._info: Optional[DeviceInfo] = None
+        self._status: Optional[DeviceStatus] = None
+        self._buttons: Optional[List[ButtonDescriptor]] = None
         self._actions: Optional[Dict[str, ActionDescriptor]] = None
         timeout = timeout if timeout is not None else self.timeout
         self._protocol = MiIOProtocol(
@@ -250,6 +252,14 @@ def status(self) -> DeviceStatus:
         """Return device status."""
         raise NotImplementedError()
 
+
+    def cached_status(self) -> DeviceStatus:
+        """Return device status from cache."""
+        if self._status is None:
+            self._status = self.status()
+
+        return self._status
+
     def actions(self) -> Dict[str, ActionDescriptor]:
         """Return device actions."""
         if self._actions is None:
@@ -263,8 +273,10 @@ def actions(self) -> Dict[str, ActionDescriptor]:
         return self._actions
 
     def settings(self) -> Dict[str, SettingDescriptor]:
-        """Return device settings."""
-        settings = self.status().settings()
+        """Return list of settings."""
+        settings = (
+            self.cached_status().settings()
+        )  # NOTE that this already does IO so schould be run in executer job in HA
         for setting in settings.values():
             # TODO: Bind setter methods, this should probably done only once during init.
             if setting.setter is None:
@@ -292,9 +304,8 @@ def settings(self) -> Dict[str, SettingDescriptor]:
         return settings
 
     def sensors(self) -> Dict[str, SensorDescriptor]:
-        """Return device sensors."""
-        # TODO: the latest status should be cached and re-used by all meta information getters
-        sensors = self.status().sensors()
+        """Return sensors."""
+        sensors = self.cached_status().sensors()
         return sensors
 
     def __repr__(self):

miio/devicestatus.py

@@ -2,16 +2,7 @@
 import logging
 import warnings
 from enum import Enum
-from typing import (
-    Callable,
-    Dict,
-    Optional,
-    Type,
-    Union,
-    get_args,
-    get_origin,
-    get_type_hints,
-)
+from typing import Dict, Optional, Type, Union, get_args, get_origin, get_type_hints
 
 from .descriptors import (
     ActionDescriptor,
@@ -173,7 +164,6 @@ def _sensor_type_for_return_type(func):
 def setting(
     name: str,
     *,
-    setter: Optional[Callable] = None,
     setter_name: Optional[str] = None,
     unit: Optional[str] = None,
     min_value: Optional[int] = None,

miio/integrations/vacuum/roborock/tests/test_vacuum.py

@@ -121,6 +121,11 @@ def __init__(self, *args, **kwargs):
             "get_multi_maps_list": lambda x: self.dummies["multi_maps"],
         }
 
+        self._multi_maps = None
+
+        self._floor_clean_details = {}
+        self._searched_clean_id = None
+
         super().__init__(args, kwargs)
 
     def change_mode(self, new_mode):

miio/integrations/vacuum/roborock/vacuum.py

@@ -7,7 +7,7 @@
 import os
 import pathlib
 import time
-from typing import List, Optional, Type, Union
+from typing import Dict, List, Optional, Tuple, Type, Union
 
 import click
 import pytz
@@ -47,6 +47,7 @@
     CleaningSummary,
     ConsumableStatus,
     DNDStatus,
+    FloorCleanDetails,
     MapList,
     SoundInstallStatus,
     SoundStatus,
@@ -137,6 +138,9 @@ def __init__(
             ip, token, start_id, debug, lazy_discover, timeout, model=model
         )
         self.manual_seqnum = -1
+        self._floor_clean_details: Dict[str, Optional[CleaningDetails]] = {}
+        self._last_clean_details: Optional[CleaningDetails] = None
+        self._searched_clean_id: Optional[int] = None
         self._maps: Optional[MapList] = None
         self._map_enum_cache = None
 
@@ -335,14 +339,20 @@ def status(self) -> VacuumStatus:
         """Return status of the vacuum."""
         status = self.vacuum_status()
         status.embed(self.consumable_status())
-        status.embed(self.clean_history())
+        clean_history = self.clean_history()
+        status.embed(clean_history)
+        (details_floors, details_last) = self.last_clean_all_floor(
+            history=clean_history
+        )
+        status.embed(details_last)
+        status.embed(details_floors)
         status.embed(self.dnd_status())
         return status
 
     @command()
     def vacuum_status(self) -> VacuumStatus:
         """Return only status of the vacuum."""
-        return VacuumStatus(self.send("get_status")[0])
+        return VacuumStatus(self.send("get_status")[0], self.get_multi_maps())
 
     def enable_log_upload(self):
         raise NotImplementedError("unknown parameters")
@@ -489,17 +499,69 @@ def clean_history(self) -> CleaningSummary:
         return CleaningSummary(self.send("get_clean_summary"))
 
     @command()
-    def last_clean_details(self) -> Optional[CleaningDetails]:
+    def last_clean_details(
+        self, history: Optional[CleaningSummary] = None
+    ) -> Optional[CleaningDetails]:
         """Return details from the last cleaning.
 
         Returns None if there has been no cleanups.
         """
-        history = self.clean_history()
+        if history is None:
+            history = self.clean_history()
         if not history.ids:
             return None
 
-        last_clean_id = history.ids.pop(0)
-        return self.clean_details(last_clean_id)
+        last_clean_id = history.ids[0]
+        if last_clean_id == self._searched_clean_id:
+            return self._last_clean_details
+
+        self._last_clean_details = self.clean_details(last_clean_id)
+        return self._last_clean_details
+
+    @command()
+    def last_clean_all_floor(
+        self, history: Optional[CleaningSummary] = None
+    ) -> Tuple[FloorCleanDetails, Optional[CleaningDetails]]:
+        """Return details from the last cleaning and for each floor.
+
+        Returns None if there has been no cleanups for that floor.
+        """
+        if history is None:
+            history = self.clean_history()
+
+        map_ids = self.get_multi_maps().map_id_list
+
+        # if cache empty, fill with None
+        if not self._floor_clean_details:
+            for id in map_ids:
+                self._floor_clean_details[str(id)] = None
+
+        if not history.ids:
+            return (
+                FloorCleanDetails(self._floor_clean_details),
+                self._last_clean_details,
+            )
+
+        last_clean_id = history.ids[0]
+        for id in history.ids:
+            # already searched this record
+            if id == self._searched_clean_id:
+                break
+
+            clean_detail = self.clean_details(id)
+            if clean_detail.multi_map_id in map_ids:
+                self._floor_clean_details[str(clean_detail.multi_map_id)] = clean_detail
+                map_ids.remove(clean_detail.multi_map_id)
+
+            if id == last_clean_id:
+                self._last_clean_details = clean_detail
+
+            # all floors found
+            if not map_ids:
+                break
+
+        self._searched_clean_id = last_clean_id
+        return (FloorCleanDetails(self._floor_clean_details), self._last_clean_details)
 
     @command(
         click.argument("id_", type=int, metavar="ID"),
@@ -514,7 +576,7 @@ def clean_details(
             _LOGGER.warning("No cleaning record found for id %s", id_)
             return None
 
-        res = CleaningDetails(details.pop())
+        res = CleaningDetails(details.pop(), self.get_multi_maps())
         return res
 
     @command()

miio/integrations/vacuum/roborock/vacuumcontainers.py

@@ -6,6 +6,7 @@
 from croniter import croniter
 from pytz import BaseTzInfo
 
+from miio.descriptors import SensorDescriptor
 from miio.device import DeviceStatus
 from miio.devicestatus import sensor, setting
 from miio.interfaces.vacuuminterface import VacuumDeviceStatus, VacuumState
@@ -136,7 +137,9 @@ def map_name_dict(self) -> Dict[str, int]:
 class VacuumStatus(VacuumDeviceStatus):
     """Container for status reports from the vacuum."""
 
-    def __init__(self, data: Dict[str, Any]) -> None:
+    def __init__(
+        self, data: Dict[str, Any], multi_maps: Optional[MultiMapList] = None
+    ) -> None:
         # {'result': [{'state': 8, 'dnd_enabled': 1, 'clean_time': 0,
         #  'msg_ver': 4, 'map_present': 1, 'error_code': 0, 'in_cleaning': 0,
         #  'clean_area': 0, 'battery': 100, 'fan_power': 20, 'msg_seq': 320}],
@@ -177,6 +180,7 @@ def __init__(self, data: Dict[str, Any]) -> None:
         # 'water_shortage_status': 0, 'dock_type': 0, 'dust_collection_status': 0,
         # 'auto_dust_collection': 1,  'mop_mode': 300, 'debug_mode': 0}]
         self.data = data
+        self._multi_maps = multi_maps
 
     @property
     @sensor("State code", entity_category="diagnostic", enabled_default=False)
@@ -337,6 +341,14 @@ def current_map_id(self) -> int:
         """
         return int((self.data["map_status"] + 1) / 4 - 1)
 
+    @property
+    def map_name(self) -> str:
+        """The name of the current map with regards to the multi map feature."""
+        if self._multi_maps is None:
+            return str(self.multi_map_id)
+
+        return self._multi_maps.map_list[self.multi_map_id]["name"]
+
     @property
     def in_zone_cleaning(self) -> bool:
         """Return True if the vacuum is in zone cleaning mode."""
@@ -494,10 +506,15 @@ def dust_collection_count(self) -> Optional[int]:
 class CleaningDetails(DeviceStatus):
     """Contains details about a specific cleaning run."""
 
-    def __init__(self, data: Union[List[Any], Dict[str, Any]]) -> None:
+    def __init__(
+        self,
+        data: Union[List[Any], Dict[str, Any]],
+        multi_maps: Optional[MultiMapList] = None,
+    ) -> None:
         # start, end, duration, area, unk, complete
         # { "result": [ [ 1488347071, 1488347123, 16, 0, 0, 0 ] ], "id": 1 }
         # newer models return a dict
+        self._multi_maps = multi_maps
         if isinstance(data, list):
             self.data = {
                 "begin": data[0],
@@ -560,6 +577,15 @@ def map_id(self) -> int:
         """Map id used (multi map feature) during the cleaning run."""
         return self.data.get("map_flag", 0)
 
+    @property
+    @sensor("Last clean map name", icon="mdi:floor-plan", entity_category="diagnostic")
+    def map_name(self) -> str:
+        """The name of the map used (multi map feature) during the cleaning run."""
+        if self._multi_maps is None:
+            return str(self.multi_map_id)
+
+        return self._multi_maps.map_list[self.multi_map_id]["name"]
+
     @property
     def error_code(self) -> int:
         """Error code."""
@@ -579,6 +605,58 @@ def complete(self) -> bool:
         return self.data["complete"] == 1
 
 
+class FloorCleanDetails(DeviceStatus):
+    """Contains details about a last cleaning run per floor."""
+
+    def __init__(self, data: Dict[str, Any]) -> None:
+        self.data = data
+
+        for map_id in self.data:
+            if self.data[map_id] is None:
+                setattr(self, f"CleanDetails_{map_id}", None)
+                setattr(self, f"start_{map_id}", None)
+                continue
+            setattr(self, f"CleanDetails_{map_id}", self.data[map_id])
+            setattr(self, f"start_{map_id}", self.data[map_id].start)
+
+    def __repr__(self):
+
+        s = f"<{self.__class__.__name__}"
+        for map_id in self.data:
+            name = f"CleanDetails_{map_id}"
+            value = getattr(self, name)
+            s += f" {name}={value}"
+
+            name = f"start_{map_id}"
+            value = getattr(self, name)
+            s += f" {name}={value}"
+
+        for name, embedded in self._embedded.items():
+            s += f" {name}={repr(embedded)}"
+
+        s += ">"
+        return s
+
+    def sensors(self) -> Dict[str, SensorDescriptor]:
+        """Return the dict of sensors exposed by the status container."""
+        self._sensors = {}  # type: ignore[attr-defined]
+
+        for map_id in self.data:
+            self._sensors[f"start_{map_id}"] = SensorDescriptor(
+                id=f"FloorCleanDetails.start_{map_id}",
+                property=f"start_{map_id}",
+                name=f"Floor {map_id} clean start",
+                type="sensor",
+                extras={
+                    "icon": "mdi:clock-time-twelve",
+                    "device_class": "timestamp",
+                    "entity_category": "diagnostic",
+                },
+            )
+
+        return self._sensors
+
+
 class ConsumableStatus(DeviceStatus):
     """Container for consumable status information, including information about brushes
     and duration until they should be changed. The methods returning time left are based