Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2026-03-15 04:57:05 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/master' into feat/device_update

Browse Source
This commit is contained in:
sdb9696
2024-05-14 18:35:23 +01:00
parent fc122eac6b 67b5d7de83
commit 06c5e55ac6
83 changed files with 3312 additions and 933 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
kasa/smart/modules/__init__.py
View File

@@ -1,49 +1,53 @@
"""Modules for SMART devices."""
from .alarmmodule import AlarmModule
from .autooffmodule import AutoOffModule
from .battery import BatterySensor
from .alarm import Alarm
from .autooff import AutoOff
from .batterysensor import BatterySensor
from .brightness import Brightness
from .childdevicemodule import ChildDeviceModule
from .cloudmodule import CloudModule
from .colormodule import ColorModule
from .colortemp import ColorTemperatureModule
from .childdevice import ChildDevice
from .cloud import Cloud
from .color import Color
from .colortemperature import ColorTemperature
from .contactsensor import ContactSensor
from .devicemodule import DeviceModule
from .energymodule import EnergyModule
from .fanmodule import FanModule
from .energy import Energy
from .fan import Fan
from .firmware import Firmware
from .frostprotection import FrostProtectionModule
from .humidity import HumiditySensor
from .ledmodule import LedModule
from .lighteffectmodule import LightEffectModule
from .lighttransitionmodule import LightTransitionModule
from .reportmodule import ReportModule
from .temperature import TemperatureSensor
from .frostprotection import FrostProtection
from .humiditysensor import HumiditySensor
from .led import Led
from .light import Light
from .lighteffect import LightEffect
from .lighttransition import LightTransition
from .reportmode import ReportMode
from .temperaturecontrol import TemperatureControl
from .timemodule import TimeModule
from .waterleak import WaterleakSensor
from .temperaturesensor import TemperatureSensor
from .time import Time
from .waterleaksensor import WaterleakSensor
__all__ = [
"AlarmModule",
"TimeModule",
"EnergyModule",
"Alarm",
"Time",
"Energy",
"DeviceModule",
"ChildDeviceModule",
"ChildDevice",
"BatterySensor",
"HumiditySensor",
"TemperatureSensor",
"TemperatureControl",
"ReportModule",
"AutoOffModule",
"LedModule",
"ReportMode",
"AutoOff",
"Led",
"Brightness",
"FanModule",
"Fan",
"Firmware",
"CloudModule",
"LightEffectModule",
"LightTransitionModule",
"ColorTemperatureModule",
"ColorModule",
"Cloud",
"Light",
"LightEffect",
"LightTransition",
"ColorTemperature",
"Color",
"WaterleakSensor",
"FrostProtectionModule",
"ContactSensor",
"FrostProtection",
]

2
kasa/smart/modules/alarmmodule.py → kasa/smart/modules/alarm.py
View File

@@ -6,7 +6,7 @@ from ...feature import Feature
from ..smartmodule import SmartModule
class AlarmModule(SmartModule):
class Alarm(SmartModule):
"""Implementation of alarm module."""
REQUIRED_COMPONENT = "alarm"

2
kasa/smart/modules/autooffmodule.py → kasa/smart/modules/autooff.py
View File

@@ -12,7 +12,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class AutoOffModule(SmartModule):
class AutoOff(SmartModule):
"""Implementation of auto off module."""
REQUIRED_COMPONENT = "auto_off"

39
kasa/smart/modules/battery.py → kasa/smart/modules/batterysensor.py
View File

@@ -2,14 +2,9 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature
from ..smartmodule import SmartModule
if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class BatterySensor(SmartModule):
"""Implementation of battery module."""
@@ -17,23 +12,11 @@ class BatterySensor(SmartModule):
REQUIRED_COMPONENT = "battery_detect"
QUERY_GETTER_NAME = "get_battery_detect_info"
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
def _initialize_features(self):
"""Initialize features."""
self._add_feature(
Feature(
device,
"battery_level",
"Battery level",
container=self,
attribute_getter="battery",
icon="mdi:battery",
unit="%",
category=Feature.Category.Info,
)
)
self._add_feature(
Feature(
device,
self._device,
"battery_low",
"Battery low",
container=self,
@@ -44,6 +27,22 @@ class BatterySensor(SmartModule):
)
)
# Some devices, like T110 contact sensor do not report the battery percentage
if "battery_percentage" in self._device.sys_info:
self._add_feature(
Feature(
self._device,
"battery_level",
"Battery level",
container=self,
attribute_getter="battery",
icon="mdi:battery",
unit="%",
category=Feature.Category.Info,
type=Feature.Type.Sensor,
)
)
@property
def battery(self):
"""Return battery level."""

