Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2026-01-10 13:52:43 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Refactor devices into subpackages and deprecate old names (#716)

Browse Source 
* Refactor devices into subpackages and deprecate old names

* Tweak and add tests

* Fix linting

* Remove duplicate implementations affecting project coverage

* Update post review

* Add device base class attributes and rename subclasses

* Rename Module to BaseModule

* Remove has_emeter_history

* Fix missing _time in init

* Update post review

* Fix test_readmeexamples

* Fix erroneously duped files

* Clean up iot and smart imports

* Update post latest review

* Tweak Device docstring
This commit is contained in:
Steven B
2024-02-04 15:20:08 +00:00
committed by GitHub
parent 6afd05be59
commit 0d119e63d0
49 changed files with 1046 additions and 606 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
kasa/iot/__init__.py Normal file
View File

@@ -0,0 +1,16 @@
"""Package for supporting legacy kasa devices."""
from .iotbulb import IotBulb
from .iotdevice import IotDevice
from .iotdimmer import IotDimmer
from .iotlightstrip import IotLightStrip
from .iotplug import IotPlug
from .iotstrip import IotStrip
__all__ = [
"IotDevice",
"IotPlug",
"IotBulb",
"IotStrip",
"IotDimmer",
"IotLightStrip",
]

526
kasa/iot/iotbulb.py Normal file
View File

@@ -0,0 +1,526 @@
"""Module for bulbs (LB*, KL*, KB*)."""
import logging
import re
from enum import Enum
from typing import Any, Dict, List, Optional, cast
try:
from pydantic.v1 import BaseModel, Field, root_validator
except ImportError:
from pydantic import BaseModel, Field, root_validator
from ..bulb import HSV, Bulb, BulbPreset, ColorTempRange
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..protocol import BaseProtocol
from .iotdevice import IotDevice, SmartDeviceException, requires_update
from .modules import Antitheft, Cloud, Countdown, Emeter, Schedule, Time, Usage
class BehaviorMode(str, Enum):
"""Enum to present type of turn on behavior."""
#: Return to the last state known state.
Last = "last_status"
#: Use chosen preset.
Preset = "customize_preset"
class TurnOnBehavior(BaseModel):
"""Model to present a single turn on behavior.
:param int preset: the index number of wanted preset.
:param BehaviorMode mode: last status or preset mode.
If you are changing existing settings, you should not set this manually.
To change the behavior, it is only necessary to change the :attr:`preset` field
to contain either the preset index, or ``None`` for the last known state.
"""
#: Index of preset to use, or ``None`` for the last known state.
preset: Optional[int] = Field(alias="index", default=None)
#: Wanted behavior
mode: BehaviorMode
@root_validator
def _mode_based_on_preset(cls, values):
"""Set the mode based on the preset value."""
if values["preset"] is not None:
values["mode"] = BehaviorMode.Preset
else:
values["mode"] = BehaviorMode.Last
return values
class Config:
"""Configuration to make the validator run when changing the values."""
validate_assignment = True
class TurnOnBehaviors(BaseModel):
"""Model to contain turn on behaviors."""
#: The behavior when the bulb is turned on programmatically.
soft: TurnOnBehavior = Field(alias="soft_on")
#: The behavior when the bulb has been off from mains power.
hard: TurnOnBehavior = Field(alias="hard_on")
TPLINK_KELVIN = {
"LB130": ColorTempRange(2500, 9000),
"LB120": ColorTempRange(2700, 6500),
"LB230": ColorTempRange(2500, 9000),
"KB130": ColorTempRange(2500, 9000),
"KL130": ColorTempRange(2500, 9000),
"KL125": ColorTempRange(2500, 6500),
"KL135": ColorTempRange(2500, 6500),
r"KL120\(EU\)": ColorTempRange(2700, 6500),
r"KL120\(US\)": ColorTempRange(2700, 5000),
r"KL430": ColorTempRange(2500, 9000),
}
NON_COLOR_MODE_FLAGS = {"transition_period", "on_off"}
_LOGGER = logging.getLogger(__name__)
class IotBulb(IotDevice, Bulb):
r"""Representation of a TP-Link Smart Bulb.
To initialize, you have to await :func:`update()` at least once.
This will allow accessing the properties using the exposed properties.
All changes to the device are done using awaitable methods,
which will not change the cached values,
so you must await :func:`update()` to fetch updates values from the device.
Errors reported by the device are raised as
:class:`SmartDeviceExceptions <kasa.exceptions.SmartDeviceException>`,
and should be handled by the user of the library.
Examples:
>>> import asyncio
>>> bulb = IotBulb("127.0.0.1")
>>> asyncio.run(bulb.update())
>>> print(bulb.alias)
Bulb2
Bulbs, like any other supported devices, can be turned on and off:
>>> asyncio.run(bulb.turn_off())
>>> asyncio.run(bulb.turn_on())
>>> asyncio.run(bulb.update())
>>> print(bulb.is_on)
True
You can use the ``is_``-prefixed properties to check for supported features:
>>> bulb.is_dimmable
True
>>> bulb.is_color
True
>>> bulb.is_variable_color_temp
True
All known bulbs support changing the brightness:
>>> bulb.brightness
30
>>> asyncio.run(bulb.set_brightness(50))
>>> asyncio.run(bulb.update())
>>> bulb.brightness
50
Bulbs supporting color temperature can be queried for the supported range:
>>> bulb.valid_temperature_range
ColorTempRange(min=2500, max=9000)
>>> asyncio.run(bulb.set_color_temp(3000))
>>> asyncio.run(bulb.update())
>>> bulb.color_temp
3000
Color bulbs can be adjusted by passing hue, saturation and value:
>>> asyncio.run(bulb.set_hsv(180, 100, 80))
>>> asyncio.run(bulb.update())
>>> bulb.hsv
HSV(hue=180, saturation=100, value=80)
If you don't want to use the default transitions,
you can pass `transition` in milliseconds.
All methods changing the state of the device support this parameter:
* :func:`turn_on`
* :func:`turn_off`
* :func:`set_hsv`
* :func:`set_color_temp`
* :func:`set_brightness`
Light strips (e.g., KL420L5) do not support this feature,
but silently ignore the parameter.
The following changes the brightness over a period of 10 seconds:
>>> asyncio.run(bulb.set_brightness(100, transition=10_000))
Bulb configuration presets can be accessed using the :func:`presets` property:
>>> bulb.presets
[BulbPreset(index=0, brightness=50, hue=0, saturation=0, color_temp=2700, custom=None, id=None, mode=None), BulbPreset(index=1, brightness=100, hue=0, saturation=75, color_temp=0, custom=None, id=None, mode=None), BulbPreset(index=2, brightness=100, hue=120, saturation=75, color_temp=0, custom=None, id=None, mode=None), BulbPreset(index=3, brightness=100, hue=240, saturation=75, color_temp=0, custom=None, id=None, mode=None)]
To modify an existing preset, pass :class:`~kasa.smartbulb.SmartBulbPreset`
instance to :func:`save_preset` method:
>>> preset = bulb.presets[0]
>>> preset.brightness
50
>>> preset.brightness = 100
>>> asyncio.run(bulb.save_preset(preset))
>>> bulb.presets[0].brightness
100
""" # noqa: E501
LIGHT_SERVICE = "smartlife.iot.smartbulb.lightingservice"
SET_LIGHT_METHOD = "transition_light_state"
emeter_type = "smartlife.iot.common.emeter"
def __init__(
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Bulb
self.add_module("schedule", Schedule(self, "smartlife.iot.common.schedule"))
self.add_module("usage", Usage(self, "smartlife.iot.common.schedule"))
self.add_module("antitheft", Antitheft(self, "smartlife.iot.common.anti_theft"))
self.add_module("time", Time(self, "smartlife.iot.common.timesetting"))
self.add_module("emeter", Emeter(self, self.emeter_type))
self.add_module("countdown", Countdown(self, "countdown"))
self.add_module("cloud", Cloud(self, "smartlife.iot.common.cloud"))
@property # type: ignore
@requires_update
def is_color(self) -> bool:
"""Whether the bulb supports color changes."""
sys_info = self.sys_info
return bool(sys_info["is_color"])
@property # type: ignore
@requires_update
def is_dimmable(self) -> bool:
"""Whether the bulb supports brightness changes."""
sys_info = self.sys_info
return bool(sys_info["is_dimmable"])
@property # type: ignore
@requires_update
def is_variable_color_temp(self) -> bool:
"""Whether the bulb supports color temperature changes."""
sys_info = self.sys_info
return bool(sys_info["is_variable_color_temp"])
@property # type: ignore
@requires_update
def valid_temperature_range(self) -> ColorTempRange:
"""Return the device-specific white temperature range (in Kelvin).
:return: White temperature range in Kelvin (minimum, maximum)
"""
if not self.is_variable_color_temp:
raise SmartDeviceException("Color temperature not supported")
for model, temp_range in TPLINK_KELVIN.items():
sys_info = self.sys_info
if re.match(model, sys_info["model"]):
return temp_range
_LOGGER.warning("Unknown color temperature range, fallback to 2700-5000")
return ColorTempRange(2700, 5000)
@property # type: ignore
@requires_update
def light_state(self) -> Dict[str, str]:
"""Query the light state."""
light_state = self.sys_info["light_state"]
if light_state is None:
raise SmartDeviceException(
"The device has no light_state or you have not called update()"
)
# if the bulb is off, its state is stored under a different key
# as is_on property depends on on_off itself, we check it here manually
is_on = light_state["on_off"]
if not is_on:
off_state = {**light_state["dft_on_state"], "on_off": is_on}
return cast(dict, off_state)
return light_state
@property # type: ignore
@requires_update
def has_effects(self) -> bool:
"""Return True if the device supports effects."""
return "lighting_effect_state" in self.sys_info
async def get_light_details(self) -> Dict[str, int]:
"""Return light details.
Example::
{'lamp_beam_angle': 290, 'min_voltage': 220, 'max_voltage': 240,
'wattage': 5, 'incandescent_equivalent': 40, 'max_lumens': 450,
'color_rendering_index': 80}
"""
return await self._query_helper(self.LIGHT_SERVICE, "get_light_details")
async def get_turn_on_behavior(self) -> TurnOnBehaviors:
"""Return the behavior for turning the bulb on."""
return TurnOnBehaviors.parse_obj(
await self._query_helper(self.LIGHT_SERVICE, "get_default_behavior")
)
async def set_turn_on_behavior(self, behavior: TurnOnBehaviors):
"""Set the behavior for turning the bulb on.
If you do not want to manually construct the behavior object,
you should use :func:`get_turn_on_behavior` to get the current settings.
"""
return await self._query_helper(
self.LIGHT_SERVICE, "set_default_behavior", behavior.dict(by_alias=True)
)
async def get_light_state(self) -> Dict[str, Dict]:
"""Query the light state."""
# TODO: add warning and refer to use light.state?
return await self._query_helper(self.LIGHT_SERVICE, "get_light_state")
async def set_light_state(
self, state: Dict, *, transition: Optional[int] = None
) -> Dict:
"""Set the light state."""
if transition is not None:
state["transition_period"] = transition
# if no on/off is defined, turn on the light
if "on_off" not in state:
state["on_off"] = 1
# If we are turning on without any color mode flags,
# we do not want to set ignore_default to ensure
# we restore the previous state.
if state["on_off"] and NON_COLOR_MODE_FLAGS.issuperset(state):
state["ignore_default"] = 0
else:
# This is necessary to allow turning on into a specific state
state["ignore_default"] = 1
light_state = await self._query_helper(
self.LIGHT_SERVICE, self.SET_LIGHT_METHOD, state
)
return light_state
@property # type: ignore
@requires_update
def hsv(self) -> HSV:
"""Return the current HSV state of the bulb.
:return: hue, saturation and value (degrees, %, %)
"""
if not self.is_color:
raise SmartDeviceException("Bulb does not support color.")
light_state = cast(dict, self.light_state)
hue = light_state["hue"]
saturation = light_state["saturation"]
value = light_state["brightness"]
return HSV(hue, saturation, value)
@requires_update
async def set_hsv(
self,
hue: int,
saturation: int,
value: Optional[int] = None,
*,
transition: Optional[int] = None,
) -> Dict:
"""Set new HSV.
:param int hue: hue in degrees
:param int saturation: saturation in percentage [0,100]
:param int value: value in percentage [0, 100]
:param int transition: transition in milliseconds.
"""
if not self.is_color:
raise SmartDeviceException("Bulb does not support color.")
if not isinstance(hue, int) or not (0 <= hue <= 360):
raise ValueError(f"Invalid hue value: {hue} (valid range: 0-360)")
if not isinstance(saturation, int) or not (0 <= saturation <= 100):
raise ValueError(
f"Invalid saturation value: {saturation} (valid range: 0-100%)"
)
light_state = {
"hue": hue,
"saturation": saturation,
"color_temp": 0,
}
if value is not None:
self._raise_for_invalid_brightness(value)
light_state["brightness"] = value
return await self.set_light_state(light_state, transition=transition)
@property # type: ignore
@requires_update
def color_temp(self) -> int:
"""Return color temperature of the device in kelvin."""
if not self.is_variable_color_temp:
raise SmartDeviceException("Bulb does not support colortemp.")
light_state = self.light_state
return int(light_state["color_temp"])
@requires_update
async def set_color_temp(
self, temp: int, *, brightness=None, transition: Optional[int] = None
) -> Dict:
"""Set the color temperature of the device in kelvin.
:param int temp: The new color temperature, in Kelvin
:param int transition: transition in milliseconds.
"""
if not self.is_variable_color_temp:
raise SmartDeviceException("Bulb does not support colortemp.")
valid_temperature_range = self.valid_temperature_range
if temp < valid_temperature_range[0] or temp > valid_temperature_range[1]:
raise ValueError(
"Temperature should be between {} and {}, was {}".format(
*valid_temperature_range, temp
)
)
light_state = {"color_temp": temp}
if brightness is not None:
light_state["brightness"] = brightness
return await self.set_light_state(light_state, transition=transition)
@property # type: ignore
@requires_update
def brightness(self) -> int:
"""Return the current brightness in percentage."""
if not self.is_dimmable: # pragma: no cover
raise SmartDeviceException("Bulb is not dimmable.")
light_state = self.light_state
return int(light_state["brightness"])
@requires_update
async def set_brightness(
self, brightness: int, *, transition: Optional[int] = None
) -> Dict:
"""Set the brightness in percentage.
:param int brightness: brightness in percent
:param int transition: transition in milliseconds.
"""
if not self.is_dimmable: # pragma: no cover
raise SmartDeviceException("Bulb is not dimmable.")
self._raise_for_invalid_brightness(brightness)
light_state = {"brightness": brightness}
return await self.set_light_state(light_state, transition=transition)
@property # type: ignore
@requires_update
def state_information(self) -> Dict[str, Any]:
"""Return bulb-specific state information."""
info: Dict[str, Any] = {
"Brightness": self.brightness,
"Is dimmable": self.is_dimmable,
}
if self.is_variable_color_temp:
info["Color temperature"] = self.color_temp
info["Valid temperature range"] = self.valid_temperature_range
if self.is_color:
info["HSV"] = self.hsv
info["Presets"] = self.presets
return info
@property # type: ignore
@requires_update
def is_on(self) -> bool:
"""Return whether the device is on."""
light_state = self.light_state
return bool(light_state["on_off"])
async def turn_off(self, *, transition: Optional[int] = None, **kwargs) -> Dict:
"""Turn the bulb off.
:param int transition: transition in milliseconds.
"""
return await self.set_light_state({"on_off": 0}, transition=transition)
async def turn_on(self, *, transition: Optional[int] = None, **kwargs) -> Dict:
"""Turn the bulb on.
:param int transition: transition in milliseconds.
"""
return await self.set_light_state({"on_off": 1}, transition=transition)
@property # type: ignore
@requires_update
def has_emeter(self) -> bool:
"""Return that the bulb has an emeter."""
return True
async def set_alias(self, alias: str) -> None:
"""Set the device name (alias).
Overridden to use a different module name.
"""
return await self._query_helper(
"smartlife.iot.common.system", "set_dev_alias", {"alias": alias}
)
@property # type: ignore
@requires_update
def presets(self) -> List[BulbPreset]:
"""Return a list of available bulb setting presets."""
return [BulbPreset(**vals) for vals in self.sys_info["preferred_state"]]
async def save_preset(self, preset: BulbPreset):
"""Save a setting preset.
You can either construct a preset object manually, or pass an existing one
obtained using :func:`presets`.
"""
if len(self.presets) == 0:
raise SmartDeviceException("Device does not supported saving presets")
if preset.index >= len(self.presets):
raise SmartDeviceException("Invalid preset index")
return await self._query_helper(
self.LIGHT_SERVICE, "set_preferred_state", preset.dict(exclude_none=True)
)
@property
def max_device_response_size(self) -> int:
"""Returns the maximum response size the device can safely construct."""
return 4096

668
kasa/iot/iotdevice.py Executable file
View File

@@ -0,0 +1,668 @@
"""Python library supporting TP-Link Smart Home devices.
The communication protocol was reverse engineered by Lubomir Stroetmann and
Tobias Esser in 'Reverse Engineering the TP-Link HS110':
https://www.softscheck.com/en/reverse-engineering-tp-link-hs110/
This library reuses codes and concepts of the TP-Link WiFi SmartPlug Client
at https://github.com/softScheck/tplink-smartplug, developed by Lubomir
Stroetmann which is licensed under the Apache License, Version 2.0.
You may obtain a copy of the license at
http://www.apache.org/licenses/LICENSE-2.0
"""
import collections.abc
import functools
import inspect
import logging
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Sequence, Set
from ..device import Device, WifiNetwork
from ..deviceconfig import DeviceConfig
from ..emeterstatus import EmeterStatus
from ..exceptions import SmartDeviceException
from ..protocol import BaseProtocol
from .modules import Emeter, IotModule
_LOGGER = logging.getLogger(__name__)
def merge(d, u):
"""Update dict recursively."""
for k, v in u.items():
if isinstance(v, collections.abc.Mapping):
d[k] = merge(d.get(k, {}), v)
else:
d[k] = v
return d
def requires_update(f):
"""Indicate that `update` should be called before accessing this method.""" # noqa: D202
if inspect.iscoroutinefunction(f):
@functools.wraps(f)
async def wrapped(*args, **kwargs):
self = args[0]
if self._last_update is None and f.__name__ not in self._sys_info:
raise SmartDeviceException(
"You need to await update() to access the data"
)
return await f(*args, **kwargs)
else:
@functools.wraps(f)
def wrapped(*args, **kwargs):
self = args[0]
if self._last_update is None and f.__name__ not in self._sys_info:
raise SmartDeviceException(
"You need to await update() to access the data"
)
return f(*args, **kwargs)
f.requires_update = True
return wrapped
@functools.lru_cache
def _parse_features(features: str) -> Set[str]:
"""Parse features string."""
return set(features.split(":"))
class IotDevice(Device):
"""Base class for all supported device types.
You don't usually want to initialize this class manually,
but either use :class:`Discover` class, or use one of the subclasses:
* :class:`IotPlug`
* :class:`IotBulb`
* :class:`IotStrip`
* :class:`IotDimmer`
* :class:`IotLightStrip`
To initialize, you have to await :func:`update()` at least once.
This will allow accessing the properties using the exposed properties.
All changes to the device are done using awaitable methods,
which will not change the cached values, but you must await update() separately.
Errors reported by the device are raised as SmartDeviceExceptions,
and should be handled by the user of the library.
Examples:
>>> import asyncio
>>> dev = IotDevice("127.0.0.1")
>>> asyncio.run(dev.update())
All devices provide several informational properties:
>>> dev.alias
Kitchen
>>> dev.model
HS110(EU)
>>> dev.rssi
-71
>>> dev.mac
50:C7:BF:00:00:00
Some information can also be changed programmatically:
>>> asyncio.run(dev.set_alias("new alias"))
>>> asyncio.run(dev.set_mac("01:23:45:67:89:ab"))
>>> asyncio.run(dev.update())
>>> dev.alias
new alias
>>> dev.mac
01:23:45:67:89:ab
When initialized using discovery or using a subclass,
you can check the type of the device:
>>> dev.is_bulb
False
>>> dev.is_strip
False
>>> dev.is_plug
True
You can also get the hardware and software as a dict,
or access the full device response:
>>> dev.hw_info
{'sw_ver': '1.2.5 Build 171213 Rel.101523',
'hw_ver': '1.0',
'mac': '01:23:45:67:89:ab',
'type': 'IOT.SMARTPLUGSWITCH',
'hwId': '00000000000000000000000000000000',
'fwId': '00000000000000000000000000000000',
'oemId': '00000000000000000000000000000000',
'dev_name': 'Wi-Fi Smart Plug With Energy Monitoring'}
>>> dev.sys_info
All devices can be turned on and off:
>>> asyncio.run(dev.turn_off())
>>> asyncio.run(dev.turn_on())
>>> asyncio.run(dev.update())
>>> dev.is_on
True
Some devices provide energy consumption meter,
and regular update will already fetch some information:
>>> dev.has_emeter
True
>>> dev.emeter_realtime
<EmeterStatus power=0.928511 voltage=231.067823 current=0.014937 total=55.139>
>>> dev.emeter_today
>>> dev.emeter_this_month
You can also query the historical data (note that these needs to be awaited),
keyed with month/day:
>>> asyncio.run(dev.get_emeter_monthly(year=2016))
{11: 1.089, 12: 1.582}
>>> asyncio.run(dev.get_emeter_daily(year=2016, month=11))
{24: 0.026, 25: 0.109}
"""
emeter_type = "emeter"
def __init__(
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
) -> None:
"""Create a new IotDevice instance."""
super().__init__(host=host, config=config, protocol=protocol)
self._sys_info: Any = None # TODO: this is here to avoid changing tests
self._features: Set[str] = set()
self._children: Sequence["IotDevice"] = []
@property
def children(self) -> Sequence["IotDevice"]:
"""Return list of children."""
return self._children
@children.setter
def children(self, children):
"""Initialize from a list of children."""
self._children = children
def add_module(self, name: str, module: IotModule):
"""Register a module."""
if name in self.modules:
_LOGGER.debug("Module %s already registered, ignoring..." % name)
return
_LOGGER.debug("Adding module %s", module)
self.modules[name] = module
def _create_request(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
):
request: Dict[str, Any] = {target: {cmd: arg}}
if child_ids is not None:
request = {"context": {"child_ids": child_ids}, target: {cmd: arg}}
return request
def _verify_emeter(self) -> None:
"""Raise an exception if there is no emeter."""
if not self.has_emeter:
raise SmartDeviceException("Device has no emeter")
if self.emeter_type not in self._last_update:
raise SmartDeviceException("update() required prior accessing emeter")
async def _query_helper(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
) -> Any:
"""Query device, return results or raise an exception.
:param target: Target system {system, time, emeter, ..}
:param cmd: Command to execute
:param arg: payload dict to be send to the device
:param child_ids: ids of child devices
:return: Unwrapped result for the call.
"""
request = self._create_request(target, cmd, arg, child_ids)
try:
response = await self._raw_query(request=request)
except Exception as ex:
raise SmartDeviceException(f"Communication error on {target}:{cmd}") from ex
if target not in response:
raise SmartDeviceException(f"No required {target} in response: {response}")
result = response[target]
if "err_code" in result and result["err_code"] != 0:
raise SmartDeviceException(f"Error on {target}.{cmd}: {result}")
if cmd not in result:
raise SmartDeviceException(f"No command in response: {response}")
result = result[cmd]
if "err_code" in result and result["err_code"] != 0:
raise SmartDeviceException(f"Error on {target} {cmd}: {result}")
if "err_code" in result:
del result["err_code"]
return result
@property # type: ignore
@requires_update
def features(self) -> Set[str]:
"""Return a set of features that the device supports."""
return self._features
@property # type: ignore
@requires_update
def supported_modules(self) -> List[str]:
"""Return a set of modules supported by the device."""
# TODO: this should rather be called `features`, but we don't want to break
# the API now. Maybe just deprecate it and point the users to use this?
return list(self.modules.keys())
@property # type: ignore
@requires_update
def has_emeter(self) -> bool:
"""Return True if device has an energy meter."""
return "ENE" in self.features
async def get_sys_info(self) -> Dict[str, Any]:
"""Retrieve system information."""
return await self._query_helper("system", "get_sysinfo")
async def update(self, update_children: bool = True):
"""Query the device to update the data.
Needed for properties that are decorated with `requires_update`.
"""
req = {}
req.update(self._create_request("system", "get_sysinfo"))
# If this is the initial update, check only for the sysinfo
# This is necessary as some devices crash on unexpected modules
# See #105, #120, #161
if self._last_update is None:
_LOGGER.debug("Performing the initial update to obtain sysinfo")
response = await self.protocol.query(req)
self._last_update = response
self._set_sys_info(response["system"]["get_sysinfo"])
await self._modular_update(req)
self._set_sys_info(self._last_update["system"]["get_sysinfo"])
async def _modular_update(self, req: dict) -> None:
"""Execute an update query."""
if self.has_emeter:
_LOGGER.debug(
"The device has emeter, querying its information along sysinfo"
)
self.add_module("emeter", Emeter(self, self.emeter_type))
request_list = []
est_response_size = 1024 if "system" in req else 0
for module in self.modules.values():
if not module.is_supported:
_LOGGER.debug("Module %s not supported, skipping" % module)
continue
est_response_size += module.estimated_query_response_size
if est_response_size > self.max_device_response_size:
request_list.append(req)
req = {}
est_response_size = module.estimated_query_response_size
q = module.query()
_LOGGER.debug("Adding query for %s: %s", module, q)
req = merge(req, q)
request_list.append(req)
responses = [
await self.protocol.query(request) for request in request_list if request
]
# Preserve the last update and merge
# responses on top of it so we remember
# which modules are not supported, otherwise
# every other update will query for them
update: Dict = self._last_update.copy() if self._last_update else {}
for response in responses:
update = {**update, **response}
self._last_update = update
def update_from_discover_info(self, info: Dict[str, Any]) -> None:
"""Update state from info from the discover call."""
self._discovery_info = info
if "system" in info and (sys_info := info["system"].get("get_sysinfo")):
self._last_update = info
self._set_sys_info(sys_info)
else:
# This allows setting of some info properties directly
# from partial discovery info that will then be found
# by the requires_update decorator
self._set_sys_info(info)
def _set_sys_info(self, sys_info: Dict[str, Any]) -> None:
"""Set sys_info."""
self._sys_info = sys_info
if features := sys_info.get("feature"):
self._features = _parse_features(features)
else:
self._features = set()
@property # type: ignore
@requires_update
def sys_info(self) -> Dict[str, Any]:
"""
Return system information.
Do not call this function from within the SmartDevice
class itself as @requires_update will be affected for other properties.
"""
return self._sys_info # type: ignore
@property # type: ignore
@requires_update
def model(self) -> str:
"""Return device model."""
sys_info = self._sys_info
return str(sys_info["model"])
@property
def has_children(self) -> bool:
"""Return true if the device has children devices."""
# Ideally we would check for the 'child_num' key in sys_info,
# but devices that speak klap do not populate this key via
# update_from_discover_info so we check for the devices
# we know have children instead.
return self.is_strip
@property # type: ignore
def alias(self) -> Optional[str]:
"""Return device name (alias)."""
sys_info = self._sys_info
return sys_info.get("alias") if sys_info else None
async def set_alias(self, alias: str) -> None:
"""Set the device name (alias)."""
return await self._query_helper("system", "set_dev_alias", {"alias": alias})
@property # type: ignore
@requires_update
def time(self) -> datetime:
"""Return current time from the device."""
return self.modules["time"].time
@property # type: ignore
@requires_update
def timezone(self) -> Dict:
"""Return the current timezone."""
return self.modules["time"].timezone
async def get_time(self) -> Optional[datetime]:
"""Return current time from the device, if available."""
_LOGGER.warning(
"Use `time` property instead, this call will be removed in the future."
)
return await self.modules["time"].get_time()
async def get_timezone(self) -> Dict:
"""Return timezone information."""
_LOGGER.warning(
"Use `timezone` property instead, this call will be removed in the future."
)
return await self.modules["time"].get_timezone()
@property # type: ignore
@requires_update
def hw_info(self) -> Dict:
"""Return hardware information.
This returns just a selection of sysinfo keys that are related to hardware.
"""
keys = [
"sw_ver",
"hw_ver",
"mac",
"mic_mac",
"type",
"mic_type",
"hwId",
"fwId",
"oemId",
"dev_name",
]
sys_info = self._sys_info
return {key: sys_info[key] for key in keys if key in sys_info}
@property # type: ignore
@requires_update
def location(self) -> Dict:
"""Return geographical location."""
sys_info = self._sys_info
loc = {"latitude": None, "longitude": None}
if "latitude" in sys_info and "longitude" in sys_info:
loc["latitude"] = sys_info["latitude"]
loc["longitude"] = sys_info["longitude"]
elif "latitude_i" in sys_info and "longitude_i" in sys_info:
loc["latitude"] = sys_info["latitude_i"] / 10000
loc["longitude"] = sys_info["longitude_i"] / 10000
else:
_LOGGER.debug("Unsupported device location.")
return loc
@property # type: ignore
@requires_update
def rssi(self) -> Optional[int]:
"""Return WiFi signal strength (rssi)."""
rssi = self._sys_info.get("rssi")
return None if rssi is None else int(rssi)
@property # type: ignore
@requires_update
def mac(self) -> str:
"""Return mac address.
:return: mac address in hexadecimal with colons, e.g. 01:23:45:67:89:ab
"""
sys_info = self._sys_info
mac = sys_info.get("mac", sys_info.get("mic_mac"))
if not mac:
raise SmartDeviceException(
"Unknown mac, please submit a bug report with sys_info output."
)
mac = mac.replace("-", ":")
# Format a mac that has no colons (usually from mic_mac field)
if ":" not in mac:
mac = ":".join(format(s, "02x") for s in bytes.fromhex(mac))
return mac
async def set_mac(self, mac):
"""Set the mac address.
:param str mac: mac in hexadecimal with colons, e.g. 01:23:45:67:89:ab
"""
return await self._query_helper("system", "set_mac_addr", {"mac": mac})
@property # type: ignore
@requires_update
def emeter_realtime(self) -> EmeterStatus:
"""Return current energy readings."""
self._verify_emeter()
return EmeterStatus(self.modules["emeter"].realtime)
async def get_emeter_realtime(self) -> EmeterStatus:
"""Retrieve current energy readings."""
self._verify_emeter()
return EmeterStatus(await self.modules["emeter"].get_realtime())
@property # type: ignore
@requires_update
def emeter_today(self) -> Optional[float]:
"""Return today's energy consumption in kWh."""
self._verify_emeter()
return self.modules["emeter"].emeter_today
@property # type: ignore
@requires_update
def emeter_this_month(self) -> Optional[float]:
"""Return this month's energy consumption in kWh."""
self._verify_emeter()
return self.modules["emeter"].emeter_this_month
async def get_emeter_daily(
self, year: Optional[int] = None, month: Optional[int] = None, kwh: bool = True
) -> Dict:
"""Retrieve daily statistics for a given month.
:param year: year for which to retrieve statistics (default: this year)
:param month: month for which to retrieve statistics (default: this
month)
:param kwh: return usage in kWh (default: True)
:return: mapping of day of month to value
"""
self._verify_emeter()
return await self.modules["emeter"].get_daystat(year=year, month=month, kwh=kwh)
@requires_update
async def get_emeter_monthly(
self, year: Optional[int] = None, kwh: bool = True
) -> Dict:
"""Retrieve monthly statistics for a given year.
:param year: year for which to retrieve statistics (default: this year)
:param kwh: return usage in kWh (default: True)
:return: dict: mapping of month to value
"""
self._verify_emeter()
return await self.modules["emeter"].get_monthstat(year=year, kwh=kwh)
@requires_update
async def erase_emeter_stats(self) -> Dict:
"""Erase energy meter statistics."""
self._verify_emeter()
return await self.modules["emeter"].erase_stats()
@requires_update
async def current_consumption(self) -> float:
"""Get the current power consumption in Watt."""
self._verify_emeter()
response = self.emeter_realtime
return float(response["power"])
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.
"""
await self._query_helper("system", "reboot", {"delay": delay})
async def turn_off(self, **kwargs) -> Dict:
"""Turn off the device."""
raise NotImplementedError("Device subclass needs to implement this.")
async def turn_on(self, **kwargs) -> Optional[Dict]:
"""Turn device on."""
raise NotImplementedError("Device subclass needs to implement this.")
@property # type: ignore
@requires_update
def is_on(self) -> bool:
"""Return True if the device is on."""
raise NotImplementedError("Device subclass needs to implement this.")
@property # type: ignore
@requires_update
def on_since(self) -> Optional[datetime]:
"""Return pretty-printed on-time, or None if not available."""
if "on_time" not in self._sys_info:
return None
if self.is_off:
return None
on_time = self._sys_info["on_time"]
return datetime.now().replace(microsecond=0) - timedelta(seconds=on_time)
@property # type: ignore
@requires_update
def state_information(self) -> Dict[str, Any]:
"""Return device-type specific, end-user friendly state information."""
raise NotImplementedError("Device subclass needs to implement this.")
@property # type: ignore
@requires_update
def device_id(self) -> str:
"""Return unique ID for the device.
If not overridden, this is the MAC address of the device.
Individual sockets on strips will override this.
"""
return self.mac
async def wifi_scan(self) -> List[WifiNetwork]: # noqa: D202
"""Scan for available wifi networks."""
async def _scan(target):
return await self._query_helper(target, "get_scaninfo", {"refresh": 1})
try:
info = await _scan("netif")
except SmartDeviceException as ex:
_LOGGER.debug(
"Unable to scan using 'netif', retrying with 'softaponboarding': %s", ex
)
info = await _scan("smartlife.iot.common.softaponboarding")
if "ap_list" not in info:
raise SmartDeviceException("Invalid response for wifi scan: %s" % info)
return [WifiNetwork(**x) for x in info["ap_list"]]
async def wifi_join(self, ssid: str, password: str, keytype: str = "3"): # noqa: D202
"""Join the given wifi network.
If joining the network fails, the device will return to AP mode after a while.
"""
async def _join(target, payload):
return await self._query_helper(target, "set_stainfo", payload)
payload = {"ssid": ssid, "password": password, "key_type": int(keytype)}
try:
return await _join("netif", payload)
except SmartDeviceException as ex:
_LOGGER.debug(
"Unable to join using 'netif', retrying with 'softaponboarding': %s", ex
)
return await _join("smartlife.iot.common.softaponboarding", payload)
@property
def max_device_response_size(self) -> int:
"""Returns the maximum response size the device can safely construct."""
return 16 * 1024
@property
def internal_state(self) -> Any:
"""Return the internal state of the instance.
The returned object contains the raw results from the last update call.
This should only be used for debugging purposes.
"""
return self._last_update or self._discovery_info

226
kasa/iot/iotdimmer.py Normal file
View File

@@ -0,0 +1,226 @@
"""Module for dimmers (currently only HS220)."""
from enum import Enum
from typing import Any, Dict, Optional
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..protocol import BaseProtocol
from .iotdevice import SmartDeviceException, requires_update
from .iotplug import IotPlug
from .modules import AmbientLight, Motion
class ButtonAction(Enum):
"""Button action."""
NoAction = "none"
Instant = "instant_on_off"
Gentle = "gentle_on_off"
Preset = "customize_preset"
class ActionType(Enum):
"""Button action."""
DoubleClick = "double_click_action"
LongPress = "long_press_action"
class FadeType(Enum):
"""Fade on/off setting."""
FadeOn = "fade_on"
FadeOff = "fade_off"
class IotDimmer(IotPlug):
r"""Representation of a TP-Link Smart Dimmer.
Dimmers work similarly to plugs, but provide also support for
adjusting the brightness. This class extends :class:`SmartPlug` interface.
To initialize, you have to await :func:`update()` at least once.
This will allow accessing the properties using the exposed properties.
All changes to the device are done using awaitable methods,
which will not change the cached values,
but you must await :func:`update()` separately.
Errors reported by the device are raised as :class:`SmartDeviceException`\s,
and should be handled by the user of the library.
Examples:
>>> import asyncio
>>> dimmer = IotDimmer("192.168.1.105")
>>> asyncio.run(dimmer.turn_on())
>>> dimmer.brightness
25
>>> asyncio.run(dimmer.set_brightness(50))
>>> asyncio.run(dimmer.update())
>>> dimmer.brightness
50
Refer to :class:`SmartPlug` for the full API.
"""
DIMMER_SERVICE = "smartlife.iot.dimmer"
def __init__(
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Dimmer
# TODO: need to be verified if it's okay to call these on HS220 w/o these
# TODO: need to be figured out what's the best approach to detect support
self.add_module("motion", Motion(self, "smartlife.iot.PIR"))
self.add_module("ambient", AmbientLight(self, "smartlife.iot.LAS"))
@property # type: ignore
@requires_update
def brightness(self) -> int:
"""Return current brightness on dimmers.
Will return a range between 0 - 100.
"""
if not self.is_dimmable:
raise SmartDeviceException("Device is not dimmable.")
sys_info = self.sys_info
return int(sys_info["brightness"])
@requires_update
async def set_brightness(
self, brightness: int, *, transition: Optional[int] = None
):
"""Set the new dimmer brightness level in percentage.
:param int transition: transition duration in milliseconds.
Using a transition will cause the dimmer to turn on.
"""
if not self.is_dimmable:
raise SmartDeviceException("Device is not dimmable.")
if not isinstance(brightness, int):
raise ValueError(
"Brightness must be integer, " "not of %s.", type(brightness)
)
if not 0 <= brightness <= 100:
raise ValueError("Brightness value %s is not valid." % brightness)
# Dimmers do not support a brightness of 0, but bulbs do.
# Coerce 0 to 1 to maintain the same interface between dimmers and bulbs.
if brightness == 0:
brightness = 1
if transition is not None:
return await self.set_dimmer_transition(brightness, transition)
return await self._query_helper(
self.DIMMER_SERVICE, "set_brightness", {"brightness": brightness}
)
async def turn_off(self, *, transition: Optional[int] = None, **kwargs):
"""Turn the bulb off.
:param int transition: transition duration in milliseconds.
"""
if transition is not None:
return await self.set_dimmer_transition(brightness=0, transition=transition)
return await super().turn_off()
@requires_update
async def turn_on(self, *, transition: Optional[int] = None, **kwargs):
"""Turn the bulb on.
:param int transition: transition duration in milliseconds.
"""
if transition is not None:
return await self.set_dimmer_transition(
brightness=self.brightness, transition=transition
)
return await super().turn_on()
async def set_dimmer_transition(self, brightness: int, transition: int):
"""Turn the bulb on to brightness percentage over transition milliseconds.
A brightness value of 0 will turn off the dimmer.
"""
if not isinstance(brightness, int):
raise ValueError(
"Brightness must be integer, " "not of %s.", type(brightness)
)
if not 0 <= brightness <= 100:
raise ValueError("Brightness value %s is not valid." % brightness)
if not isinstance(transition, int):
raise ValueError(
"Transition must be integer, " "not of %s.", type(transition)
)
if transition <= 0:
raise ValueError("Transition value %s is not valid." % transition)
return await self._query_helper(
self.DIMMER_SERVICE,
"set_dimmer_transition",
{"brightness": brightness, "duration": transition},
)
@requires_update
async def get_behaviors(self):
"""Return button behavior settings."""
behaviors = await self._query_helper(
self.DIMMER_SERVICE, "get_default_behavior", {}
)
return behaviors
@requires_update
async def set_button_action(
self, action_type: ActionType, action: ButtonAction, index: Optional[int] = None
):
"""Set action to perform on button click/hold.
:param action_type ActionType: whether to control double click or hold action.
:param action ButtonAction: what should the button do
(nothing, instant, gentle, change preset)
:param index int: in case of preset change, the preset to select
"""
action_type_setter = f"set_{action_type}"
payload: Dict[str, Any] = {"mode": str(action)}
if index is not None:
payload["index"] = index
await self._query_helper(self.DIMMER_SERVICE, action_type_setter, payload)
@requires_update
async def set_fade_time(self, fade_type: FadeType, time: int):
"""Set time for fade in / fade out."""
fade_type_setter = f"set_{fade_type}_time"
payload = {"fadeTime": time}
await self._query_helper(self.DIMMER_SERVICE, fade_type_setter, payload)
@property # type: ignore
@requires_update
def is_dimmable(self) -> bool:
"""Whether the switch supports brightness changes."""
sys_info = self.sys_info
return "brightness" in sys_info
@property # type: ignore
@requires_update
def state_information(self) -> Dict[str, Any]:
"""Return switch-specific state information."""
info = super().state_information
info["Brightness"] = self.brightness
return info

144
kasa/iot/iotlightstrip.py Normal file
View File

@@ -0,0 +1,144 @@
"""Module for light strips (KL430)."""
from typing import Any, Dict, List, Optional
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..effects import EFFECT_MAPPING_V1, EFFECT_NAMES_V1
from ..protocol import BaseProtocol
from .iotbulb import IotBulb
from .iotdevice import SmartDeviceException, requires_update
class IotLightStrip(IotBulb):
"""Representation of a TP-Link Smart light strip.
Light strips work similarly to bulbs, but use a different service for controlling,
and expose some extra information (such as length and active effect).
This class extends :class:`SmartBulb` interface.
Examples:
>>> import asyncio
>>> strip = IotLightStrip("127.0.0.1")
>>> asyncio.run(strip.update())
>>> print(strip.alias)
KL430 pantry lightstrip
Getting the length of the strip:
>>> strip.length
16
Currently active effect:
>>> strip.effect
{'brightness': 50, 'custom': 0, 'enable': 0, 'id': '', 'name': ''}
.. note::
The device supports some features that are not currently implemented,
feel free to find out how to control them and create a PR!
See :class:`SmartBulb` for more examples.
"""
LIGHT_SERVICE = "smartlife.iot.lightStrip"
SET_LIGHT_METHOD = "set_light_state"
def __init__(
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.LightStrip
@property # type: ignore
@requires_update
def length(self) -> int:
"""Return length of the strip."""
return self.sys_info["length"]
@property # type: ignore
@requires_update
def effect(self) -> Dict:
"""Return effect state.
Example:
{'brightness': 50,
'custom': 0,
'enable': 0,
'id': '',
'name': ''}
"""
return self.sys_info["lighting_effect_state"]
@property # type: ignore
@requires_update
def effect_list(self) -> Optional[List[str]]:
"""Return built-in effects list.
Example:
['Aurora', 'Bubbling Cauldron', ...]
"""
return EFFECT_NAMES_V1 if self.has_effects else None
@property # type: ignore
@requires_update
def state_information(self) -> Dict[str, Any]:
"""Return strip specific state information."""
info = super().state_information
info["Length"] = self.length
if self.has_effects:
info["Effect"] = self.effect["name"]
return info
@requires_update
async def set_effect(
self,
effect: str,
*,
brightness: Optional[int] = None,
transition: Optional[int] = None,
) -> None:
"""Set an effect on the device.
If brightness or transition is defined,
its value will be used instead of the effect-specific default.
See :meth:`effect_list` for available effects,
or use :meth:`set_custom_effect` for custom effects.
:param str effect: The effect to set
:param int brightness: The wanted brightness
:param int transition: The wanted transition time
"""
if effect not in EFFECT_MAPPING_V1:
raise SmartDeviceException(f"The effect {effect} is not a built in effect.")
effect_dict = EFFECT_MAPPING_V1[effect]
if brightness is not None:
effect_dict["brightness"] = brightness
if transition is not None:
effect_dict["transition"] = transition
await self.set_custom_effect(effect_dict)
@requires_update
async def set_custom_effect(
self,
effect_dict: Dict,
) -> None:
"""Set a custom effect on the device.
:param str effect_dict: The custom effect dict to set
"""
if not self.has_effects:
raise SmartDeviceException("Bulb does not support effects.")
await self._query_helper(
"smartlife.iot.lighting_effect",
"set_lighting_effect",
effect_dict,
)

92
kasa/iot/iotplug.py Normal file
View File

@@ -0,0 +1,92 @@
"""Module for smart plugs (HS100, HS110, ..)."""
import logging
from typing import Any, Dict, Optional
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..protocol import BaseProtocol
from .iotdevice import IotDevice, requires_update
from .modules import Antitheft, Cloud, Schedule, Time, Usage
_LOGGER = logging.getLogger(__name__)
class IotPlug(IotDevice):
r"""Representation of a TP-Link Smart Switch.
To initialize, you have to await :func:`update()` at least once.
This will allow accessing the properties using the exposed properties.
All changes to the device are done using awaitable methods,
which will not change the cached values,
but you must await :func:`update()` separately.
Errors reported by the device are raised as :class:`SmartDeviceException`\s,
and should be handled by the user of the library.
Examples:
>>> import asyncio
>>> plug = IotPlug("127.0.0.1")
>>> asyncio.run(plug.update())
>>> plug.alias
Kitchen
Setting the LED state:
>>> asyncio.run(plug.set_led(True))
>>> asyncio.run(plug.update())
>>> plug.led
True
For more examples, see the :class:`SmartDevice` class.
"""
def __init__(
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Plug
self.add_module("schedule", Schedule(self, "schedule"))
self.add_module("usage", Usage(self, "schedule"))
self.add_module("antitheft", Antitheft(self, "anti_theft"))
self.add_module("time", Time(self, "time"))
self.add_module("cloud", Cloud(self, "cnCloud"))
@property # type: ignore
@requires_update
def is_on(self) -> bool:
"""Return whether device is on."""
sys_info = self.sys_info
return bool(sys_info["relay_state"])
async def turn_on(self, **kwargs):
"""Turn the switch on."""
return await self._query_helper("system", "set_relay_state", {"state": 1})
async def turn_off(self, **kwargs):
"""Turn the switch off."""
return await self._query_helper("system", "set_relay_state", {"state": 0})
@property # type: ignore
@requires_update
def led(self) -> bool:
"""Return the state of the led."""
sys_info = self.sys_info
return bool(1 - sys_info["led_off"])
async def set_led(self, state: bool):
"""Set the state of the led (night mode)."""
return await self._query_helper(
"system", "set_led_off", {"off": int(not state)}
)
@property # type: ignore
@requires_update
def state_information(self) -> Dict[str, Any]:
"""Return switch-specific state information."""
info = {"LED state": self.led, "On since": self.on_since}
return info

378
kasa/iot/iotstrip.py Executable file
View File

@@ -0,0 +1,378 @@
"""Module for multi-socket devices (HS300, HS107, KP303, ..)."""
import logging
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any, DefaultDict, Dict, Optional
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..exceptions import SmartDeviceException
from ..protocol import BaseProtocol
from .iotdevice import (
EmeterStatus,
IotDevice,
merge,
requires_update,
)
from .iotplug import IotPlug
from .modules import Antitheft, Countdown, Emeter, Schedule, Time, Usage
_LOGGER = logging.getLogger(__name__)
def merge_sums(dicts):
"""Merge the sum of dicts."""
total_dict: DefaultDict[int, float] = defaultdict(lambda: 0.0)
for sum_dict in dicts:
for day, value in sum_dict.items():
total_dict[day] += value
return total_dict
class IotStrip(IotDevice):
r"""Representation of a TP-Link Smart Power Strip.
A strip consists of the parent device and its children.
All methods of the parent act on all children, while the child devices
share the common API with the :class:`SmartPlug` class.
To initialize, you have to await :func:`update()` at least once.
This will allow accessing the properties using the exposed properties.
All changes to the device are done using awaitable methods,
which will not change the cached values,
but you must await :func:`update()` separately.
Errors reported by the device are raised as :class:`SmartDeviceException`\s,
and should be handled by the user of the library.
Examples:
>>> import asyncio
>>> strip = IotStrip("127.0.0.1")
>>> asyncio.run(strip.update())
>>> strip.alias
TP-LINK_Power Strip_CF69
All methods act on the whole strip:
>>> for plug in strip.children:
>>> print(f"{plug.alias}: {plug.is_on}")
Plug 1: True
Plug 2: False
Plug 3: False
>>> strip.is_on
True
>>> asyncio.run(strip.turn_off())
Accessing individual plugs can be done using the `children` property:
>>> len(strip.children)
3
>>> for plug in strip.children:
>>> print(f"{plug.alias}: {plug.is_on}")
Plug 1: False
Plug 2: False
Plug 3: False
>>> asyncio.run(strip.children[1].turn_on())
>>> asyncio.run(strip.update())
>>> strip.is_on
True
For more examples, see the :class:`SmartDevice` class.
"""
def __init__(
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self.emeter_type = "emeter"
self._device_type = DeviceType.Strip
self.add_module("antitheft", Antitheft(self, "anti_theft"))
self.add_module("schedule", Schedule(self, "schedule"))
self.add_module("usage", Usage(self, "schedule"))
self.add_module("time", Time(self, "time"))
self.add_module("countdown", Countdown(self, "countdown"))
self.add_module("emeter", Emeter(self, "emeter"))
@property # type: ignore
@requires_update
def is_on(self) -> bool:
"""Return if any of the outlets are on."""
return any(plug.is_on for plug in self.children)
async def update(self, update_children: bool = True):
"""Update some of the attributes.
Needed for methods that are decorated with `requires_update`.
"""
await super().update(update_children)
# Initialize the child devices during the first update.
if not self.children:
children = self.sys_info["children"]
_LOGGER.debug("Initializing %s child sockets", len(children))
self.children = [
IotStripPlug(self.host, parent=self, child_id=child["id"])
for child in children
]
if update_children and self.has_emeter:
for plug in self.children:
await plug.update()
async def turn_on(self, **kwargs):
"""Turn the strip on."""
await self._query_helper("system", "set_relay_state", {"state": 1})
async def turn_off(self, **kwargs):
"""Turn the strip off."""
await self._query_helper("system", "set_relay_state", {"state": 0})
@property # type: ignore
@requires_update
def on_since(self) -> Optional[datetime]:
"""Return the maximum on-time of all outlets."""
if self.is_off:
return None
return max(plug.on_since for plug in self.children if plug.on_since is not None)
@property # type: ignore
@requires_update
def led(self) -> bool:
"""Return the state of the led."""
sys_info = self.sys_info
return bool(1 - sys_info["led_off"])
async def set_led(self, state: bool):
"""Set the state of the led (night mode)."""
await self._query_helper("system", "set_led_off", {"off": int(not state)})
@property # type: ignore
@requires_update
def state_information(self) -> Dict[str, Any]:
"""Return strip-specific state information.
:return: Strip information dict, keys in user-presentable form.
"""
return {
"LED state": self.led,
"Childs count": len(self.children),
"On since": self.on_since,
}
async def current_consumption(self) -> float:
"""Get the current power consumption in watts."""
return sum([await plug.current_consumption() for plug in self.children])
@requires_update
async def get_emeter_realtime(self) -> EmeterStatus:
"""Retrieve current energy readings."""
emeter_rt = await self._async_get_emeter_sum("get_emeter_realtime", {})
# Voltage is averaged since each read will result
# in a slightly different voltage since they are not atomic
emeter_rt["voltage_mv"] = int(emeter_rt["voltage_mv"] / len(self.children))
return EmeterStatus(emeter_rt)
@requires_update
async def get_emeter_daily(
self, year: Optional[int] = None, month: Optional[int] = None, kwh: bool = True
) -> Dict:
"""Retrieve daily statistics for a given month.
:param year: year for which to retrieve statistics (default: this year)
:param month: month for which to retrieve statistics (default: this
month)
:param kwh: return usage in kWh (default: True)
:return: mapping of day of month to value
"""
return await self._async_get_emeter_sum(
"get_emeter_daily", {"year": year, "month": month, "kwh": kwh}
)
@requires_update
async def get_emeter_monthly(
self, year: Optional[int] = None, kwh: bool = True
) -> Dict:
"""Retrieve monthly statistics for a given year.
:param year: year for which to retrieve statistics (default: this year)
:param kwh: return usage in kWh (default: True)
"""
return await self._async_get_emeter_sum(
"get_emeter_monthly", {"year": year, "kwh": kwh}
)
async def _async_get_emeter_sum(self, func: str, kwargs: Dict[str, Any]) -> Dict:
"""Retreive emeter stats for a time period from children."""
self._verify_emeter()
return merge_sums(
[await getattr(plug, func)(**kwargs) for plug in self.children]
)
@requires_update
async def erase_emeter_stats(self):
"""Erase energy meter statistics for all plugs."""
for plug in self.children:
await plug.erase_emeter_stats()
@property # type: ignore
@requires_update
def emeter_this_month(self) -> Optional[float]:
"""Return this month's energy consumption in kWh."""
return sum(plug.emeter_this_month for plug in self.children)
@property # type: ignore
@requires_update
def emeter_today(self) -> Optional[float]:
"""Return this month's energy consumption in kWh."""
return sum(plug.emeter_today for plug in self.children)
@property # type: ignore
@requires_update
def emeter_realtime(self) -> EmeterStatus:
"""Return current energy readings."""
emeter = merge_sums([plug.emeter_realtime for plug in self.children])
# Voltage is averaged since each read will result
# in a slightly different voltage since they are not atomic
emeter["voltage_mv"] = int(emeter["voltage_mv"] / len(self.children))
return EmeterStatus(emeter)
class IotStripPlug(IotPlug):
"""Representation of a single socket in a power strip.
This allows you to use the sockets as they were SmartPlug objects.
Instead of calling an update on any of these, you should call an update
on the parent device before accessing the properties.
The plug inherits (most of) the system information from the parent.
"""
def __init__(self, host: str, parent: "IotStrip", child_id: str) -> None:
super().__init__(host)
self.parent = parent
self.child_id = child_id
self._last_update = parent._last_update
self._set_sys_info(parent.sys_info)
self._device_type = DeviceType.StripSocket
self.modules = {}
self.protocol = parent.protocol # Must use the same connection as the parent
self.add_module("time", Time(self, "time"))
async def update(self, update_children: bool = True):
"""Query the device to update the data.
Needed for properties that are decorated with `requires_update`.
"""
await self._modular_update({})
def _create_emeter_request(
self, year: Optional[int] = None, month: Optional[int] = None
):
"""Create a request for requesting all emeter statistics at once."""
if year is None:
year = datetime.now().year
if month is None:
month = datetime.now().month
req: Dict[str, Any] = {}
merge(req, self._create_request("emeter", "get_realtime"))
merge(req, self._create_request("emeter", "get_monthstat", {"year": year}))
merge(
req,
self._create_request(
"emeter", "get_daystat", {"month": month, "year": year}
),
)
return req
def _create_request(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
):
request: Dict[str, Any] = {
"context": {"child_ids": [self.child_id]},
target: {cmd: arg},
}
return request
async def _query_helper(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
) -> Any:
"""Override query helper to include the child_ids."""
return await self.parent._query_helper(
target, cmd, arg, child_ids=[self.child_id]
)
@property # type: ignore
@requires_update
def is_on(self) -> bool:
"""Return whether device is on."""
info = self._get_child_info()
return bool(info["state"])
@property # type: ignore
@requires_update
def led(self) -> bool:
"""Return the state of the led.
This is always false for subdevices.
"""
return False
@property # type: ignore
@requires_update
def device_id(self) -> str:
"""Return unique ID for the socket.
This is a combination of MAC and child's ID.
"""
return f"{self.mac}_{self.child_id}"
@property # type: ignore
@requires_update
def alias(self) -> str:
"""Return device name (alias)."""
info = self._get_child_info()
return info["alias"]
@property # type: ignore
@requires_update
def next_action(self) -> Dict:
"""Return next scheduled(?) action."""
info = self._get_child_info()
return info["next_action"]
@property # type: ignore
@requires_update
def on_since(self) -> Optional[datetime]:
"""Return on-time, if available."""
if self.is_off:
return None
info = self._get_child_info()
on_time = info["on_time"]
return datetime.now().replace(microsecond=0) - timedelta(seconds=on_time)
@property # type: ignore
@requires_update
def model(self) -> str:
"""Return device model for a child socket."""
sys_info = self.parent.sys_info
return f"Socket for {sys_info['model']}"
def _get_child_info(self) -> Dict:
"""Return the subdevice information for this device."""
for plug in self.parent.sys_info["children"]:
if plug["id"] == self.child_id:
return plug
raise SmartDeviceException(f"Unable to find children {self.child_id}")

27
kasa/iot/modules/__init__.py Normal file
View File

@@ -0,0 +1,27 @@
"""Module for individual feature modules."""
from .ambientlight import AmbientLight
from .antitheft import Antitheft
from .cloud import Cloud
from .countdown import Countdown
from .emeter import Emeter
from .module import IotModule
from .motion import Motion
from .rulemodule import Rule, RuleModule
from .schedule import Schedule
from .time import Time
from .usage import Usage
__all__ = [
"AmbientLight",
"Antitheft",
"Cloud",
"Countdown",
"Emeter",
"IotModule",
"Motion",
"Rule",
"RuleModule",
"Schedule",
"Time",
"Usage",
]

47
kasa/iot/modules/ambientlight.py Normal file
View File

@@ -0,0 +1,47 @@
"""Implementation of the ambient light (LAS) module found in some dimmers."""
from .module import IotModule
# TODO create tests and use the config reply there
# [{"hw_id":0,"enable":0,"dark_index":1,"min_adc":0,"max_adc":2450,
# "level_array":[{"name":"cloudy","adc":490,"value":20},
# {"name":"overcast","adc":294,"value":12},
# {"name":"dawn","adc":222,"value":9},
# {"name":"twilight","adc":222,"value":9},
# {"name":"total darkness","adc":111,"value":4},
# {"name":"custom","adc":2400,"value":97}]}]
class AmbientLight(IotModule):
"""Implements ambient light controls for the motion sensor."""
def query(self):
"""Request configuration."""
return self.query_for_command("get_config")
@property
def presets(self) -> dict:
"""Return device-defined presets for brightness setting."""
return self.data["level_array"]
@property
def enabled(self) -> bool:
"""Return True if the module is enabled."""
return bool(self.data["enable"])
async def set_enabled(self, state: bool):
"""Enable/disable LAS."""
return await self.call("set_enable", {"enable": int(state)})
async def current_brightness(self) -> int:
"""Return current brightness.
Return value units.
"""
return await self.call("get_current_brt")
async def set_brightness_limit(self, value: int):
"""Set the limit when the motion sensor is inactive.
See `presets` for preset values. Custom values are also likely allowed.
"""
return await self.call("set_brt_level", {"index": 0, "value": value})

9
kasa/iot/modules/antitheft.py Normal file
View File

@@ -0,0 +1,9 @@
"""Implementation of the antitheft module."""
from .rulemodule import RuleModule
class Antitheft(RuleModule):
"""Implementation of the antitheft module.
This shares the functionality among other rule-based modules.
"""

53
kasa/iot/modules/cloud.py Normal file
View File

@@ -0,0 +1,53 @@
"""Cloud module implementation."""
try:
from pydantic.v1 import BaseModel
except ImportError:
from pydantic import BaseModel
from .module import IotModule
class CloudInfo(BaseModel):
"""Container for cloud settings."""
binded: bool
cld_connection: int
fwDlPage: str
fwNotifyType: int
illegalType: int
server: str
stopConnect: int
tcspInfo: str
tcspStatus: int
username: str
class Cloud(IotModule):
"""Module implementing support for cloud services."""
def query(self):
"""Request cloud connectivity info."""
return self.query_for_command("get_info")
@property
def info(self) -> CloudInfo:
"""Return information about the cloud connectivity."""
return CloudInfo.parse_obj(self.data["get_info"])
def get_available_firmwares(self):
"""Return list of available firmwares."""
return self.query_for_command("get_intl_fw_list")
def set_server(self, url: str):
"""Set the update server URL."""
return self.query_for_command("set_server_url", {"server": url})
def connect(self, username: str, password: str):
"""Login to the cloud using given information."""
return self.query_for_command(
"bind", {"username": username, "password": password}
)
def disconnect(self):
"""Disconnect from the cloud."""
return self.query_for_command("unbind")

6
kasa/iot/modules/countdown.py Normal file
View File

@@ -0,0 +1,6 @@
"""Implementation for the countdown timer."""
from .rulemodule import RuleModule
class Countdown(RuleModule):
"""Implementation of countdown module."""

111
kasa/iot/modules/emeter.py Normal file
View File

@@ -0,0 +1,111 @@
"""Implementation of the emeter module."""
from datetime import datetime
from typing import Dict, List, Optional, Union
from ...emeterstatus import EmeterStatus
from .usage import Usage
class Emeter(Usage):
"""Emeter module."""
@property # type: ignore
def realtime(self) -> EmeterStatus:
"""Return current energy readings."""
return EmeterStatus(self.data["get_realtime"])
@property
def emeter_today(self) -> Optional[float]:
"""Return today's energy consumption in kWh."""
raw_data = self.daily_data
today = datetime.now().day
data = self._convert_stat_data(raw_data, entry_key="day", key=today)
return data.get(today)
@property
def emeter_this_month(self) -> Optional[float]:
"""Return this month's energy consumption in kWh."""
raw_data = self.monthly_data
current_month = datetime.now().month
data = self._convert_stat_data(raw_data, entry_key="month", key=current_month)
return data.get(current_month)
async def erase_stats(self):
"""Erase all stats.
Uses different query than usage meter.
"""
return await self.call("erase_emeter_stat")
async def get_realtime(self):
"""Return real-time statistics."""
return await self.call("get_realtime")
async def get_daystat(self, *, year=None, month=None, kwh=True) -> Dict:
"""Return daily stats for the given year & month.
The return value is a dictionary of {day: energy, ...}.
"""
data = await self.get_raw_daystat(year=year, month=month)
data = self._convert_stat_data(data["day_list"], entry_key="day", kwh=kwh)
return data
async def get_monthstat(self, *, year=None, kwh=True) -> Dict:
"""Return monthly stats for the given year.
The return value is a dictionary of {month: energy, ...}.
"""
data = await self.get_raw_monthstat(year=year)
data = self._convert_stat_data(data["month_list"], entry_key="month", kwh=kwh)
return data
def _convert_stat_data(
self,
data: List[Dict[str, Union[int, float]]],
entry_key: str,
kwh: bool = True,
key: Optional[int] = None,
) -> Dict[Union[int, float], Union[int, float]]:
"""Return emeter information keyed with the day/month.
The incoming data is a list of dictionaries::
[{'year': int,
'month': int,
'day': int, <-- for get_daystat not get_monthstat
'energy_wh': int, <-- for emeter in some versions (wh)
'energy': float <-- for emeter in other versions (kwh)
}, ...]
:return: a dictionary keyed by day or month with energy as the value.
"""
if not data:
return {}
scale: float = 1
if "energy_wh" in data[0]:
value_key = "energy_wh"
if kwh:
scale = 1 / 1000
else:
value_key = "energy"
if not kwh:
scale = 1000
if key is None:
# Return all the data
return {entry[entry_key]: entry[value_key] * scale for entry in data}
# In this case we want a specific key in the data
# i.e. the current day or month.
#
# Since we usually want the data at the end of the list so we can
# optimize the search by starting at the end and avoid scaling
# the data we don't need.
#
for entry in reversed(data):
if entry[entry_key] == key:
return {entry[entry_key]: entry[value_key] * scale}
return {}

87
kasa/iot/modules/module.py Normal file
View File

@@ -0,0 +1,87 @@
"""Base class for all module implementations."""
import collections
import logging
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING
from ...exceptions import SmartDeviceException
if TYPE_CHECKING:
from kasa.iot import IotDevice
_LOGGER = logging.getLogger(__name__)
# TODO: This is used for query construcing
def merge(d, u):
"""Update dict recursively."""
for k, v in u.items():
if isinstance(v, collections.abc.Mapping):
d[k] = merge(d.get(k, {}), v)
else:
d[k] = v
return d
class IotModule(ABC):
"""Base class implemention for all modules.
The base classes should implement `query` to return the query they want to be
executed during the regular update cycle.
"""
def __init__(self, device: "IotDevice", module: str):
self._device = device
self._module = module
@abstractmethod
def query(self):
"""Query to execute during the update cycle.
The inheriting modules implement this to include their wanted
queries to the query that gets executed when Device.update() gets called.
"""
@property
def estimated_query_response_size(self):
"""Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response
will be so that queries can be split should an estimated response be too large
"""
return 256 # Estimate for modules that don't specify
@property
def data(self):
"""Return the module specific raw data from the last update."""
if self._module not in self._device._last_update:
raise SmartDeviceException(
f"You need to call update() prior accessing module data"
f" for '{self._module}'"
)
return self._device._last_update[self._module]
@property
def is_supported(self) -> bool:
"""Return whether the module is supported by the device."""
if self._module not in self._device._last_update:
_LOGGER.debug("Initial update, so consider supported: %s", self._module)
return True
return "err_code" not in self.data
def call(self, method, params=None):
"""Call the given method with the given parameters."""
return self._device._query_helper(self._module, method, params)
def query_for_command(self, query, params=None):
"""Create a request object for the given parameters."""
return self._device._create_request(self._module, query, params)
def __repr__(self) -> str:
return (
f"<Module {self.__class__.__name__} ({self._module})"
f" for {self._device.host}>"
)

74
kasa/iot/modules/motion.py Normal file
View File

@@ -0,0 +1,74 @@
"""Implementation of the motion detection (PIR) module found in some dimmers."""
from enum import Enum
from typing import Optional
from ...exceptions import SmartDeviceException
from .module import IotModule
class Range(Enum):
"""Range for motion detection."""
Far = 0
Mid = 1
Near = 2
Custom = 3
# TODO: use the config reply in tests
# {"enable":0,"version":"1.0","trigger_index":2,"cold_time":60000,
# "min_adc":0,"max_adc":4095,"array":[80,50,20,0],"err_code":0}}}
class Motion(IotModule):
"""Implements the motion detection (PIR) module."""
def query(self):
"""Request PIR configuration."""
return self.query_for_command("get_config")
@property
def range(self) -> Range:
"""Return motion detection range."""
return Range(self.data["trigger_index"])
@property
def enabled(self) -> bool:
"""Return True if module is enabled."""
return bool(self.data["enable"])
async def set_enabled(self, state: bool):
"""Enable/disable PIR."""
return await self.call("set_enable", {"enable": int(state)})
async def set_range(
self, *, range: Optional[Range] = None, custom_range: Optional[int] = None
):
"""Set the range for the sensor.
:param range: for using standard ranges
:param custom_range: range in decimeters, overrides the range parameter
"""
if custom_range is not None:
payload = {"index": Range.Custom.value, "value": custom_range}
elif range is not None:
payload = {"index": range.value}
else:
raise SmartDeviceException(
"Either range or custom_range need to be defined"
)
return await self.call("set_trigger_sens", payload)
@property
def inactivity_timeout(self) -> int:
"""Return inactivity timeout in milliseconds."""
return self.data["cold_time"]
async def set_inactivity_timeout(self, timeout: int):
"""Set inactivity timeout in milliseconds.
Note, that you need to delete the default "Smart Control" rule in the app
to avoid reverting this back to 60 seconds after a period of time.
"""
return await self.call("set_cold_time", {"cold_time": timeout})

87
kasa/iot/modules/rulemodule.py Normal file
View File

@@ -0,0 +1,87 @@
"""Base implementation for all rule-based modules."""
import logging
from enum import Enum
from typing import Dict, List, Optional
try:
from pydantic.v1 import BaseModel
except ImportError:
from pydantic import BaseModel
from .module import IotModule, merge
class Action(Enum):
"""Action to perform."""
Disabled = -1
TurnOff = 0
TurnOn = 1
Unknown = 2
class TimeOption(Enum):
"""Time when the action is executed."""
Disabled = -1
Enabled = 0
AtSunrise = 1
AtSunset = 2
class Rule(BaseModel):
"""Representation of a rule."""
id: str
name: str
enable: bool
wday: List[int]
repeat: bool
# start action
sact: Optional[Action]
stime_opt: TimeOption
smin: int
eact: Optional[Action]
etime_opt: TimeOption
emin: int
# Only on bulbs
s_light: Optional[Dict]
_LOGGER = logging.getLogger(__name__)
class RuleModule(IotModule):
"""Base class for rule-based modules, such as countdown and antitheft."""
def query(self):
"""Prepare the query for rules."""
q = self.query_for_command("get_rules")
return merge(q, self.query_for_command("get_next_action"))
@property
def rules(self) -> List[Rule]:
"""Return the list of rules for the service."""
try:
return [
Rule.parse_obj(rule) for rule in self.data["get_rules"]["rule_list"]
]
except Exception as ex:
_LOGGER.error("Unable to read rule list: %s (data: %s)", ex, self.data)
return []
async def set_enabled(self, state: bool):
"""Enable or disable the service."""
return await self.call("set_overall_enable", state)
async def delete_rule(self, rule: Rule):
"""Delete the given rule."""
return await self.call("delete_rule", {"id": rule.id})
async def delete_all_rules(self):
"""Delete all rules."""
return await self.call("delete_all_rules")

6
kasa/iot/modules/schedule.py Normal file
View File

@@ -0,0 +1,6 @@
"""Schedule module implementation."""
from .rulemodule import RuleModule
class Schedule(RuleModule):
"""Implements the scheduling interface."""

54
kasa/iot/modules/time.py Normal file
View File

@@ -0,0 +1,54 @@
"""Provides the current time and timezone information."""
from datetime import datetime
from ...exceptions import SmartDeviceException
from .module import IotModule, merge
class Time(IotModule):
"""Implements the timezone settings."""
def query(self):
"""Request time and timezone."""
q = self.query_for_command("get_time")
merge(q, self.query_for_command("get_timezone"))
return q
@property
def time(self) -> datetime:
"""Return current device time."""
res = self.data["get_time"]
return datetime(
res["year"],
res["month"],
res["mday"],
res["hour"],
res["min"],
res["sec"],
)
@property
def timezone(self):
"""Return current timezone."""
res = self.data["get_timezone"]
return res
async def get_time(self):
"""Return current device time."""
try:
res = await self.call("get_time")
return datetime(
res["year"],
res["month"],
res["mday"],
res["hour"],
res["min"],
res["sec"],
)
except SmartDeviceException:
return None
async def get_timezone(self):
"""Request timezone information from the device."""
return await self.call("get_timezone")

116
kasa/iot/modules/usage.py Normal file
View File

@@ -0,0 +1,116 @@
"""Implementation of the usage interface."""
from datetime import datetime
from typing import Dict
from .module import IotModule, merge
class Usage(IotModule):
"""Baseclass for emeter/usage interfaces."""
def query(self):
"""Return the base query."""
now = datetime.now()
year = now.year
month = now.month
req = self.query_for_command("get_realtime")
req = merge(
req, self.query_for_command("get_daystat", {"year": year, "month": month})
)
req = merge(req, self.query_for_command("get_monthstat", {"year": year}))
return req
@property
def estimated_query_response_size(self):
"""Estimated maximum query response size."""
return 2048
@property
def daily_data(self):
"""Return statistics on daily basis."""
return self.data["get_daystat"]["day_list"]
@property
def monthly_data(self):
"""Return statistics on monthly basis."""
return self.data["get_monthstat"]["month_list"]
@property
def usage_today(self):
"""Return today's usage in minutes."""
today = datetime.now().day
# Traverse the list in reverse order to find the latest entry.
for entry in reversed(self.daily_data):
if entry["day"] == today:
return entry["time"]
return None
@property
def usage_this_month(self):
"""Return usage in this month in minutes."""
this_month = datetime.now().month
# Traverse the list in reverse order to find the latest entry.
for entry in reversed(self.monthly_data):
if entry["month"] == this_month:
return entry["time"]
return None
async def get_raw_daystat(self, *, year=None, month=None) -> Dict:
"""Return raw daily stats for the given year & month."""
if year is None:
year = datetime.now().year
if month is None:
month = datetime.now().month
return await self.call("get_daystat", {"year": year, "month": month})
async def get_raw_monthstat(self, *, year=None) -> Dict:
"""Return raw monthly stats for the given year."""
if year is None:
year = datetime.now().year
return await self.call("get_monthstat", {"year": year})
async def get_daystat(self, *, year=None, month=None) -> Dict:
"""Return daily stats for the given year & month.
The return value is a dictionary of {day: time, ...}.
"""
data = await self.get_raw_daystat(year=year, month=month)
data = self._convert_stat_data(data["day_list"], entry_key="day")
return data
async def get_monthstat(self, *, year=None) -> Dict:
"""Return monthly stats for the given year.
The return value is a dictionary of {month: time, ...}.
"""
data = await self.get_raw_monthstat(year=year)
data = self._convert_stat_data(data["month_list"], entry_key="month")
return data
async def erase_stats(self):
"""Erase all stats."""
return await self.call("erase_runtime_stat")
def _convert_stat_data(self, data, entry_key) -> Dict:
"""Return usage information keyed with the day/month.
The incoming data is a list of dictionaries::
[{'year': int,
'month': int,
'day': int, <-- for get_daystat not get_monthstat
'time': int, <-- for usage (mins)
}, ...]
:return: return a dictionary keyed by day or month with time as the value.
"""
if not data:
return {}
data = {entry[entry_key]: entry["time"] for entry in data}
return data