Cache descriptors on first access (#1701)

The descriptors accessed through `sensors()`, `settings()`, and
`actions()` are now cached on the first use to avoid unnecessary I/O.

Co-authored-by: Teemu R. <tpr@iki.fi>

miio/device.py

@@ -87,6 +87,8 @@ def __init__(
         self._model: Optional[str] = model
         self._info: Optional[DeviceInfo] = None
         self._actions: Optional[Dict[str, ActionDescriptor]] = None
+        self._settings: Optional[Dict[str, SettingDescriptor]] = None
+        self._sensors: Optional[Dict[str, SensorDescriptor]] = None
         timeout = timeout if timeout is not None else self.timeout
         self._debug = debug
         self._protocol = MiIOProtocol(
@@ -177,6 +179,61 @@ 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."""
+
+        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."""
@@ -271,49 +328,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()
+            self._settings = cast(Dict[str, SettingDescriptor], self._settings)
 
-        return 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 supports_miot(self) -> bool:
         """Return True if the device supports miot commands.

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_device.py

@@ -182,3 +182,17 @@ def test_supports_miot(mocker):
 
     send.side_effect = None
     assert d.supports_miot() is True
+
+
+@pytest.mark.parametrize("getter_name", ["actions", "settings", "sensors"])
+def test_cached_descriptors(getter_name, mocker):
+    d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff")
+    getter = getattr(d, getter_name)
+    initialize_descriptors = mocker.spy(d, "_initialize_descriptors")
+    mocker.patch("miio.Device.status")
+    mocker.patch("miio.Device._sensor_descriptors_from_status", return_value={})
+    mocker.patch("miio.Device._setting_descriptors_from_status", return_value={})
+    mocker.patch("miio.Device._action_descriptors", return_value={})
+    for _i in range(5):
+        getter()
+    initialize_descriptors.assert_called_once()

miio/tests/test_devicestatus.py

@@ -141,6 +141,7 @@ def level(self) -> int:
 
     mocker.patch("miio.Device.send")
     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())
@@ -186,6 +187,7 @@ def level(self) -> int:
 
     mocker.patch("miio.Device.send")
     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())
@@ -227,6 +229,7 @@ def level(self) -> TestEnum:
 
     mocker.patch("miio.Device.send")
     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())