Skip to content

Commit

Permalink
Introduce common interfaces based on device descriptors
Browse files Browse the repository at this point in the history
This completes the common descriptor-based API for all device-inherited classes:
- status() -> DeviceStatus: returns device status
- descriptors() -> DescriptorCollection[Descriptor]: returns all defined descriptors
- actions() -> DescriptorCollection[ActionDescriptor]: returns all defined actions
- settings() -> DescriptorCollection[PropertyDescriptor]: returns all settable descriptors
- sensors() -> DescriptorCollection[PropertyDescriptor]: returns all read-only descriptors
- call_action(name, params): to call action using its name
- change_setting(name, params): to change a setting using its name

These functionalities are also provided as cli commands for all devices:
- status
- descriptors
- actions
- settings
- sensors
- call (call_action)
- set (change_setting)
  • Loading branch information
rytilahti committed Oct 20, 2023
1 parent 4390234 commit c5d6184
Show file tree
Hide file tree
Showing 9 changed files with 238 additions and 255 deletions.
123 changes: 123 additions & 0 deletions miio/descriptorcollection.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,123 @@
import logging
from collections import UserDict
from inspect import getmembers
from typing import TYPE_CHECKING, Generic, TypeVar, cast

from .descriptors import (
AccessFlags,
ActionDescriptor,
Descriptor,
EnumDescriptor,
PropertyConstraint,
PropertyDescriptor,
RangeDescriptor,
)

_LOGGER = logging.getLogger(__name__)

if TYPE_CHECKING:
from miio import Device


T = TypeVar("T")


class DescriptorCollection(UserDict, Generic[T]):
"""A container of descriptors."""

def __init__(self, *args, device: "Device"):
self._device = device
super().__init__(*args)

def descriptors_from_object(self, obj):
"""Add descriptors from an object.
This collects descriptors from the given object and adds them into the collection by:
1. Checking for '_descriptors' for descriptors created by the class itself.
2. Going through all members and looking if they have a '_descriptor' attribute set by a decorator
"""
_LOGGER.debug("Adding descriptors from %s", obj)
# 1. Check for existence of _descriptors as DeviceStatus' metaclass collects them already
if descriptors := getattr(obj, "_descriptors"): # noqa: B009
for _name, desc in descriptors.items():
self.add_descriptor(desc)

# 2. Check if object members have descriptors
for _name, method in getmembers(obj, lambda o: hasattr(o, "_descriptor")):
prop_desc = method._descriptor
if not isinstance(prop_desc, Descriptor):
_LOGGER.warning("%s %s is not a descriptor, skipping", _name, method)
continue

prop_desc.method = method
self.add_descriptor(prop_desc)

def add_descriptor(self, descriptor: Descriptor):
"""Add a descriptor to the collection.
This adds a suffix to the identifier if the name already exists.
"""
if not isinstance(descriptor, Descriptor):
raise TypeError("Tried to add non-descriptor descriptor: %s", descriptor)

if descriptor.id in self.data:
descriptor.id = descriptor.id + "-2"
_LOGGER.warning("Appended '-2' to the id of %s", descriptor)
# TODO: append suffix to dupe ids
raise ValueError(f"Duplicate descriptor id: {descriptor.id}")

if isinstance(descriptor, PropertyDescriptor):
self._handle_property_descriptor(descriptor)
elif isinstance(descriptor, ActionDescriptor):
self._handle_action_descriptor(descriptor)
else:
_LOGGER.debug("Using descriptor as is: %s", descriptor)

self.data[descriptor.id] = descriptor
_LOGGER.debug("Added descriptor: %r", descriptor)

def _handle_action_descriptor(self, prop: ActionDescriptor) -> None:
"""Bind the action method to the action."""
if prop.method_name is not None:
prop.method = getattr(self._device, prop.method_name)

if prop.method is None:
raise Exception(f"Neither method or method_name was defined for {prop}")

