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

Merge remote-tracking branch 'upstream/master' into nw/fix_non_indexed_strip

Browse Source
This commit is contained in:
Steven B
2024-12-20 09:28:43 +00:00
parent bf2e57d9e2 fe88b52e19
commit 3992a39264
474 changed files with 76947 additions and 13269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

298
kasa/iot/effects.py Normal file
View File

@@ -0,0 +1,298 @@
"""Module for light strip effects (LB*, KL*, KB*)."""
from __future__ import annotations
from typing import cast
EFFECT_AURORA = {
"custom": 0,
"id": "xqUxDhbAhNLqulcuRMyPBmVGyTOyEMEu",
"brightness": 100,
"name": "Aurora",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "sequence",
"duration": 0,
"transition": 1500,
"direction": 4,
"spread": 7,
"repeat_times": 0,
"sequence": [[120, 100, 100], [240, 100, 100], [260, 100, 100], [280, 100, 100]],
}
EFFECT_BUBBLING_CAULDRON = {
"custom": 0,
"id": "tIwTRQBqJpeNKbrtBMFCgkdPTbAQGfRP",
"brightness": 100,
"name": "Bubbling Cauldron",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [100, 270],
"saturation_range": [80, 100],
"brightness_range": [50, 100],
"duration": 0,
"transition": 200,
"init_states": [[270, 100, 100]],
"fadeoff": 1000,
"random_seed": 24,
"backgrounds": [[270, 40, 50]],
}
EFFECT_CANDY_CANE = {
"custom": 0,
"id": "HCOttllMkNffeHjEOLEgrFJjbzQHoxEJ",
"brightness": 100,
"name": "Candy Cane",
"segments": [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
"expansion_strategy": 1,
"enable": 1,
"type": "sequence",
"duration": 700,
"transition": 500,
"direction": 1,
"spread": 1,
"repeat_times": 0,
"sequence": [
[0, 0, 100],
[0, 0, 100],
[360, 81, 100],
[0, 0, 100],
[0, 0, 100],
[360, 81, 100],
[360, 81, 100],
[0, 0, 100],
[0, 0, 100],
[360, 81, 100],
[360, 81, 100],
[360, 81, 100],
[360, 81, 100],
[0, 0, 100],
[0, 0, 100],
[360, 81, 100],
],
}
EFFECT_CHRISTMAS = {
"custom": 0,
"id": "bwTatyinOUajKrDwzMmqxxJdnInQUgvM",
"brightness": 100,
"name": "Christmas",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [136, 146],
"saturation_range": [90, 100],
"brightness_range": [50, 100],
"duration": 5000,
"transition": 0,
"init_states": [[136, 0, 100]],
"fadeoff": 2000,
"random_seed": 100,
"backgrounds": [[136, 98, 75], [136, 0, 0], [350, 0, 100], [350, 97, 94]],
}
EFFECT_FLICKER = {
"custom": 0,
"id": "bCTItKETDFfrKANolgldxfgOakaarARs",
"brightness": 100,
"name": "Flicker",
"segments": [1],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [30, 40],
"saturation_range": [100, 100],
"brightness_range": [50, 100],
"duration": 0,
"transition": 0,
"transition_range": [375, 500],
"init_states": [[30, 81, 80]],
}
EFFECT_HANUKKAH = {
"custom": 0,
"id": "CdLeIgiKcQrLKMINRPTMbylATulQewLD",
"brightness": 100,
"name": "Hanukkah",
"segments": [1],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [200, 210],
"saturation_range": [0, 100],
"brightness_range": [50, 100],
"duration": 1500,
"transition": 0,
"transition_range": [400, 500],
"init_states": [[35, 81, 80]],
}
EFFECT_HAUNTED_MANSION = {
"custom": 0,
"id": "oJnFHsVQzFUTeIOBAhMRfVeujmSauhjJ",
"brightness": 80,
"name": "Haunted Mansion",
"segments": [80],
"expansion_strategy": 2,
"enable": 1,
"type": "random",
"hue_range": [45, 45],
"saturation_range": [10, 10],
"brightness_range": [0, 80],
"duration": 0,
"transition": 0,
"transition_range": [50, 1500],
"init_states": [[45, 10, 100]],
"fadeoff": 200,
"random_seed": 1,
"backgrounds": [[45, 10, 100]],
}
EFFECT_ICICLE = {
"custom": 0,
"id": "joqVjlaTsgzmuQQBAlHRkkPAqkBUiqeb",
"brightness": 70,
"name": "Icicle",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "sequence",
"duration": 0,
"transition": 400,
"direction": 4,
"spread": 3,
"repeat_times": 0,
"sequence": [
[190, 100, 70],
[190, 100, 70],
[190, 30, 50],
[190, 100, 70],
[190, 100, 70],
],
}
EFFECT_LIGHTNING = {
"custom": 0,
"id": "ojqpUUxdGHoIugGPknrUcRoyJiItsjuE",
"brightness": 100,
"name": "Lightning",
"segments": [7, 20, 23, 32, 34, 35, 49, 65, 66, 74, 80],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [240, 240],
"saturation_range": [10, 11],
"brightness_range": [90, 100],
"duration": 0,
"transition": 50,
"init_states": [[240, 30, 100]],
"fadeoff": 150,
"random_seed": 600,
"backgrounds": [[200, 100, 100], [200, 50, 10], [210, 10, 50], [240, 10, 0]],
}
EFFECT_OCEAN = {
"custom": 0,
"id": "oJjUMosgEMrdumfPANKbkFmBcAdEQsPy",
"brightness": 30,
"name": "Ocean",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "sequence",
"duration": 0,
"transition": 2000,
"direction": 3,
"spread": 16,
"repeat_times": 0,
"sequence": [[198, 84, 30], [198, 70, 30], [198, 10, 30]],
}
EFFECT_RAINBOW = {
"custom": 0,
"id": "izRhLCQNcDzIKdpMPqSTtBMuAIoreAuT",
"brightness": 100,
"name": "Rainbow",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "sequence",
"duration": 0,
"transition": 1500,
"direction": 1,
"spread": 12,
"repeat_times": 0,
"sequence": [[0, 100, 100], [100, 100, 100], [200, 100, 100], [300, 100, 100]],
}
EFFECT_RAINDROP = {
"custom": 0,
"id": "QbDFwiSFmLzQenUOPnJrsGqyIVrJrRsl",
"brightness": 30,
"name": "Raindrop",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [200, 200],
"saturation_range": [10, 20],
"brightness_range": [10, 30],
"duration": 0,
"transition": 1000,
"init_states": [[200, 40, 100]],
"fadeoff": 1000,
"random_seed": 24,
"backgrounds": [[200, 40, 0]],
}
EFFECT_SPRING = {
"custom": 0,
"id": "URdUpEdQbnOOechDBPMkKrwhSupLyvAg",
"brightness": 100,
"name": "Spring",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [0, 90],
"saturation_range": [30, 100],
"brightness_range": [90, 100],
"duration": 600,
"transition": 0,
"transition_range": [2000, 6000],
"init_states": [[80, 30, 100]],
"fadeoff": 1000,
"random_seed": 20,
"backgrounds": [[130, 100, 40]],
}
EFFECT_VALENTINES = {
"custom": 0,
"id": "QglBhMShPHUAuxLqzNEefFrGiJwahOmz",
"brightness": 100,
"name": "Valentines",
"segments": [0],
"expansion_strategy": 1,
"enable": 1,
"type": "random",
"hue_range": [340, 340],
"saturation_range": [30, 40],
"brightness_range": [90, 100],
"duration": 600,
"transition": 2000,
"init_states": [[340, 30, 100]],
"fadeoff": 3000,
"random_seed": 100,
"backgrounds": [[340, 20, 50], [20, 50, 50], [0, 100, 50]],
}
EFFECTS_LIST_V1 = [
EFFECT_AURORA,
EFFECT_BUBBLING_CAULDRON,
EFFECT_CANDY_CANE,
EFFECT_CHRISTMAS,
EFFECT_FLICKER,
EFFECT_HANUKKAH,
EFFECT_HAUNTED_MANSION,
EFFECT_ICICLE,
EFFECT_LIGHTNING,
EFFECT_OCEAN,
EFFECT_RAINBOW,
EFFECT_RAINDROP,
EFFECT_SPRING,
EFFECT_VALENTINES,
]
EFFECT_NAMES_V1: list[str] = [cast(str, effect["name"]) for effect in EFFECTS_LIST_V1]
EFFECT_MAPPING_V1 = {effect["name"]: effect for effect in EFFECTS_LIST_V1}

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

@@ -0,0 +1,534 @@
"""Module for bulbs (LB*, KL*, KB*)."""
from __future__ import annotations
import logging
import re
from dataclasses import dataclass
from enum import Enum
from typing import Annotated, cast
from mashumaro import DataClassDictMixin
from mashumaro.config import BaseConfig
from mashumaro.types import Alias
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..interfaces.light import HSV, ColorTempRange
from ..module import Module
from ..protocols import BaseProtocol
from .iotdevice import IotDevice, KasaException, requires_update
from .modules import (
Antitheft,
Cloud,
Countdown,
Emeter,
Light,
LightPreset,
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"
#: Circadian
Circadian = "circadian"
@dataclass
class TurnOnBehavior(DataClassDictMixin):
"""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.
"""
class Config(BaseConfig):
"""Serialization config."""
omit_none = True
serialize_by_alias = True
#: Wanted behavior
mode: BehaviorMode
#: Index of preset to use, or ``None`` for the last known state.
preset: Annotated[int | None, Alias("index")] = None
brightness: int | None = None
color_temp: int | None = None
hue: int | None = None
saturation: int | None = None
@dataclass
class TurnOnBehaviors(DataClassDictMixin):
"""Model to contain turn on behaviors."""
#: The behavior when the bulb is turned on programmatically.
soft: Annotated[TurnOnBehavior, Alias("soft_on")]
#: The behavior when the bulb has been off from mains power.
hard: Annotated[TurnOnBehavior, 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, 9000),
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):
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:`KasaException <kasa.exceptions.KasaException>`,
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:
>>> [ preset.to_dict() for preset in bulb.presets }
[{'brightness': 50, 'hue': 0, 'saturation': 0, 'color_temp': 2700, 'index': 0}, {'brightness': 100, 'hue': 0, 'saturation': 75, 'color_temp': 0, 'index': 1}, {'brightness': 100, 'hue': 120, 'saturation': 75, 'color_temp': 0, 'index': 2}, {'brightness': 100, 'hue': 240, 'saturation': 75, 'color_temp': 0, 'index': 3}]
To modify an existing preset, pass :class:`~kasa.interfaces.light.LightPreset`
instance to :func:`save_preset` method:
>>> preset = bulb.presets[0]
>>> preset.brightness
50
>>> preset.brightness = 100
>>> asyncio.run(bulb.save_preset(preset))
>>> asyncio.run(bulb.update())
>>> 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: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Bulb
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
await super()._initialize_modules()
self.add_module(
Module.IotSchedule, Schedule(self, "smartlife.iot.common.schedule")
)
self.add_module(Module.IotUsage, Usage(self, "smartlife.iot.common.schedule"))
self.add_module(
Module.IotAntitheft, Antitheft(self, "smartlife.iot.common.anti_theft")
)
self.add_module(Module.Time, Time(self, "smartlife.iot.common.timesetting"))
self.add_module(Module.Energy, Emeter(self, self.emeter_type))
self.add_module(Module.IotCountdown, Countdown(self, "countdown"))
self.add_module(Module.IotCloud, Cloud(self, "smartlife.iot.common.cloud"))
self.add_module(Module.Light, Light(self, self.LIGHT_SERVICE))
self.add_module(Module.LightPreset, LightPreset(self, self.LIGHT_SERVICE))
@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 KasaException("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 KasaException(
"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.from_dict(
await self._query_helper(self.LIGHT_SERVICE, "get_default_behavior")
)
async def set_turn_on_behavior(self, behavior: TurnOnBehaviors) -> dict:
"""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.to_dict()
)
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: int | None = None
) -> dict:
"""Set the light state."""
state = {**state}
if transition is not None:
state["transition_period"] = transition
if "brightness" in state:
self._raise_for_invalid_brightness(state["brightness"])
# 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 KasaException("Bulb does not support color.")
light_state = cast(dict, self.light_state)
hue = light_state["hue"]
saturation = light_state["saturation"]
value = self._brightness
# Simple HSV(hue, saturation, value) is less efficent than below
# due to the cpython implementation.
return tuple.__new__(HSV, (hue, saturation, value))
@requires_update
async def _set_hsv(
self,
hue: int,
saturation: int,
value: int | None = None,
*,
transition: int | None = 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 KasaException("Bulb does not support color.")
if not isinstance(hue, int):
raise TypeError("Hue must be an integer.")
if not (0 <= hue <= 360):
raise ValueError(f"Invalid hue value: {hue} (valid range: 0-360)")
if not isinstance(saturation, int):
raise TypeError("Saturation must be an integer.")
if 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 KasaException("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: int | None = None, transition: int | None = 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 KasaException("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)
def _raise_for_invalid_brightness(self, value: int) -> None:
if not isinstance(value, int):
raise TypeError("Brightness must be an integer")
if not (0 <= value <= 100):
raise ValueError(f"Invalid brightness value: {value} (valid range: 0-100%)")
@property # type: ignore
@requires_update
def _brightness(self) -> int:
"""Return the current brightness in percentage."""
if not self._is_dimmable: # pragma: no cover
raise KasaException("Bulb is not dimmable.")
# If the device supports effects and one is active, we get the brightness
# from the effect. This is not required when setting the brightness as
# the device handles it via set_light_state
if (
light_effect := self.modules.get(Module.IotLightEffect)
) is not None and light_effect.effect != light_effect.LIGHT_EFFECTS_OFF:
return light_effect.brightness
light_state = self.light_state
return int(light_state["brightness"])
@requires_update
async def _set_brightness(
self, brightness: int, *, transition: int | None = 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 KasaException("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 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: int | None = 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: int | None = 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) -> dict:
"""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
def max_device_response_size(self) -> int:
"""Returns the maximum response size the device can safely construct."""
return 4096

42
kasa/iot/iotcamera.py Normal file
View File

@@ -0,0 +1,42 @@
"""Module for cameras."""
from __future__ import annotations
import logging
from datetime import datetime, tzinfo
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..protocols import BaseProtocol
from .iotdevice import IotDevice
_LOGGER = logging.getLogger(__name__)
class IotCamera(IotDevice):
"""Representation of a TP-Link Camera."""
def __init__(
self,
host: str,
*,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Camera
@property
def time(self) -> datetime:
"""Get the camera's time."""
return datetime.fromtimestamp(self.sys_info["system_time"])
@property
def timezone(self) -> tzinfo:
"""Get the camera's timezone."""
return None # type: ignore
@property # type: ignore
def is_on(self) -> bool:
"""Return whether device is on."""
return True

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

@@ -0,0 +1,777 @@
"""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
"""
from __future__ import annotations
import functools
import inspect
import logging
from collections.abc import Callable, Mapping, Sequence
from datetime import datetime, timedelta, tzinfo
from typing import TYPE_CHECKING, Any, cast
from warnings import warn
from ..device import Device, DeviceInfo, WifiNetwork
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..exceptions import KasaException
from ..feature import Feature
from ..module import Module
from ..modulemapping import ModuleMapping, ModuleName
from ..protocols import BaseProtocol
from .iotmodule import IotModule, merge
from .modules import Emeter
_LOGGER = logging.getLogger(__name__)
def requires_update(f: Callable) -> Any:
"""Indicate that `update` should be called before accessing this method.""" # noqa: D202
if inspect.iscoroutinefunction(f):
@functools.wraps(f)
async def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if not self._last_update and (
self._sys_info is None or f.__name__ not in self._sys_info
):
raise KasaException("You need to await update() to access the data")
return await f(*args, **kwargs)
else:
@functools.wraps(f)
def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if not self._last_update and (
self._sys_info is None or f.__name__ not in self._sys_info
):
raise KasaException("You need to await update() to access the data")
return f(*args, **kwargs)
f.requires_update = True # type: ignore[attr-defined]
return wrapped
@functools.lru_cache
def _parse_features(features: str) -> set[str]:
"""Parse features string."""
return set(features.split(":"))
def _extract_sys_info(info: dict[str, Any]) -> dict[str, Any]:
"""Return the system info structure."""
sysinfo_default = info.get("system", {}).get("get_sysinfo", {})
sysinfo_nest = sysinfo_default.get("system", {})
if len(sysinfo_nest) > len(sysinfo_default) and isinstance(sysinfo_nest, dict):
return sysinfo_nest
return sysinfo_default
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
:class:`KasaException <kasa.exceptions.KasaException>`,
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
Bedroom Lamp Plug
>>> dev.model
HS110
>>> 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: DeviceConfig | None = None,
protocol: BaseProtocol | None = 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._supported_modules: dict[str | ModuleName[Module], IotModule] | None = None
self._legacy_features: set[str] = set()
self._children: Mapping[str, IotDevice] = {}
self._modules: dict[str | ModuleName[Module], IotModule] = {}
self._on_since: datetime | None = None
@property
def children(self) -> Sequence[IotDevice]:
"""Return list of children."""
return list(self._children.values())
@property
@requires_update
def modules(self) -> ModuleMapping[IotModule]:
"""Return the device modules."""
if TYPE_CHECKING:
return cast(ModuleMapping[IotModule], self._supported_modules)
return self._supported_modules
def add_module(self, name: str | ModuleName[Module], module: IotModule) -> None:
"""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: dict | None = None,
child_ids: list | None = None,
) -> dict:
if arg is None:
arg = {}
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 KasaException("Device has no emeter")
if self.emeter_type not in self._last_update:
raise KasaException("update() required prior accessing emeter")
async def _query_helper(
self,
target: str,
cmd: str,
arg: dict | None = None,
child_ids: list | None = None,
) -> dict:
"""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 KasaException(f"Communication error on {target}:{cmd}") from ex
if target not in response:
raise KasaException(f"No required {target} in response: {response}")
result = response[target]
if "err_code" in result and result["err_code"] != 0:
raise KasaException(f"Error on {target}.{cmd}: {result}")
if cmd not in result:
raise KasaException(f"No command in response: {response}")
result = result[cmd]
if "err_code" in result and result["err_code"] != 0:
raise KasaException(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) -> dict[str, Feature]:
"""Return a set of features that the device supports."""
return self._features
@property # type: ignore
@requires_update
def has_emeter(self) -> bool:
"""Return True if device has an energy meter."""
return "ENE" in self._legacy_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) -> None:
"""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 not self._last_update:
_LOGGER.debug("Performing the initial update to obtain sysinfo")
response = await self.protocol.query(req)
self._last_update = response
self._set_sys_info(_extract_sys_info(response))
if not self._modules:
await self._initialize_modules()
await self._modular_update(req)
self._set_sys_info(_extract_sys_info(self._last_update))
for module in self._modules.values():
await module._post_update_hook()
if not self._features:
await self._initialize_features()
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
if self.has_emeter:
_LOGGER.debug(
"The device has emeter, querying its information along sysinfo"
)
self.add_module(Module.Energy, Emeter(self, self.emeter_type))
async def _initialize_features(self) -> None:
"""Initialize common features."""
self._add_feature(
Feature(
self,
id="state",
name="State",
attribute_getter="is_on",
attribute_setter="set_state",
type=Feature.Type.Switch,
category=Feature.Category.Primary,
)
)
self._add_feature(
Feature(
device=self,
id="rssi",
name="RSSI",
attribute_getter="rssi",
icon="mdi:signal",
unit_getter=lambda: "dBm",
category=Feature.Category.Debug,
type=Feature.Type.Sensor,
)
)
# iot strips calculate on_since from the children
if "on_time" in self._sys_info or self.device_type == Device.Type.Strip:
self._add_feature(
Feature(
device=self,
id="on_since",
name="On since",
attribute_getter="on_since",
icon="mdi:clock",
category=Feature.Category.Info,
type=Feature.Type.Sensor,
)
)
self._add_feature(
Feature(
device=self,
id="reboot",
name="Reboot",
attribute_setter="reboot",
icon="mdi:restart",
category=Feature.Category.Debug,
type=Feature.Type.Action,
)
)
for module in self.modules.values():
module._initialize_features()
for module_feat in module._module_features.values():
self._add_feature(module_feat)
async def _modular_update(self, req: dict) -> None:
"""Execute an update query."""
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:
for k, v in response.items():
# The same module could have results in different responses
# i.e. smartlife.iot.common.schedule for Usage and
# Schedule, so need to call update(**v) here. If a module is
# not supported the response
# {'err_code': -1, 'err_msg': 'module not support'}
# become top level key/values of the response so check for dict
if isinstance(v, dict):
update.setdefault(k, {}).update(**v)
self._last_update = update
# IOT modules are added as default but could be unsupported post first update
if self._supported_modules is None:
supported = {}
for module_name, module in self._modules.items():
if module.is_supported:
supported[module_name] = module
self._supported_modules = supported
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
discovery_model = info["device_model"]
no_region_model, _, _ = discovery_model.partition("(")
self._set_sys_info({**info, "model": no_region_model})
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._legacy_features = _parse_features(features)
@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
@requires_update
def model(self) -> str:
"""Returns the device model."""
if self._last_update:
return self.device_info.short_name
return self._sys_info["model"]
@property # type: ignore
def alias(self) -> str | None:
"""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) -> dict:
"""Set the device name (alias)."""
return await self._query_helper("system", "set_dev_alias", {"alias": alias})
@property
@requires_update
def time(self) -> datetime:
"""Return current time from the device."""
return self.modules[Module.Time].time
@property
@requires_update
def timezone(self) -> tzinfo:
"""Return the current timezone."""
return self.modules[Module.Time].timezone
async def get_time(self) -> datetime:
"""Return current time from the device, if available."""
msg = "Use `time` property instead, this call will be removed in the future."
warn(msg, DeprecationWarning, stacklevel=2)
return self.time
async def get_timezone(self) -> tzinfo:
"""Return timezone information."""
msg = (
"Use `timezone` property instead, this call will be removed in the future."
)
warn(msg, DeprecationWarning, stacklevel=2)
return self.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) -> int | None:
"""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 KasaException(
"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: str) -> dict:
"""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})
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 factory_reset(self) -> None:
"""Reset device back to factory settings.
Note, this does not downgrade the firmware.
"""
await self._query_helper("system", "reset")
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) -> 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.")
async def set_state(self, on: bool) -> dict:
"""Set the device state."""
if on:
return await self.turn_on()
else:
return await self.turn_off()
@property # type: ignore
@requires_update
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.
"""
if self.is_off or "on_time" not in self._sys_info:
self._on_since = None
return None
on_time = self._sys_info["on_time"]
on_since = self.time - timedelta(seconds=on_time)
if not self._on_since or timedelta(
seconds=0
) < on_since - self._on_since > timedelta(seconds=5):
self._on_since = on_since
return self._on_since
@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: str) -> dict:
return await self._query_helper(target, "get_scaninfo", {"refresh": 1})
try:
info = await _scan("netif")
except KasaException 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 KasaException(f"Invalid response for wifi scan: {info}")
return [WifiNetwork(**x) for x in info["ap_list"]]
async def wifi_join(self, ssid: str, password: str, keytype: str = "3") -> dict: # 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: str, payload: dict) -> dict:
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 KasaException 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
@staticmethod
def _get_device_type_from_sys_info(info: dict[str, Any]) -> DeviceType:
"""Find SmartDevice subclass for device described by passed data."""
if "system" in info.get("system", {}).get("get_sysinfo", {}):
return DeviceType.Camera
if "system" not in info or "get_sysinfo" not in info["system"]:
raise KasaException("No 'system' or 'get_sysinfo' in response")
sysinfo: dict[str, Any] = _extract_sys_info(info)
type_: str | None = sysinfo.get("type", sysinfo.get("mic_type"))
if type_ is None:
raise KasaException("Unable to find the device type field!")
if "dev_name" in sysinfo and "Dimmer" in sysinfo["dev_name"]:
return DeviceType.Dimmer
if "smartplug" in type_.lower():
if "children" in sysinfo:
return DeviceType.Strip
if (dev_name := sysinfo.get("dev_name")) and "light" in dev_name.lower():
return DeviceType.WallSwitch
return DeviceType.Plug
if "smartbulb" in type_.lower():
if "length" in sysinfo: # strips have length
return DeviceType.LightStrip
return DeviceType.Bulb
_LOGGER.warning("Unknown device type %s, falling back to plug", type_)
return DeviceType.Plug
@staticmethod
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> DeviceInfo:
"""Get model information for a device."""
sys_info = _extract_sys_info(info)
# Get model and region info
region = None
device_model = sys_info["model"]
long_name, _, region = device_model.partition("(")
if region: # All iot devices have region but just in case
region = region.replace(")", "")
# Get other info
device_family = sys_info.get("type", sys_info.get("mic_type"))
device_type = IotDevice._get_device_type_from_sys_info(info)
fw_version_full = sys_info["sw_ver"]
firmware_version, firmware_build = fw_version_full.split(" ", maxsplit=1)
auth = bool(discovery_info and ("mgt_encrypt_schm" in discovery_info))
return DeviceInfo(
short_name=long_name,
long_name=long_name,
brand="kasa",
device_family=device_family,
device_type=device_type,
hardware_version=sys_info["hw_ver"],
firmware_version=firmware_version,
firmware_build=firmware_build,
requires_auth=auth,
region=region,
)

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

@@ -0,0 +1,241 @@
"""Module for dimmers (currently only HS220)."""
from __future__ import annotations
from enum import Enum
from typing import Any
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..module import Module
from ..protocols import BaseProtocol
from .iotdevice import KasaException, requires_update
from .iotplug import IotPlug
from .modules import AmbientLight, Light, 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:`KasaException`\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: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Dimmer
async def _initialize_modules(self) -> None:
"""Initialize modules."""
await super()._initialize_modules()
# 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(Module.IotMotion, Motion(self, "smartlife.iot.PIR"))
self.add_module(Module.IotAmbientLight, AmbientLight(self, "smartlife.iot.LAS"))
self.add_module(Module.Light, Light(self, "light"))
@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 KasaException("Device is not dimmable.")
sys_info = self.sys_info
return int(sys_info["brightness"])
@requires_update
async def _set_brightness(
self, brightness: int, *, transition: int | None = None
) -> dict:
"""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 KasaException("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(
f"Invalid brightness value: {brightness} (valid range: 0-100%)"
)
# 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: int | None = None, **kwargs) -> dict:
"""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: int | None = None, **kwargs) -> dict:
"""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) -> dict:
"""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 TypeError(f"Brightness must be an integer, not {type(brightness)}.")
if not 0 <= brightness <= 100:
raise ValueError(
f"Invalid brightness value: {brightness} (valid range: 0-100%)"
)
# If zero set to 1 millisecond
if transition == 0:
transition = 1
if not isinstance(transition, int):
raise TypeError(f"Transition must be integer, not of {type(transition)}.")
if transition <= 0:
raise ValueError(f"Transition value {transition} is not valid.")
return await self._query_helper(
self.DIMMER_SERVICE,
"set_dimmer_transition",
{"brightness": brightness, "duration": transition},
)
@requires_update
async def get_behaviors(self) -> dict:
"""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: int | None = None
) -> dict:
"""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
return await self._query_helper(
self.DIMMER_SERVICE, action_type_setter, payload
)
@requires_update
async def set_fade_time(self, fade_type: FadeType, time: int) -> dict:
"""Set time for fade in / fade out."""
fade_type_setter = f"set_{fade_type}_time"
payload = {"fadeTime": time}
return 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
def _is_variable_color_temp(self) -> bool:
"""Whether the device supports variable color temp."""
return False
@property
def _is_color(self) -> bool:
"""Whether the device supports color."""
return False

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

@@ -0,0 +1,72 @@
"""Module for light strips (KL430)."""
from __future__ import annotations
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..module import Module
from ..protocols import BaseProtocol
from .iotbulb import IotBulb
from .iotdevice import requires_update
from .modules.lighteffect import LightEffect
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)
Bedroom Lightstrip
Getting the length of the strip:
>>> strip.length
16
Currently active effect:
>>> strip.effect
{'brightness': 100, 'custom': 0, 'enable': 0,
'id': 'bCTItKETDFfrKANolgldxfgOakaarARs', 'name': 'Flicker'}
.. 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: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.LightStrip
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
await super()._initialize_modules()
self.add_module(
Module.LightEffect,
LightEffect(self, "smartlife.iot.lighting_effect"),
)
@property # type: ignore
@requires_update
def length(self) -> int:
"""Return length of the strip."""
return self.sys_info["length"]

76
kasa/iot/iotmodule.py Normal file
View File

@@ -0,0 +1,76 @@
"""Base class for IOT module implementations."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
from ..exceptions import KasaException
from ..module import Module
_LOGGER = logging.getLogger(__name__)
if TYPE_CHECKING:
from .iotdevice import IotDevice
def _merge_dict(dest: dict, source: dict) -> dict:
"""Update dict recursively."""
for k, v in source.items():
if k in dest and type(v) is dict: # noqa: E721 - only accepts `dict` type
_merge_dict(dest[k], v)
else:
dest[k] = v
return dest
merge = _merge_dict
class IotModule(Module):
"""Base class implemention for all IOT modules."""
_device: IotDevice
async def call(self, method: str, params: dict | None = None) -> dict:
"""Call the given method with the given parameters."""
return await self._device._query_helper(self._module, method, params)
def query_for_command(self, query: str, params: dict | None = None) -> dict:
"""Create a request object for the given parameters."""
return self._device._create_request(self._module, query, params)
@property
def estimated_query_response_size(self) -> int:
"""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) -> dict[str, Any]:
"""Return the module specific raw data from the last update."""
dev = self._device
q = self.query()
if not q:
return dev.sys_info
if self._module not in dev._last_update:
raise KasaException(
f"You need to call update() prior accessing module data"
f" for '{self._module}'"
)
return dev._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

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

@@ -0,0 +1,104 @@
"""Module for smart plugs (HS100, HS110, ..)."""
from __future__ import annotations
import logging
from typing import Any
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..module import Module
from ..protocols import BaseProtocol
from .iotdevice import IotDevice, requires_update
from .modules import AmbientLight, Antitheft, Cloud, Led, Motion, Schedule, Time, Usage
_LOGGER = logging.getLogger(__name__)
class IotPlug(IotDevice):
r"""Representation of a TP-Link Smart Plug.
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:`KasaException`\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
Bedroom Lamp Plug
Setting the LED state:
>>> asyncio.run(plug.set_led(True))
>>> asyncio.run(plug.update())
>>> plug.led
True
For more examples, see the :class:`Device` class.
"""
def __init__(
self,
host: str,
*,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Plug
async def _initialize_modules(self) -> None:
"""Initialize modules."""
await super()._initialize_modules()
self.add_module(Module.IotSchedule, Schedule(self, "schedule"))
self.add_module(Module.IotUsage, Usage(self, "schedule"))
self.add_module(Module.IotAntitheft, Antitheft(self, "anti_theft"))
self.add_module(Module.Time, Time(self, "time"))
self.add_module(Module.IotCloud, Cloud(self, "cnCloud"))
self.add_module(Module.Led, Led(self, "system"))
@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: Any) -> dict:
"""Turn the switch on."""
return await self._query_helper("system", "set_relay_state", {"state": 1})
async def turn_off(self, **kwargs: Any) -> dict:
"""Turn the switch off."""
return await self._query_helper("system", "set_relay_state", {"state": 0})
class IotWallSwitch(IotPlug):
"""Representation of a TP-Link Smart Wall Switch."""
def __init__(
self,
host: str,
*,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.WallSwitch
async def _initialize_modules(self) -> None:
"""Initialize modules."""
await super()._initialize_modules()
if (dev_name := self.sys_info["dev_name"]) and "PIR" in dev_name:
self.add_module(Module.IotMotion, Motion(self, "smartlife.iot.PIR"))
self.add_module(
Module.IotAmbientLight, AmbientLight(self, "smartlife.iot.LAS")
)

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

@@ -0,0 +1,489 @@
"""Module for multi-socket devices (HS300, HS107, KP303, ..)."""
from __future__ import annotations
import logging
from collections import defaultdict
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..emeterstatus import EmeterStatus
from ..exceptions import KasaException
from ..feature import Feature
from ..interfaces import Energy
from ..module import Module
from ..protocols import BaseProtocol
from .iotdevice import (
IotDevice,
requires_update,
)
from .iotmodule import IotModule
from .iotplug import IotPlug
from .modules import Antitheft, Cloud, Countdown, Emeter, Led, Schedule, Time, Usage
_LOGGER = logging.getLogger(__name__)
def merge_sums(dicts: list[dict]) -> dict:
"""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:`KasaException`\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
Bedroom Power Strip
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())
>>> asyncio.run(strip.update())
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:`Device` class.
"""
def __init__(
self,
host: str,
*,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self.emeter_type = "emeter"
self._device_type = DeviceType.Strip
async def _initialize_modules(self) -> None:
"""Initialize modules."""
# Strip has different modules to plug so do not call super
self.add_module(Module.IotAntitheft, Antitheft(self, "anti_theft"))
self.add_module(Module.IotSchedule, Schedule(self, "schedule"))
self.add_module(Module.IotUsage, Usage(self, "schedule"))
self.add_module(Module.Time, Time(self, "time"))
self.add_module(Module.IotCountdown, Countdown(self, "countdown"))
self.add_module(Module.Led, Led(self, "system"))
self.add_module(Module.IotCloud, Cloud(self, "cnCloud"))
if self.has_emeter:
_LOGGER.debug(
"The device has emeter, querying its information along sysinfo"
)
self.add_module(Module.Energy, StripEmeter(self, self.emeter_type))
@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) -> None:
"""Update some of the attributes.
Needed for methods that are decorated with `requires_update`.
"""
# Super initializes modules and features
await super().update(update_children)
initialize_children = not self.children
# Initialize the child devices during the first update.
if initialize_children:
children = self.sys_info["children"]
_LOGGER.debug("Initializing %s child sockets", len(children))
self._children = {
f"{self.mac}_{child['id']}": IotStripPlug(
self.host, parent=self, child_id=child["id"]
)
for child in children
}
for child in self._children.values():
await child._initialize_modules()
if update_children:
for plug in self.children:
if TYPE_CHECKING:
assert isinstance(plug, IotStripPlug)
await plug._update()
if not self.features:
await self._initialize_features()
async def _initialize_features(self) -> None:
"""Initialize common features."""
# Do not initialize features until children are created
if not self.children:
return
await super()._initialize_features()
async def turn_on(self, **kwargs) -> dict:
"""Turn the strip on."""
for plug in self.children:
await plug.turn_on()
return {}
async def turn_off(self, **kwargs) -> dict:
"""Turn the strip off."""
for plug in self.children:
await plug.turn_off()
return {}
@property # type: ignore
@requires_update
def on_since(self) -> datetime | None:
"""Return the maximum on-time of all outlets."""
if self.is_off:
return None
return min(plug.on_since for plug in self.children if plug.on_since is not None)
class StripEmeter(IotModule, Energy):
"""Energy module implementation to aggregate child modules."""
_supported = (
Energy.ModuleFeature.CONSUMPTION_TOTAL
| Energy.ModuleFeature.PERIODIC_STATS
| Energy.ModuleFeature.VOLTAGE_CURRENT
)
def supports(self, module_feature: Energy.ModuleFeature) -> bool:
"""Return True if module supports the feature."""
return module_feature in self._supported
def query(self) -> dict:
"""Return the base query."""
return {}
@property
def current_consumption(self) -> float | None:
"""Get the current power consumption in watts."""
return sum(
v if (v := plug.modules[Module.Energy].current_consumption) else 0.0
for plug in self._device.children
)
async def get_status(self) -> EmeterStatus:
"""Retrieve current energy readings."""
emeter_rt = await self._async_get_emeter_sum("get_status", {})
# 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._device.children)
)
return EmeterStatus(emeter_rt)
async def get_daily_stats(
self, year: int | None = None, month: int | None = 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_daily_stats", {"year": year, "month": month, "kwh": kwh}
)
async def get_monthly_stats(
self, year: int | None = 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_monthly_stats", {"year": year, "kwh": kwh}
)
async def _async_get_emeter_sum(self, func: str, kwargs: dict[str, Any]) -> dict:
"""Retrieve emeter stats for a time period from children."""
return merge_sums(
[
await getattr(plug.modules[Module.Energy], func)(**kwargs)
for plug in self._device.children
]
)
async def erase_stats(self) -> dict:
"""Erase energy meter statistics for all plugs."""
for plug in self._device.children:
await plug.modules[Module.Energy].erase_stats()
return {}
@property # type: ignore
def consumption_this_month(self) -> float | None:
"""Return this month's energy consumption in kWh."""
return sum(
v if (v := plug.modules[Module.Energy].consumption_this_month) else 0.0
for plug in self._device.children
)
@property # type: ignore
def consumption_today(self) -> float | None:
"""Return this month's energy consumption in kWh."""
return sum(
v if (v := plug.modules[Module.Energy].consumption_today) else 0.0
for plug in self._device.children
)
@property # type: ignore
def consumption_total(self) -> float | None:
"""Return total energy consumption since reboot in kWh."""
return sum(
v if (v := plug.modules[Module.Energy].consumption_total) else 0.0
for plug in self._device.children
)
@property # type: ignore
def status(self) -> EmeterStatus:
"""Return current energy readings."""
emeter = merge_sums(
[plug.modules[Module.Energy].status for plug in self._device.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._device.children))
return EmeterStatus(emeter)
@property
def current(self) -> float | None:
"""Return the current in A."""
return self.status.current
@property
def voltage(self) -> float | None:
"""Get the current voltage in V."""
return self.status.voltage
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.
"""
_parent: IotStrip
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.protocol = parent.protocol # Must use the same connection as the parent
self._on_since: datetime | None = None
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
if self.has_emeter:
self.add_module(Module.Energy, Emeter(self, self.emeter_type))
self.add_module(Module.IotUsage, Usage(self, "schedule"))
self.add_module(Module.IotAntitheft, Antitheft(self, "anti_theft"))
self.add_module(Module.IotSchedule, Schedule(self, "schedule"))
self.add_module(Module.IotCountdown, Countdown(self, "countdown"))
async def _initialize_features(self) -> None:
"""Initialize common features."""
self._add_feature(
Feature(
self,
id="state",
name="State",
attribute_getter="is_on",
attribute_setter="set_state",
type=Feature.Type.Switch,
category=Feature.Category.Primary,
)
)
self._add_feature(
Feature(
device=self,
id="on_since",
name="On since",
attribute_getter="on_since",
icon="mdi:clock",
category=Feature.Category.Info,
type=Feature.Type.Sensor,
)
)
for module in self.modules.values():
module._initialize_features()
for module_feat in module._module_features.values():
self._add_feature(module_feat)
async def update(self, update_children: bool = True) -> None:
"""Query the device to update the data.
Needed for properties that are decorated with `requires_update`.
"""
await self._update(update_children)
async def _update(self, update_children: bool = True) -> None:
"""Query the device to update the data.
Internal implementation to allow patching of public update in the cli
or test framework.
"""
await self._modular_update({})
for module in self._modules.values():
await module._post_update_hook()
if not self._features:
await self._initialize_features()
def _create_request(
self,
target: str,
cmd: str,
arg: dict | None = None,
child_ids: list | None = None,
) -> dict:
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: dict | None = None,
child_ids: list | None = None,
) -> dict:
"""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) -> datetime | None:
"""Return on-time, if available."""
if self.is_off:
self._on_since = None
return None
info = self._get_child_info()
on_time = info["on_time"]
time = self._parent.time
on_since = time - timedelta(seconds=on_time)
if not self._on_since or timedelta(
seconds=0
) < on_since - self._on_since > timedelta(seconds=5):
self._on_since = on_since
return self._on_since
@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 KasaException(
f"Unable to find children {self.child_id}"
) # pragma: no cover

