From b50f0f269e350d98b6859bae727e58e40ba25328 Mon Sep 17 00:00:00 2001 From: Teemu R Date: Sat, 21 Oct 2023 00:56:43 +0200 Subject: [PATCH] Introduce common interfaces based on device descriptors (#1845) 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) --- miio/__init__.py | 10 + miio/descriptorcollection.py | 138 +++++++++++++ miio/device.py | 189 +++++++++-------- miio/devicestatus.py | 22 +- miio/integrations/genericmiot/cli_helpers.py | 54 ----- miio/integrations/genericmiot/genericmiot.py | 111 ++-------- .../scishare/coffee/scishare_coffeemaker.py | 28 ++- miio/integrations/zhimi/fan/zhimi_miot.py | 2 +- miio/miot_device.py | 2 +- miio/tests/test_descriptorcollection.py | 191 ++++++++++++++++++ miio/tests/test_device.py | 102 +++++++++- miio/tests/test_devicestatus.py | 13 +- miio/tests/test_miotdevice.py | 4 +- 13 files changed, 590 insertions(+), 276 deletions(-) create mode 100644 miio/descriptorcollection.py delete mode 100644 miio/integrations/genericmiot/cli_helpers.py create mode 100644 miio/tests/test_descriptorcollection.py diff --git a/miio/__init__.py b/miio/__init__.py index 2e36067d0..895ae3e06 100644 --- a/miio/__init__.py +++ b/miio/__init__.py @@ -21,6 +21,16 @@ # isort: on from miio.cloud import CloudDeviceInfo, CloudException, CloudInterface +from miio.descriptorcollection import DescriptorCollection +from miio.descriptors import ( + AccessFlags, + ActionDescriptor, + Descriptor, + EnumDescriptor, + PropertyDescriptor, + RangeDescriptor, + ValidSettingRange, +) from miio.devicefactory import DeviceFactory from miio.integrations.airdog.airpurifier import AirDogX3 from miio.integrations.cgllc.airmonitor import AirQualityMonitor, AirQualityMonitorCGDN1 diff --git a/miio/descriptorcollection.py b/miio/descriptorcollection.py new file mode 100644 index 000000000..b8097d73d --- /dev/null +++ b/miio/descriptorcollection.py @@ -0,0 +1,138 @@ +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. + + This is a glorified dictionary that provides several useful features for handling + descriptors like binding names (method_name, setter_name) to *device* callables, + setting property constraints, and handling duplicate identifiers. + """ + + 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) + + def _get_free_id(id_, suffix=2): + if id_ not in self.data: + return id_ + + while f"{id_}-{suffix}" in self.data: + suffix += 1 + + return f"{id_}-{suffix}" + + descriptor.id = _get_free_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 ValueError(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 ValueError(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() + + if prop.choices is None: + raise ValueError( + f"Neither choices nor choices_attribute was defined for {prop}" + ) + + 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 diff --git a/miio/device.py b/miio/device.py index 5a53e4167..6c29d053d 100644 --- a/miio/device.py +++ b/miio/device.py @@ -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 @@ -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( @@ -179,62 +172,26 @@ 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. + + 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._properties = self._set_constraints_from_attributes(status) - self._actions = self._action_descriptors() + # Read descriptors from the status class + self._descriptors.descriptors_from_object(self.status.__annotations__["return"]) + + if not self._descriptors: + _LOGGER.warning( + "'%s' does not specify any descriptors, please considering creating a PR.", + self.__class__.__name__, + ) + + self._initialized = True @property def device_id(self) -> int: @@ -323,49 +280,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] + return self._descriptors - def properties(self) -> Dict[str, PropertyDescriptor]: - """Return all device properties.""" - if self._properties is None: - self._initialize_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. @@ -379,5 +343,38 @@ 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.""" + try: + act = self.actions()[name] + except KeyError: + raise ValueError("Unable to find action '%s'" % name) + + if params is None: + return act.method() + + 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.""" + try: + setting = self.settings()[name] + except KeyError: + raise ValueError("Unable to find setting '%s'" % name) + + params = params if params is not None else [] + + return setting.setter(params) + def __repr__(self): return f"<{self.__class__.__name__ }: {self.ip} (token: {self.token})>" diff --git a/miio/devicestatus.py b/miio/devicestatus.py index ba67f8a82..c5918ac5e 100644 --- a/miio/devicestatus.py +++ b/miio/devicestatus.py @@ -34,7 +34,7 @@ class _StatusMeta(type): def __new__(metacls, name, bases, namespace, **kwargs): cls = super().__new__(metacls, name, bases, namespace) - cls._properties: Dict[str, PropertyDescriptor] = {} + cls._descriptors: Dict[str, PropertyDescriptor] = {} cls._parent: Optional["DeviceStatus"] = None cls._embedded: Dict[str, "DeviceStatus"] = {} @@ -44,9 +44,9 @@ def __new__(metacls, name, bases, namespace, **kwargs): descriptor = getattr(prop, "_descriptor", None) if descriptor: _LOGGER.debug(f"Found descriptor for {name} {descriptor}") - if n in cls._properties: + if n in cls._descriptors: raise ValueError(f"Duplicate {n} for {name} {descriptor}") - cls._properties[n] = descriptor + cls._descriptors[n] = descriptor _LOGGER.debug("Created %s.%s: %s", name, n, descriptor) return cls @@ -91,7 +91,7 @@ def properties(self) -> Dict[str, PropertyDescriptor]: Use @sensor and @setting decorators to define properties. """ - return self._properties # type: ignore[attr-defined] + return self._descriptors # type: ignore[attr-defined] def settings(self) -> Dict[str, PropertyDescriptor]: """Return the dict of settings exposed by the status container. @@ -117,16 +117,16 @@ def embed(self, name: str, other: "DeviceStatus"): self._embedded[name] = other other._parent = self # type: ignore[attr-defined] - for prop_id, prop in other.properties().items(): - final_name = f"{name}__{prop_id}" + for property_name, prop in other.properties().items(): + final_name = f"{name}__{property_name}" - self._properties[final_name] = attr.evolve( + self._descriptors[final_name] = attr.evolve( prop, status_attribute=final_name ) def __dir__(self) -> Iterable[str]: """Overridden to include properties from embedded containers.""" - return list(super().__dir__()) + list(self._embedded) + list(self._properties) + return list(super().__dir__()) + list(self._embedded) + list(self._descriptors) @property def __cli_output__(self) -> str: @@ -255,10 +255,6 @@ def decorator_setting(func): if setter is None and setter_name is None: raise Exception("setter_name needs to be defined") - if setter_name is None: - raise NotImplementedError( - "setter not yet implemented, use setter_name instead" - ) common_values = { "id": qualified_name, @@ -318,7 +314,7 @@ def decorator_action(func): method=None, extras=kwargs, ) - func._action = descriptor + func._descriptor = descriptor return func diff --git a/miio/integrations/genericmiot/cli_helpers.py b/miio/integrations/genericmiot/cli_helpers.py deleted file mode 100644 index 5b36d32ec..000000000 --- a/miio/integrations/genericmiot/cli_helpers.py +++ /dev/null @@ -1,54 +0,0 @@ -from typing import Dict, cast - -from miio.descriptors import ActionDescriptor, PropertyDescriptor -from miio.miot_models import MiotProperty, MiotService - -# TODO: these should be moved to a generic implementation covering all actions and settings - - -def pretty_actions(result: Dict[str, ActionDescriptor]): - """Pretty print actions.""" - out = "" - service = None - for _, desc in result.items(): - miot_prop: MiotProperty = desc.extras["miot_action"] - # service is marked as optional due pydantic backrefs.. - serv = cast(MiotService, miot_prop.service) - if service is None or service.siid != serv.siid: - service = serv - out += f"[bold]{service.description} ({service.name})[/bold]\n" - - out += f"\t{desc.id}\t\t{desc.name}" - if desc.inputs: - for idx, input_ in enumerate(desc.inputs, start=1): - param = input_.extras[ - "miot_property" - ] # TODO: hack until descriptors get support for descriptions - param_desc = f"\n\t\tParameter #{idx}: {param.name} ({param.description}) ({param.format}) {param.pretty_input_constraints}" - out += param_desc - - out += "\n" - - return out - - -def pretty_properties(result: Dict[str, PropertyDescriptor]): - """Pretty print settings.""" - out = "" - verbose = False - service = None - for _, desc in result.items(): - miot_prop: MiotProperty = desc.extras["miot_property"] - # service is marked as optional due pydantic backrefs.. - serv = cast(MiotService, miot_prop.service) - if service is None or service.siid != serv.siid: - service = serv - out += f"[bold]{service.name}[/bold] ({service.description})\n" - - out += f"\t{desc.name} ({desc.id}, access: {miot_prop.pretty_access})\n" - if verbose: - out += f' urn: {repr(desc.extras["urn"])}\n' - out += f' siid: {desc.extras["siid"]}\n' - out += f' piid: {desc.extras["piid"]}\n' - - return out diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 776ad960f..8da1bf47d 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -1,23 +1,14 @@ import logging from functools import partial -from typing import Dict, List, Optional +from typing import Dict, Optional -import click - -from miio import DeviceInfo, MiotDevice -from miio.click_common import LiteralParamType, command, format_output +from miio import MiotDevice +from miio.click_common import command from miio.descriptors import AccessFlags, ActionDescriptor, PropertyDescriptor from miio.miot_cloud import MiotCloud from miio.miot_device import MiotMapping -from miio.miot_models import ( - DeviceModel, - MiotAccess, - MiotAction, - MiotProperty, - MiotService, -) - -from .cli_helpers import pretty_actions, pretty_properties +from miio.miot_models import DeviceModel, MiotAccess, MiotAction, MiotService + from .status import GenericMiotStatus _LOGGER = logging.getLogger(__name__) @@ -54,7 +45,6 @@ def __init__( self._actions: Dict[str, ActionDescriptor] = {} self._properties: Dict[str, PropertyDescriptor] = {} - self._all_properties: List[MiotProperty] = [] def initialize_model(self): """Initialize the miot model and create descriptions.""" @@ -70,12 +60,10 @@ def initialize_model(self): def status(self) -> GenericMiotStatus: """Return status based on the miot model.""" properties = [] - for prop in self._all_properties: - if MiotAccess.Read not in prop.access: - continue - - name = prop.name - q = {"siid": prop.siid, "piid": prop.piid, "did": name} + for _, prop in self.sensors().items(): + extras = prop.extras + prop = extras["miot_property"] + q = {"siid": prop.siid, "piid": prop.piid, "did": prop.name} properties.append(q) # TODO: max properties needs to be made configurable (or at least splitted to avoid too large udp datagrams @@ -89,9 +77,6 @@ def status(self) -> GenericMiotStatus: def _create_action(self, act: MiotAction) -> Optional[ActionDescriptor]: """Create action descriptor for miot action.""" desc = act.get_descriptor() - if not desc: - return None - call_action = partial(self.call_action_by, act.siid, act.aiid) desc.method = call_action @@ -101,16 +86,7 @@ def _create_actions(self, serv: MiotService): """Create action descriptors.""" for act in serv.actions: act_desc = self._create_action(act) - if act_desc is None: # skip actions we cannot handle for now.. - continue - - if ( - act_desc.name in self._actions - ): # TODO: find a way to handle duplicates, suffix maybe? - _LOGGER.warning("Got used name name, ignoring '%s': %s", act.name, act) - continue - - self._actions[act_desc.name] = act_desc + self.descriptors().add_descriptor(act_desc) def _create_properties(self, serv: MiotService): """Create sensor and setting descriptors for a service.""" @@ -134,9 +110,7 @@ def _create_properties(self, serv: MiotService): self.set_property_by, prop.siid, prop.piid, name=prop.name ) - self._properties[prop.name] = desc - # TODO: all properties is only used as the descriptors (stored in _properties) do not have siid/piid - self._all_properties.append(prop) + self.descriptors().add_descriptor(desc) def _create_descriptors(self): """Create descriptors based on the miot model.""" @@ -154,62 +128,15 @@ def _create_descriptors(self): for sensor in self._properties.values(): _LOGGER.debug(f"\t{sensor}") - def _get_action_by_name(self, name: str): - """Return action by name.""" - # TODO: cache service:action? - for act in self._actions.values(): - if act.id == name: - if act.method_name is not None: - act.method = getattr(self, act.method_name) - - return act - - raise ValueError("No action with name/id %s" % name) - - @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.""" - params = params or [] - act = self._get_action_by_name(name) - 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.""" - params = params if params is not None else [] - setting = self._properties.get(name) - if setting is None: - raise ValueError("No property found for name %s" % name) - if setting.access & AccessFlags.Write == 0: - raise ValueError("Property %s is not writable" % name) - - return setting.setter(value=params) - - def _fetch_info(self) -> DeviceInfo: - """Hook to perform the model initialization.""" - info = super()._fetch_info() - self.initialize_model() - - return info + def _initialize_descriptors(self) -> None: + """Initialize descriptors. - @command(default_output=format_output(result_msg_fmt=pretty_actions)) - def actions(self) -> Dict[str, ActionDescriptor]: - """Return available actions.""" - return self._actions - - @command(default_output=format_output(result_msg_fmt=pretty_properties)) - def properties(self) -> Dict[str, PropertyDescriptor]: - """Return available sensors.""" - # TODO: move pretty-properties to be generic for all devices - return self._properties + This will be called by the base class to initialize the descriptors. We override + it here to construct our model instead of trying to request the status and use + that to find out the available features. + """ + self.initialize_model() + self._initialized = True @property def device_type(self) -> Optional[str]: diff --git a/miio/integrations/scishare/coffee/scishare_coffeemaker.py b/miio/integrations/scishare/coffee/scishare_coffeemaker.py index 06ab8e1a7..3ce51330b 100644 --- a/miio/integrations/scishare/coffee/scishare_coffeemaker.py +++ b/miio/integrations/scishare/coffee/scishare_coffeemaker.py @@ -3,8 +3,9 @@ import click -from miio import Device +from miio import Device, DeviceStatus from miio.click_common import command, format_output +from miio.devicestatus import sensor _LOGGER = logging.getLogger(__name__) @@ -26,15 +27,13 @@ class Status(IntEnum): NoWater = 203 -class ScishareCoffee(Device): - """Main class for Scishare coffee maker (scishare.coffee.s1102).""" - - _supported_models = ["scishare.coffee.s1102"] +class ScishareCoffeeStatus(DeviceStatus): + def __init__(self, data): + self.data = data - @command() - def status(self) -> int: - """Device status.""" - status_code = self.send("Query_Machine_Status")[1] + @sensor("Status") + def state(self) -> Status: + status_code = self.data[1] try: return Status(status_code) except ValueError: @@ -44,6 +43,17 @@ def status(self) -> int: ) return Status.Unknown + +class ScishareCoffee(Device): + """Main class for Scishare coffee maker (scishare.coffee.s1102).""" + + _supported_models = ["scishare.coffee.s1102"] + + @command() + def status(self) -> ScishareCoffeeStatus: + """Device status.""" + return ScishareCoffeeStatus(self.send("Query_Machine_Status")) + @command( click.argument("temperature", type=int), default_output=format_output("Setting preheat to {temperature}"), diff --git a/miio/integrations/zhimi/fan/zhimi_miot.py b/miio/integrations/zhimi/fan/zhimi_miot.py index 9a84eb47d..0b286713e 100644 --- a/miio/integrations/zhimi/fan/zhimi_miot.py +++ b/miio/integrations/zhimi/fan/zhimi_miot.py @@ -210,7 +210,7 @@ class FanZA5(MiotDevice): "Temperature: {result.temperature}\n", ) ) - def status(self): + def status(self) -> FanStatusZA5: """Retrieve properties.""" return FanStatusZA5( { diff --git a/miio/miot_device.py b/miio/miot_device.py index 699180df8..b58391588 100644 --- a/miio/miot_device.py +++ b/miio/miot_device.py @@ -99,7 +99,7 @@ def get_properties_for_mapping(self, *, max_properties=15) -> list: click.argument("name", type=str), click.argument("params", type=LiteralParamType(), required=False), ) - def call_action(self, name: str, params=None): + def call_action_from_mapping(self, name: str, params=None): """Call an action by a name in the mapping.""" mapping = self._get_mapping() if name not in mapping: diff --git a/miio/tests/test_descriptorcollection.py b/miio/tests/test_descriptorcollection.py new file mode 100644 index 000000000..0ed97dcfb --- /dev/null +++ b/miio/tests/test_descriptorcollection.py @@ -0,0 +1,191 @@ +import pytest + +from miio import ( + AccessFlags, + ActionDescriptor, + DescriptorCollection, + Device, + DeviceStatus, + EnumDescriptor, + PropertyDescriptor, + RangeDescriptor, + ValidSettingRange, +) +from miio.devicestatus import action, sensor, setting + + +@pytest.fixture +def dev(mocker): + d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff") + mocker.patch("miio.Device.send") + mocker.patch("miio.Device.send_handshake") + yield d + + +def test_descriptors_from_device_object(mocker): + """Test descriptor collection from device class.""" + + class DummyDevice(Device): + @action(id="test", name="test") + def test_action(self): + pass + + dev = DummyDevice("127.0.0.1", "68ffffffffffffffffffffffffffffff") + mocker.patch("miio.Device.send") + mocker.patch("miio.Device.send_handshake") + + coll = DescriptorCollection(device=dev) + coll.descriptors_from_object(DummyDevice()) + assert len(coll) == 1 + assert isinstance(coll["test"], ActionDescriptor) + + +def test_descriptors_from_status_object(dev): + coll = DescriptorCollection(device=dev) + + class TestStatus(DeviceStatus): + @sensor(id="test", name="test sensor") + def test_sensor(self): + pass + + @setting(id="test-setting", name="test setting", setter=lambda _: _) + def test_setting(self): + pass + + status = TestStatus() + coll.descriptors_from_object(status) + assert len(coll) == 2 + assert isinstance(coll["test"], PropertyDescriptor) + assert isinstance(coll["test-setting"], PropertyDescriptor) + assert coll["test-setting"].access & AccessFlags.Write + + +@pytest.mark.parametrize( + "cls, params", + [ + pytest.param(ActionDescriptor, {"method": lambda _: _}, id="action"), + pytest.param(PropertyDescriptor, {"status_attribute": "foo"}), + ], +) +def test_add_descriptor(dev: Device, cls, params): + """Test that adding a descriptor works.""" + coll: DescriptorCollection = DescriptorCollection(device=dev) + coll.add_descriptor(cls(id="id", name="test name", **params)) + assert len(coll) == 1 + assert coll["id"] is not None + + +def test_handle_action_descriptor(mocker, dev): + coll = DescriptorCollection(device=dev) + invalid_desc = ActionDescriptor(id="action", name="test name") + with pytest.raises(ValueError, match="Neither method or method_name was defined"): + coll.add_descriptor(invalid_desc) + + mocker.patch.object(dev, "existing_method", create=True) + + # Test method name binding + act_with_method_name = ActionDescriptor( + id="with-method-name", name="with-method-name", method_name="existing_method" + ) + coll.add_descriptor(act_with_method_name) + assert act_with_method_name.method is not None + + # Test non-existing method + act_with_method_name_missing = ActionDescriptor( + id="with-method-name-missing", + name="with-method-name-missing", + method_name="nonexisting_method", + ) + with pytest.raises(AttributeError): + coll.add_descriptor(act_with_method_name_missing) + + +def test_handle_writable_property_descriptor(mocker, dev): + coll = DescriptorCollection(device=dev) + data = { + "name": "", + "status_attribute": "", + "access": AccessFlags.Write, + } + invalid = PropertyDescriptor(id="missing_setter", **data) + with pytest.raises(ValueError, match="Neither setter or setter_name was defined"): + coll.add_descriptor(invalid) + + mocker.patch.object(dev, "existing_method", create=True) + + # Test name binding + setter_name_desc = PropertyDescriptor( + **data, id="setter_name", setter_name="existing_method" + ) + coll.add_descriptor(setter_name_desc) + assert setter_name_desc.setter is not None + + with pytest.raises(AttributeError): + coll.add_descriptor( + PropertyDescriptor( + **data, id="missing_setter", setter_name="non_existing_setter" + ) + ) + + +def test_handle_enum_constraints(dev, mocker): + coll = DescriptorCollection(device=dev) + + data = { + "name": "enum", + "status_attribute": "attr", + } + + mocker.patch.object(dev, "choices_attr", create=True) + + # Check that error is raised if choices are missing + invalid = EnumDescriptor(id="missing", **data) + with pytest.raises( + ValueError, match="Neither choices nor choices_attribute was defined" + ): + coll.add_descriptor(invalid) + + # Check that binding works + choices_attribute = EnumDescriptor( + id="with_choices_attr", choices_attribute="choices_attr", **data + ) + coll.add_descriptor(choices_attribute) + assert len(coll) == 1 + assert coll["with_choices_attr"].choices is not None + + +def test_handle_range_constraints(dev, mocker): + coll = DescriptorCollection(device=dev) + + data = { + "name": "name", + "status_attribute": "attr", + "min_value": 0, + "max_value": 100, + "step": 1, + } + + # Check regular descriptor + desc = RangeDescriptor(id="regular", **data) + coll.add_descriptor(desc) + assert coll["regular"].max_value == 100 + + mocker.patch.object(dev, "range", create=True, new=ValidSettingRange(-1, 1000, 10)) + range_attr = RangeDescriptor(id="range_attribute", range_attribute="range", **data) + coll.add_descriptor(range_attr) + + assert coll["range_attribute"].min_value == -1 + assert coll["range_attribute"].max_value == 1000 + assert coll["range_attribute"].step == 10 + + +def test_duplicate_identifiers(dev): + coll = DescriptorCollection(device=dev) + for i in range(3): + coll.add_descriptor( + ActionDescriptor(id="action", name=f"action {i}", method=lambda _: _) + ) + + assert coll["action"] + assert coll["action-2"] + assert coll["action-3"] diff --git a/miio/tests/test_device.py b/miio/tests/test_device.py index 7face6221..506097b97 100644 --- a/miio/tests/test_device.py +++ b/miio/tests/test_device.py @@ -2,7 +2,16 @@ import pytest -from miio import Device, MiotDevice, RoborockVacuum +from miio import ( + AccessFlags, + ActionDescriptor, + DescriptorCollection, + Device, + DeviceStatus, + MiotDevice, + PropertyDescriptor, + RoborockVacuum, +) from miio.exceptions import DeviceInfoUnavailableException, PayloadDecodeException DEVICE_CLASSES = Device.__subclasses__() + MiotDevice.__subclasses__() # type: ignore @@ -171,6 +180,13 @@ def test_init_signature(cls, mocker): assert total_args == 8 +@pytest.mark.parametrize("cls", DEVICE_CLASSES) +def test_status_return_type(cls): + """Make sure that all status methods have a type hint.""" + assert "return" in cls.status.__annotations__ + assert issubclass(cls.status.__annotations__["return"], DeviceStatus) + + def test_supports_miot(mocker): from miio.exceptions import DeviceError @@ -185,14 +201,90 @@ def test_supports_miot(mocker): @pytest.mark.parametrize("getter_name", ["actions", "settings", "sensors"]) -def test_cached_descriptors(getter_name, mocker): +def test_cached_descriptors(getter_name, mocker, caplog): d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff") getter = getattr(d, getter_name) initialize_descriptors = mocker.spy(d, "_initialize_descriptors") mocker.patch("miio.Device.send") - mocker.patch("miio.Device.status") - mocker.patch("miio.Device._set_constraints_from_attributes", return_value={}) - mocker.patch("miio.Device._action_descriptors", return_value={}) + patched_status = mocker.patch("miio.Device.status") + patched_status.__annotations__ = {} + patched_status.__annotations__["return"] = DeviceStatus + mocker.patch.object(d._descriptors, "descriptors_from_object", return_value={}) for _i in range(5): getter() initialize_descriptors.assert_called_once() + assert ( + "'Device' does not specify any descriptors, please considering creating a PR" + in caplog.text + ) + + +def test_change_setting(mocker): + """Test setting changing.""" + d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff") + mocker.patch("miio.Device.send") + mocker.patch("miio.Device.send_handshake") + + descs = { + "read-only": PropertyDescriptor( + id="ro", name="ro", status_attribute="ro", access=AccessFlags.Read + ), + "write-only": PropertyDescriptor( + id="wo", name="wo", status_attribute="wo", access=AccessFlags.Write + ), + } + writable = descs["write-only"] + coll = DescriptorCollection(descs, device=d) + + mocker.patch.object(d, "descriptors", return_value=coll) + + # read-only descriptors should not appear in settings + assert len(d.settings()) == 1 + + # trying to change non-existing setting should raise an error + with pytest.raises( + ValueError, match="Unable to find setting 'non-existing-setting'" + ): + d.change_setting("non-existing-setting") + + # calling change setting should call the setter of the descriptor + setter = mocker.patch.object(writable, "setter") + d.change_setting("write-only", "new value") + setter.assert_called_with("new value") + + +def test_call_action(mocker): + """Test action calling.""" + d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff") + mocker.patch("miio.Device.send") + mocker.patch("miio.Device.send_handshake") + + descs = { + "read-only": PropertyDescriptor( + id="ro", name="ro", status_attribute="ro", access=AccessFlags.Read + ), + "write-only": PropertyDescriptor( + id="wo", name="wo", status_attribute="wo", access=AccessFlags.Write + ), + "action": ActionDescriptor(id="foo", name="action"), + } + act = descs["action"] + coll = DescriptorCollection(descs, device=d) + + mocker.patch.object(d, "descriptors", return_value=coll) + + # property descriptors should not appear in actions + assert len(d.actions()) == 1 + + # trying to execute non-existing action should raise an error + with pytest.raises(ValueError, match="Unable to find action 'non-existing-action'"): + d.call_action("non-existing-action") + + method = mocker.patch.object(act, "method") + d.call_action("action", "testinput") + method.assert_called_with("testinput") + method.reset_mock() + + # Calling without parameters executes a different code path + d.call_action("action") + method.assert_called_once() diff --git a/miio/tests/test_devicestatus.py b/miio/tests/test_devicestatus.py index 817f686ef..32bd06ac5 100644 --- a/miio/tests/test_devicestatus.py +++ b/miio/tests/test_devicestatus.py @@ -142,7 +142,9 @@ def level(self) -> int: d._protocol._device_id = b"12345678" # Patch status to return our class - mocker.patch.object(d, "status", return_value=Settings()) + status = mocker.patch.object(d, "status", return_value=Settings()) + status.__annotations__ = {} + status.__annotations__["return"] = Settings # Patch to create a new setter as defined in the status class setter = mocker.patch.object(d, "set_level", create=True) @@ -189,7 +191,10 @@ def level(self) -> int: d._protocol._device_id = b"12345678" # Patch status to return our class - mocker.patch.object(d, "status", return_value=Settings()) + status = mocker.patch.object(d, "status", return_value=Settings()) + status.__annotations__ = {} + status.__annotations__["return"] = Settings + mocker.patch.object(d, "valid_range", create=True, new=ValidSettingRange(1, 100, 2)) # Patch to create a new setter as defined in the status class setter = mocker.patch.object(d, "set_level", create=True) @@ -235,7 +240,9 @@ def level(self) -> TestEnum: d._protocol._device_id = b"12345678" # Patch status to return our class - mocker.patch.object(d, "status", return_value=Settings()) + status = mocker.patch.object(d, "status", return_value=Settings()) + status.__annotations__ = {} + status.__annotations__["return"] = Settings # Patch to create a new setter as defined in the status class setter = mocker.patch.object(d, "set_level", create=True) diff --git a/miio/tests/test_miotdevice.py b/miio/tests/test_miotdevice.py index b250a5c91..5ae88061e 100644 --- a/miio/tests/test_miotdevice.py +++ b/miio/tests/test_miotdevice.py @@ -167,10 +167,10 @@ def test_supported_models(cls): assert not cls._supported_models -def test_call_action(dev): +def test_call_action_from_mapping(dev): dev._mappings["test.model"] = {"test_action": {"siid": 1, "aiid": 1}} - dev.call_action("test_action") + dev.call_action_from_mapping("test_action") @pytest.mark.parametrize(