diff --git a/miio/integrations/roborock/vacuum/tests/test_vacuum.py b/miio/integrations/roborock/vacuum/tests/test_vacuum.py index bb057e004..553f9d9c3 100644 --- a/miio/integrations/roborock/vacuum/tests/test_vacuum.py +++ b/miio/integrations/roborock/vacuum/tests/test_vacuum.py @@ -63,6 +63,10 @@ def __init__(self, *args, **kwargs): } self._maps = None self._map_enum_cache = None + self._floor_clean_details = {} + self._last_clean_details = None + self._searched_clean_id = None + self._status_helper = UpdateHelper(self.vacuum_status) self.dummies = { "consumables": [ diff --git a/miio/integrations/roborock/vacuum/vacuum.py b/miio/integrations/roborock/vacuum/vacuum.py index 2fa6b9edf..7d1d411fd 100644 --- a/miio/integrations/roborock/vacuum/vacuum.py +++ b/miio/integrations/roborock/vacuum/vacuum.py @@ -8,7 +8,7 @@ import pathlib import time from enum import Enum -from typing import Any, List, Optional, Type +from typing import Any, Dict, List, Optional, Type import click import pytz @@ -49,6 +49,7 @@ CleaningSummary, ConsumableStatus, DNDStatus, + FloorCleanDetails, MapList, MopDryerSettings, SoundInstallStatus, @@ -143,12 +144,18 @@ def __init__( ip, token, start_id, debug, lazy_discover, timeout, model=model ) self.manual_seqnum = -1 + self._clean_history: Optional[CleaningSummary] = None + self._searched_clean_id: Optional[int] = None + self._floor_clean_details: Dict[int, Optional[CleaningDetails]] = {} + self._last_clean_details: Optional[CleaningDetails] = None self._maps: Optional[MapList] = None self._map_enum_cache = None self._status_helper = UpdateHelper(self.vacuum_status) + self._status_helper.add_update_method("map_list", self.get_maps) self._status_helper.add_update_method("consumables", self.consumable_status) self._status_helper.add_update_method("dnd_status", self.dnd_status) self._status_helper.add_update_method("clean_history", self.clean_history) + self._status_helper.add_update_method("floor_clean", self.last_clean_all_floor) self._status_helper.add_update_method("last_clean", self.last_clean_details) self._status_helper.add_update_method("mop_dryer", self.mop_dryer_settings) @@ -515,20 +522,68 @@ def enable_lab_mode(self, enable): @command() def clean_history(self) -> CleaningSummary: """Return generic cleaning history.""" - return CleaningSummary(self.send("get_clean_summary")) + self._clean_history = CleaningSummary(self.send("get_clean_summary")) + return self._clean_history @command() - def last_clean_details(self) -> Optional[CleaningDetails]: + def last_clean_details(self, skip_cache=False) -> Optional[CleaningDetails]: """Return details from the last cleaning. Returns None if there has been no cleanups. """ - history = self.clean_history() - if not history.ids: + if self._clean_history is None or skip_cache: + self.clean_history() + assert isinstance(self._clean_history, CleaningSummary) # nosec assert_used + if not self._clean_history.ids: return None - last_clean_id = history.ids.pop(0) - return self.clean_details(last_clean_id) + last_clean_id = self._clean_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, skip_cache=False) -> FloorCleanDetails: + """Return details from the last cleaning and for each floor. + + Returns None if there has been no cleanups for that floor. + """ + if self._clean_history is None or skip_cache: + self.clean_history() + assert isinstance(self._clean_history, CleaningSummary) # nosec assert_used + + map_ids = self.get_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[id] = None + + if not self._clean_history.ids: + return FloorCleanDetails(self._floor_clean_details) + + last_clean_id = self._clean_history.ids[0] + for id in self._clean_history.ids: + # already searched this record + if id == self._searched_clean_id: + break + + clean_detail = self.clean_details(id) + if clean_detail.map_id in map_ids: + self._floor_clean_details[clean_detail.map_id] = clean_detail + map_ids.remove(clean_detail.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) @command( click.argument("id_", type=int, metavar="ID"), @@ -541,8 +596,7 @@ def clean_details(self, id_: int) -> Optional[CleaningDetails]: _LOGGER.warning("No cleaning record found for id %s", id_) return None - res = CleaningDetails(details.pop()) - return res + return CleaningDetails(details.pop()) @command() @action(name="Find robot", type="vacuum") diff --git a/miio/integrations/roborock/vacuum/vacuumcontainers.py b/miio/integrations/roborock/vacuum/vacuumcontainers.py index 6b7e712f9..3a0bce465 100644 --- a/miio/integrations/roborock/vacuum/vacuumcontainers.py +++ b/miio/integrations/roborock/vacuum/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 @@ -337,6 +338,16 @@ def current_map_id(self) -> int: """ return int((self.data["map_status"] + 1) / 4 - 1) + @property + def current_map_name(self) -> str: + """The name of the current map with regards to the multi map feature.""" + try: + map_list = self.map_list.map_list + except AttributeError: + return str(self.current_map_id) + + return map_list[self.current_map_id]["name"] + @property def in_zone_cleaning(self) -> bool: """Return True if the vacuum is in zone cleaning mode.""" @@ -586,6 +597,17 @@ 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.""" + try: + map_list = self._parent.map_list.map_list + except AttributeError: + return str(self.map_id) + + return map_list[self.map_id]["name"] + @property def error_code(self) -> int: """Error code.""" @@ -605,6 +627,57 @@ 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[int, 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=datetime, + 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