185
kasa/iot/iottimezone.py Normal file
View File

@@ -0,0 +1,185 @@
"""Module for io device timezone lookups."""
from __future__ import annotations
import logging
from datetime import datetime, timedelta, tzinfo
from typing import cast
from zoneinfo import ZoneInfo
from ..cachedzoneinfo import CachedZoneInfo
_LOGGER = logging.getLogger(__name__)
async def get_timezone(index: int) -> tzinfo:
"""Get the timezone from the index."""
if index > 109:
_LOGGER.error(
"Unexpected index %s not configured as a timezone, defaulting to UTC", index
)
return await CachedZoneInfo.get_cached_zone_info("Etc/UTC")
name = TIMEZONE_INDEX[index]
return await CachedZoneInfo.get_cached_zone_info(name)
async def get_timezone_index(tzone: tzinfo) -> int:
"""Return the iot firmware index for a valid IANA timezone key."""
if isinstance(tzone, ZoneInfo):
name = tzone.key
rev = {val: key for key, val in TIMEZONE_INDEX.items()}
if name in rev:
return rev[name]
for i in range(110):
if _is_same_timezone(tzone, await get_timezone(i)):
return i
raise ValueError("Device does not support timezone %s", name)
async def get_matching_timezones(tzone: tzinfo) -> list[str]:
"""Return the iot firmware index for a valid IANA timezone key."""
matches = []
if isinstance(tzone, ZoneInfo):
name = tzone.key
vals = {val for val in TIMEZONE_INDEX.values()}
if name in vals:
matches.append(name)
for i in range(110):
fw_tz = await get_timezone(i)
if _is_same_timezone(tzone, fw_tz):
match_key = cast(ZoneInfo, fw_tz).key
if match_key not in matches:
matches.append(match_key)
return matches
def _is_same_timezone(tzone1: tzinfo, tzone2: tzinfo) -> bool:
"""Return true if the timezones have the same utcffset and dst offset.
Iot devices only support a limited static list of IANA timezones; this is used to
check if a static timezone matches the same utc offset and dst settings.
"""
now = datetime.now()
start_day = datetime(now.year, 1, 1, 12)
for i in range(365):
the_day = start_day + timedelta(days=i)
if tzone1.utcoffset(the_day) != tzone2.utcoffset(the_day):
return False
return True
TIMEZONE_INDEX = {
0: "Etc/GMT+12",
1: "Pacific/Samoa",
2: "US/Hawaii",
3: "US/Alaska",
4: "Mexico/BajaNorte",
5: "Etc/GMT+8",
6: "PST8PDT",
7: "US/Arizona",
8: "America/Mazatlan",
9: "MST",
10: "MST7MDT",
11: "Mexico/General",
12: "Etc/GMT+6",
13: "CST6CDT",
14: "America/Monterrey",
15: "Canada/Saskatchewan",
16: "America/Bogota",
17: "Etc/GMT+5",
18: "EST",
19: "America/Indiana/Indianapolis",
20: "America/Caracas",
21: "America/Asuncion",
22: "Etc/GMT+4",
23: "Canada/Atlantic",
24: "America/Cuiaba",
25: "Brazil/West",
26: "America/Santiago",
27: "Canada/Newfoundland",
28: "America/Sao_Paulo",
29: "America/Argentina/Buenos_Aires",
30: "America/Cayenne",
31: "America/Miquelon",
32: "America/Montevideo",
33: "Chile/Continental",
34: "Etc/GMT+2",
35: "Atlantic/Azores",
36: "Atlantic/Cape_Verde",
37: "Africa/Casablanca",
38: "UCT",
39: "GB",
40: "Africa/Monrovia",
41: "Europe/Amsterdam",
42: "Europe/Belgrade",
43: "Europe/Brussels",
44: "Europe/Sarajevo",
45: "Africa/Lagos",
46: "Africa/Windhoek",
47: "Asia/Amman",
48: "Europe/Athens",
49: "Asia/Beirut",
50: "Africa/Cairo",
51: "Asia/Damascus",
52: "EET",
53: "Africa/Harare",
54: "Europe/Helsinki",
55: "Asia/Istanbul",
56: "Asia/Jerusalem",
57: "Europe/Kaliningrad",
58: "Africa/Tripoli",
59: "Asia/Baghdad",
60: "Asia/Kuwait",
61: "Europe/Minsk",
62: "Europe/Moscow",
63: "Africa/Nairobi",
64: "Asia/Tehran",
65: "Asia/Muscat",
66: "Asia/Baku",
67: "Europe/Samara",
68: "Indian/Mauritius",
69: "Asia/Tbilisi",
70: "Asia/Yerevan",
71: "Asia/Kabul",
72: "Asia/Ashgabat",
73: "Asia/Yekaterinburg",
74: "Asia/Karachi",
75: "Asia/Kolkata",
76: "Asia/Colombo",
77: "Asia/Kathmandu",
78: "Asia/Almaty",
79: "Asia/Dhaka",
80: "Asia/Novosibirsk",
81: "Asia/Rangoon",
82: "Asia/Bangkok",
83: "Asia/Krasnoyarsk",
84: "Asia/Chongqing",
85: "Asia/Irkutsk",
86: "Asia/Singapore",
87: "Australia/Perth",
88: "Asia/Taipei",
89: "Asia/Ulaanbaatar",
90: "Asia/Tokyo",
91: "Asia/Seoul",
92: "Asia/Yakutsk",
93: "Australia/Adelaide",
94: "Australia/Darwin",
95: "Australia/Brisbane",
96: "Australia/Canberra",
97: "Pacific/Guam",
98: "Australia/Hobart",
99: "Antarctica/DumontDUrville",
100: "Asia/Magadan",
101: "Asia/Srednekolymsk",
102: "Etc/GMT-11",
103: "Asia/Anadyr",
104: "Pacific/Auckland",
105: "Etc/GMT-12",
106: "Pacific/Fiji",
107: "Etc/GMT-13",
108: "Pacific/Apia",
109: "Etc/GMT-14",
}

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

