-
-
Notifications
You must be signed in to change notification settings - Fork 567
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Split genericmiot into parts (#1725)
Just some janitoring to avoid piling everything into a single module. This will allow nicer test structuring when genericmiot gets some.
- Loading branch information
Showing
3 changed files
with
182 additions
and
162 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| from typing import Dict, cast | ||
|
|
||
| from miio.descriptors import ActionDescriptor, SettingDescriptor | ||
| from miio.miot_models import MiotProperty, MiotService | ||
|
|
||
| # TODO: these should be moved to a generic implementation covering all actions and settings | ||
|
|
||
|
|
||
| def pretty_actions(result: Dict[str, ActionDescriptor]): | ||
| """Pretty print actions.""" | ||
| out = "" | ||
| service = None | ||
| for _, desc in result.items(): | ||
| miot_prop: MiotProperty = desc.extras["miot_action"] | ||
| # service is marked as optional due pydantic backrefs.. | ||
| serv = cast(MiotService, miot_prop.service) | ||
| if service is None or service.siid != serv.siid: | ||
| service = serv | ||
| out += f"[bold]{service.description} ({service.name})[/bold]\n" | ||
|
|
||
| out += f"\t{desc.id}\t\t{desc.name}" | ||
| if desc.inputs: | ||
| for idx, input_ in enumerate(desc.inputs, start=1): | ||
| param = input_.extras[ | ||
| "miot_property" | ||
| ] # TODO: hack until descriptors get support for descriptions | ||
| param_desc = f"\n\t\tParameter #{idx}: {param.name} ({param.description}) ({param.format}) {param.pretty_input_constraints}" | ||
| out += param_desc | ||
|
|
||
| out += "\n" | ||
|
|
||
| return out | ||
|
|
||
|
|
||
| def pretty_settings(result: Dict[str, SettingDescriptor]): | ||
| """Pretty print settings.""" | ||
| out = "" | ||
| verbose = False | ||
| service = None | ||
| for _, desc in result.items(): | ||
| miot_prop: MiotProperty = desc.extras["miot_property"] | ||
| # service is marked as optional due pydantic backrefs.. | ||
| serv = cast(MiotService, miot_prop.service) | ||
| if service is None or service.siid != serv.siid: | ||
| service = serv | ||
| out += f"[bold]{service.name}[/bold] ({service.description})\n" | ||
|
|
||
| out += f"\t{desc.name} ({desc.id}, access: {miot_prop.pretty_access})\n" | ||
| if verbose: | ||
| out += f' urn: {repr(desc.extras["urn"])}\n' | ||
| out += f' siid: {desc.extras["siid"]}\n' | ||
| out += f' piid: {desc.extras["piid"]}\n' | ||
|
|
||
| return out |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| import logging | ||
| from typing import TYPE_CHECKING, Dict, Iterable | ||
|
|
||
| from miio import DeviceStatus | ||
| from miio.miot_models import DeviceModel, MiotAccess, MiotProperty | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| if TYPE_CHECKING: | ||
| from .genericmiot import GenericMiot | ||
|
|
||
|
|
||
| class GenericMiotStatus(DeviceStatus): | ||
| """Generic status for miot devices.""" | ||
|
|
||
| def __init__(self, response, dev): | ||
| self._model: DeviceModel = dev._miot_model | ||
| self._dev = dev | ||
| self._data = {elem["did"]: elem["value"] for elem in response} | ||
| # for hardcoded json output.. see click_common.json_output | ||
| self.data = self._data | ||
|
|
||
| self._data_by_siid_piid = { | ||
| (elem["siid"], elem["piid"]): elem["value"] for elem in response | ||
| } | ||
| self._data_by_normalized_name = { | ||
| self._normalize_name(elem["did"]): elem["value"] for elem in response | ||
| } | ||
|
|
||
| def _normalize_name(self, id_: str) -> str: | ||
| """Return a cleaned id for dict searches.""" | ||
| return id_.replace(":", "_").replace("-", "_") | ||
|
|
||
| def __getattr__(self, item): | ||
| """Return attribute for name. | ||
| This is overridden to provide access to properties using (siid, piid) tuple. | ||
| """ | ||
| # let devicestatus handle dunder methods | ||
| if item.startswith("__") and item.endswith("__"): | ||
| return super().__getattr__(item) | ||
|
|
||
| normalized_name = self._normalize_name(item) | ||
| if normalized_name in self._data_by_normalized_name: | ||
| return self._data_by_normalized_name[normalized_name] | ||
|
|
||
| # TODO: create a helper method and prohibit using non-normalized names | ||
| if ":" in item: | ||
| _LOGGER.warning("Use normalized names for accessing properties") | ||
| 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] | ||
|
|
||
| @property | ||
| def device(self) -> "GenericMiot": | ||
| """Return the device which returned this status.""" | ||
| return self._dev | ||
|
|
||
| def property_dict(self) -> Dict[str, MiotProperty]: | ||
| """Return name-keyed dictionary of properties.""" | ||
| res = {} | ||
|
|
||
| # We use (siid, piid) to locate the property as not all devices mirror the did in response | ||
| for (siid, piid), value in self._data_by_siid_piid.items(): | ||
| prop = self._model.get_property_by_siid_piid(siid, piid) | ||
| prop.value = value | ||
| res[prop.name] = prop | ||
|
|
||
| return res | ||
|
|
||
| @property | ||
| def __cli_output__(self): | ||
| """Return a CLI printable status.""" | ||
| out = "" | ||
| props = self.property_dict() | ||
| service = None | ||
| for _name, prop in props.items(): | ||
| miot_prop: MiotProperty = prop.extras["miot_property"] | ||
| if service is None or miot_prop.siid != service.siid: | ||
| service = miot_prop.service | ||
| out += f"Service [bold]{service.description} ({service.name})[/bold]\n" # type: ignore # FIXME | ||
|
|
||
| out += f"\t{prop.description} ({prop.name}, access: {prop.pretty_access}): {prop.pretty_value}" | ||
|
|
||
| if MiotAccess.Write in miot_prop.access: | ||
| out += f" ({prop.format}" | ||
| if prop.pretty_input_constraints is not None: | ||
| out += f", {prop.pretty_input_constraints}" | ||
| out += ")" | ||
|
|
||
| if self.device._debug > 1: | ||
| out += "\n\t[bold]Extras[/bold]\n" | ||
| for extra_key, extra_value in prop.extras.items(): | ||
| out += f"\t\t{extra_key} = {extra_value}\n" | ||
|
|
||
| out += "\n" | ||
|
|
||
| return out | ||
|
|
||
| def __dir__(self) -> Iterable[str]: | ||
| """Return a list of properties.""" | ||
| return list(super().__dir__()) + list(self._data_by_normalized_name.keys()) | ||
|
|
||
| def __repr__(self): | ||
| s = f"<{self.__class__.__name__}" | ||
| for name, value in self.property_dict().items(): | ||
| s += f" {name}={value}" | ||
| s += ">" | ||
|
|
||
| return s |