diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 39f2f09d3..27420fdb1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -89,3 +89,4 @@ jobs: uses: "codecov/codecov-action@v2" with: fail_ci_if_error: true + token: ${{ secrets.CODECOV_TOKEN }} diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 134b2892c..841bb3cbe 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -34,7 +34,7 @@ repos: - id: docformatter args: [--in-place, --wrap-summaries, '88', --wrap-descriptions, '88'] -- repo: https://gitlab.com/pycqa/flake8 +- repo: https://github.com/pycqa/flake8 rev: 3.9.2 hooks: - id: flake8 diff --git a/docs/contributing.rst b/docs/contributing.rst index 9b1bb0e64..6afb51989 100644 --- a/docs/contributing.rst +++ b/docs/contributing.rst @@ -194,7 +194,7 @@ Development checklist listing the known models (as reported by :meth:`~miio.device.Device.info()`). 4. Status containers is derived from :class:`~miio.devicestatus.DeviceStatus` class and all properties should have type annotations for their return values. The information that should be exposed directly - to end users should be decorated using appropriate decorators (e.g., `@sensor` or `@switch`) to make + to end users should be decorated using appropriate decorators (e.g., `@sensor` or `@setting`) to make them discoverable (:ref:`status_containers`). 5. Add tests at least for the status container handling (:ref:`adding_tests`). 6. Updating documentation is generally not needed as the API documentation @@ -293,36 +293,19 @@ This will make all decorated sensors accessible through :meth:`~miio.device.Devi device class information to Home Assistant. -Switches -"""""""" - -Use :meth:`@switch ` to create :class:`~miio.descriptors.SwitchDescriptor` objects. -This will make all decorated switches accessible through :meth:`~miio.device.Device.switches` for downstream users. - -.. code-block:: - - @property - @switch(name="Power", setter_name="set_power") - def power(self) -> bool: - """Return if device is turned on.""" - -You can either use *setter* to define a callable that can be used to adjust the value of the property, -or alternatively define *setter_name* which will be used to bind the method during the initialization -to the the :meth:`~miio.descriptors.SwitchDescriptor.setter` callable. - - Settings """""""" -Use :meth:`@switch ` to create :meth:`~miio.descriptors.SettingDescriptor` objects. +Use :meth:`@setting ` to create :meth:`~miio.descriptors.SettingDescriptor` objects. This will make all decorated settings accessible through :meth:`~miio.device.Device.settings` for downstream users. The type of the descriptor depends on the input parameters: * Passing *min_value* or *max_value* will create a :class:`~miio.descriptors.NumberSettingDescriptor`, which is useful for presenting ranges of values. - * Passing an Enum object using *choices* will create a :class:`~miio.descriptors.EnumSettingDescriptor`, - which is useful for presenting a fixed set of options. + * Passing an :class:`enum.Enum` object using *choices* will create a + :class:`~miio.descriptors.EnumSettingDescriptor`, which is useful for presenting a fixed set of options. + * Otherwise, the setting is considered to be boolean switch. You can either use *setter* to define a callable that can be used to adjust the value of the property, @@ -338,7 +321,7 @@ The *max_value* is the only mandatory parameter. If not given, *min_value* defau .. code-block:: @property - @switch(name="Fan Speed", min_value=0, max_value=100, steps=5, setter_name="set_fan_speed") + @setting(name="Fan Speed", min_value=0, max_value=100, steps=5, setter_name="set_fan_speed") def fan_speed(self) -> int: """Return the current fan speed.""" @@ -356,11 +339,33 @@ If the device has a setting with some pre-defined values, you want to use this. Off = 2 @property - @switch(name="LED Brightness", choices=SomeEnum, setter_name="set_led_brightness") + @setting(name="LED Brightness", choices=SomeEnum, setter_name="set_led_brightness") def led_brightness(self) -> LedBrightness: """Return the LED brightness.""" +Actions +""""""" + +Use :meth:`@action ` to create :class:`~miio.descriptors.ActionDescriptor` +objects for the device. +This will make all decorated actions accessible through :meth:`~miio.device.Device.actions` for downstream users. + +.. code-block:: python + + @command() + @action(name="Do Something", some_kwarg_for_downstream="hi there") + def do_something(self): + """Execute some action on the device.""" + +.. note:: + + All keywords arguments not defined in the decorator signature will be available + through the :attr:`~miio.descriptors.ActionDescriptor.extras` variable. + + This information can be used to pass information to the downstream users. + + .. _adding_tests: Adding tests diff --git a/docs/simulator.rst b/docs/simulator.rst index 83d273586..6504271d9 100644 --- a/docs/simulator.rst +++ b/docs/simulator.rst @@ -198,4 +198,60 @@ concrete example for a device using custom method names for obtaining the status MiOT Simulator -------------- -.. note:: TBD. +The ``miiocli devtools miot-simulator`` command can be used to simulate MiOT devices for a given description file. +You can command the simulated devices using the ``miiocli`` tool or any other implementation that supports the device. + +Behind the scenes, the simulator uses :class:`the push server ` to +handle the low-level protocol handling. + +The simulator implements the following methods: + + * ``miIO.info`` returns the device information + * ``get_properties`` returns randomized (leveraging the schema limits) values for the given ``siid`` and ``piid`` + * ``set_properties`` allows setting the property for the given ``siid`` and ``piid`` combination + * ``action`` to call actions that simply respond that the action succeeded + +Furthermore, two custom methods are implemented help with development: + + * ``dump_services`` returns the :ref:`list of available services ` + * ``dump_properties`` returns the :ref:`available properties and their values ` the given ``siid`` + + +Usage +""""" + +You start the simulator like this:: + + miiocli devtools miot-simulator --file some.vacuum.model.json --model some.vacuum.model + +The mandatory ``--file`` option takes a path to a MiOT description file, while ``--model`` defines the model +the simulator should report in its ``miIO.info`` response. + +.. note:: + + The default token is hardcoded to full of zeros (``00000000000000000000000000000000``). + + +.. _dump_services: + +Dump Service Information +~~~~~~~~~~~~~~~~~~~~~~~~ + +``dump_services`` method that returns a JSON dictionary keyed with the ``siid`` containing the simulated services:: + + + $ miiocli device --ip 127.0.0.1 --token 00000000000000000000000000000000 raw_command dump_services + Running command raw_command + {'services': {'1': {'siid': 1, 'description': 'Device Information'}, '2': {'siid': 2, 'description': 'Heater'}, '3': {'siid': 3, 'description': 'Countdown'}, '4': {'siid': 4, 'description': 'Environment'}, '5': {'siid': 5, 'description': 'Physical Control Locked'}, '6': {'siid': 6, 'description': 'Alarm'}, '7': {'siid': 7, 'description': 'Indicator Light'}, '8': {'siid': 8, 'description': '私有服务'}}, 'id': 2} + + +.. _dump_properties: + +Dump Service Properties +~~~~~~~~~~~~~~~~~~~~~~~ + +``dump_properties`` method can be used to return the current state of the device on service-basis:: + + $ miiocli device --ip 127.0.0.1 --token 00000000000000000000000000000000 raw_command dump_properties '{"siid": 2}' + Running command raw_command + [{'siid': 2, 'piid': 1, 'prop': 'Switch Status', 'value': False}, {'siid': 2, 'piid': 2, 'prop': 'Device Fault', 'value': 167}, {'siid': 2, 'piid': 5, 'prop': 'Target Temperature', 'value': 28}] diff --git a/miio/__init__.py b/miio/__init__.py index 45a8c9e7d..79b474eb8 100644 --- a/miio/__init__.py +++ b/miio/__init__.py @@ -12,6 +12,7 @@ from miio.exceptions import DeviceError, DeviceException, UnsupportedFeatureException from miio.miot_device import MiotDevice from miio.deviceinfo import DeviceInfo +from miio.interfaces import VacuumInterface, LightInterface, ColorTemperatureRange # isort: on diff --git a/miio/cli.py b/miio/cli.py index 653b219f0..5d722d142 100644 --- a/miio/cli.py +++ b/miio/cli.py @@ -1,4 +1,5 @@ import logging +from typing import Any, Dict import click @@ -29,11 +30,22 @@ @click.version_option() @click.pass_context def cli(ctx, debug: int, output: str): - if debug: - logging.basicConfig(level=logging.DEBUG) - _LOGGER.info("Debug mode active") - else: - logging.basicConfig(level=logging.INFO) + logging_config: Dict[str, Any] = { + "level": logging.DEBUG if debug > 0 else logging.INFO + } + try: + from rich.logging import RichHandler + + rich_config = { + "show_time": False, + } + logging_config["handlers"] = [RichHandler(**rich_config)] + logging_config["format"] = "%(message)s" + except ImportError: + pass + + # The configuration should be converted to use dictConfig, but this keeps mypy happy for now + logging.basicConfig(**logging_config) # type: ignore if output in ("json", "json_pretty"): output_func = json_output(pretty=output == "json_pretty") diff --git a/miio/click_common.py b/miio/click_common.py index 081dbc044..46ab00e0f 100644 --- a/miio/click_common.py +++ b/miio/click_common.py @@ -12,10 +12,13 @@ import click -import miio - from .exceptions import DeviceError +try: + from rich import print as echo +except ImportError: + echo = click.echo + _LOGGER = logging.getLogger(__name__) @@ -49,9 +52,8 @@ class ExceptionHandlerGroup(click.Group): def __call__(self, *args, **kwargs): try: return self.main(*args, **kwargs) - except (ValueError, miio.DeviceException) as ex: - _LOGGER.debug("Exception: %s", ex, exc_info=True) - click.echo(click.style("Error: %s" % ex, fg="red", bold=True)) + except Exception as ex: + _LOGGER.exception("Exception: %s", ex) class EnumType(click.Choice): @@ -179,10 +181,7 @@ def _wrap(self, *args, **kwargs): and self._model is None and self._info is None ): - _LOGGER.debug( - "Unknown model, trying autodetection. %s %s" - % (self._model, self._info) - ) + _LOGGER.debug("Unknown model, trying autodetection") self._fetch_info() return func(self, *args, **kwargs) @@ -304,7 +303,7 @@ def wrap(*args, **kwargs): else: msg = msg_fmt.format(**kwargs) if msg: - click.echo(msg.strip()) + echo(msg.strip()) kwargs["result"] = func(*args, **kwargs) if result_msg_fmt: if callable(result_msg_fmt): @@ -312,7 +311,7 @@ def wrap(*args, **kwargs): else: result_msg = result_msg_fmt.format(**kwargs) if result_msg: - click.echo(result_msg.strip()) + echo(result_msg.strip()) return wrap @@ -328,7 +327,7 @@ def wrap(*args, **kwargs): try: result = func(*args, **kwargs) except DeviceError as ex: - click.echo(json.dumps(ex.args[0], indent=indent)) + echo(json.dumps(ex.args[0], indent=indent)) return get_json_data_func = getattr(result, "__json__", None) @@ -337,7 +336,7 @@ def wrap(*args, **kwargs): result = get_json_data_func() elif data_variable is not None: result = data_variable - click.echo(json.dumps(result, indent=indent)) + echo(json.dumps(result, indent=indent)) return wrap diff --git a/miio/descriptors.py b/miio/descriptors.py index 7447bddb0..5b5e8ee40 100644 --- a/miio/descriptors.py +++ b/miio/descriptors.py @@ -3,28 +3,28 @@ The descriptors contain information that can be used to provide generic, dynamic user-interfaces. If you are a downstream developer, use :func:`~miio.device.Device.sensors()`, -:func:`~miio.device.Device.settings()`, :func:`~miio.device.Device.switches()`, and -:func:`~miio.device.Device.buttons()` to access the functionality exposed by the integration developer. +:func:`~miio.device.Device.settings()`, and +:func:`~miio.device.Device.actions()` to access the functionality exposed by the integration developer. -If you are developing an integration, prefer :func:`~miio.devicestatus.sensor`, :func:`~miio.devicestatus.sensor`, and -:func:`~miio.devicestatus.sensor` decorators over creating the descriptors manually. +If you are developing an integration, prefer :func:`~miio.devicestatus.sensor`, :func:`~miio.devicestatus.setting`, and +:func:`~miio.devicestatus.action` decorators over creating the descriptors manually. If needed, you can override the methods listed to add more descriptors to your integration. """ from enum import Enum, auto -from typing import Callable, Dict, Optional +from typing import Callable, Dict, Optional, Type import attr @attr.s(auto_attribs=True) -class ButtonDescriptor: +class ActionDescriptor: """Describes a button exposed by the device.""" id: str name: str - method_name: str - method: Optional[Callable] = None - extras: Optional[Dict] = None + method_name: Optional[str] = attr.ib(default=None, repr=False) + method: Optional[Callable] = attr.ib(default=None, repr=False) + extras: Dict = attr.ib(factory=dict, repr=False) @attr.s(auto_attribs=True) @@ -38,23 +38,18 @@ class SensorDescriptor: """ id: str - type: str + type: type name: str property: str unit: Optional[str] = None - extras: Optional[Dict] = None + extras: Dict = attr.ib(factory=dict, repr=False) -@attr.s(auto_attribs=True) -class SwitchDescriptor: - """Presents toggleable switch.""" - - id: str - name: str - property: str - setter_name: Optional[str] = None - setter: Optional[Callable] = None - extras: Optional[Dict] = None +class SettingType(Enum): + Undefined = auto() + Number = auto() + Boolean = auto() + Enum = auto() @attr.s(auto_attribs=True, kw_only=True) @@ -64,15 +59,27 @@ class SettingDescriptor: id: str name: str property: str - unit: str - setter: Optional[Callable] = None - setter_name: Optional[str] = None + unit: Optional[str] = None + type = SettingType.Undefined + setter: Optional[Callable] = attr.ib(default=None, repr=False) + setter_name: Optional[str] = attr.ib(default=None, repr=False) + extras: Dict = attr.ib(factory=dict, repr=False) + def cast_value(self, value: int): + """Casts value to the expected type.""" + cast_map = { + SettingType.Boolean: bool, + SettingType.Enum: int, + SettingType.Number: int, + } + return cast_map[self.type](int(value)) -class SettingType(Enum): - Number = auto() - Boolean = auto() - Enum = auto() + +@attr.s(auto_attribs=True, kw_only=True) +class BooleanSettingDescriptor(SettingDescriptor): + """Presents a settable boolean value.""" + + type: SettingType = SettingType.Boolean @attr.s(auto_attribs=True, kw_only=True) @@ -80,9 +87,8 @@ class EnumSettingDescriptor(SettingDescriptor): """Presents a settable, enum-based value.""" type: SettingType = SettingType.Enum - choices_attribute: Optional[str] = None - choices: Optional[Enum] = None - extras: Optional[Dict] = None + choices_attribute: Optional[str] = attr.ib(default=None, repr=False) + choices: Optional[Type[Enum]] = attr.ib(default=None, repr=False) @attr.s(auto_attribs=True, kw_only=True) @@ -93,4 +99,3 @@ class NumberSettingDescriptor(SettingDescriptor): max_value: int step: int type: SettingType = SettingType.Number - extras: Optional[Dict] = None diff --git a/miio/device.py b/miio/device.py index 54e98f9f2..076346cb4 100644 --- a/miio/device.py +++ b/miio/device.py @@ -1,15 +1,16 @@ import logging from enum import Enum +from inspect import getmembers from typing import Any, Dict, List, Optional, Union # noqa: F401 import click from .click_common import DeviceGroupMeta, LiteralParamType, command, format_output from .descriptors import ( - ButtonDescriptor, + ActionDescriptor, + EnumSettingDescriptor, SensorDescriptor, SettingDescriptor, - SwitchDescriptor, ) from .deviceinfo import DeviceInfo from .devicestatus import DeviceStatus @@ -62,6 +63,7 @@ def __init__( self.token: Optional[str] = token self._model: Optional[str] = model self._info: Optional[DeviceInfo] = None + self._actions: Optional[Dict[str, ActionDescriptor]] = None timeout = timeout if timeout is not None else self.timeout self._protocol = MiIOProtocol( ip, token, start_id, debug, lazy_discover, timeout @@ -246,12 +248,20 @@ def status(self) -> DeviceStatus: """Return device status.""" raise NotImplementedError() - def buttons(self) -> List[ButtonDescriptor]: - """Return a list of button-like, clickable actions of the device.""" - return [] + 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 + + return self._actions def settings(self) -> Dict[str, SettingDescriptor]: - """Return list of settings.""" + """Return device settings.""" settings = self.status().settings() for setting in settings.values(): # TODO: Bind setter methods, this should probably done only once during init. @@ -263,29 +273,20 @@ def settings(self) -> Dict[str, SettingDescriptor]: ) 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 return settings def sensors(self) -> Dict[str, SensorDescriptor]: - """Return sensors.""" + """Return device sensors.""" # TODO: the latest status should be cached and re-used by all meta information getters sensors = self.status().sensors() return sensors - def switches(self) -> Dict[str, SwitchDescriptor]: - """Return toggleable switches.""" - switches = self.status().switches() - for switch in switches.values(): - # TODO: Bind setter methods, this should probably done only once during init. - if switch.setter is None: - if switch.setter_name is None: - # TODO: this is ugly, how to fix the issue where setter_name is optional and thus not acceptable for getattr? - raise Exception( - f"Neither setter or setter_name was defined for {switch}" - ) - switch.setter = getattr(self, switch.setter_name) - - return switches - def __repr__(self): return f"<{self.__class__.__name__ }: {self.ip} (token: {self.token})>" diff --git a/miio/devicestatus.py b/miio/devicestatus.py index bef394dd5..1c2b1b2c1 100644 --- a/miio/devicestatus.py +++ b/miio/devicestatus.py @@ -14,11 +14,13 @@ ) from .descriptors import ( + ActionDescriptor, + BooleanSettingDescriptor, EnumSettingDescriptor, NumberSettingDescriptor, SensorDescriptor, SettingDescriptor, - SwitchDescriptor, + SettingType, ) _LOGGER = logging.getLogger(__name__) @@ -32,14 +34,12 @@ def __new__(metacls, name, bases, namespace, **kwargs): # TODO: clean up to contain all of these in a single container cls._sensors: Dict[str, SensorDescriptor] = {} - cls._switches: Dict[str, SwitchDescriptor] = {} cls._settings: Dict[str, SettingDescriptor] = {} cls._embedded: Dict[str, "DeviceStatus"] = {} descriptor_map = { "sensor": cls._sensors, - "switch": cls._switches, "setting": cls._settings, } for n in namespace: @@ -60,7 +60,7 @@ class DeviceStatus(metaclass=_StatusMeta): All status container classes should inherit from this class: * This class allows downstream users to access the available information in an - introspectable way. See :func:`@property`, :func:`switch`, and :func:`@setting`. + introspectable way. See :func:`@sensor` and :func:`@setting`. * :func:`embed` allows embedding other status containers. * The __repr__ implementation returns all defined properties and their values. """ @@ -93,17 +93,10 @@ def sensors(self) -> Dict[str, SensorDescriptor]: """ return self._sensors # type: ignore[attr-defined] - def switches(self) -> Dict[str, SwitchDescriptor]: - """Return the dict of sensors exposed by the status container. - - You can use @sensor decorator to define sensors inside your status class. - """ - return self._switches # type: ignore[attr-defined] - def settings(self) -> Dict[str, SettingDescriptor]: """Return the dict of settings exposed by the status container. - You can use @setting decorator to define sensors inside your status class. + You can use @setting decorator to define settings inside your status class. """ return self._settings # type: ignore[attr-defined] @@ -121,29 +114,25 @@ def embed(self, other: "DeviceStatus"): self._embedded[other_name] = other for name, sensor in other.sensors().items(): - final_name = f"{other_name}:{name}" + final_name = f"{other_name}__{name}" import attr self._sensors[final_name] = attr.evolve(sensor, property=final_name) - for name, switch in other.switches().items(): - final_name = f"{other_name}:{name}" - self._switches[final_name] = attr.evolve(switch, property=final_name) - for name, setting in other.settings().items(): - final_name = f"{other_name}:{name}" + final_name = f"{other_name}__{name}" self._settings[final_name] = attr.evolve(setting, property=final_name) - def __getattribute__(self, item): + def __getattr__(self, item): """Overridden to lookup properties from embedded containers.""" - if ":" not in item: - return super().__getattribute__(item) + if "__" not in item: + return super().__getattr__(item) - embed, prop = item.split(":") + embed, prop = item.split("__") return getattr(self._embedded[embed], prop) -def sensor(name: str, *, unit: str = "", **kwargs): +def sensor(name: str, *, unit: Optional[str] = None, **kwargs): """Syntactic sugar to create SensorDescriptor objects. The information can be used by users of the library to programmatically find out what @@ -155,22 +144,20 @@ def sensor(name: str, *, unit: str = "", **kwargs): """ def decorator_sensor(func): - property_name = func.__name__ + property_name = str(func.__name__) + qualified_name = str(func.__qualname__) def _sensor_type_for_return_type(func): rtype = get_type_hints(func).get("return") if get_origin(rtype) is Union: # Unwrap Optional[] rtype, _ = get_args(rtype) - if rtype == bool: - return "binary" - else: - return "sensor" + return rtype sensor_type = _sensor_type_for_return_type(func) descriptor = SensorDescriptor( - id=str(property_name), - property=str(property_name), + id=qualified_name, + property=property_name, name=name, unit=unit, type=sensor_type, @@ -183,34 +170,6 @@ def _sensor_type_for_return_type(func): return decorator_sensor -def switch(name: str, *, setter_name: str, **kwargs): - """Syntactic sugar to create SwitchDescriptor objects. - - The information can be used by users of the library to programmatically find out what - types of sensors are available for the device. - - The interface is kept minimal, but you can pass any extra keyword arguments. - These extras are made accessible over :attr:`~miio.descriptors.SwitchDescriptor.extras`, - and can be interpreted downstream users as they wish. - """ - - def decorator_sensor(func): - property_name = func.__name__ - - descriptor = SwitchDescriptor( - id=str(property_name), - property=str(property_name), - name=name, - setter_name=setter_name, - extras=kwargs, - ) - func._switch = descriptor - - return func - - return decorator_sensor - - def setting( name: str, *, @@ -222,6 +181,7 @@ def setting( step: Optional[int] = None, choices: Optional[Type[Enum]] = None, choices_attribute: Optional[str] = None, + type: Optional[SettingType] = None, **kwargs, ): """Syntactic sugar to create SettingDescriptor objects. @@ -235,47 +195,73 @@ def setting( """ def decorator_setting(func): - property_name = func.__name__ + property_name = str(func.__name__) + qualified_name = str(func.__qualname__) if setter is None and setter_name is None: - raise Exception("Either setter or setter_name needs to be defined") + 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, + "property": property_name, + "name": name, + "unit": unit, + "setter": setter, + "setter_name": setter_name, + "extras": kwargs, + } if min_value or max_value: descriptor = NumberSettingDescriptor( - id=str(property_name), - property=str(property_name), - name=name, - unit=unit, - setter=setter, - setter_name=setter_name, + **common_values, min_value=min_value or 0, max_value=max_value, step=step or 1, - extras=kwargs, ) elif choices or choices_attribute: - if choices_attribute is not None: - # TODO: adding choices from attribute is a bit more complex, as it requires a way to - # construct enums pointed by the attribute - raise NotImplementedError("choices_attribute is not yet implemented") descriptor = EnumSettingDescriptor( - id=str(property_name), - property=str(property_name), - name=name, - unit=unit, - setter=setter, - setter_name=setter_name, + **common_values, choices=choices, choices_attribute=choices_attribute, - extras=kwargs, ) else: - raise Exception( - "Neither {min,max}_value or choices_{attribute} was defined" - ) + descriptor = BooleanSettingDescriptor(**common_values) func._setting = descriptor return func return decorator_setting + + +def action(name: str, **kwargs): + """Syntactic sugar to create ActionDescriptor objects. + + The information can be used by users of the library to programmatically find out what + types of actions are available for the device. + + The interface is kept minimal, but you can pass any extra keyword arguments. + These extras are made accessible over :attr:`~miio.descriptors.ActionDescriptor.extras`, + and can be interpreted downstream users as they wish. + """ + + def decorator_action(func): + property_name = str(func.__name__) + qualified_name = str(func.__qualname__) + + descriptor = ActionDescriptor( + id=qualified_name, + name=name, + method_name=property_name, + method=None, + extras=kwargs, + ) + func._action = descriptor + + return func + + return decorator_action diff --git a/miio/devtools/pcapparser.py b/miio/devtools/pcapparser.py index 3cc4385f2..7def63bae 100644 --- a/miio/devtools/pcapparser.py +++ b/miio/devtools/pcapparser.py @@ -1,10 +1,17 @@ """Parse PCAP files for miio traffic.""" from collections import Counter, defaultdict from ipaddress import ip_address +from pprint import pformat as pf from typing import List import click +try: + from rich import print as echo +except ImportError: + echo = click.echo + + from miio import Message @@ -14,7 +21,7 @@ def read_payloads_from_file(file, tokens: List[str]): import dpkt from dpkt.ethernet import ETH_TYPE_IP, Ethernet except ImportError: - print("You need to install dpkt to use this tool") # noqa: T201 + echo("You need to install dpkt to use this tool") return pcap = dpkt.pcap.Reader(file) @@ -70,9 +77,9 @@ def read_payloads_from_file(file, tokens: List[str]): yield src_addr, dst_addr, payload for cat in stats: - print(f"\n== {cat} ==") # noqa: T201 + echo(f"\n== {cat} ==") for stat, value in stats[cat].items(): - print(f"\t{stat}: {value}") # noqa: T201 + echo(f"\t{stat}: {value}") @click.command() @@ -81,4 +88,4 @@ def read_payloads_from_file(file, tokens: List[str]): def parse_pcap(file, token: List[str]): """Read PCAP file and output decrypted miio communication.""" for src_addr, dst_addr, payload in read_payloads_from_file(file, token): - print(f"{src_addr:<15} -> {dst_addr:<15} {payload}") # noqa: T201 + echo(f"{src_addr:<15} -> {dst_addr:<15} {pf(payload)}") diff --git a/miio/devtools/simulators/miotsimulator.py b/miio/devtools/simulators/miotsimulator.py index f92c7e16e..81e6845a8 100644 --- a/miio/devtools/simulators/miotsimulator.py +++ b/miio/devtools/simulators/miotsimulator.py @@ -8,13 +8,16 @@ from pydantic import Field, validator from miio import PushServer +from miio.miot_cloud import MiotCloud +from miio.miot_models import DeviceModel, MiotProperty, MiotService from .common import create_info_response, mac_from_model -from .models import DeviceModel, MiotProperty, MiotService _LOGGER = logging.getLogger(__name__) UNSET = -10000 +ERR_INVALID_SETTING = -1000 + def create_random(values): """Create random value for the given mapping.""" @@ -72,14 +75,14 @@ def verify_value(cls, v, values): raise ValueError(f"{casted_value} not in range {range}") choices = values["choices"] - if choices is not None: - return choices[casted_value] + if choices is not None and not any(c.value == casted_value for c in choices): + raise ValueError(f"{casted_value} not found in {choices}") return casted_value class Config: validate_assignment = True - smart_union = True + smart_union = True # try all types before coercing class SimulatedMiotService(MiotService): @@ -109,10 +112,12 @@ def __init__(self, device_model): def initialize_state(self): """Create initial state for the device.""" for serv in self._model.services: + _LOGGER.debug("Found service: %s", serv) for act in serv.actions: _LOGGER.debug("Found action: %s", act) for prop in serv.properties: self._state[serv.siid][prop.piid] = prop + _LOGGER.debug("Found property: %s", prop) def get_properties(self, payload): """Handle get_properties method.""" @@ -121,8 +126,13 @@ def get_properties(self, payload): params = payload["params"] for p in params: res = p.copy() - res["value"] = self._state[res["siid"]][res["piid"]].current_value - res["code"] = 0 + try: + res["value"] = self._state[res["siid"]][res["piid"]].current_value + res["code"] = 0 + except Exception as ex: + res["value"] = "" + res["code"] = ERR_INVALID_SETTING + res["exception"] = str(ex) response.append(res) return {"result": response} @@ -195,12 +205,18 @@ async def main(dev, model): @click.command() -@click.option("--file", type=click.File("r"), required=True) +@click.option("--file", type=click.File("r"), required=False) @click.option("--model", type=str, required=True, default=None) def miot_simulator(file, model): """Simulate miot device.""" - data = file.read() - dev = SimulatedDeviceModel.parse_raw(data) + if file is not None: + data = file.read() + dev = SimulatedDeviceModel.parse_raw(data) + else: + cloud = MiotCloud() + # TODO: fix HACK + dev = SimulatedDeviceModel.parse_raw(cloud.get_model_schema(model)) + loop = asyncio.get_event_loop() random.seed(1) # nosec loop.run_until_complete(main(dev, model=model)) diff --git a/miio/devtools/simulators/models.py b/miio/devtools/simulators/models.py deleted file mode 100644 index f6928542a..000000000 --- a/miio/devtools/simulators/models.py +++ /dev/null @@ -1,106 +0,0 @@ -import logging -from typing import Any, List, Optional - -from pydantic import BaseModel, Field - -_LOGGER = logging.getLogger(__name__) - - -class MiotFormat(type): - """Custom type to convert textual presentation to python type.""" - - @classmethod - def __get_validators__(cls): - yield cls.convert_type - - @classmethod - def convert_type(cls, input: str): - if input.startswith("uint") or input.startswith("int"): - return int - type_map = { - "bool": bool, - "string": str, - "float": float, - } - return type_map[input] - - @classmethod - def serialize(cls, v): - return str(v) - - -class MiotEvent(BaseModel): - """Presentation of miot event.""" - - description: str - eiid: int = Field(alias="iid") - urn: str = Field(alias="type") - arguments: Any - - class Config: - extra = "forbid" - - -class MiotEnumValue(BaseModel): - """Enum value for miot.""" - - description: str - value: int - - class Config: - extra = "forbid" - - -class MiotAction(BaseModel): - """Action presentation for miot.""" - - description: str - aiid: int = Field(alias="iid") - urn: str = Field(alias="type") - inputs: Any = Field(alias="in") - output: Any = Field(alias="out") - - class Config: - extra = "forbid" - - -class MiotProperty(BaseModel): - """Property presentation for miot.""" - - description: str - piid: int = Field(alias="iid") - urn: str = Field(alias="type") - unit: str = Field(default="unknown") - format: MiotFormat - access: Any = Field(default=["read"]) - range: Optional[List[int]] = Field(alias="value-range") - choices: Optional[List[MiotEnumValue]] = Field(alias="value-list") - - class Config: - extra = "forbid" - - -class MiotService(BaseModel): - """Service presentation for miot.""" - - description: str - siid: int = Field(alias="iid") - urn: str = Field(alias="type") - properties: List[MiotProperty] = Field(default=[], repr=False) - events: Optional[List[MiotEvent]] = Field(default=[], repr=False) - actions: Optional[List[MiotAction]] = Field(default=[], repr=False) - - class Config: - extra = "forbid" - - -class DeviceModel(BaseModel): - """Device presentation for miot.""" - - description: str - urn: str = Field(alias="type") - services: List[MiotService] = Field(repr=False) - model: Optional[str] = None - - class Config: - extra = "forbid" diff --git a/miio/integrations/fan/zhimi/fan.py b/miio/integrations/fan/zhimi/fan.py index 8bdfedb02..8b4a6dec6 100644 --- a/miio/integrations/fan/zhimi/fan.py +++ b/miio/integrations/fan/zhimi/fan.py @@ -5,7 +5,7 @@ from miio import Device, DeviceStatus from miio.click_common import EnumType, command, format_output -from miio.devicestatus import sensor, setting, switch +from miio.devicestatus import sensor, setting from miio.fan_common import LedBrightness, MoveDirection _LOGGER = logging.getLogger(__name__) @@ -82,7 +82,7 @@ def power(self) -> str: return self.data["power"] @property - @switch("Power", setter_name="set_power") + @setting("Power", setter_name="set_power") def is_on(self) -> bool: """True if device is currently on.""" return self.power == "on" @@ -104,7 +104,7 @@ def temperature(self) -> Optional[float]: return None @property - @switch("LED", setter_name="set_led") + @setting("LED", setter_name="set_led") def led(self) -> Optional[bool]: """True if LED is turned on, if available.""" if "led" in self.data and self.data["led"] is not None: @@ -120,13 +120,13 @@ def led_brightness(self) -> Optional[LedBrightness]: return None @property - @switch("Buzzer", setter_name="set_buzzer") + @setting("Buzzer", setter_name="set_buzzer") def buzzer(self) -> bool: """True if buzzer is turned on.""" return self.data["buzzer"] in ["on", 1, 2] @property - @switch("Child Lock", setter_name="set_child_lock") + @setting("Child Lock", setter_name="set_child_lock") def child_lock(self) -> bool: """True if child lock is on.""" return self.data["child_lock"] == "on" @@ -148,7 +148,7 @@ def direct_speed(self) -> Optional[int]: return None @property - @switch("Oscillate", setter_name="set_oscillate") + @setting("Oscillate", setter_name="set_oscillate") def oscillate(self) -> bool: """True if oscillation is enabled.""" return self.data["angle_enable"] == "on" diff --git a/miio/integrations/humidifier/zhimi/airhumidifier.py b/miio/integrations/humidifier/zhimi/airhumidifier.py index 7a74f2ec2..3045e27a8 100644 --- a/miio/integrations/humidifier/zhimi/airhumidifier.py +++ b/miio/integrations/humidifier/zhimi/airhumidifier.py @@ -7,7 +7,7 @@ from miio import Device, DeviceError, DeviceInfo, DeviceStatus from miio.click_common import EnumType, command, format_output -from miio.devicestatus import sensor, setting, switch +from miio.devicestatus import sensor, setting _LOGGER = logging.getLogger(__name__) @@ -112,7 +112,7 @@ def humidity(self) -> int: return self.data["humidity"] @property - @switch( + @setting( name="Buzzer", icon="mdi:volume-high", setter_name="set_buzzer", @@ -138,7 +138,7 @@ def led_brightness(self) -> Optional[LedBrightness]: return None @property - @switch( + @setting( name="Child Lock", icon="mdi:lock", setter_name="set_child_lock", @@ -279,7 +279,7 @@ def water_tank_detached(self) -> Optional[bool]: return None @property - @switch( + @setting( name="Dry Mode", icon="mdi:hair-dryer", setter_name="set_dry", diff --git a/miio/integrations/light/philips/philips_bulb.py b/miio/integrations/light/philips/philips_bulb.py index cd6750323..3c675757f 100644 --- a/miio/integrations/light/philips/philips_bulb.py +++ b/miio/integrations/light/philips/philips_bulb.py @@ -11,6 +11,7 @@ MODEL_PHILIPS_LIGHT_BULB = "philips.light.bulb" MODEL_PHILIPS_LIGHT_HBULB = "philips.light.hbulb" +MODEL_PHILIPS_LIGHT_CBULB = "philips.light.cbulb" MODEL_PHILIPS_ZHIRUI_DOWNLIGHT = "philips.light.downlight" MODEL_PHILIPS_CANDLE = "philips.light.candle" MODEL_PHILIPS_CANDLE2 = "philips.light.candle2" @@ -21,6 +22,7 @@ AVAILABLE_PROPERTIES = { MODEL_PHILIPS_LIGHT_HBULB: AVAILABLE_PROPERTIES_COMMON + ["bri"], MODEL_PHILIPS_LIGHT_BULB: AVAILABLE_PROPERTIES_COLORTEMP, + MODEL_PHILIPS_LIGHT_CBULB: AVAILABLE_PROPERTIES_COLORTEMP, MODEL_PHILIPS_ZHIRUI_DOWNLIGHT: AVAILABLE_PROPERTIES_COLORTEMP, MODEL_PHILIPS_CANDLE: AVAILABLE_PROPERTIES_COLORTEMP, MODEL_PHILIPS_CANDLE2: AVAILABLE_PROPERTIES_COLORTEMP, diff --git a/miio/integrations/light/yeelight/spec_helper.py b/miio/integrations/light/yeelight/spec_helper.py index 339f3e682..aa1ac796c 100644 --- a/miio/integrations/light/yeelight/spec_helper.py +++ b/miio/integrations/light/yeelight/spec_helper.py @@ -1,11 +1,13 @@ import logging import os from enum import IntEnum -from typing import Dict, NamedTuple +from typing import Dict import attr import yaml +from miio import ColorTemperatureRange + _LOGGER = logging.getLogger(__name__) @@ -14,16 +16,9 @@ class YeelightSubLightType(IntEnum): Background = 1 -class ColorTempRange(NamedTuple): - """Color temperature range.""" - - min: int - max: int - - @attr.s(auto_attribs=True) class YeelightLampInfo: - color_temp: ColorTempRange + color_temp: ColorTemperatureRange supports_color: bool @@ -48,14 +43,14 @@ def _parse_specs_yaml(self): for key, value in models.items(): lamps = { YeelightSubLightType.Main: YeelightLampInfo( - ColorTempRange(*value["color_temp"]), + ColorTemperatureRange(*value["color_temp"]), value["supports_color"], ) } if "background" in value: lamps[YeelightSubLightType.Background] = YeelightLampInfo( - ColorTempRange(*value["background"]["color_temp"]), + ColorTemperatureRange(*value["background"]["color_temp"]), value["background"]["supports_color"], ) diff --git a/miio/integrations/light/yeelight/tests/test_yeelight_spec_helper.py b/miio/integrations/light/yeelight/tests/test_yeelight_spec_helper.py index 761cce93c..765fa3c6c 100644 --- a/miio/integrations/light/yeelight/tests/test_yeelight_spec_helper.py +++ b/miio/integrations/light/yeelight/tests/test_yeelight_spec_helper.py @@ -1,4 +1,8 @@ -from ..spec_helper import ColorTempRange, YeelightSpecHelper, YeelightSubLightType +from ..spec_helper import ( + ColorTemperatureRange, + YeelightSpecHelper, + YeelightSubLightType, +) def test_get_model_info(): @@ -6,9 +10,9 @@ def test_get_model_info(): model_info = spec_helper.get_model_info("yeelink.light.bslamp1") assert model_info.model == "yeelink.light.bslamp1" assert model_info.night_light is False - assert model_info.lamps[YeelightSubLightType.Main].color_temp == ColorTempRange( - 1700, 6500 - ) + assert model_info.lamps[ + YeelightSubLightType.Main + ].color_temp == ColorTemperatureRange(1700, 6500) assert model_info.lamps[YeelightSubLightType.Main].supports_color is True assert YeelightSubLightType.Background not in model_info.lamps @@ -18,8 +22,8 @@ def test_get_unknown_model_info(): model_info = spec_helper.get_model_info("notreal") assert model_info.model == "yeelink.light.*" assert model_info.night_light is False - assert model_info.lamps[YeelightSubLightType.Main].color_temp == ColorTempRange( - 1700, 6500 - ) + assert model_info.lamps[ + YeelightSubLightType.Main + ].color_temp == ColorTemperatureRange(1700, 6500) assert model_info.lamps[YeelightSubLightType.Main].supports_color is False assert YeelightSubLightType.Background not in model_info.lamps diff --git a/miio/integrations/light/yeelight/yeelight.py b/miio/integrations/light/yeelight/yeelight.py index 71bdeedaa..390cfb6f2 100644 --- a/miio/integrations/light/yeelight/yeelight.py +++ b/miio/integrations/light/yeelight/yeelight.py @@ -1,13 +1,18 @@ +import logging from enum import IntEnum from typing import List, Optional, Tuple import click +from miio import ColorTemperatureRange, LightInterface from miio.click_common import command, format_output from miio.device import Device, DeviceStatus +from miio.devicestatus import sensor, setting from miio.utils import int_to_rgb, rgb_to_int -from .spec_helper import ColorTempRange, YeelightSpecHelper, YeelightSubLightType +from .spec_helper import YeelightSpecHelper, YeelightSubLightType + +_LOGGER = logging.getLogger(__name__) SUBLIGHT_PROP_PREFIX = { YeelightSubLightType.Main: "", @@ -57,9 +62,12 @@ def rgb(self) -> Optional[Tuple[int, int, int]]: return None @property - def color_mode(self) -> YeelightMode: + def color_mode(self) -> Optional[YeelightMode]: """Return current color mode.""" - return YeelightMode(int(self.data[self.get_prop_name("color_mode")])) + try: + return YeelightMode(int(self.data[self.get_prop_name("color_mode")])) + except ValueError: # white only bulbs + return None @property def hsv(self) -> Optional[Tuple[int, int, int]]: @@ -105,46 +113,61 @@ def __init__(self, data): self.data = data @property + @setting("Power", setter_name="set_power") def is_on(self) -> bool: """Return whether the light is on or off.""" return self.lights[0].is_on @property + @setting("Brightness", unit="%", setter_name="set_brightness", max_value=100) def brightness(self) -> int: """Return current brightness.""" return self.lights[0].brightness @property + @sensor( + "RGB", setter_name="set_rgb" + ) # TODO: we need to extend @setting to support tuples to fix this def rgb(self) -> Optional[Tuple[int, int, int]]: """Return color in RGB if RGB mode is active.""" return self.lights[0].rgb @property - def color_mode(self) -> YeelightMode: + @sensor("Color mode") + def color_mode(self) -> Optional[YeelightMode]: """Return current color mode.""" return self.lights[0].color_mode @property + @sensor( + "HSV", setter_name="set_hsv" + ) # TODO: we need to extend @setting to support tuples to fix this def hsv(self) -> Optional[Tuple[int, int, int]]: """Return current color in HSV if HSV mode is active.""" return self.lights[0].hsv @property + @sensor( + "Color temperature", setter_name="set_color_temperature" + ) # TODO: we need to allow ranges by attribute to fix this def color_temp(self) -> Optional[int]: """Return current color temperature, if applicable.""" return self.lights[0].color_temp @property + @sensor("Color flow active") def color_flowing(self) -> bool: """Return whether the color flowing is active.""" return self.lights[0].color_flowing @property + @sensor("Color flow parameters") def color_flow_params(self) -> Optional[str]: """Return color flowing params.""" return self.lights[0].color_flow_params @property + @setting("Developer mode enabled", setter_name="set_developer_mode") def developer_mode(self) -> Optional[bool]: """Return whether the developer mode is active.""" lan_ctrl = self.data["lan_ctrl"] @@ -153,21 +176,25 @@ def developer_mode(self) -> Optional[bool]: return None @property + @setting("Save state on change enabled", setter_name="set_save_state_on_change") def save_state_on_change(self) -> bool: """Return whether the bulb state is saved on change.""" return bool(int(self.data["save_state"])) @property + @sensor("Device name") def name(self) -> str: """Return the internal name of the bulb.""" return self.data["name"] @property + @sensor("Delayed turn off in", unit="mins") def delay_off(self) -> int: """Return delay in minute before bulb is off.""" return int(self.data["delayoff"]) @property + @sensor("Music mode enabled") def music_mode(self) -> Optional[bool]: """Return whether the music mode is active.""" music_on = self.data["music_on"] @@ -176,6 +203,7 @@ def music_mode(self) -> Optional[bool]: return None @property + @sensor("Moon light mode active") def moonlight_mode(self) -> Optional[bool]: """Return whether the moonlight mode is active.""" active_mode = self.data["active_mode"] @@ -184,6 +212,7 @@ def moonlight_mode(self) -> Optional[bool]: return None @property + @sensor("Moon light mode brightness", unit="%") def moonlight_mode_brightness(self) -> Optional[int]: """Return current moonlight brightness.""" nl_br = self.data["nl_br"] @@ -218,7 +247,7 @@ def cli_format(self) -> str: s += f"{light.type.name} light\n" s += f" Power: {light.is_on}\n" s += f" Brightness: {light.brightness}\n" - s += f" Color mode: {light.color_mode.name}\n" + s += f" Color mode: {light.color_mode}\n" if light.color_mode == YeelightMode.RGB: s += f" RGB: {light.rgb}\n" elif light.color_mode == YeelightMode.HSV: @@ -236,7 +265,7 @@ def cli_format(self) -> str: return s -class Yeelight(Device): +class Yeelight(Device, LightInterface): """A rudimentary support for Yeelight bulbs. The API is the same as defined in @@ -307,7 +336,14 @@ def status(self) -> YeelightStatus: return YeelightStatus(dict(zip(properties, values))) @property - def valid_temperature_range(self) -> ColorTempRange: + def valid_temperature_range(self) -> ColorTemperatureRange: + """Return supported color temperature range.""" + _LOGGER.warning("Deprecated, use color_temperature_range instead") + return self._color_temp_range + + @property + def color_temperature_range(self) -> Optional[ColorTemperatureRange]: + """Return supported color temperature range.""" return self._color_temp_range @command( @@ -341,6 +377,13 @@ def off(self, transition=0): return self.send("set_power", ["off", "smooth", transition]) return self.send("set_power", ["off"]) + def set_power(self, on: bool, **kwargs): + """Set power on or off.""" + if on: + self.on(**kwargs) + else: + self.off(**kwargs) + @command( click.argument("level", type=int), click.option("--transition", type=int, required=False, default=0), @@ -360,6 +403,16 @@ def set_brightness(self, level, transition=0): default_output=format_output("Setting color temperature to {level}"), ) def set_color_temp(self, level, transition=500): + """Deprecated, use set_color_temperature instead.""" + _LOGGER.warning("Deprecated, use set_color_temperature instead.") + self.set_color_temperature(level, transition) + + @command( + click.argument("level", type=int), + click.option("--transition", type=int, required=False, default=0), + default_output=format_output("Setting color temperature to {level}"), + ) + def set_color_temperature(self, level, transition=500): """Set color temp in kelvin.""" if ( level > self.valid_temperature_range.max diff --git a/miio/integrations/vacuum/mijia/pro2vacuum.py b/miio/integrations/vacuum/mijia/pro2vacuum.py index 7ea511477..47b55689e 100644 --- a/miio/integrations/vacuum/mijia/pro2vacuum.py +++ b/miio/integrations/vacuum/mijia/pro2vacuum.py @@ -6,7 +6,7 @@ import click from miio.click_common import EnumType, command, format_output -from miio.devicestatus import sensor, switch +from miio.devicestatus import sensor, setting from miio.interfaces import FanspeedPresets, VacuumInterface from miio.miot_device import DeviceStatus, MiotDevice @@ -172,7 +172,7 @@ def state(self) -> DeviceState: return DeviceState(self.data["state"]) @property - @switch(name="Fan Speed", choices=FanSpeedMode, setter_name="set_fan_speed") + @setting(name="Fan Speed", choices=FanSpeedMode, setter_name="set_fan_speed") def fan_speed(self) -> FanSpeedMode: """Fan Speed.""" return FanSpeedMode(self.data["fan_speed"]) diff --git a/miio/integrations/vacuum/roborock/tests/test_vacuum.py b/miio/integrations/vacuum/roborock/tests/test_vacuum.py index af9408bd0..f88df3e4f 100644 --- a/miio/integrations/vacuum/roborock/tests/test_vacuum.py +++ b/miio/integrations/vacuum/roborock/tests/test_vacuum.py @@ -62,6 +62,15 @@ def __init__(self, *args, **kwargs): 1487548800, ], ] + self.dummies["dnd_timer"] = [ + { + "enabled": 1, + "start_minute": 0, + "end_minute": 0, + "start_hour": 22, + "end_hour": 8, + } + ] self.return_values = { "get_status": lambda x: [self.state], @@ -75,6 +84,8 @@ def __init__(self, *args, **kwargs): "app_zoned_clean": lambda x: self.change_mode("zoned clean"), "app_charge": lambda x: self.change_mode("charge"), "miIO.info": "dummy info", + "get_clean_record": lambda x: [[1488347071, 1488347123, 16, 0, 0, 0]], + "get_dnd_timer": lambda x: self.dummies["dnd_timer"], } super().__init__(args, kwargs) diff --git a/miio/integrations/vacuum/roborock/vacuum.py b/miio/integrations/vacuum/roborock/vacuum.py index e479bb745..6b4831408 100644 --- a/miio/integrations/vacuum/roborock/vacuum.py +++ b/miio/integrations/vacuum/roborock/vacuum.py @@ -1,6 +1,5 @@ import contextlib import datetime -import enum import json import logging import math @@ -21,9 +20,26 @@ command, ) from miio.device import Device, DeviceInfo +from miio.devicestatus import action from miio.exceptions import DeviceInfoUnavailableException, UnsupportedFeatureException from miio.interfaces import FanspeedPresets, VacuumInterface +from .vacuum_enums import ( + CarpetCleaningMode, + Consumable, + DustCollectionMode, + FanspeedE2, + FanspeedEnum, + FanspeedS7, + FanspeedS7_Maxv, + FanspeedV1, + FanspeedV2, + FanspeedV3, + MopIntensity, + MopMode, + TimerState, + WaterFlow, +) from .vacuumcontainers import ( CarpetModeStatus, CleaningDetails, @@ -39,112 +55,6 @@ _LOGGER = logging.getLogger(__name__) -class TimerState(enum.Enum): - On = "on" - Off = "off" - - -class Consumable(enum.Enum): - MainBrush = "main_brush_work_time" - SideBrush = "side_brush_work_time" - Filter = "filter_work_time" - SensorDirty = "sensor_dirty_time" - - -class FanspeedEnum(enum.Enum): - pass - - -class FanspeedV1(FanspeedEnum): - Silent = 38 - Standard = 60 - Medium = 77 - Turbo = 90 - - -class FanspeedV2(FanspeedEnum): - Silent = 101 - Standard = 102 - Medium = 103 - Turbo = 104 - Gentle = 105 - Auto = 106 - - -class FanspeedV3(FanspeedEnum): - Silent = 38 - Standard = 60 - Medium = 75 - Turbo = 100 - - -class FanspeedE2(FanspeedEnum): - # Original names from the app: Gentle, Silent, Standard, Strong, Max - Gentle = 41 - Silent = 50 - Standard = 68 - Medium = 79 - Turbo = 100 - - -class FanspeedS7(FanspeedEnum): - Silent = 101 - Standard = 102 - Medium = 103 - Turbo = 104 - - -class FanspeedS7_Maxv(FanspeedEnum): - Silent = 101 - Standard = 102 - Medium = 103 - Turbo = 104 - Max = 108 - - -class WaterFlow(enum.Enum): - """Water flow strength on s5 max.""" - - Minimum = 200 - Low = 201 - High = 202 - Maximum = 203 - - -class MopMode(enum.Enum): - """Mop routing on S7.""" - - Standard = 300 - Deep = 301 - - -class MopIntensity(enum.Enum): - """Mop scrub intensity on S7 + S7MAXV.""" - - Close = 200 - Mild = 201 - Moderate = 202 - Intense = 203 - - -class CarpetCleaningMode(enum.Enum): - """Type of carpet cleaning/avoidance.""" - - Avoid = 0 - Rise = 1 - Ignore = 2 - - -class DustCollectionMode(enum.Enum): - """Auto emptying mode (S7 + S7MAXV only)""" - - Smart = 0 - Quick = 1 - Daily = 2 - Strong = 3 - Max = 4 - - ROCKROBO_V1 = "rockrobo.vacuum.v1" ROCKROBO_S4 = "roborock.vacuum.s4" ROCKROBO_S4_MAX = "roborock.vacuum.a19" @@ -152,23 +62,30 @@ class DustCollectionMode(enum.Enum): ROCKROBO_S5_MAX = "roborock.vacuum.s5e" ROCKROBO_S6 = "roborock.vacuum.s6" ROCKROBO_T6 = "roborock.vacuum.t6" # cn s6 +ROCKROBO_E4 = "roborock.vacuum.a01" ROCKROBO_S6_PURE = "roborock.vacuum.a08" ROCKROBO_T7 = "roborock.vacuum.a11" # cn s7 ROCKROBO_T7S = "roborock.vacuum.a14" ROCKROBO_T7SPLUS = "roborock.vacuum.a23" ROCKROBO_S7_MAXV = "roborock.vacuum.a27" +ROCKROBO_S7_PRO_ULTRA = "roborock.vacuum.a62" ROCKROBO_Q5 = "roborock.vacuum.a34" +ROCKROBO_Q7_MAX = "roborock.vacuum.a38" ROCKROBO_G10S = "roborock.vacuum.a46" +ROCKROBO_G10 = "roborock.vacuum.a29" + ROCKROBO_S7 = "roborock.vacuum.a15" ROCKROBO_S6_MAXV = "roborock.vacuum.a10" ROCKROBO_E2 = "roborock.vacuum.e2" ROCKROBO_1S = "roborock.vacuum.m1s" ROCKROBO_C1 = "roborock.vacuum.c1" +ROCKROBO_WILD = "roborock.vacuum.*" # wildcard SUPPORTED_MODELS = [ ROCKROBO_V1, ROCKROBO_S4, ROCKROBO_S4_MAX, + ROCKROBO_E4, ROCKROBO_S5, ROCKROBO_S5_MAX, ROCKROBO_S6, @@ -179,12 +96,16 @@ class DustCollectionMode(enum.Enum): ROCKROBO_T7SPLUS, ROCKROBO_S7, ROCKROBO_S7_MAXV, + ROCKROBO_S7_PRO_ULTRA, ROCKROBO_Q5, + ROCKROBO_Q7_MAX, + ROCKROBO_G10, ROCKROBO_G10S, ROCKROBO_S6_MAXV, ROCKROBO_E2, ROCKROBO_1S, ROCKROBO_C1, + ROCKROBO_WILD, ] AUTO_EMPTY_MODELS = [ @@ -217,6 +138,7 @@ def start(self): return self.send("app_start") @command() + @action(name="Stop cleaning", type="vacuum") def stop(self): """Stop cleaning. @@ -226,16 +148,19 @@ def stop(self): return self.send("app_stop") @command() + @action(name="Spot cleaning", type="vacuum") def spot(self): """Start spot cleaning.""" return self.send("app_spot") @command() + @action(name="Pause cleaning", type="vacuum") def pause(self): """Pause cleaning.""" return self.send("app_pause") @command() + @action(name="Start cleaning", type="vacuum") def resume_or_start(self): """A shortcut for resuming or starting cleaning.""" status = self.status() @@ -288,6 +213,7 @@ def create_dummy_mac(addr): return self._info @command() + @action(name="Home", type="vacuum") def home(self): """Stop cleaning and return home.""" @@ -399,11 +325,17 @@ def manual_control( @command() def status(self) -> VacuumStatus: """Return status of the vacuum.""" - status = VacuumStatus(self.send("get_status")[0]) + status = self.vacuum_status() status.embed(self.consumable_status()) status.embed(self.clean_history()) + status.embed(self.dnd_status()) return status + @command() + def vacuum_status(self) -> VacuumStatus: + """Return only status of the vacuum.""" + return VacuumStatus(self.send("get_status")[0]) + def enable_log_upload(self): raise NotImplementedError("unknown parameters") # return self.send("enable_log_upload") @@ -544,6 +476,7 @@ def clean_details( return res @command() + @action(name="Find robot", type="vacuum") def find(self): """Find the robot.""" return self.send("find_me", [""]) @@ -721,6 +654,7 @@ def set_sound_volume(self, vol: int): return self.send("change_sound_volume", [vol]) @command() + @action(name="Test sound volume", type="vacuum") def test_sound_volume(self): """Test current sound volume.""" return self.send("test_sound_volume") @@ -855,12 +789,14 @@ def set_dust_collection_mode(self, mode: DustCollectionMode) -> bool: return self.send("set_dust_collection_mode", {"mode": mode.value})[0] == "ok" @command() + @action(name="Start dust collection", icon="mdi:turbine") def start_dust_collection(self): """Activate automatic dust collection.""" self._verify_auto_empty_support() return self.send("app_start_collect_dust") @command() + @action(name="Stop dust collection", icon="mdi:turbine") def stop_dust_collection(self): """Abort in progress dust collection.""" self._verify_auto_empty_support() @@ -953,7 +889,7 @@ def set_mop_mode(self, mop_mode: MopMode): @command() def mop_intensity(self) -> MopIntensity: """Get mop scrub intensity setting.""" - if self.model != ROCKROBO_S7: + if self.model not in [ROCKROBO_S7, ROCKROBO_S7_MAXV]: raise UnsupportedFeatureException( "Mop scrub intensity not supported by %s", self.model ) @@ -963,7 +899,7 @@ def mop_intensity(self) -> MopIntensity: @command(click.argument("mop_intensity", type=EnumType(MopIntensity))) def set_mop_intensity(self, mop_intensity: MopIntensity): """Set mop scrub intensity setting.""" - if self.model != ROCKROBO_S7: + if self.model not in [ROCKROBO_S7, ROCKROBO_S7_MAXV]: raise UnsupportedFeatureException( "Mop scrub intensity not supported by %s", self.model ) diff --git a/miio/integrations/vacuum/roborock/vacuum_enums.py b/miio/integrations/vacuum/roborock/vacuum_enums.py new file mode 100644 index 000000000..3cf0cab94 --- /dev/null +++ b/miio/integrations/vacuum/roborock/vacuum_enums.py @@ -0,0 +1,110 @@ +import enum + + +class TimerState(enum.Enum): + On = "on" + Off = "off" + + +class Consumable(enum.Enum): + MainBrush = "main_brush_work_time" + SideBrush = "side_brush_work_time" + Filter = "filter_work_time" + SensorDirty = "sensor_dirty_time" + + +class FanspeedEnum(enum.Enum): + pass + + +class FanspeedV1(FanspeedEnum): + Silent = 38 + Standard = 60 + Medium = 77 + Turbo = 90 + + +class FanspeedV2(FanspeedEnum): + Silent = 101 + Standard = 102 + Medium = 103 + Turbo = 104 + Gentle = 105 + Auto = 106 + + +class FanspeedV3(FanspeedEnum): + Silent = 38 + Standard = 60 + Medium = 75 + Turbo = 100 + + +class FanspeedE2(FanspeedEnum): + # Original names from the app: Gentle, Silent, Standard, Strong, Max + Gentle = 41 + Silent = 50 + Standard = 68 + Medium = 79 + Turbo = 100 + + +class FanspeedS7(FanspeedEnum): + Silent = 101 + Standard = 102 + Medium = 103 + Turbo = 104 + + +class FanspeedS7_Maxv(FanspeedEnum): + # Original names from the app: Quiet, Balanced, Turbo, Max, Max+ + Off = 105 + Silent = 101 + Standard = 102 + Medium = 103 + Turbo = 104 + Max = 108 + + +class WaterFlow(enum.Enum): + """Water flow strength on s5 max.""" + + Minimum = 200 + Low = 201 + High = 202 + Maximum = 203 + + +class MopMode(enum.Enum): + """Mop routing on S7 + S7MAXV.""" + + Standard = 300 + Deep = 301 + DeepPlus = 303 + + +class MopIntensity(enum.Enum): + """Mop scrub intensity on S7 + S7MAXV.""" + + Off = 200 + Mild = 201 + Moderate = 202 + Intense = 203 + + +class CarpetCleaningMode(enum.Enum): + """Type of carpet cleaning/avoidance.""" + + Avoid = 0 + Rise = 1 + Ignore = 2 + + +class DustCollectionMode(enum.Enum): + """Auto emptying mode (S7 + S7MAXV only)""" + + Smart = 0 + Quick = 1 + Daily = 2 + Strong = 3 + Max = 4 diff --git a/miio/integrations/vacuum/roborock/vacuumcontainers.py b/miio/integrations/vacuum/roborock/vacuumcontainers.py index 6e62e3fb0..a7d81a92b 100644 --- a/miio/integrations/vacuum/roborock/vacuumcontainers.py +++ b/miio/integrations/vacuum/roborock/vacuumcontainers.py @@ -10,6 +10,8 @@ from miio.interfaces.vacuuminterface import VacuumDeviceStatus, VacuumState from miio.utils import pretty_seconds, pretty_time +from .vacuum_enums import MopIntensity, MopMode + def pretty_area(x: float) -> float: return int(x) / 1000000 @@ -80,6 +82,15 @@ def pretty_area(x: float) -> float: 22: "Clean the dock charging contacts", 23: "Docking station not reachable", 24: "No-go zone or invisible wall detected", + 26: "Wall sensor is dirty", + 27: "VibraRise system is jammed", + 28: "Roborock is on carpet", +} + +dock_error_codes = { # from vacuum_cleaner-EN.pdf + 0: "No error", + 38: "Clean water tank empty", + 39: "Dirty water tank full", } @@ -129,13 +140,13 @@ def __init__(self, data: Dict[str, Any]) -> None: self.data = data @property - @sensor("State Code") + @sensor("State code", entity_category="diagnostic", enabled_default=False) def state_code(self) -> int: """State code as returned by the device.""" return int(self.data["state"]) @property - @sensor("State message") + @sensor("State", entity_category="diagnostic") def state(self) -> str: """Human readable state description, see also :func:`state_code`.""" return STATE_CODE_TO_STRING.get( @@ -148,13 +159,23 @@ def vacuum_state(self) -> VacuumState: return STATE_CODE_TO_VACUUMSTATE.get(self.state_code, VacuumState.Unknown) @property - @sensor("Error Code", icon="mdi:alert") + @sensor( + "Error code", + icon="mdi:alert", + entity_category="diagnostic", + enabled_default=False, + ) def error_code(self) -> int: """Error code as returned by the device.""" return int(self.data["error_code"]) @property - @sensor("Error", icon="mdi:alert") + @sensor( + "Error string", + icon="mdi:alert", + entity_category="diagnostic", + enabled_default=False, + ) def error(self) -> str: """Human readable error description, see also :func:`error_code`.""" try: @@ -163,7 +184,36 @@ def error(self) -> str: return "Definition missing for error %s" % self.error_code @property - @sensor("Battery", unit="%", device_class="battery") + @sensor( + "Dock error code", + icon="mdi:alert", + entity_category="diagnostic", + enabled_default=False, + ) + def dock_error_code(self) -> Optional[int]: + """Dock error status as returned by the device.""" + if "dock_error_status" in self.data: + return int(self.data["dock_error_status"]) + return None + + @property + @sensor( + "Dock error string", + icon="mdi:alert", + entity_category="diagnostic", + enabled_default=False, + ) + def dock_error(self) -> Optional[str]: + """Human readable dock error description, see also :func:`dock_error_code`.""" + if self.dock_error_code is None: + return None + try: + return dock_error_codes[self.dock_error_code] + except KeyError: + return "Definition missing for dock error %s" % self.dock_error_code + + @property + @sensor("Battery", unit="%", device_class="battery", enabled_default=False) def battery(self) -> int: """Remaining battery in percentage.""" return int(self.data["battery"]) @@ -178,18 +228,53 @@ def battery(self) -> int: step=1, icon="mdi:fan", ) - def fanspeed(self) -> int: + def fanspeed(self) -> Optional[int]: """Current fan speed.""" - return int(self.data["fan_power"]) + fan_power = int(self.data["fan_power"]) + if fan_power > 100: + # values 100+ are reserved for presets + return None + return fan_power + + @property + @setting( + "Mop scrub intensity", + choices=MopIntensity, + setter_name="set_mop_intensity", + icon="mdi:checkbox-multiple-blank-circle-outline", + ) + def mop_intensity(self) -> Optional[int]: + """Current mop intensity.""" + if "water_box_mode" in self.data: + return int(self.data["water_box_mode"]) + return None @property - @sensor("Clean Duration", unit="s", icon="mdi:timer-sand") + @setting( + "Mop route", + choices=MopMode, + setter_name="set_mop_mode", + icon="mdi:swap-horizontal-variant", + ) + def mop_route(self) -> Optional[int]: + """Current mop route.""" + if "mop_mode" in self.data: + return int(self.data["mop_mode"]) + return None + + @property + @sensor( + "Current clean duration", + unit="s", + icon="mdi:timer-sand", + device_class="duration", + ) def clean_time(self) -> timedelta: """Time used for cleaning (if finished, shows how long it took).""" return pretty_seconds(self.data["clean_time"]) @property - @sensor("Cleaned Area", unit="m2", icon="mdi:texture-box") + @sensor("Current clean area", unit="m²", icon="mdi:texture-box") def clean_area(self) -> float: """Cleaned area in m2.""" return pretty_area(self.data["clean_area"]) @@ -226,7 +311,7 @@ def is_on(self) -> bool: ) @property - @sensor("Water Box Attached") + @sensor("Water box attached", icon="mdi:cup-water") def is_water_box_attached(self) -> Optional[bool]: """Return True is water box is installed.""" if "water_box_status" in self.data: @@ -234,7 +319,7 @@ def is_water_box_attached(self) -> Optional[bool]: return None @property - @sensor("Mop Attached") + @sensor("Mop attached") def is_water_box_carriage_attached(self) -> Optional[bool]: """Return True if water box carriage (mop) is installed, None if sensor not present.""" @@ -243,7 +328,7 @@ def is_water_box_carriage_attached(self) -> Optional[bool]: return None @property - @sensor("Water Level Low", icon="mdi:alert") + @sensor("Water level low", icon="mdi:water-alert-outline") def is_water_shortage(self) -> Optional[bool]: """Returns True if water is low in the tank, None if sensor not present.""" if "water_shortage_status" in self.data: @@ -251,7 +336,23 @@ def is_water_shortage(self) -> Optional[bool]: return None @property - @sensor("Error", icon="mdi:alert") + @setting( + "Auto dust collection", + setter_name="set_dust_collection", + icon="mdi:turbine", + entity_category="config", + ) + def auto_dust_collection(self) -> Optional[bool]: + """Returns True if auto dust collection is enabled, None if sensor not + present.""" + if "auto_dust_collection" in self.data: + return self.data["auto_dust_collection"] == 1 + return None + + @property + @sensor( + "Error", icon="mdi:alert", entity_category="diagnostic", enabled_default=False + ) def got_error(self) -> bool: """True if an error has occurred.""" return self.error_code != 0 @@ -283,30 +384,52 @@ def __init__(self, data: Union[List[Any], Dict[str, Any]]) -> None: self.data["records"] = [] @property - @sensor("Total Cleaning Time", icon="mdi:timer-sand") + @sensor( + "Total clean duration", + unit="s", + icon="mdi:timer-sand", + device_class="duration", + entity_category="diagnostic", + ) def total_duration(self) -> timedelta: """Total cleaning duration.""" return pretty_seconds(self.data["clean_time"]) @property - @sensor("Total Cleaning Area", icon="mdi:texture-box") + @sensor( + "Total clean area", + unit="m²", + icon="mdi:texture-box", + entity_category="diagnostic", + ) def total_area(self) -> float: """Total cleaned area.""" return pretty_area(self.data["clean_area"]) @property - @sensor("Total Clean Count") + @sensor( + "Total clean count", + icon="mdi:counter", + state_class="total_increasing", + entity_category="diagnostic", + ) def count(self) -> int: """Number of cleaning runs.""" return int(self.data["clean_count"]) @property def ids(self) -> List[int]: - """A list of available cleaning IDs, see also :class:`CleaningDetails`.""" + """A list of available cleaning IDs, see also + :class:`CleaningDetails`.""" return list(self.data["records"]) @property - @sensor("Dust Collection Count") + @sensor( + "Total dust collection count", + icon="mdi:counter", + state_class="total_increasing", + entity_category="diagnostic", + ) def dust_collection_count(self) -> Optional[int]: """Total number of dust collections.""" if "dust_collection_count" in self.data: @@ -335,21 +458,46 @@ def __init__(self, data: Union[List[Any], Dict[str, Any]]) -> None: self.data = data @property + @sensor( + "Last clean start", + icon="mdi:clock-time-twelve", + device_class="timestamp", + entity_category="diagnostic", + ) def start(self) -> datetime: """When cleaning was started.""" return pretty_time(self.data["begin"]) @property + @sensor( + "Last clean end", + icon="mdi:clock-time-twelve", + device_class="timestamp", + entity_category="diagnostic", + ) def end(self) -> datetime: """When cleaning was finished.""" return pretty_time(self.data["end"]) @property + @sensor( + "Last clean duration", + unit="s", + icon="mdi:timer-sand", + device_class="duration", + entity_category="diagnostic", + ) def duration(self) -> timedelta: """Total duration of the cleaning run.""" return pretty_seconds(self.data["duration"]) @property + @sensor( + "Last clean area", + unit="m²", + icon="mdi:texture-box", + entity_category="diagnostic", + ) def area(self) -> float: """Total cleaned area.""" return pretty_area(self.data["area"]) @@ -397,46 +545,117 @@ def __init__(self, data: Dict[str, Any]) -> None: self.sensor_dirty_total = timedelta(hours=30) @property - @sensor("Main Brush Usage", unit="s") + @sensor( + "Main brush used", + unit="s", + icon="mdi:brush", + device_class="duration", + entity_category="diagnostic", + enabled_default=False, + ) def main_brush(self) -> timedelta: """Main brush usage time.""" return pretty_seconds(self.data["main_brush_work_time"]) @property - @sensor("Main Brush Remaining", unit="s") + @sensor( + "Main brush left", + unit="s", + icon="mdi:brush", + device_class="duration", + entity_category="diagnostic", + ) def main_brush_left(self) -> timedelta: """How long until the main brush should be changed.""" return self.main_brush_total - self.main_brush @property + @sensor( + "Side brush used", + unit="s", + icon="mdi:brush", + device_class="duration", + entity_category="diagnostic", + enabled_default=False, + ) def side_brush(self) -> timedelta: """Side brush usage time.""" return pretty_seconds(self.data["side_brush_work_time"]) @property + @sensor( + "Side brush left", + unit="s", + icon="mdi:brush", + device_class="duration", + entity_category="diagnostic", + ) def side_brush_left(self) -> timedelta: """How long until the side brush should be changed.""" return self.side_brush_total - self.side_brush @property + @sensor( + "Filter used", + unit="s", + icon="mdi:air-filter", + device_class="duration", + entity_category="diagnostic", + enabled_default=False, + ) def filter(self) -> timedelta: """Filter usage time.""" return pretty_seconds(self.data["filter_work_time"]) @property + @sensor( + "Filter left", + unit="s", + icon="mdi:air-filter", + device_class="duration", + entity_category="diagnostic", + ) def filter_left(self) -> timedelta: """How long until the filter should be changed.""" return self.filter_total - self.filter @property + @sensor( + "Sensor dirty used", + unit="s", + icon="mdi:eye-outline", + device_class="duration", + entity_category="diagnostic", + enabled_default=False, + ) def sensor_dirty(self) -> timedelta: """Return ``sensor_dirty_time``""" return pretty_seconds(self.data["sensor_dirty_time"]) @property + @sensor( + "Sensor dirty left", + unit="s", + icon="mdi:eye-outline", + device_class="duration", + entity_category="diagnostic", + ) def sensor_dirty_left(self) -> timedelta: return self.sensor_dirty_total - self.sensor_dirty + @property + @sensor( + "Dustbin times auto-empty used", + icon="mdi:delete", + entity_category="diagnostic", + enabled_default=False, + ) + def dustbin_auto_empty_used(self) -> Optional[int]: + """Return ``dust_collection_work_times``""" + if "dust_collection_work_times" in self.data: + return self.data["dust_collection_work_times"] + return None + class DNDStatus(DeviceStatus): """A container for the do-not-disturb status.""" @@ -447,17 +666,31 @@ def __init__(self, data: Dict[str, Any]): self.data = data @property - @sensor("Do Not Disturb") + @sensor("Do not disturb", icon="mdi:minus-circle-off", entity_category="diagnostic") def enabled(self) -> bool: """True if DnD is enabled.""" return bool(self.data["enabled"]) @property + @sensor( + "Do not disturb start", + icon="mdi:minus-circle-off", + device_class="timestamp", + entity_category="diagnostic", + enabled_default=False, + ) def start(self) -> time: """Start time of DnD.""" return time(hour=self.data["start_hour"], minute=self.data["start_minute"]) @property + @sensor( + "Do not disturb end", + icon="mdi:minus-circle-off", + device_class="timestamp", + entity_category="diagnostic", + enabled_default=False, + ) def end(self) -> time: """End time of DnD.""" return time(hour=self.data["end_hour"], minute=self.data["end_minute"]) @@ -616,7 +849,7 @@ def __init__(self, data): self.data = data @property - @sensor("Carpet Mode") + @sensor("Carpet mode") def enabled(self) -> bool: """True if carpet mode is enabled.""" return self.data["enable"] == 1 diff --git a/miio/interfaces/__init__.py b/miio/interfaces/__init__.py index df788c7f8..4f0247c41 100644 --- a/miio/interfaces/__init__.py +++ b/miio/interfaces/__init__.py @@ -1,5 +1,11 @@ """Interfaces API.""" +from .lightinterface import ColorTemperatureRange, LightInterface from .vacuuminterface import FanspeedPresets, VacuumInterface -__all__ = ["FanspeedPresets", "VacuumInterface"] +__all__ = [ + "FanspeedPresets", + "VacuumInterface", + "LightInterface", + "ColorTemperatureRange", +] diff --git a/miio/interfaces/lightinterface.py b/miio/interfaces/lightinterface.py new file mode 100644 index 000000000..316686918 --- /dev/null +++ b/miio/interfaces/lightinterface.py @@ -0,0 +1,37 @@ +"""`LightInterface` is an interface (abstract class) for light devices.""" +from abc import abstractmethod +from typing import NamedTuple, Optional, Tuple + + +class ColorTemperatureRange(NamedTuple): + """Color temperature range.""" + + min: int + max: int + + +class LightInterface: + """Light interface.""" + + @abstractmethod + def set_power(self, on: bool, **kwargs): + """Turn device on or off.""" + + @abstractmethod + def set_brightness(self, level: int, **kwargs): + """Set the light brightness [0,100].""" + + @property + def color_temperature_range(self) -> Optional[ColorTemperatureRange]: + """Return the color temperature range, if supported.""" + return None + + def set_color_temperature(self, level: int, **kwargs): + """Set color temperature in kelvin.""" + raise NotImplementedError( + "Called set_color_temperature on device that does not support it" + ) + + def set_rgb(self, rgb: Tuple[int, int, int], **kwargs): + """Set color in RGB.""" + raise NotImplementedError("Called set_rgb on device that does not support it") diff --git a/miio/miioprotocol.py b/miio/miioprotocol.py index 958a62423..90e4d21e5 100644 --- a/miio/miioprotocol.py +++ b/miio/miioprotocol.py @@ -8,6 +8,7 @@ import logging import socket from datetime import datetime, timedelta +from pprint import pformat as pf from typing import Any, Dict, List import construct @@ -172,7 +173,7 @@ def send( msg = {"data": {"value": request}, "header": {"value": header}, "checksum": 0} m = Message.build(msg, token=self.token) - _LOGGER.debug("%s:%s >>: %s", self.ip, self.port, request) + _LOGGER.debug("%s:%s >>: %s", self.ip, self.port, pf(request)) if self.debug > 1: _LOGGER.debug( "send (timeout %s): %s", @@ -208,7 +209,7 @@ def send( self.port, header["ts"], payload["id"], - payload, + pf(payload), ) if "error" in payload: self._handle_error(payload["error"]) diff --git a/miio/miot_cloud.py b/miio/miot_cloud.py new file mode 100644 index 000000000..410833b23 --- /dev/null +++ b/miio/miot_cloud.py @@ -0,0 +1,112 @@ +"""Module implementing handling of miot schema files.""" +import logging +from datetime import datetime, timedelta +from operator import attrgetter +from pathlib import Path +from typing import List + +import appdirs +import requests # TODO: externalize HTTP requests to avoid direct dependency +from pydantic import BaseModel + +from miio.miot_models import DeviceModel + +_LOGGER = logging.getLogger(__name__) + + +class ReleaseInfo(BaseModel): + model: str + status: str + type: str + version: int + + @property + def filename(self) -> str: + return f"{self.model}_{self.status}_{self.version}.json" + + +class ReleaseList(BaseModel): + instances: List[ReleaseInfo] + + def info_for_model(self, model: str, *, status_filter="released") -> ReleaseInfo: + matches = [inst for inst in self.instances if inst.model == model] + + if len(matches) > 1: + _LOGGER.warning( + "more than a single match for model %s: %s, filtering with status=%s", + model, + matches, + status_filter, + ) + + released_versions = [inst for inst in matches if inst.status == status_filter] + if not released_versions: + raise Exception(f"No releases for {model}, adjust status_filter?") + + _LOGGER.debug("Got %s releases, picking the newest one", released_versions) + + match = max(released_versions, key=attrgetter("version")) + _LOGGER.debug("Using %s", match) + + return match + + +class MiotCloud: + def __init__(self): + self._cache_dir = Path(appdirs.user_cache_dir("python-miio")) + + def get_device_model(self, model: str) -> DeviceModel: + """Get device model for model name.""" + file = self._cache_dir / f"{model}.json" + if file.exists(): + _LOGGER.debug("Using cached %s", file) + return DeviceModel.parse_raw(file.read_text()) + + return DeviceModel.parse_raw(self.get_model_schema(model)) + + def get_model_schema(self, model: str) -> str: + """Get the preferred schema for the model.""" + instances = self.fetch_release_list() + release_info = instances.info_for_model(model) + + model_file = self._cache_dir / f"{release_info.model}.json" + url = f"https://miot-spec.org/miot-spec-v2/instance?type={release_info.type}" + + data = self._fetch(url, model_file) + + return data + + def fetch_release_list(self): + """Fetch a list of available schemas.""" + mapping_file = "model-to-urn.json" + url = "http://miot-spec.org/miot-spec-v2/instances?status=all" + data = self._fetch(url, self._cache_dir / mapping_file) + + return ReleaseList.parse_raw(data) + + def _fetch(self, url: str, target_file: Path, cache_hours=6): + """Fetch the URL and cache results, if expired.""" + + def valid_cache(): + expiration = timedelta(hours=cache_hours) + if ( + datetime.fromtimestamp(target_file.stat().st_mtime) + expiration + > datetime.utcnow() + ): + return True + + return False + + if target_file.exists() and valid_cache(): + _LOGGER.debug("Returning data from cache: %s", target_file) + return target_file.read_text() + + _LOGGER.debug("Going to download %s to %s", url, target_file) + content = requests.get(url) + content.raise_for_status() + + response = content.text + written = target_file.write_text(response) + _LOGGER.debug("Written %s bytes to %s", written, target_file) + + return response diff --git a/miio/miot_device.py b/miio/miot_device.py index 1e4cbee50..0a471c612 100644 --- a/miio/miot_device.py +++ b/miio/miot_device.py @@ -150,13 +150,16 @@ def get_property_by(self, siid: int, piid: int): click.argument( "value_type", type=EnumType(MiotValueType), required=False, default=None ), + click.option("--name", required=False), ) def set_property_by( self, siid: int, piid: int, value: Union[int, float, str, bool], + *, value_type: Any = None, + name: str = None, ): """Set a single property (siid/piid) to given value. @@ -166,9 +169,12 @@ def set_property_by( if value_type is not None: value = value_type.value(value) + if name is None: + name = f"set-{siid}-{piid}" + return self.send( "set_properties", - [{"did": f"set-{siid}-{piid}", "siid": siid, "piid": piid, "value": value}], + [{"did": name, "siid": siid, "piid": piid, "value": value}], ) def set_property(self, property_key: str, value): diff --git a/miio/miot_models.py b/miio/miot_models.py new file mode 100644 index 000000000..509c6f888 --- /dev/null +++ b/miio/miot_models.py @@ -0,0 +1,280 @@ +import logging +from datetime import timedelta +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field, PrivateAttr, root_validator + +_LOGGER = logging.getLogger(__name__) + + +class URN(BaseModel): + """Parsed type URN.""" + + namespace: str + type: str + name: str + internal_id: str + model: str + version: int + + @classmethod + def __get_validators__(cls): + yield cls.validate + + @classmethod + def validate(cls, v): + if not isinstance(v, str) or ":" not in v: + raise TypeError("invalid type") + + _, namespace, type, name, id_, model, version = v.split(":") + + return cls( + namespace=namespace, + type=type, + name=name, + internal_id=id_, + model=model, + version=version, + ) + + def __repr__(self): + return f"urn:{self.namespace}:{self.type}:{self.name}:{self.internal_id}:{self.model}:{self.version}" + + +class MiotFormat(type): + """Custom type to convert textual presentation to python type.""" + + @classmethod + def __get_validators__(cls): + yield cls.convert_type + + @classmethod + def convert_type(cls, input: str): + if input.startswith("uint") or input.startswith("int"): + return int + type_map = { + "bool": bool, + "string": str, + "float": float, + } + return type_map[input] + + +class MiotEvent(BaseModel): + """Presentation of miot event.""" + + eiid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + arguments: Any + + service: Optional["MiotService"] = None # backref to containing service + + class Config: + extra = "forbid" + + +class MiotEnumValue(BaseModel): + """Enum value for miot.""" + + description: str + value: int + + @root_validator + def description_from_value(cls, values): + """If description is empty, use the value instead.""" + if not values["description"]: + values["description"] = str(values["value"]) + return values + + class Config: + extra = "forbid" + + +class MiotAction(BaseModel): + """Action presentation for miot.""" + + aiid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + inputs: Any = Field(alias="in") + outputs: Any = Field(alias="out") + + extras: Dict = Field(default_factory=dict, repr=False) + + service: Optional["MiotService"] = None # backref to containing service + + @property + def siid(self) -> Optional[int]: + """Return siid.""" + if self.service is not None: + return self.service.siid + + return None + + @property + def plain_name(self) -> str: + """Return plain name.""" + return self.urn.name + + @property + def name(self) -> str: + """Return combined name of the service and the action.""" + return f"{self.service.name}:{self.urn.name}" # type: ignore + + class Config: + extra = "forbid" + + +class MiotProperty(BaseModel): + """Property presentation for miot.""" + + piid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + format: MiotFormat + access: Any = Field(default=["read"]) + unit: Optional[str] = None + + range: Optional[List[int]] = Field(alias="value-range") + choices: Optional[List[MiotEnumValue]] = Field(alias="value-list") + + extras: Dict[str, Any] = Field(default_factory=dict, repr=False) + + service: Optional["MiotService"] = None # backref to containing service + + # TODO: currently just used to pass the data for miiocli + # there must be a better way to do this.. + value: Optional[Any] = None + + @property + def siid(self) -> Optional[int]: + """Return siid.""" + if self.service is not None: + return self.service.siid + + return None + + @property + def plain_name(self): + """Return plain name.""" + return self.urn.name + + @property + def name(self) -> str: + """Return combined name of the service and the property.""" + return f"{self.service.name}:{self.urn.name}" # type: ignore + + @property + def pretty_value(self): + value = self.value + + if self.choices is not None: + # TODO: find a nicer way to get the choice by value + selected = next(c.description for c in self.choices if c.value == value) + current = f"{selected} (value: {value})" + return current + + if self.format == bool: + return bool(value) + + unit_map = { + "none": "", + "percentage": "%", + "minutes": timedelta(minutes=1), + "hours": timedelta(hours=1), + "days": timedelta(days=1), + } + + unit = unit_map.get(self.unit) + if isinstance(unit, timedelta): + value = value * unit + else: + value = f"{value} {unit}" + + return value + + class Config: + extra = "forbid" + + +class MiotService(BaseModel): + """Service presentation for miot.""" + + siid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + properties: List[MiotProperty] = Field(default_factory=list, repr=False) + events: List[MiotEvent] = Field(default_factory=list, repr=False) + actions: List[MiotAction] = Field(default_factory=list, repr=False) + + def __init__(self, *args, **kwargs): + """Initialize a service. + + Overridden to propagate the siid to the children. + """ + super().__init__(*args, **kwargs) + + for prop in self.properties: + prop.service = self + for act in self.actions: + act.service = self + for ev in self.events: + ev.service = self + + @property + def name(self) -> str: + """Return service name.""" + return self.urn.name + + class Config: + extra = "forbid" + + +class DeviceModel(BaseModel): + """Device presentation for miot.""" + + description: str + urn: URN = Field(alias="type") + services: List[MiotService] = Field(repr=False) + + # internal mappings to simplify accesses to a specific (siid, piid) + _properties_by_id: Dict[int, Dict[int, MiotProperty]] = PrivateAttr( + default_factory=dict + ) + _properties_by_name: Dict[str, Dict[str, MiotProperty]] = PrivateAttr( + default_factory=dict + ) + + def __init__(self, *args, **kwargs): + """Presentation of a miot device model scehma. + + Overridden to implement internal (siid, piid) mapping. + """ + super().__init__(*args, **kwargs) + for serv in self.services: + self._properties_by_name[serv.name] = dict() + self._properties_by_id[serv.siid] = dict() + for prop in serv.properties: + self._properties_by_name[serv.name][prop.plain_name] = prop + self._properties_by_id[serv.siid][prop.piid] = prop + + @property + def device_type(self) -> str: + """Return device type as string.""" + return self.urn.type + + def get_property(self, service: str, prop_name: str) -> MiotProperty: + """Return the property model for given service and property name.""" + return self._properties_by_name[service][prop_name] + + def get_property_by_siid_piid(self, siid: int, piid: int) -> MiotProperty: + """Return the property model for given siid, piid.""" + return self._properties_by_id[siid][piid] + + class Config: + extra = "forbid" diff --git a/miio/powerstrip.py b/miio/powerstrip.py index 8235fbc4b..91337175f 100644 --- a/miio/powerstrip.py +++ b/miio/powerstrip.py @@ -7,7 +7,7 @@ from .click_common import EnumType, command, format_output from .device import Device -from .devicestatus import DeviceStatus, sensor, switch +from .devicestatus import DeviceStatus, sensor, setting from .utils import deprecated _LOGGER = logging.getLogger(__name__) @@ -61,7 +61,7 @@ def power(self) -> str: return self.data["power"] @property - @switch(name="Power", setter_name="set_power", device_class="outlet") + @setting(name="Power", setter_name="set_power", device_class="outlet") def is_on(self) -> bool: """True if the device is turned on.""" return self.power == "on" @@ -105,7 +105,7 @@ def wifi_led(self) -> Optional[bool]: return self.led @property - @switch( + @setting( name="LED", icon="mdi:led-outline", setter_name="set_led", device_class="switch" ) def led(self) -> Optional[bool]: diff --git a/miio/push_server/server.py b/miio/push_server/server.py index 9a21cef31..ddd906eab 100644 --- a/miio/push_server/server.py +++ b/miio/push_server/server.py @@ -17,6 +17,7 @@ FAKE_DEVICE_MODEL = "chuangmi.plug.v3" PushServerCallback = Callable[[str, str, str], None] +MethodDict = Dict[str, Union[Dict, Callable]] def calculated_token_enc(token): @@ -66,7 +67,7 @@ def __init__(self, device_ip=None): self._listen_couroutine = None self._registered_devices = {} - self._methods = {} + self._methods: MethodDict = {} self._event_id = 1000000 @@ -325,6 +326,6 @@ def server_model(self): return self._server_model @property - def methods(self): + def methods(self) -> MethodDict: """Return a dict of implemented methods.""" return self._methods diff --git a/miio/push_server/serverprotocol.py b/miio/push_server/serverprotocol.py index 2bc43597d..6e2459e08 100644 --- a/miio/push_server/serverprotocol.py +++ b/miio/push_server/serverprotocol.py @@ -11,6 +11,10 @@ "21310020ffffffffffffffffffffffffffffffffffffffffffffffffffffffff" ) +ERR_INVALID = -1 +ERR_UNSUPPORTED = -2 +ERR_METHOD_EXEC_FAILED = -3 + class ServerProtocol: """Handle responding to UDP packets.""" @@ -73,11 +77,11 @@ def send_response(self, host, port, msg_id, token, payload=None): if payload is None: payload = {} - result = {**payload, "id": msg_id} - msg = self._create_message(result, token, device_id=self.server.server_id) + data = {**payload, "id": msg_id} + msg = self._create_message(data, token, device_id=self.server.server_id) self.transport.sendto(msg, (host, port)) - _LOGGER.debug(">> %s:%s: %s", host, port, result) + _LOGGER.debug(">> %s:%s: %s", host, port, data) def send_error(self, host, port, msg_id, token, code, message): """Send error message with given code and message to the client.""" @@ -121,19 +125,36 @@ def _handle_datagram_from_client(self, host: str, port: int, data): msg_value, ) + if "method" not in msg_value: + return self.send_error( + host, port, msg_id, token, ERR_INVALID, "missing method" + ) + methods = self.server.methods if msg_value["method"] not in methods: - return self.send_error(host, port, msg_id, token, -1, "unsupported method") + return self.send_error( + host, port, msg_id, token, ERR_UNSUPPORTED, "unsupported method" + ) + _LOGGER.debug("Got method call: %s", msg_value["method"]) method = methods[msg_value["method"]] if callable(method): try: response = method(msg_value) except Exception as ex: - return self.send_error(host, port, msg_id, token, -1, str(ex)) + _LOGGER.exception(ex) + return self.send_error( + host, + port, + msg_id, + token, + ERR_METHOD_EXEC_FAILED, + f"Exception {type(ex)}: {ex}", + ) else: response = method + _LOGGER.debug("Responding %s with %s", msg_id, response) return self.send_response(host, port, msg_id, token, payload=response) def datagram_received(self, data, addr): diff --git a/miio/push_server/test_serverprotocol.py b/miio/push_server/test_serverprotocol.py index 37ce3bd63..42fa18132 100644 --- a/miio/push_server/test_serverprotocol.py +++ b/miio/push_server/test_serverprotocol.py @@ -2,7 +2,12 @@ from miio import Message -from .serverprotocol import ServerProtocol +from .serverprotocol import ( + ERR_INVALID, + ERR_METHOD_EXEC_FAILED, + ERR_UNSUPPORTED, + ServerProtocol, +) HOST = "127.0.0.1" PORT = 1234 @@ -108,15 +113,44 @@ def test_datagram_with_known_method(protocol: ServerProtocol, mocker): assert cargs["payload"] == response_payload -def test_datagram_with_unknown_method(protocol: ServerProtocol, mocker): - """Test that regular client messages are handled properly.""" +@pytest.mark.parametrize( + "method,err_code", [("unknown_method", ERR_UNSUPPORTED), (None, ERR_INVALID)] +) +def test_datagram_with_unknown_method( + method, err_code, protocol: ServerProtocol, mocker +): + """Test that invalid payloads are erroring out correctly.""" protocol.send_error = mocker.Mock() # type: ignore[assignment] protocol.server.methods = {} - msg = protocol._create_message({"id": 1, "method": "miIO.info"}, DUMMY_TOKEN, 1234) + data = {"id": 1} + + if method is not None: + data["method"] = method + + msg = protocol._create_message(data, DUMMY_TOKEN, 1234) + protocol._handle_datagram_from_client(HOST, PORT, msg) + + protocol.send_error.assert_called() # type: ignore + cargs = protocol.send_error.call_args[0] # type: ignore + assert cargs[4] == err_code + + +def test_datagram_with_exception_raising(protocol: ServerProtocol, mocker): + """Test that exception raising callbacks are .""" + protocol.send_error = mocker.Mock() # type: ignore[assignment] + + def _raise(*args, **kwargs): + raise Exception("error message") + + protocol.server.methods = {"raise": _raise} + + data = {"id": 1, "method": "raise"} + + msg = protocol._create_message(data, DUMMY_TOKEN, 1234) protocol._handle_datagram_from_client(HOST, PORT, msg) protocol.send_error.assert_called() # type: ignore cargs = protocol.send_error.call_args[0] # type: ignore - assert cargs[4] == -1 - assert cargs[5] == "unsupported method" + assert cargs[4] == ERR_METHOD_EXEC_FAILED + assert "error message" in cargs[5] diff --git a/miio/tests/test_devicestatus.py b/miio/tests/test_devicestatus.py index cbb7049c2..67f785e72 100644 --- a/miio/tests/test_devicestatus.py +++ b/miio/tests/test_devicestatus.py @@ -4,7 +4,7 @@ from miio import Device, DeviceStatus from miio.descriptors import EnumSettingDescriptor, NumberSettingDescriptor -from miio.devicestatus import sensor, setting, switch +from miio.devicestatus import sensor, setting def test_multiple(): @@ -102,29 +102,6 @@ def unknown(self): assert "unknown_kwarg" in sensors["unknown"].extras -def test_switch_decorator(mocker): - class DecoratedSwitches(DeviceStatus): - @property - @switch(name="Power", setter_name="set_power") - def power(self): - pass - - mocker.patch("miio.Device.send") - d = Device("127.0.0.1", "68ffffffffffffffffffffffffffffff") - - # Patch status to return our class - mocker.patch.object(d, "status", return_value=DecoratedSwitches()) - # Patch to create a new setter as defined in the status class - set_power = mocker.patch.object(d, "set_power", create=True, return_value=1) - - sensors = d.switches() - assert len(sensors) == 1 - assert sensors["power"].name == "Power" - - sensors["power"].setter(True) - set_power.assert_called_with(True) - - def test_setting_decorator_number(mocker): """Tests for setting decorator with numbers.""" @@ -224,7 +201,7 @@ def sub_sensor(self): assert len(sensors) == 2 assert getattr(main, sensors["main_sensor"].property) == "main" - assert getattr(main, sensors["SubStatus:sub_sensor"].property) == "sub" + assert getattr(main, sensors["SubStatus__sub_sensor"].property) == "sub" with pytest.raises(KeyError): main.sensors()["nonexisting_sensor"] diff --git a/miio/tests/test_miot_models.py b/miio/tests/test_miot_models.py new file mode 100644 index 000000000..dcca795cc --- /dev/null +++ b/miio/tests/test_miot_models.py @@ -0,0 +1,155 @@ +"""Tests for miot model parsing.""" + +import pytest +from pydantic import BaseModel + +from miio.miot_models import ( + URN, + MiotAction, + MiotEnumValue, + MiotEvent, + MiotFormat, + MiotProperty, + MiotService, +) + + +def test_enum(): + """Test that enum parsing works.""" + data = """ + { + "value": 1, + "description": "dummy" + }""" + en = MiotEnumValue.parse_raw(data) + assert en.value == 1 + assert en.description == "dummy" + + +def test_enum_missing_description(): + """Test that missing description gets replaced by the value.""" + data = '{"value": 1, "description": ""}' + en = MiotEnumValue.parse_raw(data) + assert en.value == 1 + assert en.description == "1" + + +TYPES_FOR_FORMAT = [ + ("bool", bool), + ("string", str), + ("float", float), + ("uint8", int), + ("uint16", int), + ("uint32", int), + ("int8", int), + ("int16", int), + ("int32", int), +] + + +@pytest.mark.parametrize("format,expected_type", TYPES_FOR_FORMAT) +def test_format(format, expected_type): + class Wrapper(BaseModel): + """Need to wrap as plain string is not valid json.""" + + format: MiotFormat + + data = f'{{"format": "{format}"}}' + f = Wrapper.parse_raw(data) + assert f.format == expected_type + + +def test_action(): + """Test the public properties of action.""" + simple_action = """ + { + "iid": 1, + "type": "urn:miot-spec-v2:action:dummy-action:0000001:dummy:1", + "description": "Description", + "in": [], + "out": [] + }""" + act = MiotAction.parse_raw(simple_action) + assert act.aiid == 1 + assert act.urn.type == "action" + assert act.description == "Description" + assert act.inputs == [] + assert act.outputs == [] + + assert act.plain_name == "dummy-action" + + +def test_urn(): + """Test the parsing of URN strings.""" + urn_string = "urn:namespace:type:name:41414141:dummy.model:1" + example_urn = f'{{"urn": "{urn_string}"}}' + + class Wrapper(BaseModel): + """Need to wrap as plain string is not valid json.""" + + urn: URN + + wrapper = Wrapper.parse_raw(example_urn) + urn = wrapper.urn + assert urn.namespace == "namespace" + assert urn.type == "type" + assert urn.name == "name" + assert urn.internal_id == "41414141" + assert urn.model == "dummy.model" + assert urn.version == 1 + + # Check that the serialization works, too + assert repr(urn) == urn_string + + +def test_service(): + data = """ + { + "iid": 1, + "description": "test service", + "type": "urn:miot-spec-v2:service:device-information:00000001:dummy:1" + } + """ + serv = MiotService.parse_raw(data) + assert serv.siid == 1 + assert serv.urn.type == "service" + assert serv.actions == [] + assert serv.properties == [] + assert serv.events == [] + + +def test_event(): + data = '{"iid": 1, "type": "urn:spect:event:example_event:00000001:dummymodel:1", "description": "dummy", "arguments": []}' + ev = MiotEvent.parse_raw(data) + assert ev.eiid == 1 + assert ev.urn.type == "event" + assert ev.description == "dummy" + assert ev.arguments == [] + + +def test_property(): + data = """ + { + "iid": 1, + "type": "urn:miot-spec-v2:property:manufacturer:00000001:dummy:1", + "description": "Device Manufacturer", + "format": "string", + "access": [ + "read" + ] + } + """ + prop: MiotProperty = MiotProperty.parse_raw(data) + assert prop.piid == 1 + assert prop.urn.type == "property" + assert prop.format == str + assert prop.access == ["read"] + assert prop.description == "Device Manufacturer" + + assert prop.plain_name == "manufacturer" + + +@pytest.mark.xfail(reason="not implemented") +def test_property_pretty_value(): + """Test the pretty value conversions.""" + raise NotImplementedError() diff --git a/miio/tests/test_miotdevice.py b/miio/tests/test_miotdevice.py index d2bab5ba6..bca77505e 100644 --- a/miio/tests/test_miotdevice.py +++ b/miio/tests/test_miotdevice.py @@ -61,7 +61,7 @@ def test_get_property_by(dev): def test_set_property_by(dev, value_type, value): siid = 1 piid = 1 - _ = dev.set_property_by(siid, piid, value, value_type) + _ = dev.set_property_by(siid, piid, value, value_type=value_type) if value_type is not None: value = value_type.value(value) @@ -72,6 +72,18 @@ def test_set_property_by(dev, value_type, value): ) +def test_set_property_by_name(dev): + siid = 1 + piid = 1 + value = 1 + _ = dev.set_property_by(siid, piid, value, name="test-name") + + dev.send.assert_called_with( + "set_properties", + [{"did": "test-name", "siid": siid, "piid": piid, "value": value}], + ) + + def test_call_action_by(dev): siid = 1 aiid = 1