@@ -0,0 +1,35 @@
"""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 .led import Led
from .light import Light
from .lighteffect import LightEffect
from .lightpreset import IotLightPreset, LightPreset
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",
"Led",
"Light",
"LightEffect",
"LightPreset",
"IotLightPreset",
"Motion",
"Rule",
"RuleModule",
"Schedule",
"Time",
"Usage",
]

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

@@ -0,0 +1,93 @@
"""Implementation of the ambient light (LAS) module found in some dimmers."""
import logging
from ...feature import Feature
from ..iotmodule import IotModule, merge
_LOGGER = logging.getLogger(__name__)
class AmbientLight(IotModule):
"""Implements ambient light controls for the motion sensor."""
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
device=self._device,
container=self,
id="ambient_light_enabled",
name="Ambient light enabled",
icon="mdi:brightness-percent",
attribute_getter="enabled",
attribute_setter="set_enabled",
type=Feature.Type.Switch,
category=Feature.Category.Config,
)
)
self._add_feature(
Feature(
device=self._device,
container=self,
id="ambient_light",
name="Ambient Light",
icon="mdi:brightness-percent",
attribute_getter="ambientlight_brightness",
type=Feature.Type.Sensor,
category=Feature.Category.Primary,
unit_getter=lambda: "%",
)
)
def query(self) -> dict:
"""Request configuration."""
req = merge(
self.query_for_command("get_config"),
self.query_for_command("get_current_brt"),
)
return req
@property
def config(self) -> dict:
"""Return current ambient light config."""
config = self.data["get_config"]
devs = config["devs"]
if len(devs) != 1:
_LOGGER.error("Unexpected number of devs in config: %s", config)
return devs[0]
@property
def presets(self) -> dict:
"""Return device-defined presets for brightness setting."""
return self.config["level_array"]
@property
def enabled(self) -> bool:
"""Return True if the module is enabled."""
return bool(self.config["enable"])
@property
def ambientlight_brightness(self) -> int:
"""Return True if the module is enabled."""
return int(self.data["get_current_brt"]["value"])
async def set_enabled(self, state: bool) -> dict:
"""Enable/disable LAS."""
return await self.call("set_enable", {"enable": int(state)})
async def current_brightness(self) -> dict:
"""Return current brightness.
Return value units.
"""
return await self.call("get_current_brt")
async def set_brightness_limit(self, value: int) -> dict:
"""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})

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

