Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 68 additions & 34 deletions miio/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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.
Expand Down
3 changes: 3 additions & 0 deletions miio/tests/dummies.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
14 changes: 14 additions & 0 deletions miio/tests/test_device.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just for the future, the proper way to do this is to use plain _ here. No need to fix that now though.

getter()
initialize_descriptors.assert_called_once()
3 changes: 3 additions & 0 deletions miio/tests/test_devicestatus.py
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down Expand Up @@ -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())
Expand Down Expand Up @@ -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())
Expand Down