24
kasa/smart/modules/brightness.py
View File

@@ -2,16 +2,10 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature
from ..smartmodule import SmartModule
if TYPE_CHECKING:
from ..smartdevice import SmartDevice
BRIGHTNESS_MIN = 1
BRIGHTNESS_MIN = 0
BRIGHTNESS_MAX = 100
@@ -20,8 +14,11 @@ class Brightness(SmartModule):
REQUIRED_COMPONENT = "brightness"
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
def _initialize_features(self):
"""Initialize features."""
super()._initialize_features()
device = self._device
self._add_feature(
Feature(
device,
@@ -47,8 +44,11 @@ class Brightness(SmartModule):
"""Return current brightness."""
return self.data["brightness"]
async def set_brightness(self, brightness: int):
"""Set the brightness."""
async def set_brightness(self, brightness: int, *, transition: int | None = None):
"""Set the brightness. A brightness value of 0 will turn off the light.
Note, transition is not supported and will be ignored.
"""
if not isinstance(brightness, int) or not (
BRIGHTNESS_MIN <= brightness <= BRIGHTNESS_MAX
):
@@ -57,6 +57,8 @@ class Brightness(SmartModule):
f"(valid range: {BRIGHTNESS_MIN}-{BRIGHTNESS_MAX}%)"
)
if brightness == 0:
return await self._device.turn_off()
return await self.call("set_device_info", {"brightness": brightness})
async def _check_supported(self):

2
kasa/smart/modules/childdevicemodule.py → kasa/smart/modules/childdevice.py
View File

@@ -3,7 +3,7 @@
from ..smartmodule import SmartModule
class ChildDeviceModule(SmartModule):
class ChildDevice(SmartModule):
"""Implementation for child devices."""
REQUIRED_COMPONENT = "child_device"

2
kasa/smart/modules/cloudmodule.py → kasa/smart/modules/cloud.py
View File

@@ -12,7 +12,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class CloudModule(SmartModule):
class Cloud(SmartModule):
"""Implementation of cloud module."""
QUERY_GETTER_NAME = "get_connect_cloud_state"

4
kasa/smart/modules/colormodule.py → kasa/smart/modules/color.py
View File

@@ -4,15 +4,15 @@ from __future__ import annotations
from typing import TYPE_CHECKING
from ...bulb import HSV
from ...feature import Feature
from ...interfaces.light import HSV
from ..smartmodule import SmartModule
if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class ColorModule(SmartModule):
class Color(SmartModule):
"""Implementation of color module."""
REQUIRED_COMPONENT = "color"

4
kasa/smart/modules/colortemp.py → kasa/smart/modules/colortemperature.py
View File

@@ -5,8 +5,8 @@ from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from ...bulb import ColorTempRange
from ...feature import Feature
from ...interfaces.light import ColorTempRange
from ..smartmodule import SmartModule
if TYPE_CHECKING:
@@ -18,7 +18,7 @@ _LOGGER = logging.getLogger(__name__)
DEFAULT_TEMP_RANGE = [2500, 6500]
class ColorTemperatureModule(SmartModule):
class ColorTemperature(SmartModule):
"""Implementation of color temp module."""
REQUIRED_COMPONENT = "color_temperature"

42
kasa/smart/modules/contactsensor.py Normal file
View File

@@ -0,0 +1,42 @@
"""Implementation of contact sensor module."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature
from ..smartmodule import SmartModule
if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class ContactSensor(SmartModule):
"""Implementation of contact sensor module."""
REQUIRED_COMPONENT = None # we depend on availability of key
REQUIRED_KEY_ON_PARENT = "open"
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
device,
id="is_open",
name="Open",
container=self,
attribute_getter="is_open",
icon="mdi:door",
category=Feature.Category.Primary,
type=Feature.Type.BinarySensor,
)
)
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {}
@property
def is_open(self):
"""Return True if the contact sensor is open."""
return self._device.sys_info["open"]

2
kasa/smart/modules/energymodule.py → kasa/smart/modules/energy.py
View File

@@ -12,7 +12,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class EnergyModule(SmartModule):
class Energy(SmartModule):
"""Implementation of energy monitoring module."""
REQUIRED_COMPONENT = "energy_monitoring"

2
kasa/smart/modules/fanmodule.py → kasa/smart/modules/fan.py
View File

@@ -11,7 +11,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class FanModule(SmartModule):
class Fan(SmartModule):
"""Implementation of fan_control module."""
REQUIRED_COMPONENT = "fan_control"

98
kasa/smart/modules/firmware.py
View File

@@ -5,10 +5,8 @@ from __future__ import annotations
import asyncio
import logging
from datetime import date
from typing import TYPE_CHECKING, Any, Optional
from typing import TYPE_CHECKING, Any, Callable, Coroutine, Optional
# When support for cpython older than 3.11 is dropped
# async_timeout can be replaced with asyncio.timeout
# When support for cpython older than 3.11 is dropped
# async_timeout can be replaced with asyncio.timeout
from async_timeout import timeout as asyncio_timeout
@@ -16,9 +14,12 @@ from pydantic.v1 import BaseModel, Field, validator
from ...exceptions import SmartErrorCode
from ...feature import Feature
from ...firmware import Firmware as FirmwareInterface
from ...firmware import FirmwareUpdate as FirmwareUpdateInterface
from ...firmware import UpdateResult
from ...interfaces import Firmware as FirmwareInterface
from ...interfaces.firmware import (
FirmwareDownloadState as FirmwareDownloadStateInterface,
)
from ...interfaces.firmware import FirmwareUpdateInfo as FirmwareUpdateInfoInterface
from ...interfaces.firmware import UpdateResult
from ..smartmodule import SmartModule
if TYPE_CHECKING:
@@ -28,7 +29,20 @@ if TYPE_CHECKING:
_LOGGER = logging.getLogger(__name__)
class FirmwareUpdate(BaseModel):
class DownloadState(BaseModel):
"""Download state."""
# Example:
# {'status': 0, 'download_progress': 0, 'reboot_time': 5,
# 'upgrade_time': 5, 'auto_upgrade': False}
status: int
progress: int = Field(alias="download_progress")
reboot_time: int
upgrade_time: int
auto_upgrade: bool
class FirmwareUpdateInfo(BaseModel):
"""Update info status object."""
status: int = Field(alias="type")
@@ -91,7 +105,7 @@ class Firmware(SmartModule, FirmwareInterface):
name="Current firmware version",
container=self,
attribute_getter="current_firmware",
category=Feature.Category.Info,
category=Feature.Category.Debug,
)
)
self._add_feature(
@@ -101,7 +115,7 @@ class Firmware(SmartModule, FirmwareInterface):
name="Available firmware version",
container=self,
attribute_getter="latest_firmware",
category=Feature.Category.Info,
category=Feature.Category.Debug,
)
)
@@ -128,9 +142,9 @@ class Firmware(SmartModule, FirmwareInterface):
fw = self.data.get("get_latest_fw") or self.data
if not self._device.is_cloud_connected or isinstance(fw, SmartErrorCode):
# Error in response, probably disconnected from the cloud.
return FirmwareUpdate(type=0, need_to_upgrade=False)
return FirmwareUpdateInfo(type=0, need_to_upgrade=False)
return FirmwareUpdate.parse_obj(fw)
return FirmwareUpdateInfo.parse_obj(fw)
@property
def update_available(self) -> bool | None:
@@ -139,31 +153,60 @@ class Firmware(SmartModule, FirmwareInterface):
return None
return self.firmware_update_info.update_available
async def get_update_state(self):
async def get_update_state(self) -> DownloadState:
"""Return update state."""
return await self.call("get_fw_download_state")
resp = await self.call("get_fw_download_state")
state = resp["get_fw_download_state"]
return DownloadState(**state)
async def update(self):
async def update(
self, progress_cb: Callable[[DownloadState], Coroutine] | None = None
):
"""Update the device firmware."""
current_fw = self.current_firmware
_LOGGER.debug(
_LOGGER.info(
"Going to upgrade from %s to %s",
current_fw,
self.firmware_update_info.version,
)
resp = await self.call("fw_download")
_LOGGER.debug("Update request response: %s", resp)
await self.call("fw_download")
# TODO: read timeout from get_auto_update_info or from get_fw_download_state?
async with asyncio_timeout(60 * 5):
while True:
await asyncio.sleep(0.5)
state = await self.get_update_state()
_LOGGER.debug("Update state: %s" % state)
# TODO: this could await a given callable for progress
try:
state = await self.get_update_state()
except Exception as ex:
_LOGGER.warning(
"Got exception, maybe the device is rebooting? %s", ex
)
continue
if self.firmware_update_info.version != current_fw:
_LOGGER.info("Updated to %s", self.firmware_update_info.version)
_LOGGER.debug("Update state: %s" % state)
if progress_cb is not None:
asyncio.create_task(progress_cb(state))
if state.status == 0:
_LOGGER.info(
"Update idle, hopefully updated to %s",
self.firmware_update_info.version,
)
break
elif state.status == 2:
_LOGGER.info("Downloading firmware, progress: %s", state.progress)
elif state.status == 3:
upgrade_sleep = state.upgrade_time
_LOGGER.info(
"Flashing firmware, sleeping for %s before checking status",
upgrade_sleep,
)
await asyncio.sleep(upgrade_sleep)
elif state.status < 0:
_LOGGER.error("Got error: %s", state.status)
break
else:
_LOGGER.warning("Unhandled state code: %s", state)
@property
def auto_update_enabled(self):
@@ -178,16 +221,21 @@ class Firmware(SmartModule, FirmwareInterface):
data = {**self.data["get_auto_update_info"], "enable": enabled}
await self.call("set_auto_update_info", data)
async def update_firmware(self, *, progress_cb) -> UpdateResult:
async def update_firmware(
self,
*,
progress_cb: Callable[[FirmwareDownloadStateInterface], Coroutine]
| None = None,
) -> UpdateResult:
"""Update the firmware."""
# TODO: implement, this is part of the common firmware API
raise NotImplementedError
async def check_for_updates(self) -> FirmwareUpdateInterface:
async def check_for_updates(self) -> FirmwareUpdateInfoInterface:
"""Return firmware update information."""
# TODO: naming of the common firmware API methods
info = self.firmware_update_info
return FirmwareUpdateInterface(
return FirmwareUpdateInfoInterface(
current_version=self.current_firmware,
update_available=info.update_available,
available_version=info.version,

2
kasa/smart/modules/frostprotection.py
View File

@@ -12,7 +12,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class FrostProtectionModule(SmartModule):
class FrostProtection(SmartModule):
"""Implementation for frost protection module.
This basically turns the thermostat on and off.

0
kasa/smart/modules/humidity.py → kasa/smart/modules/humiditysensor.py
View File

27
kasa/smart/modules/ledmodule.py → kasa/smart/modules/led.py
View File

@@ -2,37 +2,16 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature
from ...interfaces.led import Led as LedInterface
from ..smartmodule import SmartModule
if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class LedModule(SmartModule):
class Led(SmartModule, LedInterface):
"""Implementation of led controls."""
REQUIRED_COMPONENT = "led"
QUERY_GETTER_NAME = "get_led_info"
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
device=device,
container=self,
id="led",
name="LED",
icon="mdi:led-{state}",
attribute_getter="led",
attribute_setter="set_led",
type=Feature.Type.Switch,
category=Feature.Category.Config,
)
)
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {self.QUERY_GETTER_NAME: {"led_rule": None}}
@@ -56,7 +35,7 @@ class LedModule(SmartModule):
This should probably be a select with always/never/nightmode.
"""
rule = "always" if enable else "never"
return await self.call("set_led_info", self.data | {"led_rule": rule})
return await self.call("set_led_info", dict(self.data, **{"led_rule": rule}))
@property
def night_mode_settings(self):

126
kasa/smart/modules/light.py Normal file
View File

@@ -0,0 +1,126 @@
"""Module for led controls."""
from __future__ import annotations
from ...exceptions import KasaException
from ...interfaces.light import HSV, ColorTempRange
from ...interfaces.light import Light as LightInterface
from ...module import Module
from ..smartmodule import SmartModule
class Light(SmartModule, LightInterface):
"""Implementation of a light."""
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {}
@property
def is_color(self) -> bool:
"""Whether the bulb supports color changes."""
return Module.Color in self._device.modules
@property
def is_dimmable(self) -> bool:
"""Whether the bulb supports brightness changes."""
return Module.Brightness in self._device.modules
@property
def is_variable_color_temp(self) -> bool:
"""Whether the bulb supports color temperature changes."""
return Module.ColorTemperature in self._device.modules
@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 not self.is_variable_color_temp:
raise KasaException("Color temperature not supported")
return self._device.modules[Module.ColorTemperature].valid_temperature_range
@property
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.")
return self._device.modules[Module.Color].hsv
@property
def color_temp(self) -> int:
"""Whether the bulb supports color temperature changes."""
if not self.is_variable_color_temp:
raise KasaException("Bulb does not support colortemp.")
return self._device.modules[Module.ColorTemperature].color_temp
@property
def brightness(self) -> int:
"""Return the current brightness in percentage."""
if not self.is_dimmable: # pragma: no cover
raise KasaException("Bulb is not dimmable.")
return self._device.modules[Module.Brightness].brightness
async def set_hsv(
self,
hue: int,
saturation: int,
value: int | None = None,
*,
transition: int | None = None,
) -> dict:
"""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 between 1 and 100
:param int transition: transition in milliseconds.
"""
if not self.is_color:
raise KasaException("Bulb does not support color.")
return await self._device.modules[Module.Color].set_hsv(hue, saturation, value)
async def set_color_temp(
self, temp: int, *, brightness=None, transition: int | None = None
) -> dict:
"""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 not self.is_variable_color_temp:
raise KasaException("Bulb does not support colortemp.")
return await self._device.modules[Module.ColorTemperature].set_color_temp(temp)
async def set_brightness(
self, brightness: int, *, transition: int | None = None
) -> dict:
"""Set the brightness in percentage.
Note, transition is not supported and will be ignored.
: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.")
return await self._device.modules[Module.Brightness].set_brightness(brightness)
@property
def has_effects(self) -> bool:
"""Return True if the device supports effects."""
return Module.LightEffect in self._device.modules

45
kasa/smart/modules/lighteffectmodule.py → kasa/smart/modules/lighteffect.py
View File

@@ -6,14 +6,14 @@ import base64
import copy
from typing import TYPE_CHECKING, Any
from ...feature import Feature
from ...interfaces.lighteffect import LightEffect as LightEffectInterface
from ..smartmodule import SmartModule
if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class LightEffectModule(SmartModule):
class LightEffect(SmartModule, LightEffectInterface):
"""Implementation of dynamic light effects."""
REQUIRED_COMPONENT = "light_effect"
@@ -22,29 +22,11 @@ class LightEffectModule(SmartModule):
"L1": "Party",
"L2": "Relax",
}
LIGHT_EFFECTS_OFF = "Off"
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._scenes_names_to_id: dict[str, str] = {}
def _initialize_features(self):
"""Initialize features."""
device = self._device
self._add_feature(
Feature(
device,
id="light_effect",
name="Light effect",
container=self,
attribute_getter="effect",
attribute_setter="set_effect",
category=Feature.Category.Config,
type=Feature.Type.Choice,
choices_getter="effect_list",
)
)
def _initialize_effects(self) -> dict[str, dict[str, Any]]:
"""Return built-in effects."""
# Copy the effects so scene name updates do not update the underlying dict.
@@ -64,7 +46,7 @@ class LightEffectModule(SmartModule):
return effects
@property
def effect_list(self) -> list[str] | None:
def effect_list(self) -> list[str]:
"""Return built-in effects list.
Example:
@@ -90,6 +72,9 @@ class LightEffectModule(SmartModule):
async def set_effect(
self,
effect: str,
*,
brightness: int | None = None,
transition: int | None = None,
) -> None:
"""Set an effect for the device.
@@ -108,6 +93,24 @@ class LightEffectModule(SmartModule):
params["id"] = effect_id
return await self.call("set_dynamic_light_effect_rule_enable", params)
async def set_custom_effect(
self,
effect_dict: dict,
) -> None:
"""Set a custom effect on the device.
:param str effect_dict: The custom effect dict to set
"""
raise NotImplementedError(
"Device does not support setting custom effects. "
"Use has_custom_effects to check for support."
)
@property
def has_custom_effects(self) -> bool:
"""Return True if the device supports setting custom effects."""
return False
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {self.QUERY_GETTER_NAME: {"start_index": 0}}

2
kasa/smart/modules/lighttransitionmodule.py → kasa/smart/modules/lighttransition.py
View File

@@ -12,7 +12,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class LightTransitionModule(SmartModule):
class LightTransition(SmartModule):
"""Implementation of gradual on/off."""
REQUIRED_COMPONENT = "on_off_gradually"

2
kasa/smart/modules/reportmodule.py → kasa/smart/modules/reportmode.py
View File

@@ -11,7 +11,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class ReportModule(SmartModule):
class ReportMode(SmartModule):
"""Implementation of report module."""
REQUIRED_COMPONENT = "report_mode"

0
kasa/smart/modules/temperature.py → kasa/smart/modules/temperaturesensor.py
View File

2
kasa/smart/modules/timemodule.py → kasa/smart/modules/time.py
View File

@@ -13,7 +13,7 @@ if TYPE_CHECKING:
from ..smartdevice import SmartDevice
class TimeModule(SmartModule):
class Time(SmartModule):
"""Implementation of device_local_time."""
REQUIRED_COMPONENT = "time"

0
kasa/smart/modules/waterleak.py → kasa/smart/modules/waterleaksensor.py
View File

1
kasa/smart/smartchilddevice.py
View File

@@ -49,6 +49,7 @@ class SmartChildDevice(SmartDevice):
"""Return child device type."""
child_device_map = {
"plug.powerstrip.sub-plug": DeviceType.Plug,
"subg.trigger.contact-sensor": DeviceType.Sensor,
"subg.trigger.temp-hmdt-sensor": DeviceType.Sensor,
"subg.trigger.water-leak-sensor": DeviceType.Sensor,
"kasa.switch.outlet.sub-fan": DeviceType.Fan,

265
kasa/smart/smartdevice.py
View File

@@ -5,32 +5,24 @@ from __future__ import annotations
import base64
import logging
from datetime import datetime, timedelta
from typing import Any, Mapping, Sequence, cast, overload
from typing import TYPE_CHECKING, Any, Mapping, Sequence, cast
from ..aestransport import AesTransport
from ..bulb import HSV, Bulb, BulbPreset, ColorTempRange
from ..device import Device, WifiNetwork
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..emeterstatus import EmeterStatus
from ..exceptions import AuthenticationError, DeviceError, KasaException, SmartErrorCode
from ..fan import Fan
from ..feature import Feature
from ..firmware import Firmware
from ..module import ModuleT
from ..module import Module
from ..modulemapping import ModuleMapping, ModuleName
from ..smartprotocol import SmartProtocol
from .modules import (
Brightness,
CloudModule,
ColorModule,
ColorTemperatureModule,
Cloud,
DeviceModule,
EnergyModule,
FanModule,
TimeModule,
)
from .modules import (
Firmware as FirmwareModule,
Firmware,
Light,
Time,
)
from .smartmodule import SmartModule
@@ -41,12 +33,12 @@ _LOGGER = logging.getLogger(__name__)
# the child but only work on the parent. See longer note below in _initialize_modules.
# This list should be updated when creating new modules that could have the
# same issue, homekit perhaps?
WALL_SWITCH_PARENT_ONLY_MODULES = [DeviceModule, TimeModule, Firmware, CloudModule]
WALL_SWITCH_PARENT_ONLY_MODULES = [DeviceModule, Time, Firmware, Cloud]
# Device must go last as the other interfaces also inherit Device
# and python needs a consistent method resolution order.
class SmartDevice(Bulb, Fan, Device):
class SmartDevice(Device):
"""Base class to represent a SMART protocol based device."""
def __init__(
@@ -64,7 +56,7 @@ class SmartDevice(Bulb, Fan, Device):
self._components_raw: dict[str, Any] | None = None
self._components: dict[str, int] = {}
self._state_information: dict[str, Any] = {}
self._modules: dict[str, SmartModule] = {}
self._modules: dict[str | ModuleName[Module], SmartModule] = {}
self._exposes_child_modules = False
self._parent: SmartDevice | None = None
self._children: Mapping[str, SmartDevice] = {}
@@ -105,8 +97,20 @@ class SmartDevice(Bulb, Fan, Device):
return list(self._children.values())
@property
def modules(self) -> dict[str, SmartModule]:
def modules(self) -> ModuleMapping[SmartModule]:
"""Return the device modules."""
if self._exposes_child_modules:
modules = {k: v for k, v in self._modules.items()}
for child in self._children.values():
for k, v in child._modules.items():
if k not in modules:
modules[k] = v
if TYPE_CHECKING:
return cast(ModuleMapping[SmartModule], modules)
return modules
if TYPE_CHECKING: # Needed for python 3.8
return cast(ModuleMapping[SmartModule], self._modules)
return self._modules
def _try_get_response(self, responses: dict, request: str, default=None) -> dict:
@@ -213,7 +217,10 @@ class SmartDevice(Bulb, Fan, Device):
skip_parent_only_modules and mod in WALL_SWITCH_PARENT_ONLY_MODULES
) or mod.__name__ in child_modules_to_skip:
continue
if mod.REQUIRED_COMPONENT in self._components:
if (
mod.REQUIRED_COMPONENT in self._components
or self.sys_info.get(mod.REQUIRED_KEY_ON_PARENT) is not None
):
_LOGGER.debug(
"Found required %s, adding %s to modules.",
mod.REQUIRED_COMPONENT,
@@ -223,6 +230,13 @@ class SmartDevice(Bulb, Fan, Device):
if await module._check_supported():
self._modules[module.name] = module
if (
Module.Brightness in self._modules
or Module.Color in self._modules
or Module.ColorTemperature in self._modules
):
self._modules[Light.__name__] = Light(self, "light")
async def _initialize_features(self):
"""Initialize device features."""
self._add_feature(
@@ -310,41 +324,20 @@ class SmartDevice(Bulb, Fan, Device):
)
)
for module in self._modules.values():
module._initialize_features()
for module in self.modules.values():
# Check if module features have already been initialized.
# i.e. when _exposes_child_modules is true
if not module._module_features:
module._initialize_features()
for feat in module._module_features.values():
self._add_feature(feat)
@overload
def get_module(self, module_type: type[ModuleT]) -> ModuleT | None: ...
@overload
def get_module(self, module_type: str) -> SmartModule | None: ...
def get_module(
self, module_type: type[ModuleT] | str
) -> ModuleT | SmartModule | None:
"""Return the module from the device modules or None if not present."""
if isinstance(module_type, str):
module_name = module_type
elif issubclass(module_type, SmartModule):
module_name = module_type.__name__
else:
return None
if module_name in self.modules:
return self.modules[module_name]
elif self._exposes_child_modules:
for child in self._children.values():
if module_name in child.modules:
return child.modules[module_name]
return None
@property
def is_cloud_connected(self):
def is_cloud_connected(self) -> bool:
"""Returns if the device is connected to the cloud."""
if "CloudModule" not in self.modules:
if Module.Cloud not in self.modules:
return False
return self.modules["CloudModule"].is_connected
return self.modules[Module.Cloud].is_connected
@property
def sys_info(self) -> dict[str, Any]:
@@ -368,10 +361,10 @@ class SmartDevice(Bulb, Fan, Device):
def time(self) -> datetime:
"""Return the time."""
# TODO: Default to parent's time module for child devices
if self._parent and "TimeModule" in self.modules:
_timemod = cast(TimeModule, self._parent.modules["TimeModule"]) # noqa: F405
if self._parent and Module.Time in self.modules:
_timemod = self._parent.modules[Module.Time]
else:
_timemod = cast(TimeModule, self.modules["TimeModule"]) # noqa: F405
_timemod = self.modules[Module.Time]
return _timemod.time
@@ -448,12 +441,7 @@ class SmartDevice(Bulb, Fan, Device):
@property
def has_emeter(self) -> bool:
"""Return if the device has emeter."""
return "EnergyModule" in self.modules
@property
def is_dimmer(self) -> bool:
"""Whether the device acts as a dimmer."""
return self.is_dimmable
return Module.Energy in self.modules
@property
def is_on(self) -> bool:
@@ -490,19 +478,19 @@ class SmartDevice(Bulb, Fan, Device):
@property
def emeter_realtime(self) -> EmeterStatus:
"""Get the emeter status."""
energy = cast(EnergyModule, self.modules["EnergyModule"])
energy = self.modules[Module.Energy]
return energy.emeter_realtime
@property
def emeter_this_month(self) -> float | None:
"""Get the emeter value for this month."""
energy = cast(EnergyModule, self.modules["EnergyModule"])
energy = self.modules[Module.Energy]
return energy.emeter_this_month
@property
def emeter_today(self) -> float | None:
"""Get the emeter value for today."""
energy = cast(EnergyModule, self.modules["EnergyModule"])
energy = self.modules[Module.Energy]
return energy.emeter_today
@property
@@ -514,8 +502,7 @@ class SmartDevice(Bulb, Fan, Device):
):
return None
on_time = cast(float, on_time)
if (timemod := self.modules.get("TimeModule")) is not None:
timemod = cast(TimeModule, timemod) # noqa: F405
if (timemod := self.modules.get(Module.Time)) is not None:
return timemod.time - timedelta(seconds=on_time)
else: # We have no device time, use current local time.
return datetime.now().replace(microsecond=0) - timedelta(seconds=on_time)
@@ -628,12 +615,6 @@ class SmartDevice(Bulb, Fan, Device):
return self._device_type
@property
def firmware(self) -> Firmware:
"""Return firmware module."""
# TODO: open question: does it make sense to expose common modules?
return cast(Firmware, self.get_module(FirmwareModule))
@staticmethod
def _get_device_type_from_components(
components: list[str], device_type: str
@@ -661,149 +642,3 @@ class SmartDevice(Bulb, Fan, Device):
return DeviceType.Thermostat
_LOGGER.warning("Unknown device type, falling back to plug")
return DeviceType.Plug
# Fan interface methods
@property
def is_fan(self) -> bool:
"""Return True if the device is a fan."""
return "FanModule" in self.modules
@property
def fan_speed_level(self) -> int:
"""Return fan speed level."""
if not self.is_fan:
raise KasaException("Device is not a Fan")
return cast(FanModule, self.modules["FanModule"]).fan_speed_level
async def set_fan_speed_level(self, level: int):
"""Set fan speed level."""
if not self.is_fan:
raise KasaException("Device is not a Fan")
await cast(FanModule, self.modules["FanModule"]).set_fan_speed_level(level)
# Bulb interface methods
@property
def is_color(self) -> bool:
"""Whether the bulb supports color changes."""
return "ColorModule" in self.modules
@property
def is_dimmable(self) -> bool:
"""Whether the bulb supports brightness changes."""
return "Brightness" in self.modules
@property
def is_variable_color_temp(self) -> bool:
"""Whether the bulb supports color temperature changes."""
return "ColorTemperatureModule" in self.modules
@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 not self.is_variable_color_temp:
raise KasaException("Color temperature not supported")
return cast(
ColorTemperatureModule, self.modules["ColorTemperatureModule"]
).valid_temperature_range
@property
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.")
return cast(ColorModule, self.modules["ColorModule"]).hsv
@property
def color_temp(self) -> int:
"""Whether the bulb supports color temperature changes."""
if not self.is_variable_color_temp:
raise KasaException("Bulb does not support colortemp.")
return cast(
ColorTemperatureModule, self.modules["ColorTemperatureModule"]
).color_temp
@property
def brightness(self) -> int:
"""Return the current brightness in percentage."""
if not self.is_dimmable: # pragma: no cover
raise KasaException("Bulb is not dimmable.")
return cast(Brightness, self.modules["Brightness"]).brightness
async def set_hsv(
self,
hue: int,
saturation: int,
value: int | None = None,
*,
transition: int | None = None,
) -> dict:
"""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 between 1 and 100
:param int transition: transition in milliseconds.
"""
if not self.is_color:
raise KasaException("Bulb does not support color.")
return await cast(ColorModule, self.modules["ColorModule"]).set_hsv(
hue, saturation, value
)
async def set_color_temp(
self, temp: int, *, brightness=None, transition: int | None = None
) -> dict:
"""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 not self.is_variable_color_temp:
raise KasaException("Bulb does not support colortemp.")
return await cast(
ColorTemperatureModule, self.modules["ColorTemperatureModule"]
).set_color_temp(temp)
async def set_brightness(
self, brightness: int, *, transition: int | None = None
) -> dict:
"""Set the brightness in percentage.
Note, transition is not supported and will be ignored.
: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.")
return await cast(Brightness, self.modules["Brightness"]).set_brightness(
brightness
)
@property
def presets(self) -> list[BulbPreset]:
"""Return a list of available bulb setting presets."""
return []
@property
def has_effects(self) -> bool:
"""Return True if the device supports effects."""
return "LightEffectModule" in self.modules

18
kasa/smart/smartmodule.py
View File

@@ -18,8 +18,13 @@ class SmartModule(Module):
"""Base class for SMART modules."""
NAME: str
REQUIRED_COMPONENT: str
#: Module is initialized, if the given component is available
REQUIRED_COMPONENT: str | None = None
#: Module is initialized, if the given key available in the main sysinfo
REQUIRED_KEY_ON_PARENT: str | None = None
#: Query to execute during the main update cycle
QUERY_GETTER_NAME: str
REGISTERED_MODULES: dict[str, type[SmartModule]] = {}
def __init__(self, device: SmartDevice, module: str):
@@ -27,8 +32,6 @@ class SmartModule(Module):
super().__init__(device, module)
def __init_subclass__(cls, **kwargs):
assert cls.REQUIRED_COMPONENT is not None # noqa: S101
name = getattr(cls, "NAME", cls.__name__)
_LOGGER.debug("Registering %s" % cls)
cls.REGISTERED_MODULES[name] = cls
@@ -91,8 +94,13 @@ class SmartModule(Module):
@property
def supported_version(self) -> int:
"""Return version supported by the device."""
return self._device._components[self.REQUIRED_COMPONENT]
"""Return version supported by the device.
If the module has no required component, this will return -1.
"""
if self.REQUIRED_COMPONENT is not None:
return self._device._components[self.REQUIRED_COMPONENT]
return -1
async def _check_supported(self) -> bool:
"""Additional check to see if the module is supported by the device.