@@ -0,0 +1,10 @@
"""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.
"""

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

@@ -0,0 +1,77 @@
"""Cloud module implementation."""
from dataclasses import dataclass
from typing import Annotated
from mashumaro import DataClassDictMixin
from mashumaro.types import Alias
from ...feature import Feature
from ..iotmodule import IotModule
@dataclass
class CloudInfo(DataClassDictMixin):
"""Container for cloud settings."""
provisioned: Annotated[int, Alias("binded")]
cloud_connected: Annotated[int, Alias("cld_connection")]
firmware_download_page: Annotated[str, Alias("fwDlPage")]
firmware_notify_type: Annotated[int, Alias("fwNotifyType")]
illegal_type: Annotated[int, Alias("illegalType")]
server: str
stop_connect: Annotated[int, Alias("stopConnect")]
tcsp_info: Annotated[str, Alias("tcspInfo")]
tcsp_status: Annotated[int, Alias("tcspStatus")]
username: str
class Cloud(IotModule):
"""Module implementing support for cloud services."""
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
device=self._device,
container=self,
id="cloud_connection",
name="Cloud connection",
icon="mdi:cloud",
attribute_getter="is_connected",
type=Feature.Type.BinarySensor,
category=Feature.Category.Info,
)
)
@property
def is_connected(self) -> bool:
"""Return true if device is connected to the cloud."""
return bool(self.info.cloud_connected)
def query(self) -> dict:
"""Request cloud connectivity info."""
return self.query_for_command("get_info")
@property
def info(self) -> CloudInfo:
"""Return information about the cloud connectivity."""
return CloudInfo.from_dict(self.data["get_info"])
def get_available_firmwares(self) -> dict:
"""Return list of available firmwares."""
return self.query_for_command("get_intl_fw_list")
def set_server(self, url: str) -> dict:
"""Set the update server URL."""
return self.query_for_command("set_server_url", {"server": url})
def connect(self, username: str, password: str) -> dict:
"""Login to the cloud using given information."""
return self.query_for_command(
"bind", {"username": username, "password": password}
)
def disconnect(self) -> dict:
"""Disconnect from the cloud."""
return self.query_for_command("unbind")

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

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

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

