-
-
Notifications
You must be signed in to change notification settings - Fork 273
Expand file tree
/
Copy pathdevice.py
More file actions
590 lines (487 loc) · 18.3 KB
/
device.py
File metadata and controls
590 lines (487 loc) · 18.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
"""Interact with TPLink Smart Home devices.
Once you have a device via :ref:`Discovery <discover_target>` or
:ref:`Connect <connect_target>` you can start interacting with a device.
>>> from kasa import Discover
>>>
>>> dev = await Discover.discover_single(
>>> "127.0.0.2",
>>> username="[email protected]",
>>> password="great_password"
>>> )
>>>
Most devices can be turned on and off
>>> await dev.turn_on()
>>> await dev.update()
>>> print(dev.is_on)
True
>>> await dev.turn_off()
>>> await dev.update()
>>> print(dev.is_on)
False
All devices provide several informational properties:
>>> dev.alias
Bedroom Lamp Plug
>>> dev.model
HS110(EU)
>>> dev.rssi
-71
>>> dev.mac
50:C7:BF:00:00:00
Some information can also be changed programmatically:
>>> await dev.set_alias("new alias")
>>> await dev.update()
>>> dev.alias
new alias
Devices support different functionality that are exposed via
:ref:`modules <module_target>` that you can access via :attr:`~kasa.Device.modules`:
>>> for module_name in dev.modules:
>>> print(module_name)
Energy
schedule
usage
anti_theft
Time
cloud
Led
>>> led_module = dev.modules["Led"]
>>> print(led_module.led)
False
>>> await led_module.set_led(True)
>>> await dev.update()
>>> print(led_module.led)
True
Individual pieces of functionality are also exposed via :ref:`features <feature_target>`
which you can access via :attr:`~kasa.Device.features` and will only be present if
they are supported.
Features are similar to modules in that they provide functionality that may or may
not be present.
Whereas modules group functionality into a common interface, features expose a single
function that may or may not be part of a module.
The advantage of features is that they have a simple common interface of `id`, `name`,
`value` and `set_value` so no need to learn the module API.
They are useful if you want write code that dynamically adapts as new features are
added to the API.
>>> for feature_name in dev.features:
>>> print(feature_name)
state
rssi
on_since
reboot
current_consumption
consumption_today
consumption_this_month
consumption_total
voltage
current
cloud_connection
led
>>> led_feature = dev.features["led"]
>>> print(led_feature.value)
True
>>> await led_feature.set_value(False)
>>> await dev.update()
>>> print(led_feature.value)
False
"""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, tzinfo
from typing import TYPE_CHECKING, Any, TypeAlias
from warnings import warn
from .credentials import Credentials as _Credentials
from .device_type import DeviceType
from .deviceconfig import (
DeviceConfig,
DeviceConnectionParameters,
DeviceEncryptionType,
DeviceFamily,
)
from .exceptions import KasaException
from .feature import Feature
from .module import Module
from .protocols import BaseProtocol, IotProtocol
from .transports import XorTransport
if TYPE_CHECKING:
from .modulemapping import ModuleMapping, ModuleName
@dataclass
class WifiNetwork:
"""Wifi network container."""
ssid: str
key_type: int
# These are available only on softaponboarding
cipher_type: int | None = None
bssid: str | None = None
channel: int | None = None
rssi: int | None = None
# For SMART devices
signal_level: int | None = None
_LOGGER = logging.getLogger(__name__)
@dataclass
class _DeviceInfo:
"""Device Model Information."""
short_name: str
long_name: str
brand: str
device_family: str
device_type: DeviceType
hardware_version: str
firmware_version: str
firmware_build: str
requires_auth: bool
region: str | None
class Device(ABC):
"""Common device interface.
Do not instantiate this class directly, instead get a device instance from
:func:`Device.connect()`, :func:`Discover.discover()`
or :func:`Discover.discover_single()`.
"""
# All types required to create devices directly via connect are aliased here
# to avoid consumers having to do multiple imports.
#: The type of device
Type: TypeAlias = DeviceType
#: The credentials for authentication
Credentials: TypeAlias = _Credentials
#: Configuration for connecting to the device
Config: TypeAlias = DeviceConfig
#: The family of the device, e.g. SMART.KASASWITCH.
Family: TypeAlias = DeviceFamily
#: The encryption for the device, e.g. Klap or Aes
EncryptionType: TypeAlias = DeviceEncryptionType
#: The connection type for the device.
ConnectionParameters: TypeAlias = DeviceConnectionParameters
def __init__(
self,
host: str,
*,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
"""Create a new Device instance.
:param str host: host name or IP address of the device
:param DeviceConfig config: device configuration
:param BaseProtocol protocol: protocol for communicating with the device
"""
if config and protocol:
protocol._transport._config = config
self.protocol: BaseProtocol = protocol or IotProtocol(
transport=XorTransport(config=config or DeviceConfig(host=host)),
)
self._last_update: Any = None
_LOGGER.debug("Initializing %s of type %s", host, type(self))
self._device_type = DeviceType.Unknown
# TODO: typing Any is just as using dict | None would require separate
# checks in accessors. the @updated_required decorator does not ensure
# mypy that these are not accessed incorrectly.
self._discovery_info: dict[str, Any] | None = None
self._features: dict[str, Feature] = {}
self._parent: Device | None = None
self._children: Mapping[str, Device] = {}
@staticmethod
async def connect(
*,
host: str | None = None,
config: DeviceConfig | None = None,
) -> Device:
"""Connect to a single device by the given hostname or device configuration.
This method avoids the UDP based discovery process and
will connect directly to the device.
It is generally preferred to avoid :func:`discover_single()` and
use this function instead as it should perform better when
the WiFi network is congested or the device is not responding
to discovery requests.
:param host: Hostname of device to query
:param config: Connection parameters to ensure the correct protocol
and connection options are used.
:rtype: SmartDevice
:return: Object for querying/controlling found device.
"""
from .device_factory import connect # pylint: disable=import-outside-toplevel
return await connect(host=host, config=config) # type: ignore[arg-type]
@abstractmethod
async def update(self, update_children: bool = True) -> None:
"""Update the device."""
async def disconnect(self) -> None:
"""Disconnect and close any underlying connection resources."""
await self.protocol.close()
@property
@abstractmethod
def modules(self) -> ModuleMapping[Module]:
"""Return the device modules."""
@property
@abstractmethod
def is_on(self) -> bool:
"""Return true if the device is on."""
@property
def is_off(self) -> bool:
"""Return True if device is off."""
return not self.is_on
@abstractmethod
async def turn_on(self, **kwargs) -> dict:
"""Turn on the device."""
@abstractmethod
async def turn_off(self, **kwargs) -> dict:
"""Turn off the device."""
@abstractmethod
async def set_state(self, on: bool) -> dict:
"""Set the device state to *on*.
This allows turning the device on and off.
See also *turn_off* and *turn_on*.
"""
@property
def host(self) -> str:
"""The device host."""
return self.protocol._transport._host
@host.setter
def host(self, value: str) -> None:
"""Set the device host.
Generally used by discovery to set the hostname after ip discovery.
"""
self.protocol._transport._host = value
self.protocol._transport._config.host = value
@property
def port(self) -> int:
"""The device port."""
return self.protocol._transport._port
@property
def credentials(self) -> _Credentials | None:
"""The device credentials."""
return self.protocol._transport._credentials
@property
def credentials_hash(self) -> str | None:
"""The protocol specific hash of the credentials the device is using."""
return self.protocol._transport.credentials_hash
@property
def device_type(self) -> DeviceType:
"""Return the device type."""
return self._device_type
@abstractmethod
def update_from_discover_info(self, info: dict) -> None:
"""Update state from info from the discover call."""
@property
def config(self) -> DeviceConfig:
"""Return the device configuration."""
return self.protocol.config
@property
@abstractmethod
def model(self) -> str:
"""Returns the device model."""
@property
@abstractmethod
def _model_region(self) -> str:
"""Return device full model name and region."""
@property
@abstractmethod
def alias(self) -> str | None:
"""Returns the device alias or nickname."""
async def _raw_query(self, request: str | dict) -> dict:
"""Send a raw query to the device."""
return await self.protocol.query(request=request)
@property
def parent(self) -> Device | None:
"""Return the parent on child devices."""
return self._parent
@property
def children(self) -> Sequence[Device]:
"""Returns the child devices."""
return list(self._children.values())
def get_child_device(self, name_or_id: str) -> Device | None:
"""Return child device by its device_id or alias."""
if name_or_id in self._children:
return self._children[name_or_id]
name_lower = name_or_id.lower()
for child in self.children:
if child.alias and child.alias.lower() == name_lower:
return child
return None
@property
@abstractmethod
def sys_info(self) -> dict[str, Any]:
"""Returns the device info."""
def get_plug_by_name(self, name: str) -> Device:
"""Return child device for the given name."""
for p in self.children:
if p.alias == name:
return p
raise KasaException(f"Device has no child with {name}")
def get_plug_by_index(self, index: int) -> Device:
"""Return child device for the given index."""
if index + 1 > len(self.children) or index < 0:
raise KasaException(
f"Invalid index {index}, device has {len(self.children)} plugs"
)
return self.children[index]
@property
@abstractmethod
def time(self) -> datetime:
"""Return the time."""
@property
@abstractmethod
def timezone(self) -> tzinfo:
"""Return the timezone and time_difference."""
@property
@abstractmethod
def hw_info(self) -> dict:
"""Return hardware info for the device."""
@property
@abstractmethod
def location(self) -> dict:
"""Return the device location."""
@property
@abstractmethod
def rssi(self) -> int | None:
"""Return the rssi."""
@property
@abstractmethod
def mac(self) -> str:
"""Return the mac formatted with colons."""
@property
@abstractmethod
def device_id(self) -> str:
"""Return the device id."""
@property
@abstractmethod
def internal_state(self) -> dict:
"""Return all the internal state data."""
@property
def state_information(self) -> dict[str, Any]:
"""Return available features and their values."""
return {feat.name: feat.value for feat in self._features.values()}
@property
def features(self) -> dict[str, Feature]:
"""Return the list of supported features."""
return self._features
def _add_feature(self, feature: Feature) -> None:
"""Add a new feature to the device."""
if feature.id in self._features:
raise KasaException(f"Duplicate feature id {feature.id}")
assert feature.id is not None # TODO: hack for typing # noqa: S101
self._features[feature.id] = feature
@property
@abstractmethod
def has_emeter(self) -> bool:
"""Return if the device has emeter."""
@property
@abstractmethod
def on_since(self) -> datetime | None:
"""Return the time that the device was turned on or None if turned off.
This returns a cached value if the device reported value difference is under
five seconds to avoid device-caused jitter.
"""
@abstractmethod
async def wifi_scan(self) -> list[WifiNetwork]:
"""Scan for available wifi networks."""
@abstractmethod
async def wifi_join(
self, ssid: str, password: str, keytype: str = "wpa2_psk"
) -> dict:
"""Join the given wifi network."""
@abstractmethod
async def set_alias(self, alias: str) -> dict:
"""Set the device name (alias)."""
@abstractmethod
async def reboot(self, delay: int = 1) -> None:
"""Reboot the device.
Note that giving a delay of zero causes this to block,
as the device reboots immediately without responding to the call.
"""
@abstractmethod
async def factory_reset(self) -> None:
"""Reset device back to factory settings.
Note, this does not downgrade the firmware.
"""
def __repr__(self) -> str:
update_needed = " - update() needed" if not self._last_update else ""
if not self._last_update and not self._discovery_info:
return f"<{self.device_type} at {self.host}{update_needed}>"
return (
f"<{self.device_type} at {self.host} -"
f" {self.alias} ({self.model}){update_needed}>"
)
_deprecated_device_type_attributes = {
# is_type
"is_bulb": (None, DeviceType.Bulb),
"is_dimmer": (None, DeviceType.Dimmer),
"is_light_strip": (None, DeviceType.LightStrip),
"is_plug": (None, DeviceType.Plug),
"is_wallswitch": (None, DeviceType.WallSwitch),
"is_strip": (None, DeviceType.Strip),
"is_strip_socket": (None, DeviceType.StripSocket),
}
def _get_replacing_attr(
self, module_name: ModuleName | None, *attrs: Any
) -> str | None:
# If module name is None check self
if not module_name:
check = self
elif (check := self.modules.get(module_name)) is None:
return None
for attr in attrs:
# Use dir() as opposed to hasattr() to avoid raising exceptions
# from properties
if attr in dir(check):
return attr
return None
_deprecated_other_attributes = {
# light attributes
"is_color": (Module.Light, ["is_color"]),
"is_dimmable": (Module.Light, ["is_dimmable"]),
"is_variable_color_temp": (Module.Light, ["is_variable_color_temp"]),
"brightness": (Module.Light, ["brightness"]),
"set_brightness": (Module.Light, ["set_brightness"]),
"hsv": (Module.Light, ["hsv"]),
"set_hsv": (Module.Light, ["set_hsv"]),
"color_temp": (Module.Light, ["color_temp"]),
"set_color_temp": (Module.Light, ["set_color_temp"]),
"valid_temperature_range": (Module.Light, ["valid_temperature_range"]),
"has_effects": (Module.Light, ["has_effects"]),
"_deprecated_set_light_state": (Module.Light, ["has_effects"]),
# led attributes
"led": (Module.Led, ["led"]),
"set_led": (Module.Led, ["set_led"]),
# light effect attributes
# The return values for effect is a str instead of dict so the lightstrip
# modules have a _deprecated method to return the value as before.
"effect": (Module.LightEffect, ["_deprecated_effect", "effect"]),
# The return values for effect_list includes the Off effect so the lightstrip
# modules have a _deprecated method to return the values as before.
"effect_list": (Module.LightEffect, ["_deprecated_effect_list", "effect_list"]),
"set_effect": (Module.LightEffect, ["set_effect"]),
"set_custom_effect": (Module.LightEffect, ["set_custom_effect"]),
# light preset attributes
"presets": (Module.LightPreset, ["_deprecated_presets", "preset_states_list"]),
"save_preset": (Module.LightPreset, ["_deprecated_save_preset"]),
# Emeter attribues
"get_emeter_realtime": (Module.Energy, ["get_status"]),
"emeter_realtime": (Module.Energy, ["status"]),
"emeter_today": (Module.Energy, ["consumption_today"]),
"emeter_this_month": (Module.Energy, ["consumption_this_month"]),
"current_consumption": (Module.Energy, ["current_consumption"]),
"get_emeter_daily": (Module.Energy, ["get_daily_stats"]),
"get_emeter_monthly": (Module.Energy, ["get_monthly_stats"]),
# Other attributes
"supported_modules": (None, ["modules"]),
}
if not TYPE_CHECKING:
def __getattr__(self, name: str) -> Any:
# is_device_type
if dep_device_type_attr := self._deprecated_device_type_attributes.get(
name
):
msg = f"{name} is deprecated, use device_type property instead"
warn(msg, DeprecationWarning, stacklevel=2)
return self.device_type == dep_device_type_attr[1]
# Other deprecated attributes
if (dep_attr := self._deprecated_other_attributes.get(name)) and (
(replacing_attr := self._get_replacing_attr(dep_attr[0], *dep_attr[1]))
is not None
):
mod = dep_attr[0]
dev_or_mod = self.modules[mod] if mod else self
replacing = f"Module.{mod} in device.modules" if mod else replacing_attr
msg = f"{name} is deprecated, use: {replacing} instead"
warn(msg, DeprecationWarning, stacklevel=2)
return getattr(dev_or_mod, replacing_attr)
raise AttributeError(f"Device has no attribute {name!r}")