def _handle_property_descriptor(self, prop: PropertyDescriptor) -> None:
"""Bind the setter method to the property."""
if prop.setter_name is not None:
prop.setter = getattr(self._device, prop.setter_name)

if prop.access & AccessFlags.Write and prop.setter is None:
raise Exception(f"Neither setter or setter_name was defined for {prop}")

self._handle_constraints(prop)

def _handle_constraints(self, prop: PropertyDescriptor) -> None:
"""Set attribute-based constraints for the descriptor."""
if prop.constraint == PropertyConstraint.Choice:
prop = cast(EnumDescriptor, prop)
if prop.choices_attribute is not None:
retrieve_choices_function = getattr(
self._device, prop.choices_attribute
)
prop.choices = retrieve_choices_function()

elif prop.constraint == PropertyConstraint.Range:
prop = cast(RangeDescriptor, prop)
if prop.range_attribute is not None:
range_def = getattr(self._device, prop.range_attribute)
prop.min_value = range_def.min_value
prop.max_value = range_def.max_value
prop.step = range_def.step

# A property without constraints, nothing to do here.

@property
def __cli_output__(self):
"""Return a string presentation for the cli."""
s = ""
for d in self.data.values():
s += f"{d.__cli_output__}\n"
return s
176 changes: 80 additions & 96 deletions miio/device.py
Original file line number Diff line number Diff line change
@@ -1,26 +1,18 @@
import logging
from enum import Enum
from inspect import getmembers
from typing import Any, Dict, List, Optional, Union, cast, final # noqa: F401

import click

from .click_common import DeviceGroupMeta, LiteralParamType, command, format_output
from .descriptors import (
AccessFlags,
ActionDescriptor,
EnumDescriptor,
PropertyConstraint,
PropertyDescriptor,
RangeDescriptor,
)
from .descriptorcollection import DescriptorCollection
from .descriptors import AccessFlags, ActionDescriptor, Descriptor, PropertyDescriptor
from .deviceinfo import DeviceInfo
from .devicestatus import DeviceStatus
from .exceptions import (
DeviceError,
DeviceInfoUnavailableException,
PayloadDecodeException,
UnsupportedFeatureException,
)
from .miioprotocol import MiIOProtocol

