2024-06-19 08:53:40 +00:00
|
|
|
"""Interact with modules.
|
|
|
|
|
|
|
|
Modules are implemented by devices to encapsulate sets of functionality like
|
|
|
|
Light, AutoOff, Firmware etc.
|
|
|
|
|
|
|
|
>>> from kasa import Discover, Module
|
|
|
|
>>>
|
|
|
|
>>> dev = await Discover.discover_single(
|
|
|
|
>>> "127.0.0.3",
|
|
|
|
>>> username="user@example.com",
|
|
|
|
>>> password="great_password"
|
|
|
|
>>> )
|
|
|
|
>>> await dev.update()
|
|
|
|
>>> print(dev.alias)
|
|
|
|
Living Room Bulb
|
|
|
|
|
|
|
|
To see whether a device supports functionality check for the existence of the module:
|
|
|
|
|
|
|
|
>>> if light := dev.modules.get("Light"):
|
|
|
|
>>> print(light.hsv)
|
|
|
|
HSV(hue=0, saturation=100, value=100)
|
|
|
|
|
|
|
|
If you know or expect the module to exist you can access by index:
|
|
|
|
|
|
|
|
>>> light_preset = dev.modules["LightPreset"]
|
|
|
|
>>> print(light_preset.preset_list)
|
|
|
|
['Not set', 'Light preset 1', 'Light preset 2', 'Light preset 3',\
|
|
|
|
'Light preset 4', 'Light preset 5', 'Light preset 6', 'Light preset 7']
|
|
|
|
|
|
|
|
Modules support typing via the Module names in Module:
|
|
|
|
|
2024-11-18 18:46:36 +00:00
|
|
|
>>> from typing import reveal_type, TYPE_CHECKING
|
2024-06-19 08:53:40 +00:00
|
|
|
>>> light_effect = dev.modules.get("LightEffect")
|
|
|
|
>>> light_effect_typed = dev.modules.get(Module.LightEffect)
|
|
|
|
>>> if TYPE_CHECKING:
|
|
|
|
>>> reveal_type(light_effect) # Static checker will reveal: str
|
|
|
|
>>> reveal_type(light_effect_typed) # Static checker will reveal: LightEffect
|
|
|
|
|
|
|
|
"""
|
2024-04-16 18:21:20 +00:00
|
|
|
|
2024-04-17 13:39:24 +00:00
|
|
|
from __future__ import annotations
|
|
|
|
|
2024-02-19 17:01:31 +00:00
|
|
|
import logging
|
|
|
|
from abc import ABC, abstractmethod
|
2024-11-22 07:52:23 +00:00
|
|
|
from collections.abc import Callable
|
|
|
|
from functools import cache
|
2024-05-03 15:01:21 +00:00
|
|
|
from typing import (
|
|
|
|
TYPE_CHECKING,
|
2024-05-10 18:29:28 +00:00
|
|
|
Final,
|
2024-05-03 15:01:21 +00:00
|
|
|
TypeVar,
|
2024-11-22 07:52:23 +00:00
|
|
|
get_type_hints,
|
2024-05-03 15:01:21 +00:00
|
|
|
)
|
2024-02-19 17:01:31 +00:00
|
|
|
|
2024-02-21 15:52:55 +00:00
|
|
|
from .exceptions import KasaException
|
2024-02-19 17:01:31 +00:00
|
|
|
from .feature import Feature
|
2024-05-10 18:29:28 +00:00
|
|
|
from .modulemapping import ModuleName
|
2024-02-19 17:01:31 +00:00
|
|
|
|
2024-04-29 16:34:20 +00:00
|
|
|
if TYPE_CHECKING:
|
2024-05-13 16:34:44 +00:00
|
|
|
from . import interfaces
|
2024-05-11 18:28:18 +00:00
|
|
|
from .device import Device
|
2024-05-10 18:29:28 +00:00
|
|
|
from .iot import modules as iot
|
|
|
|
from .smart import modules as smart
|
2024-11-23 08:07:47 +00:00
|
|
|
from .smartcam import modules as smartcam
|
2024-04-29 16:34:20 +00:00
|
|
|
|
2024-02-19 17:01:31 +00:00
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
2024-05-03 15:01:21 +00:00
|
|
|
ModuleT = TypeVar("ModuleT", bound="Module")
|
|
|
|
|
2024-02-19 17:01:31 +00:00
|
|
|
|
2024-11-22 07:52:23 +00:00
|
|
|
class FeatureAttribute:
|
|
|
|
"""Class for annotating attributes bound to feature."""
|
|
|
|
|
|
|
|
|
2024-02-19 17:01:31 +00:00
|
|
|
class Module(ABC):
|
|
|
|
"""Base class implemention for all modules.
|
|
|
|
|
|
|
|
The base classes should implement `query` to return the query they want to be
|
|
|
|
executed during the regular update cycle.
|
|
|
|
"""
|
|
|
|
|
2024-05-10 18:29:28 +00:00
|
|
|
# Common Modules
|
2024-06-17 10:22:05 +00:00
|
|
|
Energy: Final[ModuleName[interfaces.Energy]] = ModuleName("Energy")
|
|
|
|
Fan: Final[ModuleName[interfaces.Fan]] = ModuleName("Fan")
|
2024-05-13 16:34:44 +00:00
|
|
|
LightEffect: Final[ModuleName[interfaces.LightEffect]] = ModuleName("LightEffect")
|
|
|
|
Led: Final[ModuleName[interfaces.Led]] = ModuleName("Led")
|
|
|
|
Light: Final[ModuleName[interfaces.Light]] = ModuleName("Light")
|
2024-05-19 10:20:18 +00:00
|
|
|
LightPreset: Final[ModuleName[interfaces.LightPreset]] = ModuleName("LightPreset")
|
2024-10-15 07:59:25 +00:00
|
|
|
Time: Final[ModuleName[interfaces.Time]] = ModuleName("Time")
|
2024-05-10 18:29:28 +00:00
|
|
|
|
|
|
|
# IOT only Modules
|
|
|
|
IotAmbientLight: Final[ModuleName[iot.AmbientLight]] = ModuleName("ambient")
|
|
|
|
IotAntitheft: Final[ModuleName[iot.Antitheft]] = ModuleName("anti_theft")
|
|
|
|
IotCountdown: Final[ModuleName[iot.Countdown]] = ModuleName("countdown")
|
|
|
|
IotMotion: Final[ModuleName[iot.Motion]] = ModuleName("motion")
|
|
|
|
IotSchedule: Final[ModuleName[iot.Schedule]] = ModuleName("schedule")
|
|
|
|
IotUsage: Final[ModuleName[iot.Usage]] = ModuleName("usage")
|
|
|
|
IotCloud: Final[ModuleName[iot.Cloud]] = ModuleName("cloud")
|
|
|
|
|
|
|
|
# SMART only Modules
|
2024-05-11 18:28:18 +00:00
|
|
|
Alarm: Final[ModuleName[smart.Alarm]] = ModuleName("Alarm")
|
|
|
|
AutoOff: Final[ModuleName[smart.AutoOff]] = ModuleName("AutoOff")
|
2024-05-10 18:29:28 +00:00
|
|
|
BatterySensor: Final[ModuleName[smart.BatterySensor]] = ModuleName("BatterySensor")
|
|
|
|
Brightness: Final[ModuleName[smart.Brightness]] = ModuleName("Brightness")
|
2024-05-11 18:28:18 +00:00
|
|
|
ChildDevice: Final[ModuleName[smart.ChildDevice]] = ModuleName("ChildDevice")
|
|
|
|
Cloud: Final[ModuleName[smart.Cloud]] = ModuleName("Cloud")
|
|
|
|
Color: Final[ModuleName[smart.Color]] = ModuleName("Color")
|
|
|
|
ColorTemperature: Final[ModuleName[smart.ColorTemperature]] = ModuleName(
|
|
|
|
"ColorTemperature"
|
2024-05-10 18:29:28 +00:00
|
|
|
)
|
|
|
|
ContactSensor: Final[ModuleName[smart.ContactSensor]] = ModuleName("ContactSensor")
|
2024-05-11 18:28:18 +00:00
|
|
|
DeviceModule: Final[ModuleName[smart.DeviceModule]] = ModuleName("DeviceModule")
|
2024-05-10 18:29:28 +00:00
|
|
|
Firmware: Final[ModuleName[smart.Firmware]] = ModuleName("Firmware")
|
2024-05-11 18:28:18 +00:00
|
|
|
FrostProtection: Final[ModuleName[smart.FrostProtection]] = ModuleName(
|
|
|
|
"FrostProtection"
|
|
|
|
)
|
|
|
|
HumiditySensor: Final[ModuleName[smart.HumiditySensor]] = ModuleName(
|
|
|
|
"HumiditySensor"
|
2024-05-10 18:29:28 +00:00
|
|
|
)
|
2024-05-11 18:28:18 +00:00
|
|
|
LightTransition: Final[ModuleName[smart.LightTransition]] = ModuleName(
|
|
|
|
"LightTransition"
|
2024-05-10 18:29:28 +00:00
|
|
|
)
|
2024-07-24 13:48:33 +00:00
|
|
|
MotionSensor: Final[ModuleName[smart.MotionSensor]] = ModuleName("MotionSensor")
|
2024-05-11 18:28:18 +00:00
|
|
|
ReportMode: Final[ModuleName[smart.ReportMode]] = ModuleName("ReportMode")
|
2024-07-01 11:59:24 +00:00
|
|
|
SmartLightEffect: Final[ModuleName[smart.SmartLightEffect]] = ModuleName(
|
|
|
|
"LightEffect"
|
|
|
|
)
|
2024-07-31 14:58:48 +00:00
|
|
|
IotLightEffect: Final[ModuleName[iot.LightEffect]] = ModuleName("LightEffect")
|
2024-05-11 18:28:18 +00:00
|
|
|
TemperatureSensor: Final[ModuleName[smart.TemperatureSensor]] = ModuleName(
|
2024-05-10 18:29:28 +00:00
|
|
|
"TemperatureSensor"
|
|
|
|
)
|
2024-05-11 18:28:18 +00:00
|
|
|
TemperatureControl: Final[ModuleName[smart.TemperatureControl]] = ModuleName(
|
2024-05-10 18:29:28 +00:00
|
|
|
"TemperatureControl"
|
|
|
|
)
|
|
|
|
WaterleakSensor: Final[ModuleName[smart.WaterleakSensor]] = ModuleName(
|
|
|
|
"WaterleakSensor"
|
|
|
|
)
|
2024-11-01 19:46:36 +00:00
|
|
|
ChildProtection: Final[ModuleName[smart.ChildProtection]] = ModuleName(
|
|
|
|
"ChildProtection"
|
|
|
|
)
|
2024-11-01 15:36:09 +00:00
|
|
|
TriggerLogs: Final[ModuleName[smart.TriggerLogs]] = ModuleName("TriggerLogs")
|
2024-05-10 18:29:28 +00:00
|
|
|
|
2024-11-23 08:07:47 +00:00
|
|
|
# SMARTCAM only modules
|
|
|
|
Camera: Final[ModuleName[smartcam.Camera]] = ModuleName("Camera")
|
2024-10-24 16:22:45 +00:00
|
|
|
|
2024-11-10 18:55:13 +00:00
|
|
|
def __init__(self, device: Device, module: str) -> None:
|
2024-02-19 17:01:31 +00:00
|
|
|
self._device = device
|
|
|
|
self._module = module
|
2024-04-17 13:39:24 +00:00
|
|
|
self._module_features: dict[str, Feature] = {}
|
2024-02-19 17:01:31 +00:00
|
|
|
|
2024-11-22 07:52:23 +00:00
|
|
|
def has_feature(self, attribute: str | property | Callable) -> bool:
|
|
|
|
"""Return True if the module attribute feature is supported."""
|
|
|
|
return bool(self.get_feature(attribute))
|
|
|
|
|
|
|
|
def get_feature(self, attribute: str | property | Callable) -> Feature | None:
|
|
|
|
"""Get Feature for a module attribute or None if not supported."""
|
|
|
|
return _get_bound_feature(self, attribute)
|
|
|
|
|
2024-02-19 17:01:31 +00:00
|
|
|
@abstractmethod
|
2024-11-10 18:55:13 +00:00
|
|
|
def query(self) -> dict:
|
2024-02-19 17:01:31 +00:00
|
|
|
"""Query to execute during the update cycle.
|
|
|
|
|
|
|
|
The inheriting modules implement this to include their wanted
|
|
|
|
queries to the query that gets executed when Device.update() gets called.
|
|
|
|
"""
|
|
|
|
|
|
|
|
@property
|
|
|
|
@abstractmethod
|
2024-11-10 18:55:13 +00:00
|
|
|
def data(self) -> dict:
|
2024-02-19 17:01:31 +00:00
|
|
|
"""Return the module specific raw data from the last update."""
|
|
|
|
|
2024-11-10 18:55:13 +00:00
|
|
|
def _initialize_features(self) -> None: # noqa: B027
|
2024-04-30 06:56:09 +00:00
|
|
|
"""Initialize features after the initial update.
|
|
|
|
|
|
|
|
This can be implemented if features depend on module query responses.
|
2024-05-19 09:18:17 +00:00
|
|
|
It will only be called once per module and will always be called
|
|
|
|
after *_post_update_hook* has been called for every device module and its
|
|
|
|
children's modules.
|
|
|
|
"""
|
|
|
|
|
2024-11-10 18:55:13 +00:00
|
|
|
async def _post_update_hook(self) -> None: # noqa: B027
|
2024-05-19 09:18:17 +00:00
|
|
|
"""Perform actions after a device update.
|
|
|
|
|
|
|
|
This can be implemented if a module needs to perform actions each time
|
|
|
|
the device has updated like generating collections for property access.
|
|
|
|
It will be called after every update and will be called prior to
|
|
|
|
*_initialize_features* on the first update.
|
2024-04-30 06:56:09 +00:00
|
|
|
"""
|
|
|
|
|
2024-11-10 18:55:13 +00:00
|
|
|
def _add_feature(self, feature: Feature) -> None:
|
2024-02-19 17:01:31 +00:00
|
|
|
"""Add module feature."""
|
2024-05-07 09:13:35 +00:00
|
|
|
id_ = feature.id
|
|
|
|
if id_ in self._module_features:
|
2024-11-10 18:55:13 +00:00
|
|
|
raise KasaException(f"Duplicate id detected {id_}")
|
2024-05-07 09:13:35 +00:00
|
|
|
self._module_features[id_] = feature
|
2024-02-19 17:01:31 +00:00
|
|
|
|
|
|
|
def __repr__(self) -> str:
|
|
|
|
return (
|
|
|
|
f"<Module {self.__class__.__name__} ({self._module})"
|
|
|
|
f" for {self._device.host}>"
|
|
|
|
)
|
2024-11-22 07:52:23 +00:00
|
|
|
|
|
|
|
|
|
|
|
def _is_bound_feature(attribute: property | Callable) -> bool:
|
|
|
|
"""Check if an attribute is bound to a feature with FeatureAttribute."""
|
|
|
|
if isinstance(attribute, property):
|
|
|
|
hints = get_type_hints(attribute.fget, include_extras=True)
|
|
|
|
else:
|
|
|
|
hints = get_type_hints(attribute, include_extras=True)
|
|
|
|
|
|
|
|
if (return_hints := hints.get("return")) and hasattr(return_hints, "__metadata__"):
|
|
|
|
metadata = hints["return"].__metadata__
|
|
|
|
for meta in metadata:
|
|
|
|
if isinstance(meta, FeatureAttribute):
|
|
|
|
return True
|
|
|
|
|
|
|
|
return False
|
|
|
|
|
|
|
|
|
|
|
|
@cache
|
|
|
|
def _get_bound_feature(
|
|
|
|
module: Module, attribute: str | property | Callable
|
|
|
|
) -> Feature | None:
|
|
|
|
"""Get Feature for a bound property or None if not supported."""
|
|
|
|
if not isinstance(attribute, str):
|
|
|
|
if isinstance(attribute, property):
|
|
|
|
# Properties have __name__ in 3.13 so this could be simplified
|
|
|
|
# when only 3.13 supported
|
|
|
|
attribute_name = attribute.fget.__name__ # type: ignore[union-attr]
|
|
|
|
else:
|
|
|
|
attribute_name = attribute.__name__
|
|
|
|
attribute_callable = attribute
|
|
|
|
else:
|
|
|
|
if TYPE_CHECKING:
|
|
|
|
assert isinstance(attribute, str)
|
|
|
|
attribute_name = attribute
|
|
|
|
attribute_callable = getattr(module.__class__, attribute, None) # type: ignore[assignment]
|
|
|
|
if not attribute_callable:
|
|
|
|
raise KasaException(
|
|
|
|
f"No attribute named {attribute_name} in "
|
|
|
|
f"module {module.__class__.__name__}"
|
|
|
|
)
|
|
|
|
|
|
|
|
if not _is_bound_feature(attribute_callable):
|
|
|
|
raise KasaException(
|
|
|
|
f"Attribute {attribute_name} of module {module.__class__.__name__}"
|
|
|
|
" is not bound to a feature"
|
|
|
|
)
|
|
|
|
|
|
|
|
check = {attribute_name, attribute_callable}
|
|
|
|
for feature in module._module_features.values():
|
|
|
|
if (getter := feature.attribute_getter) and getter in check:
|
|
|
|
return feature
|
|
|
|
|
|
|
|
if (setter := feature.attribute_setter) and setter in check:
|
|
|
|
return feature
|
|
|
|
|
|
|
|
return None
|