GeezerVison is showing github.com — Add generic interface (descriptors) for obtaining device features · python-kasa/python-kasa@f5e180a
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: kasa/cli.py +36Lines changed: 36 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -565,6 +565,10 @@ async def state(ctx, dev: Device): | 565 | 565 | else: | 566 | 566 | echo(f"\t{info_name}: {info_data}") | 567 | 567 |
| | 568 | + echo("\n\t[bold]== Descriptors == [/bold]") | | 569 | + for id_, descriptor in dev.descriptors.items(): | | 570 | + echo(f"\t{descriptor.name} ({id_}): {descriptor.value}") | | 571 | + | 568 | 572 | if dev.has_emeter: | 569 | 573 | echo("\n\t[bold]== Current State ==[/bold]") | 570 | 574 | emeter_status = dev.emeter_realtime | @@ -1102,5 +1106,37 @@ async def shell(dev: Device): | 1102 | 1106 | loop.stop() | 1103 | 1107 |
| 1104 | 1108 |
| | 1109 | +@cli.command(name="descriptor") | | 1110 | +@click.argument("name", required=False) | | 1111 | +@click.argument("value", required=False) | | 1112 | +@pass_dev | | 1113 | +async def descriptor(dev, name: str, value): | | 1114 | + """Access and modify descriptor values. | | 1115 | +
| | 1116 | + If no *name* is given, lists available descriptors and their values. | | 1117 | + If only *name* is given, the value of named descriptor is returned. | | 1118 | + If both *name* and *value* are set, the described setting is changed. | | 1119 | + """ | | 1120 | + if not name: | | 1121 | + echo("[bold]== Descriptors ==[/bold]") | | 1122 | + for name, desc in dev.descriptors.items(): | | 1123 | + echo(f"{desc.name} ({name}): {desc.value}") | | 1124 | + return | | 1125 | + | | 1126 | + if name not in dev.descriptors: | | 1127 | + echo(f"No descriptor by name {name}") | | 1128 | + return | | 1129 | + | | 1130 | + desc = dev.descriptors[name] | | 1131 | + | | 1132 | + if value is None: | | 1133 | + echo(f"{desc.name} ({name}): {desc.value}") | | 1134 | + return desc.value | | 1135 | + | | 1136 | + echo(f"Setting {name} to {value}") | | 1137 | + value = ast.literal_eval(value) | | 1138 | + return await dev.descriptors[name].set_value(value) | | 1139 | + | | 1140 | + | 1105 | 1141 | if __name__ == "__main__": | 1106 | 1142 | cli() |
Collapse file Copy file name to clipboard +54Lines changed: 54 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| | 1 | +"""Generic interface for defining device features.""" | | 2 | +from dataclasses import dataclass | | 3 | +from enum import Enum, auto | | 4 | +from typing import Any, Callable | | 5 | + | | 6 | + | | 7 | +class DescriptorCategory(Enum): | | 8 | + """Descriptor category.""" | | 9 | + | | 10 | + # TODO: we could probably do better than using the scheme homeassistant is using | | 11 | + Config = auto() | | 12 | + Diagnostic = auto() | | 13 | + | | 14 | + | | 15 | +class DescriptorType(Enum): | | 16 | + """Type of the information defined by the descriptor.""" | | 17 | + | | 18 | + Sensor = auto() | | 19 | + BinarySensor = auto() | | 20 | + Switch = auto() | | 21 | + Button = auto() | | 22 | + | | 23 | + | | 24 | +@dataclass | | 25 | +class Descriptor: | | 26 | + """Descriptor defines a generic interface for device features.""" | | 27 | + | | 28 | + device: Any # TODO: rename to something else, this can also be a module. | | 29 | + #: User-friendly short description | | 30 | + name: str | | 31 | + #: Name of the property that allows accessing the value | | 32 | + attribute_getter: str | Callable | | 33 | + #: Name of the method that allows changing the value | | 34 | + attribute_setter: str | None = None | | 35 | + #: Type of the information | | 36 | + icon: str | None = None | | 37 | + #: Unit of the descriptor | | 38 | + unit: str | None = None | | 39 | + #: Hint for homeassistant | | 40 | + #: TODO: Replace with a set of flags to allow homeassistant make its own decision? | | 41 | + show_in_hass: bool = True | | 42 | + category: DescriptorCategory = DescriptorCategory.Diagnostic | | 43 | + type: DescriptorType = DescriptorType.Sensor | | 44 | + | | 45 | + @property | | 46 | + def value(self): | | 47 | + """Return the current value.""" | | 48 | + if isinstance(self.attribute_getter, Callable): | | 49 | + return self.attribute_getter(self.device) | | 50 | + return getattr(self.device, self.attribute_getter) | | 51 | + | | 52 | + async def set_value(self, value): | | 53 | + """Set the value.""" | | 54 | + return await getattr(self.device, self.attribute_setter)(value) |
Collapse file Copy file name to clipboardExpand all lines: kasa/device.py +14Lines changed: 14 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 6 | 6 | from typing import Any, Dict, List, Optional, Sequence, Set, Union | 7 | 7 |
| 8 | 8 | from .credentials import Credentials | | 9 | +from .descriptors import Descriptor | 9 | 10 | from .device_type import DeviceType | 10 | 11 | from .deviceconfig import DeviceConfig | 11 | 12 | from .emeterstatus import EmeterStatus | @@ -69,6 +70,7 @@ def __init__( | 69 | 70 | self._discovery_info: Optional[Dict[str, Any]] = None | 70 | 71 |
| 71 | 72 | self.modules: Dict[str, Any] = {} | | 73 | + self._descriptors: Dict[str, Descriptor] = {} | 72 | 74 |
| 73 | 75 | @staticmethod | 74 | 76 | async def connect( | @@ -343,6 +345,18 @@ async def wifi_join(self, ssid: str, password: str, keytype: str = "wpa2_psk"): | 343 | 345 | async def set_alias(self, alias: str): | 344 | 346 | """Set the device name (alias).""" | 345 | 347 |
| | 348 | + @property | | 349 | + def descriptors(self) -> Dict[str, Descriptor]: | | 350 | + """Return the list of descriptors.""" | | 351 | + return self._descriptors | | 352 | + | | 353 | + def add_descriptor(self, descriptor: "Descriptor"): | | 354 | + """Add a new descriptor to the device.""" | | 355 | + desc_name = descriptor.name.lower().replace(" ", "_") | | 356 | + if desc_name in self._descriptors: | | 357 | + raise SmartDeviceException("Duplicate descriptor name %s" % desc_name) | | 358 | + self._descriptors[desc_name] = descriptor | | 359 | + | 346 | 360 | def __repr__(self): | 347 | 361 | if self._last_update is None: | 348 | 362 | return f"<{self._device_type} at {self.host} - update() needed>" | |
Collapse file Copy file name to clipboardExpand all lines: kasa/iot/iotdevice.py +38Lines changed: 38 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 18 | 18 | from datetime import datetime, timedelta | 19 | 19 | from typing import Any, Dict, List, Optional, Sequence, Set | 20 | 20 |
| | 21 | +from ..descriptors import Descriptor | 21 | 22 | from ..device import Device, WifiNetwork | 22 | 23 | from ..deviceconfig import DeviceConfig | 23 | 24 | from ..emeterstatus import EmeterStatus | @@ -186,6 +187,7 @@ def __init__( | 186 | 187 | self._sys_info: Any = None # TODO: this is here to avoid changing tests | 187 | 188 | self._features: Set[str] = set() | 188 | 189 | self._children: Sequence["IotDevice"] = [] | | 190 | + self._supported_modules: Optional[Dict[str, IotModule]] = None | 189 | 191 |
| 190 | 192 | @property | 191 | 193 | def children(self) -> Sequence["IotDevice"]: | @@ -299,9 +301,33 @@ async def update(self, update_children: bool = True): | 299 | 301 | self._last_update = response | 300 | 302 | self._set_sys_info(response["system"]["get_sysinfo"]) | 301 | 303 |
| | 304 | + if not self._descriptors: | | 305 | + await self._initialize_descriptors() | | 306 | + | 302 | 307 | await self._modular_update(req) | 303 | 308 | self._set_sys_info(self._last_update["system"]["get_sysinfo"]) | 304 | 309 |
| | 310 | + async def _initialize_descriptors(self): | | 311 | + self.add_descriptor( | | 312 | + Descriptor( | | 313 | + device=self, name="RSSI", attribute_getter="rssi", icon="mdi:signal" | | 314 | + ) | | 315 | + ) | | 316 | + self.add_descriptor( | | 317 | + Descriptor( | | 318 | + device=self, name="Time", attribute_getter="time", show_in_hass=False | | 319 | + ) | | 320 | + ) | | 321 | + if "on_time" in self._sys_info: | | 322 | + self.add_descriptor( | | 323 | + Descriptor( | | 324 | + device=self, | | 325 | + name="On since", | | 326 | + attribute_getter="on_since", | | 327 | + icon="mdi:clock", | | 328 | + ) | | 329 | + ) | | 330 | + | 305 | 331 | async def _modular_update(self, req: dict) -> None: | 306 | 332 | """Execute an update query.""" | 307 | 333 | if self.has_emeter: | @@ -310,6 +336,18 @@ async def _modular_update(self, req: dict) -> None: | 310 | 336 | ) | 311 | 337 | self.add_module("emeter", Emeter(self, self.emeter_type)) | 312 | 338 |
| | 339 | + # TODO: perhaps modules should not have unsupported modules, | | 340 | + # making separate handling for this unnecessary | | 341 | + if self._supported_modules is None: | | 342 | + supported = {} | | 343 | + for module in self.modules.values(): | | 344 | + if module.is_supported: | | 345 | + supported[module._module] = module | | 346 | + for _, module_desc in module._module_descriptors.items(): | | 347 | + self.add_descriptor(module_desc) | | 348 | + | | 349 | + self._supported_modules = supported | | 350 | + | 313 | 351 | request_list = [] | 314 | 352 | est_response_size = 1024 if "system" in req else 0 | 315 | 353 | for module in self.modules.values(): | |
Collapse file Copy file name to clipboardExpand all lines: kasa/iot/iotplug.py +14-2Lines changed: 14 additions & 2 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 2 | 2 | import logging | 3 | 3 | from typing import Any, Dict, Optional | 4 | 4 |
| | 5 | +from ..descriptors import Descriptor, DescriptorCategory, DescriptorType | 5 | 6 | from ..device_type import DeviceType | 6 | 7 | from ..deviceconfig import DeviceConfig | 7 | 8 | from ..protocol import BaseProtocol | @@ -56,6 +57,18 @@ def __init__( | 56 | 57 | self.add_module("time", Time(self, "time")) | 57 | 58 | self.add_module("cloud", Cloud(self, "cnCloud")) | 58 | 59 |
| | 60 | + self.add_descriptor( | | 61 | + Descriptor( | | 62 | + device=self, | | 63 | + name="LED", | | 64 | + icon="mdi:led-{state}", | | 65 | + attribute_getter="led", | | 66 | + attribute_setter="set_led", | | 67 | + category=DescriptorCategory.Config, | | 68 | + type=DescriptorType.Switch, | | 69 | + ) | | 70 | + ) | | 71 | + | 59 | 72 | @property # type: ignore | 60 | 73 | @requires_update | 61 | 74 | def is_on(self) -> bool: | @@ -88,5 +101,4 @@ async def set_led(self, state: bool): | 88 | 101 | @requires_update | 89 | 102 | def state_information(self) -> Dict[str, Any]: | 90 | 103 | """Return switch-specific state information.""" | 91 | | - info = {"LED state": self.led, "On since": self.on_since} | 92 | | - return info | | 104 | + return {} |
Collapse file Copy file name to clipboardExpand all lines: kasa/iot/modules/cloud.py +18Lines changed: 18 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 4 | 4 | except ImportError: | 5 | 5 | from pydantic import BaseModel | 6 | 6 |
| | 7 | +from ...descriptors import Descriptor, DescriptorType | 7 | 8 | from .module import IotModule | 8 | 9 |
| 9 | 10 |
| @@ -25,6 +26,23 @@ class CloudInfo(BaseModel): | 25 | 26 | class Cloud(IotModule): | 26 | 27 | """Module implementing support for cloud services.""" | 27 | 28 |
| | 29 | + def __init__(self, device, module): | | 30 | + super().__init__(device, module) | | 31 | + self.add_descriptor( | | 32 | + Descriptor( | | 33 | + device=self, | | 34 | + name="Cloud Connection", | | 35 | + icon="mdi:cloud", | | 36 | + attribute_getter="is_connected", | | 37 | + type=DescriptorType.BinarySensor, | | 38 | + ) | | 39 | + ) | | 40 | + | | 41 | + @property | | 42 | + def is_connected(self) -> bool: | | 43 | + """Return true if device is connected to the cloud.""" | | 44 | + return self.info.binded | | 45 | + | 28 | 46 | def query(self): | 29 | 47 | """Request cloud connectivity info.""" | 30 | 48 | return self.query_for_command("get_info") | |
Collapse file Copy file name to clipboardExpand all lines: kasa/iot/modules/module.py +10-1Lines changed: 10 additions & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
| 2 | 2 | import collections | 3 | 3 | import logging | 4 | 4 | from abc import ABC, abstractmethod | 5 | | -from typing import TYPE_CHECKING | | 5 | +from typing import TYPE_CHECKING, Dict | 6 | 6 |
| | 7 | +from ...descriptors import Descriptor | 7 | 8 | from ...exceptions import SmartDeviceException | 8 | 9 |
| 9 | 10 | if TYPE_CHECKING: | @@ -34,6 +35,14 @@ class IotModule(ABC): | 34 | 35 | def __init__(self, device: "IotDevice", module: str): | 35 | 36 | self._device = device | 36 | 37 | self._module = module | | 38 | + self._module_descriptors: Dict[str, Descriptor] = {} | | 39 | + | | 40 | + def add_descriptor(self, desc): | | 41 | + """Add module descriptor.""" | | 42 | + module_desc_name = f"{self._module}_{desc.name}" | | 43 | + if module_desc_name in self._module_descriptors: | | 44 | + raise Exception("Duplicate name detected %s" % module_desc_name) | | 45 | + self._module_descriptors[module_desc_name] = desc | 37 | 46 |
| 38 | 47 | @abstractmethod | 39 | 48 | def query(self): | |
Collapse file Copy file name to clipboardExpand all lines: kasa/smart/smartdevice.py +57-3Lines changed: 57 additions & 3 deletions | Original file line number | Diff line number | Diff line change |
|---|
| 5 | 5 | from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Set, cast | 6 | 6 |
| 7 | 7 | from ..aestransport import AesTransport | | 8 | +from ..descriptors import Descriptor, DescriptorType | 8 | 9 | from ..device import Device, WifiNetwork | 9 | 10 | from ..device_type import DeviceType | 10 | 11 | from ..deviceconfig import DeviceConfig | @@ -118,13 +119,62 @@ async def update(self, update_children: bool = True): | 118 | 119 | for info in child_info["child_device_list"]: | 119 | 120 | self._children[info["device_id"]].update_internal_state(info) | 120 | 121 |
| | 122 | + # We can first initialize the descriptors after the first update. | | 123 | + # We make here an assumption that every device has at least a single descriptor. | | 124 | + if not self._descriptors: | | 125 | + await self._initialize_descriptors() | | 126 | + | 121 | 127 | _LOGGER.debug("Got an update: %s", self._last_update) | 122 | 128 |
| 123 | 129 | async def _initialize_modules(self): | 124 | 130 | """Initialize modules based on component negotiation response.""" | 125 | 131 | if "energy_monitoring" in self._components: | 126 | 132 | self.emeter_type = "emeter" | 127 | 133 |
| | 134 | + async def _initialize_descriptors(self): | | 135 | + """Initialize device descriptors.""" | | 136 | + self.add_descriptor( | | 137 | + Descriptor( | | 138 | + self, | | 139 | + "Signal Level", | | 140 | + attribute_getter=lambda x: x._info["signal_level"], | | 141 | + icon="mdi:signal", | | 142 | + ) | | 143 | + ) | | 144 | + self.add_descriptor( | | 145 | + Descriptor( | | 146 | + device=self, name="Time", attribute_getter="time", show_in_hass=False | | 147 | + ) | | 148 | + ) | | 149 | + self.add_descriptor( | | 150 | + Descriptor( | | 151 | + device=self, name="SSID", attribute_getter="ssid", icon="mdi:wifi" | | 152 | + ) | | 153 | + ) | | 154 | + | | 155 | + if "overheated" in self._info: | | 156 | + self.add_descriptor( | | 157 | + Descriptor( | | 158 | + self, | | 159 | + "Overheated", | | 160 | + attribute_getter=lambda x: x._info["overheated"], | | 161 | + icon="mdi:heat-wave", | | 162 | + type=DescriptorType.BinarySensor, | | 163 | + ) | | 164 | + ) | | 165 | + | | 166 | + # We check for the key available, and not for the property truthiness, | | 167 | + # as the value is falsy when the device is off. | | 168 | + if "on_since" in self._info: | | 169 | + self.add_descriptor( | | 170 | + Descriptor( | | 171 | + device=self, | | 172 | + name="On since", | | 173 | + attribute_getter="on_since", | | 174 | + icon="mdi:clock", | | 175 | + ) | | 176 | + ) | | 177 | + | 128 | 178 | @property | 129 | 179 | def sys_info(self) -> Dict[str, Any]: | 130 | 180 | """Returns the device info.""" | @@ -215,15 +265,19 @@ async def _query_helper( | 215 | 265 | return res | 216 | 266 |
| 217 | 267 | @property | 218 | | - def state_information(self) -> Dict[str, Any]: | 219 | | - """Return the key state information.""" | | 268 | + def ssid(self) -> str: | | 269 | + """Return ssid of the connected wifi ap.""" | 220 | 270 | ssid = self._info.get("ssid") | 221 | 271 | ssid = base64.b64decode(ssid).decode() if ssid else "No SSID" | | 272 | + return ssid | 222 | 273 |
| | 274 | + @property | | 275 | + def state_information(self) -> Dict[str, Any]: | | 276 | + """Return the key state information.""" | 223 | 277 | return { | 224 | 278 | "overheated": self._info.get("overheated"), | 225 | 279 | "signal_level": self._info.get("signal_level"), | 226 | | - "SSID": ssid, | | 280 | + "SSID": self.ssid, | 227 | 281 | } | 228 | 282 |
| 229 | 283 | @property | |
|