diff --git a/miio/device.py b/miio/device.py index caa1287ea..2a88c4fe8 100644 --- a/miio/device.py +++ b/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): diff --git a/miio/devicestatus.py b/miio/devicestatus.py index c12d0491d..8a0bb0431 100644 --- a/miio/devicestatus.py +++ b/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, diff --git a/miio/integrations/vacuum/roborock/tests/test_vacuum.py b/miio/integrations/vacuum/roborock/tests/test_vacuum.py index 8608fa680..326edc986 100644 --- a/miio/integrations/vacuum/roborock/tests/test_vacuum.py +++ b/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): diff --git a/miio/integrations/vacuum/roborock/vacuum.py b/miio/integrations/vacuum/roborock/vacuum.py index 5983956c0..cb7e768af 100644 --- a/miio/integrations/vacuum/roborock/vacuum.py +++ b/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() diff --git a/miio/integrations/vacuum/roborock/vacuumcontainers.py b/miio/integrations/vacuum/roborock/vacuumcontainers.py index 04de1c1d0..dadf677b0 100644 --- a/miio/integrations/vacuum/roborock/vacuumcontainers.py +++ b/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