diff --git a/README.rst b/README.rst index c9a3a8dda..0db12969c 100644 --- a/README.rst +++ b/README.rst @@ -9,12 +9,13 @@ This library (and its accompanying cli tool) can be used to interface with devic Getting started --------------- -If you already have a token for your device and the device type, you can directly start using `miiocli` tool. -If you don't have a token for your device, refer to `Getting started `__ section of `the manual `__ for instructions how to obtain it. +The ``miiocli`` command allows controlling supported devices from the command line, +given that you know their IP addresses and tokens. +You can use ``miiocli cloud`` command to obtain this information from the cloud. +Refer to `Getting started `__ section of `the manual `__ for more detailed instructions. -The `miiocli` is the main way to execute commands from command line. -You can always use `--help` to get more information about the available commands. -For example, executing it without any extra arguments will print out options and available commands:: +You can always use ``--help`` to get more information about available commands, subcommands, and their options. +For example, to print out options and available commands:: $ miiocli --help Usage: miiocli [OPTIONS] COMMAND [ARGS]... @@ -28,7 +29,7 @@ For example, executing it without any extra arguments will print out options and airconditioningcompanion .. -You can get some information from any miIO/MIoT device, including its device model, using the `info` command:: +You can get some information from any miIO/MIoT device, including its device model, using the ``info`` command:: miiocli device --ip --token info @@ -38,8 +39,28 @@ You can get some information from any miIO/MIoT device, including its device mod Network: {'localIp': '', 'mask': '255.255.255.0', 'gw': ''} AP: {'rssi': -73, 'ssid': '', 'primary': 11, 'bssid': ''} -Different devices are supported by their corresponding modules (e.g., `roborockvacuum` or `fan`). -You can get the list of available commands for any given module by passing `--help` argument to it:: + +Controlling MIoT devices +^^^^^^^^^^^^^^^^^^^^^^^^ + +MiOT devices are supported by the ``genericmiot`` integration which provides basic support for all MiOT devices. +Internally, it downloads ``miot-spec`` files to find out about supported features. +All features of supported devices are available using these common commands:: + +- ``miiocli genericmiot status`` to print the device status information, including settings (prefixed with ``[S]``). +- ``miiocli genericmiot set`` to change settings. +- ``miiocli genericmiot actions`` to list available actions. +- ``miiocli genericmiot call`` to execute actions. + +Use ``miiocli genericmiot --help`` for more available commands. + + +Controlling other devices +^^^^^^^^^^^^^^^^^^^^^^^^^ + +Older devices are mainly supported by their corresponding modules (e.g., ``roborockvacuum`` or ``fan``). + +You can get the list of available commands for any given module by passing ``--help`` argument to it:: $ miiocli roborockvacuum --help @@ -63,25 +84,30 @@ You can avoid this by specifying the model manually:: API usage --------- -All functionality is accessible through the `miio` module:: +All functionalities of this library are accessible through the ``miio`` module. +While you can initialize individual integration classes manually, +the simplest way to obtain a device instance is to use ``DeviceFactory``:: - from miio import RoborockVacuum + from miio import DeviceFactory - vac = RoborockVacuum("", "") - vac.start() + dev = DeviceFactory.create("", "") + dev.info() -Each separate device type inherits from `miio.Device` -(and in case of MIoT devices, `miio.MiotDevice`) which provides a common API. -Each command invocation will automatically detect (and cache) the device model necessary for some actions -by querying the device. -You can avoid this by specifying the model manually:: +This will perform an ``info`` query to the device to detect its model information, +which is crucial especially for MiOT devices. - from miio import RoborockVacuum +Introspecting supported features +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ - vac = RoborockVacuum("", "", model="roborock.vacuum.s5") +You can introspect device classes using the following methods:: -Please refer to `API documentation `__ for more information. +- ``actions()`` to return information about available device actions. +- ``settings()`` to obtain information about available settings that can be changed. +- ``sensors()`` to obtain information about sensors. + +Each of these return `device descriptor objects `__, +which contain the necessary metadata about the available features to allow constructing generic interfaces. Troubleshooting @@ -90,6 +116,7 @@ You can find some solutions for the most common problems can be found in `Troubl If you have any questions, or simply want to join up for a chat, check `our Matrix room `__. + Contributing ------------ @@ -100,11 +127,14 @@ To ease the process of setting up a development environment we have prepared `a Supported devices ----------------- +While all MIoT devices are supported through the ``genericmiot`` integration, +this library supports also the following devices:: + - Xiaomi Mi Robot Vacuum V1, S4, S4 MAX, S5, S5 MAX, S6 Pure, M1S, S7 - Xiaomi Mi Home Air Conditioner Companion - Xiaomi Mi Smart Air Conditioner A (xiaomi.aircondition.mc1, mc2, mc4, mc5) - Xiaomi Mi Air Purifier 2, 3H, 3C, 4, Pro, Pro H, 4 Pro (zhimi.airpurifier.m2, mb3, mb4, mb5, v7, vb2, va2), 4 Lite -- Xiaomi Mi Air (Purifier) Dog X3, X5, X7SM (airdog.airpurifier.x3, airdog.airpurifier.x5, airdog.airpurifier.x7sm) +- Xiaomi Mi Air (Purifier) Dog X3, X5, X7SM (airdog.airpurifier.x3, x5, x7sm) - Xiaomi Mi Air Humidifier - Smartmi Air Purifier - Xiaomi Aqara Camera @@ -139,9 +169,8 @@ Supported devices - Xiaomi Smart WiFi Speaker - Xiaomi Mi WiFi Repeater 2 - Xiaomi Mi Smart Rice Cooker -- Xiaomi Smartmi Fresh Air System VA2 (zhimi.airfresh.va2), VA4 (zhimi.airfresh.va4), - A1 (dmaker.airfresh.a1), T2017 (dmaker.airfresh.t2017) -- Yeelight lights (basic support, we recommend using `python-yeelight `__) +- Xiaomi Smartmi Fresh Air System VA2 (zhimi.airfresh.va2), VA4 (va4), T2017 (t2017), A1 (dmaker.airfresh.a1) +- Yeelight lights (see also `python-yeelight `__) - Xiaomi Mi Air Dehumidifier - Xiaomi Tinymu Smart Toilet Cover - Xiaomi 16 Relays Module diff --git a/miio/__init__.py b/miio/__init__.py index 79b474eb8..8789776d8 100644 --- a/miio/__init__.py +++ b/miio/__init__.py @@ -47,6 +47,7 @@ AirPurifierMiot, ) from miio.integrations.fan import Fan, Fan1C, FanLeshow, FanMiot, FanP5, FanZA5 +from miio.integrations.genericmiot import GenericMiot from miio.integrations.humidifier import ( AirHumidifier, AirHumidifierJsq, diff --git a/miio/device.py b/miio/device.py index caa1287ea..3d05a477b 100644 --- a/miio/device.py +++ b/miio/device.py @@ -145,7 +145,8 @@ def _fetch_info(self) -> DeviceInfo: self._info = devinfo _LOGGER.debug("Detected model %s", devinfo.model) cls = self.__class__.__name__ - bases = ["Device", "MiotDevice"] + # Ignore bases and generic classes + bases = ["Device", "MiotDevice", "GenericMiot"] if devinfo.model not in self.supported_models and cls not in bases: _LOGGER.warning( "Found an unsupported model '%s' for class '%s'. If this is working for you, please open an issue at https://github.com/rytilahti/python-miio/", diff --git a/miio/devicefactory.py b/miio/devicefactory.py index e99f5de68..055cdab92 100644 --- a/miio/devicefactory.py +++ b/miio/devicefactory.py @@ -33,11 +33,12 @@ def register(cls, integration_cls: Type[Device]): for model in integration_cls.supported_models: # type: ignore if model in cls._supported_models: _LOGGER.debug( - "Got duplicate of %s for %s, previously registered by %s", + "Ignoring duplicate of %s for %s, previously registered by %s", model, integration_cls, cls._supported_models[model], ) + continue _LOGGER.debug(" * %s => %s", model, integration_cls) cls._supported_models[model] = integration_cls @@ -62,7 +63,11 @@ def class_for_model(cls, model: str): wildcard_models = { m: impl for m, impl in cls._supported_models.items() if m.endswith("*") } - for wildcard_model, impl in wildcard_models.items(): + # We sort here to return the implementation with most specific prefix + sorted_by_longest_prefix = sorted( + wildcard_models.items(), key=lambda item: len(item[0]), reverse=True + ) + for wildcard_model, impl in sorted_by_longest_prefix: m = wildcard_model.rstrip("*") if model.startswith(m): _LOGGER.debug( @@ -76,13 +81,27 @@ def class_for_model(cls, model: str): raise DeviceException("No implementation found for model %s" % model) @classmethod - def create(self, host: str, token: str, model: Optional[str] = None) -> Device: + def create( + self, + host: str, + token: str, + model: Optional[str] = None, + *, + force_generic_miot=False, + ) -> Device: """Return instance for the given host and token, with optional model override. The optional model parameter can be used to override the model detection. """ + dev: Device + if force_generic_miot: # TODO: find a better way to handle this. + from .integrations.genericmiot import GenericMiot + + dev = GenericMiot(host, token, model=model) + dev.info() + return dev if model is None: - dev: Device = Device(host, token) + dev = Device(host, token) info = dev.info() model = info.model diff --git a/miio/integrations/genericmiot/__init__.py b/miio/integrations/genericmiot/__init__.py new file mode 100644 index 000000000..59f29d119 --- /dev/null +++ b/miio/integrations/genericmiot/__init__.py @@ -0,0 +1,3 @@ +from .genericmiot import GenericMiot + +__all__ = ["GenericMiot"] diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py new file mode 100644 index 000000000..b2b42217b --- /dev/null +++ b/miio/integrations/genericmiot/genericmiot.py @@ -0,0 +1,445 @@ +import logging +from enum import Enum +from functools import partial +from typing import Dict, List, Optional, Union + +import click + +from miio import DeviceInfo, DeviceStatus, MiotDevice +from miio.click_common import LiteralParamType, command, format_output +from miio.descriptors import ( + ActionDescriptor, + BooleanSettingDescriptor, + EnumSettingDescriptor, + NumberSettingDescriptor, + SensorDescriptor, + SettingDescriptor, +) +from miio.miot_cloud import MiotCloud +from miio.miot_device import MiotMapping +from miio.miot_models import DeviceModel, MiotAction, MiotProperty, MiotService + +_LOGGER = logging.getLogger(__name__) + + +def pretty_status(result: "GenericMiotStatus"): + """Pretty print status information.""" + out = "" + props = result.property_dict() + for _name, prop in props.items(): + pretty_value = prop.pretty_value + + if "write" in prop.access: + out += "[S] " + + out += f"{prop.description} ({prop.name}): {pretty_value}" + + if prop.choices is not None: # TODO: hide behind verbose flag? + out += ( + " (from: " + + ", ".join([f"{c.description} ({c.value})" for c in prop.choices]) + + ")" + ) + + if prop.range is not None: # TODO: hide behind verbose flag? + out += ( + f" (min: {prop.range[0]}, max: {prop.range[1]}, step: {prop.range[2]})" + ) + + out += "\n" + + return out + + +def pretty_actions(result: Dict[str, ActionDescriptor]): + """Pretty print actions.""" + out = "" + for _, desc in result.items(): + out += f"{desc.id}\t\t{desc.name}\n" + + return out + + +def pretty_settings(result: Dict[str, SettingDescriptor]): + """Pretty print settings.""" + out = "" + for _, desc in result.items(): + out += f"# {desc.id} ({desc.name})" + out += f' urn: {repr(desc.extras["urn"])}\n' + out += f' siid: {desc.extras["siid"]}\n' + out += f' piid: {desc.extras["piid"]}\n' + + return out + + +class GenericMiotStatus(DeviceStatus): + """Generic status for miot devices.""" + + def __init__(self, response, dev): + self._model = dev._miot_model + self._dev = dev + self._data = {elem["did"]: elem["value"] for elem in response} + + def __getattr__(self, item): + """Return attribute for name. + + This is overridden to provide access to properties using (siid, piid) tuple. + """ + # TODO: find a better way to encode the property information + serv, prop = item.split(":") + prop = self._model.get_property(serv, prop) + value = self._data[item] + + # TODO: this feels like a wrong place to convert value to enum.. + if prop.choices is not None: + for choice in prop.choices: + if choice.value == value: + return choice.description + + _LOGGER.warning( + "Unable to find choice for value: %s: %s", value, prop.choices + ) + + return self._data[item] + + def property_dict(self) -> Dict[str, MiotProperty]: + """Return (siid, piid)-keyed dictionary of properties.""" + res = {} + for did, value in self._data.items(): + service, prop_name = did.split(":") + prop = self._model.get_property(service, prop_name) + prop.value = value + res[did] = prop + + return res + + def __repr__(self): + s = f"<{self.__class__.__name__}" + for name, value in self.property_dict().items(): + s += f" {name}={value}" + s += ">" + + return s + + +class GenericMiot(MiotDevice): + _supported_models = [ + "*" + ] # we support all devices, if not, it is a responsibility of caller to verify that + + def __init__( + self, + ip: str = None, + token: str = None, + start_id: int = 0, + debug: int = 0, + lazy_discover: bool = True, + timeout: int = None, + *, + model: str = None, + mapping: MiotMapping = None, + ): + super().__init__( + ip, + token, + start_id, + debug, + lazy_discover, + timeout, + model=model, + mapping=mapping, + ) + self._model = model + self._miot_model: Optional[DeviceModel] = None + + self._actions: Dict[str, ActionDescriptor] = {} + self._sensors: Dict[str, SensorDescriptor] = {} + self._settings: Dict[str, SettingDescriptor] = {} + self._properties: List[MiotProperty] = [] + + def initialize_model(self): + """Initialize the miot model and create descriptions.""" + if self._miot_model is not None: + return + + miotcloud = MiotCloud() + self._miot_model = miotcloud.get_device_model(self.model) + _LOGGER.debug("Initialized: %s", self._miot_model) + self._create_descriptors() + + @command(default_output=format_output(result_msg_fmt=pretty_status)) + def status(self) -> GenericMiotStatus: + """Return status based on the miot model.""" + properties = [] + for prop in self._properties: + if "read" not in prop.access: + _LOGGER.debug("Property has no read access, skipping: %s", prop) + continue + + siid = prop.siid + piid = prop.piid + name = prop.name # f"{prop.service.urn.name}:{prop.name}" + q = {"siid": siid, "piid": piid, "did": name} + properties.append(q) + + # TODO: max properties needs to be made configurable (or at least splitted to avoid too large udp datagrams + # some devices are stricter: https://github.com/rytilahti/python-miio/issues/1550#issuecomment-1303046286 + response = self.get_properties( + properties, property_getter="get_properties", max_properties=10 + ) + + return GenericMiotStatus(response, self) + + def _create_action(self, act: MiotAction) -> Optional[ActionDescriptor]: + """Create action descriptor for miot action.""" + if act.inputs: + # TODO: need to figure out how to expose input parameters for downstreams + _LOGGER.warning( + "Got inputs for action, skipping as handling is unknown: %s", act + ) + return None + + call_action = partial(self.call_action_by, act.siid, act.aiid) + + id_ = act.name + + # TODO: move extras handling to the model + extras = act.extras + extras["urn"] = act.urn + extras["siid"] = act.siid + extras["aiid"] = act.aiid + + return ActionDescriptor( + id=id_, + name=act.description, + method=call_action, + extras=extras, + ) + + def _create_actions(self, serv: MiotService): + """Create action descriptors.""" + for act in serv.actions: + act_desc = self._create_action(act) + if act_desc is None: # skip actions we cannot handle for now.. + continue + + if ( + act_desc.name in self._actions + ): # TODO: find a way to handle duplicates, suffix maybe? + _LOGGER.warning("Got used name name, ignoring '%s': %s", act.name, act) + continue + + self._actions[act_desc.name] = act_desc + + def _create_sensor(self, prop: MiotProperty) -> SensorDescriptor: + """Create sensor descriptor for a property.""" + property_name = prop.name + + return SensorDescriptor( + id=property_name, + name=prop.description, + property=property_name, + type=prop.format, + extras=prop.extras, + ) + + def _create_sensors_and_settings(self, serv: MiotService): + """Create sensor and setting descriptors for a service.""" + for prop in serv.properties: + if prop.access == ["notify"]: + _LOGGER.debug("Skipping notify-only property: %s", prop) + continue + if "read" not in prop.access: # TODO: handle write-only properties + _LOGGER.warning("Skipping write-only: %s", prop) + continue + + desc = self._descriptor_for_property(prop) + if isinstance(desc, SensorDescriptor): + self._sensors[prop.name] = desc + elif isinstance(desc, SettingDescriptor): + self._settings[prop.name] = desc + else: + raise Exception("unknown descriptor type") + + self._properties.append(prop) + + def _descriptor_for_property(self, prop: MiotProperty): + """Create a descriptor based on the property information.""" + desc: SettingDescriptor + name = prop.description + property_name = prop.name + + setter = partial(self.set_property_by, prop.siid, prop.piid, name=property_name) + + # TODO: move extras handling to the model + extras = prop.extras + extras["urn"] = prop.urn + extras["siid"] = prop.siid + extras["piid"] = prop.piid + + # Handle settable ranged properties + if prop.range is not None: + return self._create_range_setting(name, prop, property_name, setter, extras) + + # Handle settable enums + elif prop.choices is not None: + # TODO: handle two-value enums as booleans? + return self._create_choices_setting( + name, prop, property_name, setter, extras + ) + + # Handle settable booleans + elif "write" in prop.access and prop.format == bool: + return BooleanSettingDescriptor( + id=property_name, + name=name, + property=property_name, + setter=setter, + unit=prop.unit, + extras=extras, + ) + + # Fallback to sensors + return self._create_sensor(prop) + + def _create_choices_setting( + self, name, prop, property_name, setter, extras + ) -> Union[SensorDescriptor, EnumSettingDescriptor]: + """Create a descriptor for enum-based setting.""" + try: + choices = Enum( + prop.description, {c.description: c.value for c in prop.choices} + ) + _LOGGER.debug("Created enum %s", choices) + except ValueError as ex: + _LOGGER.error("Unable to create enum for %s: %s", prop, ex) + raise + + desc = EnumSettingDescriptor( + id=property_name, + name=name, + property=property_name, + unit=prop.unit, + choices=choices, + extras=extras, + ) + if "write" in prop.access: + desc.setter = setter + return desc + else: + return self._create_sensor(prop) + + def _create_range_setting(self, name, prop, property_name, setter, extras): + """Create a descriptor for range-based setting.""" + desc = NumberSettingDescriptor( + id=property_name, + name=name, + property=property_name, + min_value=prop.range[0], + max_value=prop.range[1], + step=prop.range[2], + unit=prop.unit, + extras=extras, + ) + if "write" in prop.access: + desc.setter = setter + return desc + else: + return self._create_sensor(prop) + + def _create_descriptors(self): + """Create descriptors based on the miot model.""" + for serv in self._miot_model.services: + if serv.siid == 1: + continue # Skip device details + + self._create_actions(serv) + self._create_sensors_and_settings(serv) + + _LOGGER.debug("Created %s actions", len(self._actions)) + for act in self._actions.values(): + _LOGGER.debug(f"\t{act}") + _LOGGER.debug("Created %s sensors", len(self._sensors)) + for sensor in self._sensors.values(): + _LOGGER.debug(f"\t{sensor}") + _LOGGER.debug("Created %s settings", len(self._settings)) + for setting in self._settings.values(): + _LOGGER.debug(f"\t{setting}") + + def _get_action_by_name(self, name: str): + """Return action by name.""" + # TODO: cache service:action? + for act in self._actions.values(): + if act.id == name: + if act.method_name is not None: + act.method = getattr(self, act.method_name) + + return act + + raise ValueError("No action with name/id %s" % name) + + @command( + click.argument("name"), + click.argument("params", type=LiteralParamType(), required=False), + name="call", + ) + def call_action(self, name: str, params=None): + """Call action by name.""" + params = params or [] + act = self._get_action_by_name(name) + return act.method(params) + + @command( + click.argument("name"), + click.argument("params", type=LiteralParamType(), required=True), + name="set", + ) + def change_setting(self, name: str, params=None): + """Change setting value.""" + params = params if params is not None else [] + # TODO: create a name/plain name getter to the device model + service, prop_name = name.split(":") + # prop = self._miot_model.get_property(service, prop) + setting = self._settings.get(name, None) + if setting is None: + raise ValueError("No setting found for name %s" % name) + + return setting.setter(value=setting.cast_value(params)) + + def _fetch_info(self) -> DeviceInfo: + """Hook to perform the model initialization.""" + info = super()._fetch_info() + self.initialize_model() + + return info + + @command(default_output=format_output(result_msg_fmt=pretty_actions)) + def actions(self) -> Dict[str, ActionDescriptor]: + """Return available actions.""" + return self._actions + + @command() + def sensors(self) -> Dict[str, SensorDescriptor]: + """Return available sensors.""" + return self._sensors + + @command(default_output=format_output(result_msg_fmt=pretty_settings)) + def settings(self) -> Dict[str, SettingDescriptor]: + """Return available settings.""" + return self._settings + + @property + def device_type(self) -> Optional[str]: + """Return device type.""" + # TODO: this should be probably mapped to an enum + if self._miot_model is not None: + return self._miot_model.urn.type + return None + + @classmethod + def get_device_group(cls): + """Return device command group. + + TODO: insert the actions from the model for better click integration + """ + return super().get_device_group() diff --git a/miio/miot_device.py b/miio/miot_device.py index 0a471c612..92c431f83 100644 --- a/miio/miot_device.py +++ b/miio/miot_device.py @@ -77,9 +77,6 @@ def __init__( ip, token, start_id, debug, lazy_discover, timeout, model=model ) - if mapping is None and not hasattr(self, "mapping") and not self._mappings: - _LOGGER.warning("Neither the class nor the parameter defines the mapping") - if mapping is not None: self.mapping = mapping diff --git a/miio/miot_models.py b/miio/miot_models.py index 509c6f888..6490334e1 100644 --- a/miio/miot_models.py +++ b/miio/miot_models.py @@ -17,6 +17,8 @@ class URN(BaseModel): model: str version: int + parent_urn: Optional["URN"] = Field(None, repr=False) + @classmethod def __get_validators__(cls): yield cls.validate @@ -37,9 +39,14 @@ def validate(cls, v): version=version, ) - def __repr__(self): + @property + def urn_string(self) -> str: + """Return string presentation of the URN.""" return f"urn:{self.namespace}:{self.type}:{self.name}:{self.internal_id}:{self.model}:{self.version}" + def __repr__(self): + return f"" + class MiotFormat(type): """Custom type to convert textual presentation to python type.""" @@ -60,21 +67,6 @@ def convert_type(cls, input: str): 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.""" @@ -92,20 +84,21 @@ class Config: extra = "forbid" -class MiotAction(BaseModel): - """Action presentation for miot.""" +class MiotBaseModel(BaseModel): + """Base model for all other miot models.""" - 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 + def fill_from_parent(self, service: "MiotService"): + """Fill some information from the parent service.""" + # TODO: this could be done using a validator + self.service = service + self.urn.parent_urn = service.urn + @property def siid(self) -> Optional[int]: """Return siid.""" @@ -122,18 +115,33 @@ def plain_name(self) -> str: @property def name(self) -> str: """Return combined name of the service and the action.""" - return f"{self.service.name}:{self.urn.name}" # type: ignore + if self.service is not None and self.urn.name is not None: + return f"{self.service.name}:{self.urn.name}" # type: ignore + return "unitialized" + + +class MiotAction(MiotBaseModel): + """Action presentation for miot.""" + + aiid: int = Field(alias="iid") + + inputs: Any = Field(alias="in") + outputs: Any = Field(alias="out") + + def fill_from_parent(self, service: "MiotService"): + """Overridden to convert inputs and outputs to property references.""" + super().fill_from_parent(service) + self.inputs = [service.get_property_by_id(piid) for piid in self.inputs] + self.outputs = [service.get_property_by_id(piid) for piid in self.outputs] class Config: extra = "forbid" -class MiotProperty(BaseModel): +class MiotProperty(MiotBaseModel): """Property presentation for miot.""" piid: int = Field(alias="iid") - urn: URN = Field(alias="type") - description: str format: MiotFormat access: Any = Field(default=["read"]) @@ -142,32 +150,10 @@ class MiotProperty(BaseModel): 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 @@ -201,6 +187,16 @@ class Config: extra = "forbid" +class MiotEvent(MiotBaseModel): + """Presentation of miot event.""" + + eiid: int = Field(alias="iid") + arguments: Any + + class Config: + extra = "forbid" + + class MiotService(BaseModel): """Service presentation for miot.""" @@ -212,19 +208,32 @@ class MiotService(BaseModel): events: List[MiotEvent] = Field(default_factory=list, repr=False) actions: List[MiotAction] = Field(default_factory=list, repr=False) + _property_by_id: Dict[int, MiotProperty] = PrivateAttr(default_factory=dict) + _action_by_id: Dict[int, MiotAction] = PrivateAttr(default_factory=dict) + def __init__(self, *args, **kwargs): """Initialize a service. - Overridden to propagate the siid to the children. + Overridden to propagate the service to the children. """ super().__init__(*args, **kwargs) for prop in self.properties: - prop.service = self + self._property_by_id[prop.piid] = prop + prop.fill_from_parent(self) for act in self.actions: - act.service = self + self._action_by_id[act.aiid] = act + act.fill_from_parent(self) for ev in self.events: - ev.service = self + ev.fill_from_parent(self) + + def get_property_by_id(self, piid): + """Return property by id.""" + return self._property_by_id[piid] + + def get_action_by_id(self, aiid): + """Return action by id.""" + return self._action_by_id[aiid] @property def name(self) -> str: diff --git a/miio/tests/test_devicefactory.py b/miio/tests/test_devicefactory.py index dd9a5a9e0..75b754f38 100644 --- a/miio/tests/test_devicefactory.py +++ b/miio/tests/test_devicefactory.py @@ -1,6 +1,6 @@ import pytest -from miio import Device, DeviceException, DeviceFactory, Gateway, MiotDevice +from miio import Device, DeviceFactory, DeviceInfo, Gateway, GenericMiot, MiotDevice DEVICE_CLASSES = Device.__subclasses__() + MiotDevice.__subclasses__() # type: ignore DEVICE_CLASSES.remove(MiotDevice) @@ -37,6 +37,44 @@ class _DummyDevice(Device): def test_device_class_for_model_unknown(): - """Test that unknown model raises an exception.""" - with pytest.raises(DeviceException): - DeviceFactory.class_for_model("foo.foo.xyz") + """Test that unknown model returns genericmiot.""" + assert DeviceFactory.class_for_model("foo.foo.xyz.invalid") == GenericMiot + + +@pytest.mark.parametrize("cls", DEVICE_CLASSES) +@pytest.mark.parametrize("force_model", [True, False]) +def test_create(cls, force_model, mocker): + """Test create for both forced and autodetected models.""" + mocker.patch("miio.Device.send") + + model = None + first_supported_model = next(iter(cls.supported_models)) + if force_model: + model = first_supported_model + + dummy_info = DeviceInfo({"model": first_supported_model}) + info = mocker.patch("miio.Device.info", return_value=dummy_info) + + device = DeviceFactory.create("127.0.0.1", 32 * "0", model=model) + device_class = DeviceFactory.class_for_model(device.model) + assert isinstance(device, device_class) + + if force_model: + info.assert_not_called() + else: + info.assert_called() + + +@pytest.mark.parametrize("cls", DEVICE_CLASSES) +def test_create_force_miot(cls, mocker): + """Test that force_generic_miot works.""" + mocker.patch("miio.Device.send") + mocker.patch("miio.Device.info") + class_for_model = mocker.patch("miio.DeviceFactory.class_for_model") + + assert isinstance( + DeviceFactory.create("127.0.0.1", 32 * "0", force_generic_miot=True), + GenericMiot, + ) + + class_for_model.assert_not_called() diff --git a/miio/tests/test_miot_models.py b/miio/tests/test_miot_models.py index dcca795cc..f87f86f4c 100644 --- a/miio/tests/test_miot_models.py +++ b/miio/tests/test_miot_models.py @@ -13,6 +13,42 @@ MiotService, ) +DUMMY_SERVICE = """ + { + "iid": 1, + "description": "test service", + "type": "urn:miot-spec-v2:service:device-information:00000001:dummy:1", + "properties": [ + { + "iid": 4, + "type": "urn:miot-spec-v2:property:firmware-revision:00000005:dummy:1", + "description": "Current Firmware Version", + "format": "string", + "access": [ + "read" + ] + } + ], + "actions": [ + { + "iid": 1, + "type": "urn:miot-spec-v2:action:start-sweep:00000004:dummy:1", + "description": "Start Sweep", + "in": [], + "out": [] + } + ], + "events": [ + { + "iid": 1, + "type": "urn:miot-spec-v2:event:low-battery:00000003:dummy:1", + "description": "Low Battery", + "arguments": [] + } + ] + } +""" + def test_enum(): """Test that enum parsing works.""" @@ -98,8 +134,9 @@ class Wrapper(BaseModel): assert urn.model == "dummy.model" assert urn.version == 1 - # Check that the serialization works, too - assert repr(urn) == urn_string + # Check that the serialization works + assert urn.urn_string == urn_string + assert repr(urn) == f"" def test_service(): @@ -118,6 +155,32 @@ def test_service(): assert serv.events == [] +@pytest.mark.parametrize("entity_type", ["actions", "properties", "events"]) +def test_service_back_references(entity_type): + """Check that backrefs are created correctly for properties, actions, and events.""" + serv = MiotService.parse_raw(DUMMY_SERVICE) + assert serv.siid == 1 + assert serv.urn.type == "service" + + entities = getattr(serv, entity_type) + assert len(entities) == 1 + entity_to_test = entities[0] + + assert entity_to_test.service.siid == serv.siid + + +@pytest.mark.parametrize("entity_type", ["actions", "properties", "events"]) +def test_entity_names(entity_type): + """Check that entity name consists of service name and entity's plain name.""" + serv = MiotService.parse_raw(DUMMY_SERVICE) + + entities = getattr(serv, entity_type) + assert len(entities) == 1 + entity_to_test = entities[0] + + assert entity_to_test.name == f"{serv.name}:{entity_to_test.plain_name}" + + def test_event(): data = '{"iid": 1, "type": "urn:spect:event:example_event:00000001:dummymodel:1", "description": "dummy", "arguments": []}' ev = MiotEvent.parse_raw(data) diff --git a/miio/tests/test_miotdevice.py b/miio/tests/test_miotdevice.py index bca77505e..3f8fdb720 100644 --- a/miio/tests/test_miotdevice.py +++ b/miio/tests/test_miotdevice.py @@ -3,6 +3,7 @@ import pytest from miio import Huizuo, MiotDevice +from miio.integrations.genericmiot import GenericMiot from miio.miot_device import MiotValueType, _filter_request_fields MIOT_DEVICES = MiotDevice.__subclasses__() @@ -17,18 +18,11 @@ def dev(module_mocker): device = MiotDevice( "127.0.0.1", "68ffffffffffffffffffffffffffffff", mapping=DUMMY_MAPPING ) - device._model = "testmodel" + device._model = "test.model" module_mocker.patch.object(device, "send") return device -def test_missing_mapping(caplog): - """Make sure ctor raises exception if neither class nor parameter defines the - mapping.""" - _ = MiotDevice("127.0.0.1", "68ffffffffffffffffffffffffffffff") - assert "Neither the class nor the parameter defines the mapping" in caplog.text - - def test_ctor_mapping(): """Make sure the constructor accepts the mapping parameter.""" test_mapping = {} @@ -115,13 +109,13 @@ def test_call_action_by(dev): @pytest.mark.parametrize( "model,expected_mapping,expected_log", [ - ("some_model", {"x": {"y": 1}}, ""), - ("unknown_model", {"x": {"y": 1}}, "Unable to find mapping"), + ("some.model", {"x": {"y": 1}}, ""), + ("unknown.model", {"x": {"y": 1}}, "Unable to find mapping"), ], ) def test_get_mapping(dev, caplog, model, expected_mapping, expected_log): """Test _get_mapping logic for fallbacks.""" - dev._mappings["some_model"] = {"x": {"y": 1}} + dev._mappings["some.model"] = {"x": {"y": 1}} dev._model = model assert dev._get_mapping() == expected_mapping @@ -145,6 +139,9 @@ def test_mapping_deprecation(cls): @pytest.mark.parametrize("cls", MIOT_DEVICES) def test_mapping_structure(cls): """Check that mappings are structured correctly.""" + if cls == GenericMiot: + pytest.skip("Skipping genericmiot as it provides no mapping") + assert cls._mappings model, contents = next(iter(cls._mappings.items())) @@ -163,13 +160,15 @@ def test_mapping_structure(cls): @pytest.mark.parametrize("cls", MIOT_DEVICES) def test_supported_models(cls): assert cls.supported_models == list(cls._mappings.keys()) + if cls == GenericMiot: + pytest.skip("Skipping genericmiot as it uses supported_models for now") # make sure that that _supported_models is not defined assert not cls._supported_models def test_call_action(dev): - dev._mappings["testmodel"] = {"test_action": {"siid": 1, "aiid": 1}} + dev._mappings["test.model"] = {"test_action": {"siid": 1, "aiid": 1}} dev.call_action("test_action") @@ -188,7 +187,7 @@ def test_call_action(dev): def test_get_properties_for_mapping_readables(mocker, dev, props, included_in_request): base_props = {"readable_property": {"siid": 1, "piid": 1}} base_request = [{"did": k, **v} for k, v in base_props.items()] - dev._mappings["testmodel"] = mapping = { + dev._mappings["test.model"] = mapping = { **base_props, "property_under_test": {"siid": 1, "piid": 2, **props}, }