Initialize and cache descriptors only once

miio/airdehumidifier.py

@@ -185,7 +185,7 @@ def status(self) -> AirDehumidifierStatus:
         values = self.get_properties(properties, max_properties=1)
 
         return AirDehumidifierStatus(
-            defaultdict(lambda: None, zip(properties, values)), self.info()
+            defaultdict(lambda: None, zip(properties, values)), self._fetch_info()
         )
 
     @command(default_output=format_output("Powering on"))

miio/airqualitymonitor.py

@@ -170,9 +170,13 @@ def status(self) -> AirQualityMonitorStatus:
             self.model, AVAILABLE_PROPERTIES[MODEL_AIRQUALITYMONITOR_V1]
         )
 
+        firmware_version = self._fetch_info().firmware_version
+        if firmware_version is None:
+            firmware_version = "unknown"
+
         is_s1_firmware_version_4 = (
             self.model == MODEL_AIRQUALITYMONITOR_S1
-            and self.info().firmware_version.startswith("4")
+            and firmware_version.startswith("4")
         )
         if is_s1_firmware_version_4 and "battery" in properties:
             properties.remove("battery")

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._settings: Optional[Dict[str, SettingDescriptor]] = None
+        self._sensors: Optional[Dict[str, SensorDescriptor]] = None
         self._actions: Optional[Dict[str, ActionDescriptor]] = None
         timeout = timeout if timeout is not None else self.timeout
         self._protocol = MiIOProtocol(
@@ -133,13 +135,18 @@ def info(self, *, skip_cache=False) -> DeviceInfo:
 
         :param skip_cache bool: Skip the cache
         """
+        self._initialize_descriptors()
+
         if self._info is not None and not skip_cache:
             return self._info
 
         return self._fetch_info()
 
-    def _fetch_info(self) -> DeviceInfo:
+    def _fetch_info(self, *, skip_cache=False) -> DeviceInfo:
         """Perform miIO.info query on the device and cache the result."""
+        if self._info is not None and not skip_cache:
+            return self._info
+
         try:
             devinfo = DeviceInfo(self.send("miIO.info"))
             self._info = devinfo
@@ -159,6 +166,63 @@ def _fetch_info(self) -> DeviceInfo:
                 "Unable to request miIO.info from the device"
             ) from ex
 
+    def _setting_descriptors_from_status(
+        self, status: DeviceStatus
+    ) -> Dict[str, SettingDescriptor]:
+        """Get the setting descriptors from a DeviceStatus."""
+        settings = status.settings()
+        for setting in settings.values():
+            if setting.setter_name is not None:
+                setting.setter = getattr(self, setting.setter_name)
+            if setting.setter is None:
+                raise Exception(
+                    f"Neither setter or setter_name was defined for {setting}"
+                )
+            setting = cast(EnumSettingDescriptor, setting)
+            if (
+                setting.type == SettingType.Enum
+                and setting.choices_attribute is not None
+            ):
+                retrieve_choices_function = getattr(self, setting.choices_attribute)
+                setting.choices = retrieve_choices_function()
+            if setting.type == SettingType.Number:
+                setting = cast(NumberSettingDescriptor, setting)
+                if setting.range_attribute is not None:
+                    range_def = getattr(self, setting.range_attribute)
+                    setting.min_value = range_def.min_value
+                    setting.max_value = range_def.max_value
+                    setting.step = range_def.step
+
+        return settings
+
+    def _sensor_descriptors_from_status(
+        self, status: DeviceStatus
+    ) -> Dict[str, SensorDescriptor]:
+        """Get the sensor descriptors from a DeviceStatus."""
+        return status.sensors()
+
+    def _action_descriptors(self) -> Dict[str, ActionDescriptor]:
+        """Get the action descriptors from a DeviceStatus."""
+        actions = {}
+        for action_tuple in getmembers(self, lambda o: hasattr(o, "_action")):
+            method_name, method = action_tuple
+            action = method._action
+            action.method = method  # bind the method
+            actions[method_name] = action
+
+        return actions
+
+    def _initialize_descriptors(self) -> None:
+        """Cache all the descriptors once on the first call."""
+        if self._sensors is not None:
+            return
+
+        status = self.status()
+
+        self._sensors = self._sensor_descriptors_from_status(status)
+        self._settings = self._setting_descriptors_from_status(status)
+        self._actions = self._action_descriptors()
+
     @property
     def device_id(self) -> int:
         """Return device id (did), if available."""
@@ -182,7 +246,7 @@ def model(self) -> str:
         if self._model is not None:
             return self._model
 
-        return self.info().model
+        return cast(str, self._fetch_info().model)
 
     def update(self, url: str, md5: str):
         """Start an OTA update."""
@@ -253,49 +317,26 @@ def status(self) -> DeviceStatus:
     def actions(self) -> Dict[str, ActionDescriptor]:
         """Return device actions."""
         if self._actions is None:
-            self._actions = {}
-            for action_tuple in getmembers(self, lambda o: hasattr(o, "_action")):
-                method_name, method = action_tuple
-                action = method._action
-                action.method = method  # bind the method
-                self._actions[method_name] = action
+            self._initialize_descriptors()
 
+        self._actions = cast(Dict[str, ActionDescriptor], self._actions)
         return self._actions
 
     def settings(self) -> Dict[str, SettingDescriptor]:
         """Return device settings."""
-        settings = self.status().settings()
-        for setting in settings.values():
-            # TODO: Bind setter methods, this should probably done only once during init.
-            if setting.setter is None:
-                # TODO: this is ugly, how to fix the issue where setter_name is optional and thus not acceptable for getattr?
-                if setting.setter_name is None:
-                    raise Exception(
-                        f"Neither setter or setter_name was defined for {setting}"
-                    )
-
-                setting.setter = getattr(self, setting.setter_name)
-            if (
-                isinstance(setting, EnumSettingDescriptor)
-                and setting.choices_attribute is not None
-            ):
-                retrieve_choices_function = getattr(self, setting.choices_attribute)
-                setting.choices = retrieve_choices_function()  # This can do IO
-            if setting.type == SettingType.Number:
-                setting = cast(NumberSettingDescriptor, setting)
-                if setting.range_attribute is not None:
-                    range_def = getattr(self, setting.range_attribute)
-                    setting.min_value = range_def.min_value
-                    setting.max_value = range_def.max_value
-                    setting.step = range_def.step
+        if self._settings is None:
+            self._initialize_descriptors()
 
-        return settings
+        self._settings = cast(Dict[str, SettingDescriptor], self._settings)
+        return self._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
+        if self._sensors is None:
+            self._initialize_descriptors()
+
+        self._sensors = cast(Dict[str, SensorDescriptor], self._sensors)
+        return self._sensors
 
     def __repr__(self):
         return f"<{self.__class__.__name__ }: {self.ip} (token: {self.token})>"

miio/gateway/gateway.py

@@ -153,7 +153,7 @@ def devices(self):
     def mac(self):
         """Return the mac address of the gateway."""
         if self._info is None:
-            self._info = self.info()
+            self._info = self._fetch_info()
         return self._info.mac_address
 
     @property

miio/integrations/humidifier/zhimi/airhumidifier.py

@@ -370,7 +370,7 @@ def status(self) -> AirHumidifierStatus:
         values = self.get_properties(properties, max_properties=_props_per_request)
 
         return AirHumidifierStatus(
-            defaultdict(lambda: None, zip(properties, values)), self.info()
+            defaultdict(lambda: None, zip(properties, values)), self._fetch_info()
         )
 
     @command(default_output=format_output("Powering on"))

miio/integrations/vacuum/roborock/vacuum.py

@@ -175,12 +175,15 @@ def resume_or_start(self):
 
         return self.start()
 
-    def _fetch_info(self) -> DeviceInfo:
+    def _fetch_info(self, *, skip_cache=False) -> DeviceInfo:
         """Return info about the device.
 
         This is overrides the base class info to account for gen1 devices that do not
         respond to info query properly when not connected to the cloud.
         """
+        if self._info is not None and not skip_cache:
+            return self._info
+
         try:
             info = super()._fetch_info()
             return info
@@ -596,13 +599,16 @@ def _enum_as_dict(cls):
 
         if self.model == ROCKROBO_V1:
             _LOGGER.debug("Got robov1, checking for firmware version")
-            fw_version = self.info().firmware_version
-            version, build = fw_version.split("_")
-            version = tuple(map(int, version.split(".")))
-            if version >= (3, 5, 8):
-                fanspeeds = FanspeedV3
-            elif version == (3, 5, 7):
-                fanspeeds = FanspeedV2
+            fw_version = self._fetch_info().firmware_version
+            if fw_version is not None:
+                version, build = fw_version.split("_")
+                version_tuple = tuple(map(int, version.split(".")))
+                if version_tuple >= (3, 5, 8):
+                    fanspeeds = FanspeedV3
+                elif version_tuple == (3, 5, 7):
+                    fanspeeds = FanspeedV2
+                else:
+                    fanspeeds = FanspeedV1
             else:
                 fanspeeds = FanspeedV1
         elif self.model == ROCKROBO_E2:

miio/tests/dummies.py

@@ -37,6 +37,9 @@ def __init__(self, *args, **kwargs):
         self.start_state = self.state.copy()
         self._protocol = DummyMiIOProtocol(self)
         self._info = None
+        self._settings = {}
+        self._sensors = {}
+        self._actions = {}
         # TODO: ugly hack to check for pre-existing _model
         if getattr(self, "_model", None) is None:
             self._model = "dummy.model"

miio/tests/test_airqualitymonitor.py

@@ -16,6 +16,7 @@
 class DummyAirQualityMonitorV1(DummyDevice, AirQualityMonitor):
     def __init__(self, *args, **kwargs):
         self._model = MODEL_AIRQUALITYMONITOR_V1
+        self.version = "3.1.8_9999"
         self.state = {
             "power": "on",
             "aqi": 34,
@@ -35,6 +36,9 @@ def __init__(self, *args, **kwargs):
         }
         super().__init__(args, kwargs)
 
+    def _fetch_info(self):
+        return DummyDeviceInfo(version=self.version)
+
 
 @pytest.fixture(scope="class")
 def airqualitymonitorv1(request):
@@ -100,7 +104,7 @@ def _get_state(self, props):
         """Return wanted properties."""
         return self.state
 
-    def info(self):
+    def _fetch_info(self):
         return DummyDeviceInfo(version=self.version)
 
 
@@ -185,6 +189,7 @@ def test_status(self):
 class DummyAirQualityMonitorB1(DummyDevice, AirQualityMonitor):
     def __init__(self, *args, **kwargs):
         self._model = MODEL_AIRQUALITYMONITOR_B1
+        self.version = "3.1.8_9999"
         self.state = {
             "co2e": 1466,
             "humidity": 59.79999923706055,
@@ -201,6 +206,9 @@ def _get_state(self, props):
         """Return wanted properties."""
         return self.state
 
+    def _fetch_info(self):
+        return DummyDeviceInfo(version=self.version)
+
 
 @pytest.fixture(scope="class")
 def airqualitymonitorb1(request):

miio/tests/test_devicestatus.py

@@ -122,6 +122,7 @@ def level(self) -> int:
             return 1
 
     mocker.patch("miio.Device.send")
+    mocker.patch("miio.Device.send_handshake")
     d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff")
 
     # Patch status to return our class
@@ -167,7 +168,9 @@ def level(self) -> int:
             return 1
 
     mocker.patch("miio.Device.send")
+    mocker.patch("miio.Device.send_handshake")
     d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff")
+    d._protocol._device_id = b"12345678"
 
     # Patch status to return our class
     mocker.patch.object(d, "status", return_value=Settings())
@@ -208,7 +211,9 @@ def level(self) -> TestEnum:
             return TestEnum.First
 
     mocker.patch("miio.Device.send")
+    mocker.patch("miio.Device.send_handshake")
     d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff")
+    d._protocol._device_id = b"12345678"
 
     # Patch status to return our class
     mocker.patch.object(d, "status", return_value=Settings())

miio/wifirepeater.py

@@ -132,9 +132,9 @@ def set_configuration(self, ssid: str, password: str, ssid_hidden: bool = False)
     )
     def wifi_roaming(self) -> bool:
         """Return the roaming setting."""
-        return self.info().raw["desc"]["wifi_explorer"] == 1
+        return self._fetch_info().raw["desc"]["wifi_explorer"] == 1
 
     @command(default_output=format_output("RSSI of the accesspoint: {result}"))
     def rssi_accesspoint(self) -> int:
         """Received signal strength indicator of the accesspoint."""
-        return self.info().accesspoint["rssi"]
+        return self._fetch_info().accesspoint["rssi"]