@@ -0,0 +1,155 @@
"""Implementation of the emeter module."""
from __future__ import annotations
from datetime import datetime
from ...emeterstatus import EmeterStatus
from ...interfaces.energy import Energy as EnergyInterface
from .usage import Usage
class Emeter(Usage, EnergyInterface):
"""Emeter module."""
async def _post_update_hook(self) -> None:
self._supported = EnergyInterface.ModuleFeature.PERIODIC_STATS
if (
"voltage_mv" in self.data["get_realtime"]
or "voltage" in self.data["get_realtime"]
):
self._supported = (
self._supported | EnergyInterface.ModuleFeature.VOLTAGE_CURRENT
)
if (
"total_wh" in self.data["get_realtime"]
or "total" in self.data["get_realtime"]
):
self._supported = (
self._supported | EnergyInterface.ModuleFeature.CONSUMPTION_TOTAL
)
@property # type: ignore
def status(self) -> EmeterStatus:
"""Return current energy readings."""
return EmeterStatus(self.data["get_realtime"])
@property
def consumption_today(self) -> float | None:
"""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, 0.0)
@property
def consumption_this_month(self) -> float | None:
"""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, 0.0)
@property
def current_consumption(self) -> float | None:
"""Get the current power consumption in Watt."""
return self.status.power
@property
def consumption_total(self) -> float | None:
"""Return total consumption since last reboot in kWh."""
return self.status.total
@property
def current(self) -> float | None:
"""Return the current in A."""
return self.status.current
@property
def voltage(self) -> float | None:
"""Get the current voltage in V."""
return self.status.voltage
async def erase_stats(self) -> dict:
"""Erase all stats.
Uses different query than usage meter.
"""
return await self.call("erase_emeter_stat")
async def get_status(self) -> EmeterStatus:
"""Return real-time statistics."""
return EmeterStatus(await self.call("get_realtime"))
async def get_daily_stats(
self, *, year: int | None = None, month: int | None = None, kwh: bool = 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_monthly_stats(
self, *, year: int | None = None, kwh: bool = 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, int | float]],
entry_key: str,
kwh: bool = True,
key: int | None = None,
) -> dict[int | float, 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 {}

37
kasa/iot/modules/led.py Normal file
View File

@@ -0,0 +1,37 @@
"""Module for led controls."""
from __future__ import annotations
from ...interfaces.led import Led as LedInterface
from ..iotmodule import IotModule
class Led(IotModule, LedInterface):
"""Implementation of led controls."""
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {}
@property
def mode(self) -> str:
"""LED mode setting.
"always", "never"
"""
return "always" if self.led else "never"
@property
def led(self) -> bool:
"""Return the state of the led."""
sys_info = self.data
return bool(1 - sys_info["led_off"])
async def set_led(self, state: bool) -> dict:
"""Set the state of the led (night mode)."""
return await self.call("set_led_off", {"off": int(not state)})
@property
def is_supported(self) -> bool:
"""Return whether the module is supported by the device."""
return "led_off" in self.data

268
kasa/iot/modules/light.py Normal file
View File

@@ -0,0 +1,268 @@
"""Implementation of brightness module."""
from __future__ import annotations
from dataclasses import asdict
from typing import TYPE_CHECKING, Annotated, cast
from ...device_type import DeviceType
from ...exceptions import KasaException
from ...feature import Feature
from ...interfaces.light import HSV, ColorTempRange, LightState
from ...interfaces.light import Light as LightInterface
from ...module import FeatureAttribute
from ..iotmodule import IotModule
if TYPE_CHECKING:
from ..iotbulb import IotBulb
from ..iotdimmer import IotDimmer
BRIGHTNESS_MIN = 0
BRIGHTNESS_MAX = 100
class Light(IotModule, LightInterface):
"""Implementation of brightness module."""
_device: IotBulb | IotDimmer
_light_state: LightState
def _initialize_features(self) -> None:
"""Initialize features."""
super()._initialize_features()
device = self._device
if device._is_dimmable:
self._add_feature(
Feature(
device,
id="brightness",
name="Brightness",
container=self,
attribute_getter="brightness",
attribute_setter="set_brightness",
range_getter=lambda: (BRIGHTNESS_MIN, BRIGHTNESS_MAX),
type=Feature.Type.Number,
category=Feature.Category.Primary,
)
)
if device._is_variable_color_temp:
self._add_feature(
Feature(
device=device,
id="color_temperature",
name="Color temperature",
container=self,
attribute_getter="color_temp",
attribute_setter="set_color_temp",
range_getter="valid_temperature_range",
category=Feature.Category.Primary,
type=Feature.Type.Number,
)
)
if device._is_color:
self._add_feature(
Feature(
device=device,
id="hsv",
name="HSV",
container=self,
attribute_getter="hsv",
attribute_setter="set_hsv",
# TODO proper type for setting hsv
type=Feature.Type.Unknown,
)
)
def query(self) -> dict:
"""Query to execute during the update cycle."""
# Brightness is contained in the main device info response.
return {}
def _get_bulb_device(self) -> IotBulb | None:
"""For type checker this gets an IotBulb.
IotDimmer is not a subclass of IotBulb and using isinstance
here at runtime would create a circular import.
"""
if self._device.device_type in {DeviceType.Bulb, DeviceType.LightStrip}:
return cast("IotBulb", self._device)
return None
@property # type: ignore
def is_dimmable(self) -> int:
"""Whether the bulb supports brightness changes."""
return self._device._is_dimmable
@property # type: ignore
def brightness(self) -> Annotated[int, FeatureAttribute()]:
"""Return the current brightness in percentage."""
return self._device._brightness
async def set_brightness(
self, brightness: int, *, transition: int | None = None
) -> Annotated[dict, FeatureAttribute()]:
"""Set the brightness in percentage. A value of 0 will turn off the light.
:param int brightness: brightness in percent
:param int transition: transition in milliseconds.
"""
return await self.set_state(
LightState(brightness=brightness, transition=transition)
)
@property
def is_color(self) -> bool:
"""Whether the light supports color changes."""
if (bulb := self._get_bulb_device()) is None:
return False
return bulb._is_color
@property
def is_variable_color_temp(self) -> bool:
"""Whether the bulb supports color temperature changes."""
if (bulb := self._get_bulb_device()) is None:
return False
return bulb._is_variable_color_temp
@property
def has_effects(self) -> bool:
"""Return True if the device supports effects."""
if (bulb := self._get_bulb_device()) is None:
return False
return bulb._has_effects
@property
def hsv(self) -> Annotated[HSV, FeatureAttribute()]:
"""Return the current HSV state of the bulb.
:return: hue, saturation and value (degrees, %, %)
"""
if (bulb := self._get_bulb_device()) is None or not bulb._is_color:
raise KasaException("Light does not support color.")
return bulb._hsv
async def set_hsv(
self,
hue: int,
saturation: int,
value: int | None = None,
*,
transition: int | None = None,
) -> Annotated[dict, FeatureAttribute()]:
"""Set new HSV.
Note, transition is not supported and will be ignored.
: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 (bulb := self._get_bulb_device()) is None or not bulb._is_color:
raise KasaException("Light does not support color.")
return await bulb._set_hsv(hue, saturation, value, transition=transition)
@property
def valid_temperature_range(self) -> ColorTempRange:
"""Return the device-specific white temperature range (in Kelvin).
:return: White temperature range in Kelvin (minimum, maximum)
"""
if (
bulb := self._get_bulb_device()
) is None or not bulb._is_variable_color_temp:
raise KasaException("Light does not support colortemp.")
return bulb._valid_temperature_range
@property
def color_temp(self) -> Annotated[int, FeatureAttribute()]:
"""Whether the bulb supports color temperature changes."""
if (
bulb := self._get_bulb_device()
) is None or not bulb._is_variable_color_temp:
raise KasaException("Light does not support colortemp.")
return bulb._color_temp
async def set_color_temp(
self, temp: int, *, brightness: int | None = None, transition: int | None = None
) -> Annotated[dict, FeatureAttribute()]:
"""Set the color temperature of the device in kelvin.
Note, transition is not supported and will be ignored.
:param int temp: The new color temperature, in Kelvin
:param int transition: transition in milliseconds.
"""
if (
bulb := self._get_bulb_device()
) is None or not bulb._is_variable_color_temp:
raise KasaException("Light does not support colortemp.")
return await bulb._set_color_temp(
temp, brightness=brightness, transition=transition
)
async def set_state(self, state: LightState) -> dict:
"""Set the light state."""
# iot protocol Dimmers and smart protocol devices do not support
# brightness of 0 so 0 will turn off all devices for consistency
if (bulb := self._get_bulb_device()) is None: # Dimmer
if TYPE_CHECKING:
assert isinstance(self._device, IotDimmer)
if state.brightness == 0 or state.light_on is False:
return await self._device.turn_off(transition=state.transition)
elif state.brightness:
# set_dimmer_transition will turn on the device
return await self._device.set_dimmer_transition(
state.brightness, state.transition or 0
)
return await self._device.turn_on(transition=state.transition)
else:
transition = state.transition
state_dict = asdict(state)
state_dict = {k: v for k, v in state_dict.items() if v is not None}
if "transition" in state_dict:
del state_dict["transition"]
state_dict["on_off"] = 1 if state.light_on is None else int(state.light_on)
if state_dict.get("brightness") == 0:
state_dict["on_off"] = 0
del state_dict["brightness"]
# If light on state not set default to on.
elif state.light_on is None:
state_dict["on_off"] = 1
else:
state_dict["on_off"] = int(state.light_on)
# Remove the light_on from the dict
state_dict.pop("light_on", None)
return await bulb._set_light_state(state_dict, transition=transition)
@property
def state(self) -> LightState:
"""Return the current light state."""
return self._light_state
async def _post_update_hook(self) -> None:
device = self._device
if device.is_on is False:
state = LightState(light_on=False)
else:
state = LightState(light_on=True)
if device._is_dimmable:
state.brightness = self.brightness
if device._is_color:
hsv = self.hsv
state.hue = hsv.hue
state.saturation = hsv.saturation
if device._is_variable_color_temp:
state.color_temp = self.color_temp
self._light_state = state
async def _deprecated_set_light_state(
self, state: dict, *, transition: int | None = None
) -> dict:
"""Set the light state."""
if (bulb := self._get_bulb_device()) is None:
raise KasaException("Device does not support set_light_state")
else:
return await bulb._set_light_state(state, transition=transition)

135
kasa/iot/modules/lighteffect.py Normal file
View File

@@ -0,0 +1,135 @@
"""Module for light effects."""
from __future__ import annotations
from ...interfaces.lighteffect import LightEffect as LightEffectInterface
from ..effects import EFFECT_MAPPING_V1, EFFECT_NAMES_V1
from ..iotmodule import IotModule
class LightEffect(IotModule, LightEffectInterface):
"""Implementation of dynamic light effects."""
@property
def effect(self) -> str:
"""Return effect state.
Example:
{'brightness': 50,
'custom': 0,
'enable': 0,
'id': '',
'name': ''}
"""
eff = self.data["lighting_effect_state"]
name = eff["name"]
if eff["enable"]:
return name
return self.LIGHT_EFFECTS_OFF
@property
def brightness(self) -> int:
"""Return light effect brightness."""
return self.data["lighting_effect_state"]["brightness"]
@property
def effect_list(self) -> list[str]:
"""Return built-in effects list.
Example:
['Aurora', 'Bubbling Cauldron', ...]
"""
effect_list = [self.LIGHT_EFFECTS_OFF]
effect_list.extend(EFFECT_NAMES_V1)
return effect_list
async def set_effect(
self,
effect: str,
*,
brightness: int | None = None,
transition: int | None = None,
) -> dict:
"""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 == self.LIGHT_EFFECTS_OFF:
if self.effect in EFFECT_MAPPING_V1:
# TODO: We could query get_lighting_effect here to
# get the custom effect although not sure how to find
# custom effects
effect_dict = EFFECT_MAPPING_V1[self.effect]
else:
effect_dict = EFFECT_MAPPING_V1["Aurora"]
effect_dict = {**effect_dict}
effect_dict["enable"] = 0
return await self.set_custom_effect(effect_dict)
elif effect not in EFFECT_MAPPING_V1:
raise ValueError(f"The effect {effect} is not a built in effect.")
else:
effect_dict = EFFECT_MAPPING_V1[effect]
effect_dict = {**effect_dict}
if brightness is not None:
effect_dict["brightness"] = brightness
if transition is not None:
effect_dict["transition"] = transition
return await self.set_custom_effect(effect_dict)
async def set_custom_effect(
self,
effect_dict: dict,
) -> dict:
"""Set a custom effect on the device.
:param str effect_dict: The custom effect dict to set
"""
return await self.call(
"set_lighting_effect",
effect_dict,
)
@property
def has_custom_effects(self) -> bool:
"""Return True if the device supports setting custom effects."""
return True
def query(self) -> dict:
"""Return the base query."""
return {}
@property # type: ignore
def _deprecated_effect(self) -> dict:
"""Return effect state.
Example:
{'brightness': 50,
'custom': 0,
'enable': 0,
'id': '',
'name': ''}
"""
# LightEffectModule returns the current effect name
# so return the dict here for backwards compatibility
return self.data["lighting_effect_state"]
@property # type: ignore
def _deprecated_effect_list(self) -> list[str] | None:
"""Return built-in effects list.
Example:
['Aurora', 'Bubbling Cauldron', ...]
"""
# LightEffectModule returns effect names along with a LIGHT_EFFECTS_OFF value
# so return the original effect names here for backwards compatibility
return EFFECT_NAMES_V1

169
kasa/iot/modules/lightpreset.py Normal file
View File

@@ -0,0 +1,169 @@
"""Light preset module."""
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import asdict, dataclass
from typing import TYPE_CHECKING
from mashumaro.config import BaseConfig
from ...exceptions import KasaException
from ...interfaces import LightPreset as LightPresetInterface
from ...interfaces import LightState
from ...json import DataClassJSONMixin
from ...module import Module
from ..iotmodule import IotModule
if TYPE_CHECKING:
pass
# type ignore can be removed after migration mashumaro:
# error: Signature of "__replace__" incompatible with supertype "LightState"
@dataclass(kw_only=True, repr=False)
class IotLightPreset(DataClassJSONMixin, LightState): # type: ignore[override]
"""Light configuration preset."""
class Config(BaseConfig):
"""Config class."""
omit_none = True
index: int
brightness: int
# These are not available for effect mode presets on light strips
hue: int | None = None
saturation: int | None = None
color_temp: int | None = None
# Variables for effect mode presets
custom: int | None = None
id: str | None = None
mode: int | None = None
class LightPreset(IotModule, LightPresetInterface):
"""Class for setting light presets."""
_presets: dict[str, IotLightPreset]
_preset_list: list[str]
async def _post_update_hook(self) -> None:
"""Update the internal presets."""
self._presets = {
f"Light preset {index+1}": IotLightPreset.from_dict(vals)
for index, vals in enumerate(self.data["preferred_state"])
# Devices may list some light effects along with normal presets but these
# are handled by the LightEffect module so exclude preferred states with id
if "id" not in vals
}
self._preset_list = [self.PRESET_NOT_SET]
self._preset_list.extend(self._presets.keys())
@property
def preset_list(self) -> list[str]:
"""Return built-in effects list.
Example:
['Off', 'Preset 1', 'Preset 2', ...]
"""
return self._preset_list
@property
def preset_states_list(self) -> Sequence[IotLightPreset]:
"""Return built-in effects list.
Example:
['Off', 'Preset 1', 'Preset 2', ...]
"""
return list(self._presets.values())
@property
def preset(self) -> str:
"""Return current preset name."""
light = self._device.modules[Module.Light]
is_color = light.has_feature("hsv")
is_variable_color_temp = light.has_feature("color_temp")
brightness = light.brightness
color_temp = light.color_temp if is_variable_color_temp else None
h, s = (light.hsv.hue, light.hsv.saturation) if is_color else (None, None)
for preset_name, preset in self._presets.items():
if (
preset.brightness == brightness
and (preset.color_temp == color_temp or not is_variable_color_temp)
and (preset.hue == h or not is_color)
and (preset.saturation == s or not is_color)
):
return preset_name
return self.PRESET_NOT_SET
async def set_preset(
self,
preset_name: str,
) -> dict:
"""Set a light preset for the device."""
light = self._device.modules[Module.Light]
if preset_name == self.PRESET_NOT_SET:
if light.has_feature("hsv"):
preset = LightState(hue=0, saturation=0, brightness=100)
else:
preset = LightState(brightness=100)
elif (preset := self._presets.get(preset_name)) is None: # type: ignore[assignment]
raise ValueError(f"{preset_name} is not a valid preset: {self.preset_list}")
return await light.set_state(preset)
@property
def has_save_preset(self) -> bool:
"""Return True if the device supports updating presets."""
return True
async def save_preset(
self,
preset_name: str,
preset_state: LightState,
) -> dict:
"""Update the preset with preset_name with the new preset_info."""
if len(self._presets) == 0:
raise KasaException("Device does not supported saving presets")
if preset_name not in self._presets:
raise ValueError(f"{preset_name} is not a valid preset: {self.preset_list}")
index = list(self._presets.keys()).index(preset_name)
state = asdict(preset_state)
state = {k: v for k, v in state.items() if v is not None}
state["index"] = index
return await self.call("set_preferred_state", state)
def query(self) -> dict:
"""Return the base query."""
return {}
@property # type: ignore
def _deprecated_presets(self) -> list[IotLightPreset]:
"""Return a list of available bulb setting presets."""
return [
IotLightPreset(**vals)
for vals in self._device.sys_info["preferred_state"]
if "id" not in vals
]
async def _deprecated_save_preset(self, preset: IotLightPreset) -> dict:
"""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 KasaException("Device does not supported saving presets")
if preset.index >= len(self._presets):
raise KasaException("Invalid preset index")
return await self.call("set_preferred_state", preset.to_dict())

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

@@ -0,0 +1,102 @@
"""Implementation of the motion detection (PIR) module found in some dimmers."""
from __future__ import annotations
import logging
from enum import Enum
from ...exceptions import KasaException
from ...feature import Feature
from ..iotmodule import IotModule
_LOGGER = logging.getLogger(__name__)
class Range(Enum):
"""Range for motion detection."""
Far = 0
Mid = 1
Near = 2
Custom = 3
class Motion(IotModule):
"""Implements the motion detection (PIR) module."""
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
# Only add features if the device supports the module
if "get_config" not in self.data:
return
if "enable" not in self.config:
_LOGGER.warning("%r initialized, but no enable in response")
return
self._add_feature(
Feature(
device=self._device,
container=self,
id="pir_enabled",
name="PIR enabled",
icon="mdi:motion-sensor",
attribute_getter="enabled",
attribute_setter="set_enabled",
type=Feature.Type.Switch,
category=Feature.Category.Config,
)
)
def query(self) -> dict:
"""Request PIR configuration."""
return self.query_for_command("get_config")
@property
def config(self) -> dict:
"""Return current configuration."""
return self.data["get_config"]
@property
def range(self) -> Range:
"""Return motion detection range."""
return Range(self.config["trigger_index"])
@property
def enabled(self) -> bool:
"""Return True if module is enabled."""
return bool(self.config["enable"])
async def set_enabled(self, state: bool) -> dict:
"""Enable/disable PIR."""
return await self.call("set_enable", {"enable": int(state)})
async def set_range(
self, *, range: Range | None = None, custom_range: int | None = None
) -> dict:
"""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 KasaException("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.config["cold_time"]
async def set_inactivity_timeout(self, timeout: int) -> dict:
"""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."""
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from mashumaro import DataClassDictMixin
from ..iotmodule 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
@dataclass
class Rule(DataClassDictMixin):
"""Representation of a rule."""
id: str
name: str
enable: int
wday: list[int]
repeat: int
# start action
sact: Action | None = None
stime_opt: TimeOption | None = None
smin: int | None = None
eact: Action | None = None
etime_opt: TimeOption | None = None
emin: int | None = None
# Only on bulbs
s_light: dict | None = None
_LOGGER = logging.getLogger(__name__)
class RuleModule(IotModule):
"""Base class for rule-based modules, such as countdown and antitheft."""
def query(self) -> dict:
"""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.from_dict(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) -> dict:
"""Enable or disable the service."""
return await self.call("set_overall_enable", {"enable": state})
async def delete_rule(self, rule: Rule) -> dict:
"""Delete the given rule."""
return await self.call("delete_rule", {"id": rule.id})
async def delete_all_rules(self) -> dict:
"""Delete all rules."""
return await self.call("delete_all_rules")

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

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

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

@@ -0,0 +1,93 @@
"""Provides the current time and timezone information."""
from __future__ import annotations
from datetime import UTC, datetime, tzinfo
from ...exceptions import KasaException
from ...interfaces import Time as TimeInterface
from ..iotmodule import IotModule, merge
from ..iottimezone import get_timezone, get_timezone_index
class Time(IotModule, TimeInterface):
"""Implements the timezone settings."""
_timezone: tzinfo = UTC
def query(self) -> dict:
"""Request time and timezone."""
q = self.query_for_command("get_time")
merge(q, self.query_for_command("get_timezone"))
return q
async def _post_update_hook(self) -> None:
"""Perform actions after a device update."""
if res := self.data.get("get_timezone"):
self._timezone = await get_timezone(res.get("index"))
@property
def time(self) -> datetime:
"""Return current device time."""
res = self.data["get_time"]
time = datetime(
res["year"],
res["month"],
res["mday"],
res["hour"],
res["min"],
res["sec"],
tzinfo=self.timezone,
)
return time
@property
def timezone(self) -> tzinfo:
"""Return current timezone."""
return self._timezone
async def get_time(self) -> datetime | None:
"""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"],
tzinfo=self.timezone,
)
except KasaException:
return None
async def set_time(self, dt: datetime) -> dict:
"""Set the device time."""
params = {
"year": dt.year,
"month": dt.month,
"mday": dt.day,
"hour": dt.hour,
"min": dt.minute,
"sec": dt.second,
}
if dt.tzinfo:
index = await get_timezone_index(dt.tzinfo)
current_index = self.data.get("get_timezone", {}).get("index", -1)
if current_index != -1 and current_index != index:
params["index"] = index
method = "set_timezone"
else:
method = "set_time"
else:
method = "set_time"
try:
return await self.call(method, params)
except Exception as ex:
raise KasaException(ex) from ex
async def get_timezone(self) -> dict:
"""Request timezone information from the device."""
return await self.call("get_timezone")

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

@@ -0,0 +1,122 @@
"""Implementation of the usage interface."""
from __future__ import annotations
from datetime import datetime
from ..iotmodule import IotModule, merge
class Usage(IotModule):
"""Baseclass for emeter/usage interfaces."""
def query(self) -> dict:
"""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) -> int:
"""Estimated maximum query response size."""
return 2048
@property
def daily_data(self) -> list[dict]:
"""Return statistics on daily basis."""
return self.data["get_daystat"]["day_list"]
@property
def monthly_data(self) -> list[dict]:
"""Return statistics on monthly basis."""
return self.data["get_monthstat"]["month_list"]
@property
def usage_today(self) -> int | None:
"""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) -> int | None:
"""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: int | None = None, month: int | None = 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: int | None = 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: int | None = None, month: int | None = 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: int | None = 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) -> dict:
"""Erase all stats."""
return await self.call("erase_runtime_stat")
def _convert_stat_data(self, data: list[dict], entry_key: str) -> 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 {}
res = {entry[entry_key]: entry["time"] for entry in data}
return res