Expand Down Expand Up @@ -87,8 +79,9 @@ def __init__(
self.token: Optional[str] = token
self._model: Optional[str] = model
self._info: Optional[DeviceInfo] = None
self._actions: Optional[Dict[str, ActionDescriptor]] = None
self._properties: Optional[Dict[str, PropertyDescriptor]] = None
# TODO: use _info's noneness instead?
self._initialized: bool = False
self._descriptors: DescriptorCollection = DescriptorCollection(device=self)
timeout = timeout if timeout is not None else self.timeout
self._debug = debug
self._protocol = MiIOProtocol(
Expand Down Expand Up @@ -179,62 +172,19 @@ def _fetch_info(self) -> DeviceInfo:
"Unable to request miIO.info from the device"
) from ex

def _set_constraints_from_attributes(
self, status: DeviceStatus
) -> Dict[str, PropertyDescriptor]:
"""Get the setting descriptors from a DeviceStatus."""
properties = status.properties()
unsupported_settings = []
for key, prop in properties.items():
if prop.setter_name is not None:
prop.setter = getattr(self, prop.setter_name)
if prop.setter is None:
raise Exception(f"Neither setter or setter_name was defined for {prop}")

if prop.constraint == PropertyConstraint.Choice:
prop = cast(EnumDescriptor, prop)
if prop.choices_attribute is not None:
retrieve_choices_function = getattr(self, prop.choices_attribute)
try:
prop.choices = retrieve_choices_function()
except UnsupportedFeatureException:
# TODO: this should not be done here
unsupported_settings.append(key)
continue

elif prop.constraint == PropertyConstraint.Range:
prop = cast(RangeDescriptor, prop)
if prop.range_attribute is not None:
range_def = getattr(self, prop.range_attribute)
prop.min_value = range_def.min_value
prop.max_value = range_def.max_value
prop.step = range_def.step

else:
_LOGGER.debug("Got a regular setting without constraints: %s", prop)

for unsupp_key in unsupported_settings:
properties.pop(unsupp_key)

return properties

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[action.id] = action

return actions

def _initialize_descriptors(self) -> None:
"""Cache all the descriptors once on the first call."""
status = self.status()
"""Initialize the device descriptors.
self._properties = self._set_constraints_from_attributes(status)
self._actions = self._action_descriptors()
This will add descriptors defined in the implementation class and the status class.
This can be overridden to add additional descriptors to the device.
If you do so, do not forget to call this method.
"""
self._descriptors.descriptors_from_object(self)

#
self._descriptors.descriptors_from_object(self.status.__annotations__["return"])
self._initialized = True

@property
def device_id(self) -> int:
Expand Down Expand Up @@ -323,49 +273,56 @@ def get_properties(

return values

@command()
def status(self) -> DeviceStatus:
"""Return device status."""
raise NotImplementedError()

def actions(self) -> Dict[str, ActionDescriptor]:
"""Return device actions."""
if self._actions is None:
@command()
def descriptors(self) -> DescriptorCollection[Descriptor]:
"""Return a collection containing all descriptors for the device."""
if not self._initialized:
self._initialize_descriptors()

# TODO: we ignore the return value for now as these should always be initialized
return self._actions # type: ignore[return-value]

def properties(self) -> Dict[str, PropertyDescriptor]:
"""Return all device properties."""
if self._properties is None:
self._initialize_descriptors()
return self._descriptors

# TODO: we ignore the return value for now as these should always be initialized
return self._properties # type: ignore[return-value]
@command()
def actions(self) -> DescriptorCollection[ActionDescriptor]:
"""Return device actions."""
return DescriptorCollection(
{
k: v
for k, v in self.descriptors().items()
if isinstance(v, ActionDescriptor)
},
device=self,
)

@final
def settings(self) -> Dict[str, PropertyDescriptor]:
@command()
def settings(self) -> DescriptorCollection[PropertyDescriptor]:
"""Return settable properties."""
if self._properties is None:
self._initialize_descriptors()

return {
prop.id: prop
for prop in self.properties().values()
if prop.access & AccessFlags.Write
}
return DescriptorCollection(
{
k: v
for k, v in self.descriptors().items()
if isinstance(v, PropertyDescriptor) and v.access & AccessFlags.Write
},
device=self,
)

@final
def sensors(self) -> Dict[str, PropertyDescriptor]:
@command()
def sensors(self) -> DescriptorCollection[PropertyDescriptor]:
"""Return read-only properties."""
if self._properties is None:
self._initialize_descriptors()

return {
prop.id: prop
for prop in self.properties().values()
if prop.access ^ AccessFlags.Write
}
return DescriptorCollection(
{
k: v
for k, v in self.descriptors().items()
if isinstance(v, PropertyDescriptor) and v.access & AccessFlags.Read
},
device=self,
)

def supports_miot(self) -> bool:
"""Return True if the device supports miot commands.
Expand All @@ -379,5 +336,32 @@ def supports_miot(self) -> bool:
return False
return True

@command(
click.argument("name"),
click.argument("params", type=LiteralParamType(), required=False),
name="call",
)
def call_action(self, name: str, params=None):
"""Call action by name."""
act = self.actions()[name]
params = params or []

return act.method(params)

@command(
click.argument("name"),
click.argument("params", type=LiteralParamType(), required=True),
name="set",
)
def change_setting(self, name: str, params=None):
"""Change setting value."""
setting = self.settings()[name]
params = params if params is not None else []

if setting.access & AccessFlags.Write == 0:
raise ValueError("Property %s is not writable" % name)

return setting.setter(params)

def __repr__(self):
return f"<{self.__class__.__name__ }: {self.ip} (token: {self.token})>"
Loading

0 comments on commit c5d6184

Please sign in to comment.