Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2024-12-22 19:23:34 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Steven B 2024-12-15 16:23:56 +00:00
commit cddac30479
No known key found for this signature in database
GPG Key ID: 6D5B46B3679F2A43
45 changed files with 592 additions and 304 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -223,6 +223,7 @@ See [supported devices in our documentation](SUPPORTED.md) for more detailed inf
* [Home Assistant](https://www.home-assistant.io/integrations/tplink/)
* [MQTT access to TP-Link devices, using python-kasa](https://github.com/flavio-fernandes/mqtt2kasa)
* [Homebridge Kasa Python Plug-In](https://github.com/ZeliardM/homebridge-kasa-python)
### Other related projects

13
docs/source/featureattributes.md Normal file
View File

@ -0,0 +1,13 @@
Some modules have attributes that may not be supported by the device.
These attributes will be annotated with a `FeatureAttribute` return type.
For example:
```py
@property
def hsv(self) -> Annotated[HSV, FeatureAttribute()]:
"""Return the current HSV state of the bulb."""
```
You can test whether a `FeatureAttribute` is supported by the device with {meth}`kasa.Module.has_feature`
or {meth}`kasa.Module.get_feature` which will return `None` if not supported.
Calling these methods on attributes not annotated with a `FeatureAttribute` return type will return an error.

64
docs/source/reference.md
View File

@ -13,11 +13,13 @@
## Device
% N.B. Credentials clashes with autodoc
```{eval-rst}
.. autoclass:: Device
:members:
:undoc-members:
:exclude-members: Credentials
```
@ -28,7 +30,6 @@
.. autoclass:: Credentials
:members:
:undoc-members:
:noindex:
```
@ -61,15 +62,11 @@
```{eval-rst}
.. autoclass:: Module
:noindex:
:members:
:inherited-members:
:undoc-members:
```
```{eval-rst}
.. autoclass:: Feature
:noindex:
:members:
:inherited-members:
:undoc-members:
@ -77,7 +74,6 @@
```{eval-rst}
.. automodule:: kasa.interfaces
:noindex:
:members:
:inherited-members:
:undoc-members:
@ -85,64 +81,29 @@
## Protocols and transports
```{eval-rst}
.. autoclass:: kasa.protocols.BaseProtocol
.. automodule:: kasa.protocols
:members:
:inherited-members:
:imported-members:
:undoc-members:
:exclude-members: SmartErrorCode
:no-index:
```
```{eval-rst}
.. autoclass:: kasa.protocols.IotProtocol
.. automodule:: kasa.transports
:members:
:inherited-members:
:imported-members:
:undoc-members:
:no-index:
```
```{eval-rst}
.. autoclass:: kasa.protocols.SmartProtocol
:members:
:inherited-members:
:undoc-members:
```
```{eval-rst}
.. autoclass:: kasa.transports.BaseTransport
:members:
:inherited-members:
:undoc-members:
```
```{eval-rst}
.. autoclass:: kasa.transports.XorTransport
:members:
:inherited-members:
:undoc-members:
```
```{eval-rst}
.. autoclass:: kasa.transports.KlapTransport
:members:
:inherited-members:
:undoc-members:
```
```{eval-rst}
.. autoclass:: kasa.transports.KlapTransportV2
:members:
:inherited-members:
:undoc-members:
```
```{eval-rst}
.. autoclass:: kasa.transports.AesTransport
:members:
:inherited-members:
:undoc-members:
```
## Errors and exceptions
```{eval-rst}
.. autoclass:: kasa.exceptions.KasaException
:members:
@ -171,3 +132,4 @@
.. autoclass:: kasa.exceptions.TimeoutError
:members:
:undoc-members:
```

110
docs/source/topics.md
View File

@ -80,14 +80,17 @@ This can be done using the {attr}`~kasa.Device.internal_state` property.
## Modules and Features
The functionality provided by all {class}`~kasa.Device` instances is (mostly) done inside separate modules.
While the individual device-type specific classes provide an easy access for the most import features,
you can also access individual modules through {attr}`kasa.Device.modules`.
You can get the list of supported modules for a given device instance using {attr}`~kasa.Device.supported_modules`.
While the device class provides easy access for most device related attributes,
for components like `light` and `camera` you can access the module through {attr}`kasa.Device.modules`.
The module names are handily available as constants on {class}`~kasa.Module` and will return type aware values from the collection.
```{note}
If you only need some module-specific information,
you can call the wanted method on the module to avoid using {meth}`~kasa.Device.update`.
```
Features represent individual pieces of functionality within a module like brightness, hsv and temperature within a light module.
They allow for instrospection and can be accessed through {attr}`kasa.Device.features`.
Attributes can be accessed via a `Feature` or a module attribute depending on the use case.
Modules tend to provide richer functionality but using the features does not require an understanding of the module api.
:::{include} featureattributes.md
:::
(topics-protocols-and-transports)=
## Protocols and Transports
@ -137,96 +140,3 @@ The base exception for all library errors is {class}`KasaException <kasa.excepti
- If the library encounters and unsupported deviceit raises an {class}`UnsupportedDeviceError <kasa.exceptions.UnsupportedDeviceError>`.
- If the device fails to respond within a timeout the library raises a {class}`TimeoutError <kasa.exceptions.TimeoutError>`.
- All other failures will raise the base {class}`KasaException <kasa.exceptions.KasaException>` class.
<!-- Commenting out this section keeps git seeing the change as a rename.
API documentation for modules and features
******************************************
.. autoclass:: kasa.Module
:noindex:
:members:
:inherited-members:
:undoc-members:
.. automodule:: kasa.interfaces
:noindex:
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.Feature
:noindex:
:members:
:inherited-members:
:undoc-members:
API documentation for protocols and transports
**********************************************
.. autoclass:: kasa.protocols.BaseProtocol
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.protocols.IotProtocol
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.protocols.SmartProtocol
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.transports.BaseTransport
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.transports.XorTransport
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.transports.KlapTransport
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.transports.KlapTransportV2
:members:
:inherited-members:
:undoc-members:
.. autoclass:: kasa.transports.AesTransport
:members:
:inherited-members:
:undoc-members:
API documentation for errors and exceptions
*******************************************
.. autoclass:: kasa.exceptions.KasaException
:members:
:undoc-members:
.. autoclass:: kasa.exceptions.DeviceError
:members:
:undoc-members:
.. autoclass:: kasa.exceptions.AuthenticationError
:members:
:undoc-members:
.. autoclass:: kasa.exceptions.UnsupportedDeviceError
:members:
:undoc-members:
.. autoclass:: kasa.exceptions.TimeoutError
:members:
:undoc-members:
-->

6
docs/tutorial.py
View File

@ -40,7 +40,7 @@ Different groups of functionality are supported by modules which you can access
key from :class:`~kasa.Module`.
Modules will only be available on the device if they are supported but some individual features of a module may not be available for your device.
You can check the availability using ``is_``-prefixed properties like `is_color`.
You can check the availability using ``has_feature()`` method.
>>> from kasa import Module
>>> Module.Light in dev.modules
@ -52,9 +52,9 @@ True
>>> await dev.update()
>>> light.brightness
50
>>> light.is_color
>>> light.has_feature("hsv")
True
>>> if light.is_color:
>>> if light.has_feature("hsv"):
>>> print(light.hsv)
HSV(hue=0, saturation=100, value=50)

10
kasa/cli/device.py
View File

@ -41,8 +41,14 @@ async def state(ctx, dev: Device):
echo(f"Device state: {dev.is_on}")
echo(f"Time: {dev.time} (tz: {dev.timezone})")
echo(f"Hardware: {dev.hw_info['hw_ver']}")
echo(f"Software: {dev.hw_info['sw_ver']}")
echo(
f"Hardware: {dev.device_info.hardware_version}"
f"{' (' + dev.region + ')' if dev.region else ''}"
)
echo(
f"Firmware: {dev.device_info.firmware_version}"
f" {dev.device_info.firmware_build}"
)
echo(f"MAC (rssi): {dev.mac} ({dev.rssi})")
if verbose:
echo(f"Location: {dev.location}")

14
kasa/cli/light.py
View File

@ -25,7 +25,9 @@ def light(dev) -> None:
@pass_dev_or_child
async def brightness(dev: Device, brightness: int, transition: int):
"""Get or set brightness."""
if not (light := dev.modules.get(Module.Light)) or not light.is_dimmable:
if not (light := dev.modules.get(Module.Light)) or not light.has_feature(
"brightness"
):
error("This device does not support brightness.")
return
@ -45,13 +47,15 @@ async def brightness(dev: Device, brightness: int, transition: int):
@pass_dev_or_child
async def temperature(dev: Device, temperature: int, transition: int):
"""Get or set color temperature."""
if not (light := dev.modules.get(Module.Light)) or not light.is_variable_color_temp:
if not (light := dev.modules.get(Module.Light)) or not (
color_temp_feat := light.get_feature("color_temp")
):
error("Device does not support color temperature")
return
if temperature is None:
echo(f"Color temperature: {light.color_temp}")
valid_temperature_range = light.valid_temperature_range
valid_temperature_range = color_temp_feat.range
if valid_temperature_range != (0, 0):
echo("(min: {}, max: {})".format(*valid_temperature_range))
else:
@ -59,7 +63,7 @@ async def temperature(dev: Device, temperature: int, transition: int):
"Temperature range unknown, please open a github issue"
f" or a pull request for model '{dev.model}'"
)
return light.valid_temperature_range
return color_temp_feat.range
else:
echo(f"Setting color temperature to {temperature}")
return await light.set_color_temp(temperature, transition=transition)
@ -99,7 +103,7 @@ async def effect(dev: Device, ctx, effect):
@pass_dev_or_child
async def hsv(dev: Device, ctx, h, s, v, transition):
"""Get or set color in HSV."""
if not (light := dev.modules.get(Module.Light)) or not light.is_color:
if not (light := dev.modules.get(Module.Light)) or not light.has_feature("hsv"):
error("Device does not support colors")
return

22
kasa/device.py
View File

@ -29,7 +29,7 @@ All devices provide several informational properties:
>>> dev.alias
Bedroom Lamp Plug
>>> dev.model
HS110(EU)
HS110
>>> dev.rssi
-71
>>> dev.mac
@ -151,7 +151,7 @@ _LOGGER = logging.getLogger(__name__)
@dataclass
class _DeviceInfo:
class DeviceInfo:
"""Device Model Information."""
short_name: str
@ -208,7 +208,7 @@ class Device(ABC):
self.protocol: BaseProtocol = protocol or IotProtocol(
transport=XorTransport(config=config or DeviceConfig(host=host)),
)
self._last_update: Any = None
self._last_update: dict[str, Any] = {}
_LOGGER.debug("Initializing %s of type %s", host, type(self))
self._device_type = DeviceType.Unknown
# TODO: typing Any is just as using dict | None would require separate
@ -334,9 +334,21 @@ class Device(ABC):
"""Returns the device model."""
@property
def region(self) -> str | None:
"""Returns the device region."""
return self.device_info.region
@property
def device_info(self) -> DeviceInfo:
"""Return device info."""
return self._get_device_info(self._last_update, self._discovery_info)
@staticmethod
@abstractmethod
def _model_region(self) -> str:
"""Return device full model name and region."""
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> DeviceInfo:
"""Get device info."""
@property
@abstractmethod

12
kasa/discover.py
View File

@ -22,7 +22,7 @@ Discovery returns a dict of {ip: discovered devices}:
>>>
>>> found_devices = await Discover.discover()
>>> [dev.model for dev in found_devices.values()]
['KP303(UK)', 'HS110(EU)', 'L530E', 'KL430(US)', 'HS220(US)']
['KP303', 'HS110', 'L530E', 'KL430', 'HS220']
You can pass username and password for devices requiring authentication
@ -65,17 +65,17 @@ It is also possible to pass a coroutine to be executed for each found device:
>>> print(f"Discovered {dev.alias} (model: {dev.model})")
>>>
>>> devices = await Discover.discover(on_discovered=print_dev_info, credentials=creds)
Discovered Bedroom Power Strip (model: KP303(UK))
Discovered Bedroom Lamp Plug (model: HS110(EU))
Discovered Bedroom Power Strip (model: KP303)
Discovered Bedroom Lamp Plug (model: HS110)
Discovered Living Room Bulb (model: L530)
Discovered Bedroom Lightstrip (model: KL430(US))
Discovered Living Room Dimmer Switch (model: HS220(US))
Discovered Bedroom Lightstrip (model: KL430)
Discovered Living Room Dimmer Switch (model: HS220)
Discovering a single device returns a kasa.Device object.
>>> device = await Discover.discover_single("127.0.0.1", credentials=creds)
>>> device.model
'KP303(UK)'
'KP303'
"""

13
kasa/interfaces/light.py
View File

@ -23,13 +23,13 @@ Get the light module to interact:
>>> light = dev.modules[Module.Light]
You can use the ``is_``-prefixed properties to check for supported features:
You can use the ``has_feature()`` method to check for supported features:
>>> light.is_dimmable
>>> light.has_feature("brightness")
True
>>> light.is_color
>>> light.has_feature("hsv")
True
>>> light.is_variable_color_temp
>>> light.has_feature("color_temp")
True
All known bulbs support changing the brightness:
@ -43,8 +43,9 @@ All known bulbs support changing the brightness:
Bulbs supporting color temperature can be queried for the supported range:
>>> light.valid_temperature_range
ColorTempRange(min=2500, max=6500)
>>> if color_temp_feature := light.get_feature("color_temp"):
>>> print(f"{color_temp_feature.minimum_value}, {color_temp_feature.maximum_value}")
2500, 6500
>>> await light.set_color_temp(3000)
>>> await dev.update()
>>> light.color_temp

3
kasa/interfaces/lighteffect.py
View File

@ -13,8 +13,7 @@ Living Room Bulb
Light effects are accessed via the LightPreset module. To list available presets
>>> if dev.modules[Module.Light].has_effects:
>>> light_effect = dev.modules[Module.LightEffect]
>>> light_effect = dev.modules[Module.LightEffect]
>>> light_effect.effect_list
['Off', 'Party', 'Relax']

33
kasa/iot/iotdevice.py
View File

@ -22,7 +22,7 @@ from datetime import datetime, timedelta, tzinfo
from typing import TYPE_CHECKING, Any, cast
from warnings import warn
from ..device import Device, WifiNetwork, _DeviceInfo
from ..device import Device, DeviceInfo, WifiNetwork
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..exceptions import KasaException
@ -43,7 +43,7 @@ def requires_update(f: Callable) -> Any:
@functools.wraps(f)
async def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if self._last_update is None and (
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")
@ -54,7 +54,7 @@ def requires_update(f: Callable) -> Any:
@functools.wraps(f)
def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if self._last_update is None and (
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")
@ -112,7 +112,7 @@ class IotDevice(Device):
>>> dev.alias
Bedroom Lamp Plug
>>> dev.model
HS110(EU)
HS110
>>> dev.rssi
-71
>>> dev.mac
@ -310,7 +310,7 @@ class IotDevice(Device):
# If this is the initial update, check only for the sysinfo
# This is necessary as some devices crash on unexpected modules
# See #105, #120, #161
if self._last_update is None:
if not self._last_update:
_LOGGER.debug("Performing the initial update to obtain sysinfo")
response = await self.protocol.query(req)
self._last_update = response
@ -452,7 +452,9 @@ class IotDevice(Device):
# This allows setting of some info properties directly
# from partial discovery info that will then be found
# by the requires_update decorator
self._set_sys_info(info)
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."""
@ -471,18 +473,13 @@ class IotDevice(Device):
"""
return self._sys_info # type: ignore
@property # type: ignore
@requires_update
def model(self) -> str:
"""Return device model."""
sys_info = self._sys_info
return str(sys_info["model"])
@property
@requires_update
def _model_region(self) -> str:
"""Return device full model name and region."""
return self.model
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:
@ -748,7 +745,7 @@ class IotDevice(Device):
@staticmethod
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> _DeviceInfo:
) -> DeviceInfo:
"""Get model information for a device."""
sys_info = _extract_sys_info(info)
@ -766,7 +763,7 @@ class IotDevice(Device):
firmware_version, firmware_build = fw_version_full.split(" ", maxsplit=1)
auth = bool(discovery_info and ("mgt_encrypt_schm" in discovery_info))
return _DeviceInfo(
return DeviceInfo(
short_name=long_name,
long_name=long_name,
brand="kasa",

30
kasa/iot/modules/light.py
View File

@ -3,13 +3,14 @@
from __future__ import annotations
from dataclasses import asdict
from typing import TYPE_CHECKING, cast
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:
@ -32,7 +33,7 @@ class Light(IotModule, LightInterface):
super()._initialize_features()
device = self._device
if self._device._is_dimmable:
if device._is_dimmable:
self._add_feature(
Feature(
device,
@ -46,7 +47,7 @@ class Light(IotModule, LightInterface):
category=Feature.Category.Primary,
)
)
if self._device._is_variable_color_temp:
if device._is_variable_color_temp:
self._add_feature(
Feature(
device=device,
@ -60,7 +61,7 @@ class Light(IotModule, LightInterface):
type=Feature.Type.Number,
)
)
if self._device._is_color:
if device._is_color:
self._add_feature(
Feature(
device=device,
@ -95,13 +96,13 @@ class Light(IotModule, LightInterface):
return self._device._is_dimmable
@property # type: ignore
def brightness(self) -> int:
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
) -> dict:
) -> Annotated[dict, FeatureAttribute()]:
"""Set the brightness in percentage. A value of 0 will turn off the light.
:param int brightness: brightness in percent
@ -133,7 +134,7 @@ class Light(IotModule, LightInterface):
return bulb._has_effects
@property
def hsv(self) -> HSV:
def hsv(self) -> Annotated[HSV, FeatureAttribute()]:
"""Return the current HSV state of the bulb.
:return: hue, saturation and value (degrees, %, %)
@ -149,7 +150,7 @@ class Light(IotModule, LightInterface):
value: int | None = None,
*,
transition: int | None = None,
) -> dict:
) -> Annotated[dict, FeatureAttribute()]:
"""Set new HSV.
Note, transition is not supported and will be ignored.
@ -176,7 +177,7 @@ class Light(IotModule, LightInterface):
return bulb._valid_temperature_range
@property
def color_temp(self) -> int:
def color_temp(self) -> Annotated[int, FeatureAttribute()]:
"""Whether the bulb supports color temperature changes."""
if (
bulb := self._get_bulb_device()
@ -186,7 +187,7 @@ class Light(IotModule, LightInterface):
async def set_color_temp(
self, temp: int, *, brightness: int | None = None, transition: int | None = None
) -> dict:
) -> Annotated[dict, FeatureAttribute()]:
"""Set the color temperature of the device in kelvin.
Note, transition is not supported and will be ignored.
@ -242,17 +243,18 @@ class Light(IotModule, LightInterface):
return self._light_state
async def _post_update_hook(self) -> None:
if self._device.is_on is False:
device = self._device
if device.is_on is False:
state = LightState(light_on=False)
else:
state = LightState(light_on=True)
if self.is_dimmable:
if device._is_dimmable:
state.brightness = self.brightness
if self.is_color:
if device._is_color:
hsv = self.hsv
state.hue = hsv.hue
state.saturation = hsv.saturation
if self.is_variable_color_temp:
if device._is_variable_color_temp:
state.color_temp = self.color_temp
self._light_state = state

18
kasa/iot/modules/lightpreset.py
View File

@ -85,17 +85,19 @@ class LightPreset(IotModule, LightPresetInterface):
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 light.is_variable_color_temp else None
h, s = (light.hsv.hue, light.hsv.saturation) if light.is_color else (None, None)
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 light.is_variable_color_temp
)
and (preset.hue == h or not light.is_color)
and (preset.saturation == s or not light.is_color)
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
@ -107,7 +109,7 @@ class LightPreset(IotModule, LightPresetInterface):
"""Set a light preset for the device."""
light = self._device.modules[Module.Light]
if preset_name == self.PRESET_NOT_SET:
if light.is_color:
if light.has_feature("hsv"):
preset = LightState(hue=0, saturation=0, brightness=100)
else:
preset = LightState(brightness=100)

6
kasa/module.py
View File

@ -21,6 +21,9 @@ check for the existence of the module:
>>> print(light.brightness)
100
.. include:: ../featureattributes.md
:parser: myst_parser.sphinx_
To see whether a device supports specific functionality, you can check whether the
module has that feature:
@ -151,6 +154,9 @@ class Module(ABC):
)
TriggerLogs: Final[ModuleName[smart.TriggerLogs]] = ModuleName("TriggerLogs")
HomeKit: Final[ModuleName[smart.HomeKit]] = ModuleName("HomeKit")
Matter: Final[ModuleName[smart.Matter]] = ModuleName("Matter")
# SMARTCAM only modules
Camera: Final[ModuleName[smartcam.Camera]] = ModuleName("Camera")

28
kasa/protocols/smartprotocol.py
View File

@ -71,6 +71,13 @@ REDACTORS: dict[str, Callable[[Any], Any] | None] = {
"map_data": lambda x: "#SCRUBBED_MAPDATA#" if x else "",
}
# Queries that are known not to work properly when sent as a
# multiRequest. They will not return the `method` key.
FORCE_SINGLE_REQUEST = {
"getConnectStatus",
"scanApList",
}
class SmartProtocol(BaseProtocol):
"""Class for the new TPLink SMART protocol."""
@ -91,6 +98,7 @@ class SmartProtocol(BaseProtocol):
self._transport._config.batch_size or self.DEFAULT_MULTI_REQUEST_BATCH_SIZE
)
self._redact_data = True
self._method_missing_logged = False
def get_smart_request(self, method: str, params: dict | None = None) -> str:
"""Get a request message as a string."""
@ -180,6 +188,7 @@ class SmartProtocol(BaseProtocol):
multi_requests = [
{"method": method, "params": params} if params else {"method": method}
for method, params in requests.items()
if method not in FORCE_SINGLE_REQUEST
]
end = len(multi_requests)
@ -249,17 +258,18 @@ class SmartProtocol(BaseProtocol):
responses = response_step["result"]["responses"]
for response in responses:
# some smartcam devices calls do not populate the method key
# which we can only handle if there's a single request.
# these should be defined in DO_NOT_SEND_AS_MULTI_REQUEST.
if not (method := response.get("method")):
if len(requests) == 1:
method = next(iter(requests))
else:
_LOGGER.debug(
if not self._method_missing_logged:
# Avoid spamming the logs
self._method_missing_logged = True
_LOGGER.error(
"No method key in response for %s, skipping: %s",
self._host,
response,
response_step,
)
continue
# These will end up being queried individually
continue
self._handle_response_error_code(
response, method, raise_on_error=raise_on_error
@ -269,7 +279,9 @@ class SmartProtocol(BaseProtocol):
result, method, retry_count=retry_count
)
multi_result[method] = result
# Multi requests don't continue after errors so requery any missing
# Multi requests don't continue after errors so requery any missing.
# Will also query individually any DO_NOT_SEND_AS_MULTI_REQUEST.
for method, params in requests.items():
if method not in multi_result:
resp = await self._transport.send(

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

@ -16,6 +16,7 @@ from .energy import Energy
from .fan import Fan
from .firmware import Firmware
from .frostprotection import FrostProtection
from .homekit import HomeKit
from .humiditysensor import HumiditySensor
from .led import Led
from .light import Light
@ -23,6 +24,7 @@ from .lighteffect import LightEffect
from .lightpreset import LightPreset
from .lightstripeffect import LightStripEffect
from .lighttransition import LightTransition
from .matter import Matter
from .motionsensor import MotionSensor
from .overheatprotection import OverheatProtection
from .reportmode import ReportMode
@ -66,4 +68,6 @@ __all__ = [
"Thermostat",
"SmartLightEffect",
"OverheatProtection",
"HomeKit",
"Matter",
]

32
kasa/smart/modules/homekit.py Normal file
View File

@ -0,0 +1,32 @@
"""Implementation of homekit module."""
from __future__ import annotations
from ...feature import Feature
from ..smartmodule import SmartModule
class HomeKit(SmartModule):
"""Implementation of homekit module."""
QUERY_GETTER_NAME: str = "get_homekit_info"
REQUIRED_COMPONENT = "homekit"
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
self._device,
id="homekit_setup_code",
name="Homekit setup code",
container=self,
attribute_getter=lambda x: x.info["mfi_setup_code"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)
@property
def info(self) -> dict[str, str]:
"""Homekit mfi setup info."""
return self.data

23
kasa/smart/modules/light.py
View File

@ -55,7 +55,7 @@ class Light(SmartModule, LightInterface):
:return: White temperature range in Kelvin (minimum, maximum)
"""
if not self.is_variable_color_temp:
if Module.ColorTemperature not in self._device.modules:
raise KasaException("Color temperature not supported")
return self._device.modules[Module.ColorTemperature].valid_temperature_range
@ -66,7 +66,7 @@ class Light(SmartModule, LightInterface):
:return: hue, saturation and value (degrees, %, %)
"""
if not self.is_color:
if Module.Color not in self._device.modules:
raise KasaException("Bulb does not support color.")
return self._device.modules[Module.Color].hsv
@ -74,7 +74,7 @@ class Light(SmartModule, LightInterface):
@property
def color_temp(self) -> Annotated[int, FeatureAttribute()]:
"""Whether the bulb supports color temperature changes."""
if not self.is_variable_color_temp:
if Module.ColorTemperature not in self._device.modules:
raise KasaException("Bulb does not support colortemp.")
return self._device.modules[Module.ColorTemperature].color_temp
@ -82,7 +82,7 @@ class Light(SmartModule, LightInterface):
@property
def brightness(self) -> Annotated[int, FeatureAttribute()]:
"""Return the current brightness in percentage."""
if not self.is_dimmable: # pragma: no cover
if Module.Brightness not in self._device.modules:
raise KasaException("Bulb is not dimmable.")
return self._device.modules[Module.Brightness].brightness
@ -104,7 +104,7 @@ class Light(SmartModule, LightInterface):
:param int value: value between 1 and 100
:param int transition: transition in milliseconds.
"""
if not self.is_color:
if Module.Color not in self._device.modules:
raise KasaException("Bulb does not support color.")
return await self._device.modules[Module.Color].set_hsv(hue, saturation, value)
@ -119,7 +119,7 @@ class Light(SmartModule, LightInterface):
:param int temp: The new color temperature, in Kelvin
:param int transition: transition in milliseconds.
"""
if not self.is_variable_color_temp:
if Module.ColorTemperature not in self._device.modules:
raise KasaException("Bulb does not support colortemp.")
return await self._device.modules[Module.ColorTemperature].set_color_temp(
temp, brightness=brightness
@ -135,7 +135,7 @@ class Light(SmartModule, LightInterface):
:param int brightness: brightness in percent
:param int transition: transition in milliseconds.
"""
if not self.is_dimmable: # pragma: no cover
if Module.Brightness not in self._device.modules:
raise KasaException("Bulb is not dimmable.")
return await self._device.modules[Module.Brightness].set_brightness(brightness)
@ -167,16 +167,17 @@ class Light(SmartModule, LightInterface):
return self._light_state
async def _post_update_hook(self) -> None:
if self._device.is_on is False:
device = self._device
if device.is_on is False:
state = LightState(light_on=False)
else:
state = LightState(light_on=True)
if self.is_dimmable:
if Module.Brightness in device.modules:
state.brightness = self.brightness
if self.is_color:
if Module.Color in device.modules:
hsv = self.hsv
state.hue = hsv.hue
state.saturation = hsv.saturation
if self.is_variable_color_temp:
if Module.ColorTemperature in device.modules:
state.color_temp = self.color_temp
self._light_state = state

13
kasa/smart/modules/lightpreset.py
View File

@ -96,13 +96,18 @@ class LightPreset(SmartModule, LightPresetInterface):
"""Return current preset name."""
light = self._device.modules[SmartModule.Light]
brightness = light.brightness
color_temp = light.color_temp if light.is_variable_color_temp else None
h, s = (light.hsv.hue, light.hsv.saturation) if light.is_color else (None, None)
color_temp = light.color_temp if light.has_feature("color_temp") else None
h, s = (
(light.hsv.hue, light.hsv.saturation)
if light.has_feature("hsv")
else (None, None)
)
for preset_name, preset in self._presets.items():
if (
preset.brightness == brightness
and (
preset.color_temp == color_temp or not light.is_variable_color_temp
preset.color_temp == color_temp
or not light.has_feature("color_temp")
)
and preset.hue == h
and preset.saturation == s
@ -117,7 +122,7 @@ class LightPreset(SmartModule, LightPresetInterface):
"""Set a light preset for the device."""
light = self._device.modules[SmartModule.Light]
if preset_name == self.PRESET_NOT_SET:
if light.is_color:
if light.has_feature("hsv"):
preset = LightState(hue=0, saturation=0, brightness=100)
else:
preset = LightState(brightness=100)

43
kasa/smart/modules/matter.py Normal file
View File

@ -0,0 +1,43 @@
"""Implementation of matter module."""
from __future__ import annotations
from ...feature import Feature
from ..smartmodule import SmartModule
class Matter(SmartModule):
"""Implementation of matter module."""
QUERY_GETTER_NAME: str = "get_matter_setup_info"
REQUIRED_COMPONENT = "matter"
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
self._device,
id="matter_setup_code",
name="Matter setup code",
container=self,
attribute_getter=lambda x: x.info["setup_code"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)
self._add_feature(
Feature(
self._device,
id="matter_setup_payload",
name="Matter setup payload",
container=self,
attribute_getter=lambda x: x.info["setup_payload"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)
@property
def info(self) -> dict[str, str]:
"""Matter setup info."""
return self.data

17
kasa/smart/smartchilddevice.py
View File

@ -6,6 +6,7 @@ import logging
import time
from typing import Any
from ..device import DeviceInfo
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper
@ -50,6 +51,22 @@ class SmartChildDevice(SmartDevice):
self._components_raw = component_info_raw
self._components = self._parse_components(self._components_raw)
@property
def device_info(self) -> DeviceInfo:
"""Return device info.
Child device does not have it info and components in _last_update so
this overrides the base implementation to call _get_device_info with
info and components combined as they would be in _last_update.
"""
return self._get_device_info(
{
"get_device_info": self._info,
"component_nego": self._components_raw,
},
None,
)
async def update(self, update_children: bool = True) -> None:
"""Update child module info.

24
kasa/smart/smartdevice.py
View File

@ -9,7 +9,7 @@ from collections.abc import Mapping, Sequence
from datetime import UTC, datetime, timedelta, tzinfo
from typing import TYPE_CHECKING, Any, TypeAlias, cast
from ..device import Device, WifiNetwork, _DeviceInfo
from ..device import Device, DeviceInfo, WifiNetwork
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
from ..exceptions import AuthenticationError, DeviceError, KasaException, SmartErrorCode
@ -69,7 +69,6 @@ class SmartDevice(Device):
self._modules: dict[str | ModuleName[Module], SmartModule] = {}
self._parent: SmartDevice | None = None
self._children: Mapping[str, SmartDevice] = {}
self._last_update = {}
self._last_update_time: float | None = None
self._on_since: datetime | None = None
self._info: dict[str, Any] = {}
@ -497,18 +496,13 @@ class SmartDevice(Device):
@property
def model(self) -> str:
"""Returns the device model."""
return str(self._info.get("model"))
# If update hasn't been called self._device_info can't be used
if self._last_update:
return self.device_info.short_name
@property
def _model_region(self) -> str:
"""Return device full model name and region."""
if (disco := self._discovery_info) and (
disco_model := disco.get("device_model")
):
return disco_model
# Some devices have the region in the specs element.
region = f"({specs})" if (specs := self._info.get("specs")) else ""
return f"{self.model}{region}"
disco_model = str(self._info.get("device_model"))
long_name, _, _ = disco_model.partition("(")
return long_name
@property
def alias(self) -> str | None:
@ -808,7 +802,7 @@ class SmartDevice(Device):
@staticmethod
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> _DeviceInfo:
) -> DeviceInfo:
"""Get model information for a device."""
di = info["get_device_info"]
components = [comp["id"] for comp in info["component_nego"]["component_list"]]
@ -837,7 +831,7 @@ class SmartDevice(Device):
# Brand inferred from SMART.KASAPLUG/SMART.TAPOPLUG etc.
brand = devicetype[:4].lower()
return _DeviceInfo(
return DeviceInfo(
short_name=short_name,
long_name=long_name,
brand=brand,

6
kasa/smart/smartmodule.py
View File

@ -57,7 +57,7 @@ class SmartModule(Module):
#: Module is initialized, if any of the given keys exists in the sysinfo
SYSINFO_LOOKUP_KEYS: list[str] = []
#: Query to execute during the main update cycle
QUERY_GETTER_NAME: str
QUERY_GETTER_NAME: str = ""
REGISTERED_MODULES: dict[str, type[SmartModule]] = {}
@ -138,7 +138,9 @@ class SmartModule(Module):
Default implementation uses the raw query getter w/o parameters.
"""
return {self.QUERY_GETTER_NAME: None}
if self.QUERY_GETTER_NAME:
return {self.QUERY_GETTER_NAME: None}
return {}
async def call(self, method: str, params: dict | None = None) -> dict:
"""Call a method.

4
kasa/smartcam/modules/__init__.py
View File

@ -4,7 +4,9 @@ from .alarm import Alarm
from .camera import Camera
from .childdevice import ChildDevice
from .device import DeviceModule
from .homekit import HomeKit
from .led import Led
from .matter import Matter
from .pantilt import PanTilt
from .time import Time
@ -16,4 +18,6 @@ __all__ = [
"Led",
"PanTilt",
"Time",
"HomeKit",
"Matter",
]

16
kasa/smartcam/modules/homekit.py Normal file
View File

@ -0,0 +1,16 @@
"""Implementation of homekit module."""
from __future__ import annotations
from ..smartcammodule import SmartCamModule
class HomeKit(SmartCamModule):
"""Implementation of homekit module."""
REQUIRED_COMPONENT = "homekit"
@property
def info(self) -> dict[str, str]:
"""Not supported, return empty dict."""
return {}

44
kasa/smartcam/modules/matter.py Normal file
View File

@ -0,0 +1,44 @@
"""Implementation of matter module."""
from __future__ import annotations
from ...feature import Feature
from ..smartcammodule import SmartCamModule
class Matter(SmartCamModule):
"""Implementation of matter module."""
QUERY_GETTER_NAME = "getMatterSetupInfo"
QUERY_MODULE_NAME = "matter"
REQUIRED_COMPONENT = "matter"
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
self._device,
id="matter_setup_code",
name="Matter setup code",
container=self,
attribute_getter=lambda x: x.info["setup_code"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)
self._add_feature(
Feature(
self._device,
id="matter_setup_payload",
name="Matter setup payload",
container=self,
attribute_getter=lambda x: x.info["setup_payload"],
type=Feature.Type.Sensor,
category=Feature.Category.Debug,
)
)
@property
def info(self) -> dict[str, str]:
"""Matter setup info."""
return self.data

10
kasa/smartcam/smartcamdevice.py
View File

@ -5,7 +5,7 @@ from __future__ import annotations
import logging
from typing import Any, cast
from ..device import _DeviceInfo
from ..device import DeviceInfo
from ..device_type import DeviceType
from ..module import Module
from ..protocols.smartcamprotocol import _ChildCameraProtocolWrapper
@ -37,7 +37,7 @@ class SmartCamDevice(SmartDevice):
@staticmethod
def _get_device_info(
info: dict[str, Any], discovery_info: dict[str, Any] | None
) -> _DeviceInfo:
) -> DeviceInfo:
"""Get model information for a device."""
basic_info = info["getDeviceInfo"]["device_info"]["basic_info"]
short_name = basic_info["device_model"]
@ -45,7 +45,7 @@ class SmartCamDevice(SmartDevice):
device_type = SmartCamDevice._get_device_type_from_sysinfo(basic_info)
fw_version_full = basic_info["sw_version"]
firmware_version, firmware_build = fw_version_full.split(" ", maxsplit=1)
return _DeviceInfo(
return DeviceInfo(
short_name=basic_info["device_model"],
long_name=long_name,
brand="tapo",
@ -248,8 +248,8 @@ class SmartCamDevice(SmartDevice):
def hw_info(self) -> dict:
"""Return hardware info for the device."""
return {
"sw_ver": self._info.get("hw_ver"),
"hw_ver": self._info.get("fw_ver"),
"sw_ver": self._info.get("fw_ver"),
"hw_ver": self._info.get("hw_ver"),
"mac": self._info.get("mac"),
"type": self._info.get("type"),
"hwId": self._info.get("hwId"),

7
kasa/smartcam/smartcammodule.py
View File

@ -21,8 +21,6 @@ class SmartCamModule(SmartModule):
SmartCamAlarm: Final[ModuleName[modules.Alarm]] = ModuleName("SmartCamAlarm")
#: Query to execute during the main update cycle
QUERY_GETTER_NAME: str
#: Module name to be queried
QUERY_MODULE_NAME: str
#: Section name or names to be queried
@ -37,6 +35,8 @@ class SmartCamModule(SmartModule):
Default implementation uses the raw query getter w/o parameters.
"""
if not self.QUERY_GETTER_NAME:
return {}
section_names = (
{"name": self.QUERY_SECTION_NAMES} if self.QUERY_SECTION_NAMES else {}
)
@ -86,7 +86,8 @@ class SmartCamModule(SmartModule):
f" for '{self._module}'"
)
return query_resp.get(self.QUERY_MODULE_NAME)
# Some calls return the data under the module, others not
return query_resp.get(self.QUERY_MODULE_NAME, query_resp)
else:
found = {key: val for key, val in dev._last_update.items() if key in q}
for key in q:

8
tests/device_fixtures.py
View File

@ -473,8 +473,12 @@ def get_nearest_fixture_to_ip(dev):
assert protocol_fixtures, "Unknown device type"
# This will get the best fixture with a match on model region
if model_region_fixtures := filter_fixtures(
"", model_filter={dev._model_region}, fixture_list=protocol_fixtures
if (di := dev.device_info) and (
model_region_fixtures := filter_fixtures(
"",
model_filter={di.long_name + (f"({di.region})" if di.region else "")},
fixture_list=protocol_fixtures,
)
):
return next(iter(model_region_fixtures))

16
tests/fakeprotocol_smart.py
View File

@ -114,6 +114,15 @@ class FakeSmartTransport(BaseTransport):
"type": 0,
},
),
"get_homekit_info": (
"homekit",
{
"mfi_setup_code": "000-00-000",
"mfi_setup_id": "0000",
"mfi_token_token": "000000000000000000000000000000000",
"mfi_token_uuid": "00000000-0000-0000-0000-000000000000",
},
),
"get_auto_update_info": (
"firmware",
{"enable": True, "random_range": 120, "time": 180},
@ -151,6 +160,13 @@ class FakeSmartTransport(BaseTransport):
"energy_monitoring",
{"igain": 10861, "vgain": 118657},
),
"get_matter_setup_info": (
"matter",
{
"setup_code": "00000000000",
"setup_payload": "00:0000000-0000.00.000",
},
),
}
async def send(self, request: str):

35
tests/fakeprotocol_smartcam.py
View File

@ -34,6 +34,7 @@ class FakeSmartCamTransport(BaseTransport):
list_return_size=10,
is_child=False,
verbatim=False,
components_not_included=False,
):
super().__init__(
config=DeviceConfig(
@ -44,6 +45,7 @@ class FakeSmartCamTransport(BaseTransport):
),
),
)
self.fixture_name = fixture_name
# When True verbatim will bypass any extra processing of missing
# methods and is used to test the fixture creation itself.
@ -58,6 +60,17 @@ class FakeSmartCamTransport(BaseTransport):
# self.child_protocols = self._get_child_protocols()
self.list_return_size = list_return_size
# Setting this flag allows tests to create dummy transports without
# full fixture info for testing specific cases like list handling etc
self.components_not_included = (components_not_included,)
if not components_not_included:
self.components = {
comp["name"]: comp["version"]
for comp in self.info["getAppComponentList"]["app_component"][
"app_component_list"
]
}
@property
def default_port(self):
"""Default port for the transport."""
@ -112,6 +125,15 @@ class FakeSmartCamTransport(BaseTransport):
info = info[key]
info[set_keys[-1]] = value
FIXTURE_MISSING_MAP = {
"getMatterSetupInfo": (
"matter",
{
"setup_code": "00000000000",
"setup_payload": "00:0000000-0000.00.000",
},
)
}
# Setters for when there's not a simple mapping of setters to getters
SETTERS = {
("system", "sys", "dev_alias"): [
@ -217,8 +239,17 @@ class FakeSmartCamTransport(BaseTransport):
start_index : start_index + self.list_return_size
]
return {"result": result, "error_code": 0}
else:
return {"error_code": -1}
if (
# FIXTURE_MISSING is for service calls not in place when
# SMART fixtures started to be generated
missing_result := self.FIXTURE_MISSING_MAP.get(method)
) and missing_result[0] in self.components:
# Copy to info so it will work with update methods
info[method] = copy.deepcopy(missing_result[1])
result = copy.deepcopy(info[method])
return {"result": result, "error_code": 0}
return {"error_code": -1}
return {"error_code": -1}
async def close(self) -> None:

19
tests/fixtureinfo.py
View File

@ -145,12 +145,21 @@ def filter_fixtures(
def _component_match(
fixture_data: FixtureInfo, component_filter: str | ComponentFilter
):
if (component_nego := fixture_data.data.get("component_nego")) is None:
components = {}
if component_nego := fixture_data.data.get("component_nego"):
components = {
component["id"]: component["ver_code"]
for component in component_nego["component_list"]
}
if get_app_component_list := fixture_data.data.get("getAppComponentList"):
components = {
component["name"]: component["version"]
for component in get_app_component_list["app_component"][
"app_component_list"
]
}
if not components:
return False
components = {
component["id"]: component["ver_code"]
for component in component_nego["component_list"]
}
if isinstance(component_filter, str):
return component_filter in components
else:

4
tests/iot/test_iotbulb.py
View File

@ -91,7 +91,9 @@ async def test_unknown_temp_range(dev: IotBulb, monkeypatch, caplog):
monkeypatch.setitem(dev._sys_info, "model", "unknown bulb")
light = dev.modules.get(Module.Light)
assert light
assert light.valid_temperature_range == (2700, 5000)
color_temp_feat = light.get_feature("color_temp")
assert color_temp_feat
assert color_temp_feat.range == (2700, 5000)
assert "Unknown color temperature range, fallback to 2700-5000" in caplog.text

4
tests/iot/test_iotdevice.py
View File

@ -99,7 +99,7 @@ async def test_invalid_connection(mocker, dev):
@has_emeter_iot
async def test_initial_update_emeter(dev, mocker):
"""Test that the initial update performs second query if emeter is available."""
dev._last_update = None
dev._last_update = {}
dev._legacy_features = set()
spy = mocker.spy(dev.protocol, "query")
await dev.update()
@ -111,7 +111,7 @@ async def test_initial_update_emeter(dev, mocker):
@no_emeter_iot
async def test_initial_update_no_emeter(dev, mocker):
"""Test that the initial update performs second query if emeter is available."""
dev._last_update = None
dev._last_update = {}
dev._legacy_features = set()
spy = mocker.spy(dev.protocol, "query")
await dev.update()

80
tests/protocols/test_smartprotocol.py
View File

@ -2,6 +2,7 @@ import logging
import pytest
import pytest_mock
from pytest_mock import MockerFixture
from kasa.exceptions import (
SMART_RETRYABLE_ERRORS,
@ -14,6 +15,7 @@ from kasa.smart import SmartDevice
from ..conftest import device_smart
from ..fakeprotocol_smart import FakeSmartTransport
from ..fakeprotocol_smartcam import FakeSmartCamTransport
DUMMY_QUERY = {"foobar": {"foo": "bar", "bar": "foo"}}
DUMMY_MULTIPLE_QUERY = {
@ -448,3 +450,81 @@ async def test_smart_queries_redaction(
await dev.update()
assert device_id not in caplog.text
assert "REDACTED_" + device_id[9::] in caplog.text
async def test_no_method_returned_multiple(
mocker: MockerFixture, caplog: pytest.LogCaptureFixture
):
"""Test protocol handles multiple requests that don't return the method."""
req = {
"getDeviceInfo": {"device_info": {"name": ["basic_info", "info"]}},
"getAppComponentList": {"app_component": {"name": "app_component_list"}},
}
res = {
"result": {
"responses": [
{
"method": "getDeviceInfo",
"result": {
"device_info": {
"basic_info": {
"device_model": "C210",
},
}
},
"error_code": 0,
},
{
"result": {"app_component": {"app_component_list": []}},
"error_code": 0,
},
]
},
"error_code": 0,
}
transport = FakeSmartCamTransport(
{},
"dummy-name",
components_not_included=True,
)
protocol = SmartProtocol(transport=transport)
mocker.patch.object(protocol._transport, "send", return_value=res)
await protocol.query(req)
assert "No method key in response" in caplog.text
caplog.clear()
await protocol.query(req)
assert "No method key in response" not in caplog.text
async def test_no_multiple_methods(
mocker: MockerFixture, caplog: pytest.LogCaptureFixture
):
"""Test protocol sends NO_MULTI methods as single call."""
req = {
"getDeviceInfo": {"device_info": {"name": ["basic_info", "info"]}},
"getConnectStatus": {"onboarding": {"get_connect_status": {}}},
}
info = {
"getDeviceInfo": {
"device_info": {
"basic_info": {
"avatar": "Home",
}
}
},
"getConnectStatus": {
"onboarding": {
"get_connect_status": {"current_ssid": "", "err_code": 0, "status": 0}
}
},
}
transport = FakeSmartCamTransport(
info,
"dummy-name",
components_not_included=True,
)
protocol = SmartProtocol(transport=transport)
send_spy = mocker.spy(protocol._transport, "send")
await protocol.query(req)
assert send_spy.call_count == 2

16
tests/smart/modules/test_homekit.py Normal file
View File

@ -0,0 +1,16 @@
from kasa import Module
from kasa.smart import SmartDevice
from ...device_fixtures import parametrize
homekit = parametrize(
"has homekit", component_filter="homekit", protocol_filter={"SMART"}
)
@homekit
async def test_info(dev: SmartDevice):
"""Test homekit info."""
homekit = dev.modules.get(Module.HomeKit)
assert homekit
assert homekit.info

20
tests/smart/modules/test_matter.py Normal file
View File

@ -0,0 +1,20 @@
from kasa import Module
from kasa.smart import SmartDevice
from ...device_fixtures import parametrize
matter = parametrize(
"has matter", component_filter="matter", protocol_filter={"SMART", "SMARTCAM"}
)
@matter
async def test_info(dev: SmartDevice):
"""Test matter info."""
matter = dev.modules.get(Module.Matter)
assert matter
assert matter.info
setup_code = dev.features.get("matter_setup_code")
assert setup_code
setup_payload = dev.features.get("matter_setup_payload")
assert setup_payload

24
tests/smart/test_smartdevice.py
View File

@ -62,11 +62,14 @@ async def test_device_type_no_update(discovery_mock, caplog: pytest.LogCaptureFi
assert repr(dev) == f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - update() needed>"
discovery_result = copy.deepcopy(discovery_mock.discovery_data["result"])
disco_model = discovery_result["device_model"]
short_model, _, _ = disco_model.partition("(")
dev.update_from_discover_info(discovery_result)
assert dev.device_type is DeviceType.Unknown
assert (
repr(dev)
== f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - None (None) - update() needed>"
== f"<DeviceType.Unknown at {DISCOVERY_MOCK_IP} - None ({short_model}) - update() needed>"
)
discovery_result["device_type"] = "SMART.FOOBAR"
dev.update_from_discover_info(discovery_result)
@ -74,7 +77,7 @@ async def test_device_type_no_update(discovery_mock, caplog: pytest.LogCaptureFi
assert dev.device_type is DeviceType.Plug
assert (
repr(dev)
== f"<DeviceType.Plug at {DISCOVERY_MOCK_IP} - None (None) - update() needed>"
== f"<DeviceType.Plug at {DISCOVERY_MOCK_IP} - None ({short_model}) - update() needed>"
)
assert "Unknown device type, falling back to plug" in caplog.text
@ -469,7 +472,9 @@ async def test_smartdevice_cloud_connection(dev: SmartDevice, mocker: MockerFixt
async def test_smart_temp_range(dev: Device):
light = dev.modules.get(Module.Light)
assert light
assert light.valid_temperature_range
color_temp_feat = light.get_feature("color_temp")
assert color_temp_feat
assert color_temp_feat.range
@device_smart
@ -528,3 +533,16 @@ async def test_initialize_modules_required_component(
assert "AvailableComponent" in dev.modules
assert "NonExistingComponent" not in dev.modules
async def test_smartmodule_query():
"""Test that a module that doesn't set QUERY_GETTER_NAME has empty query."""
class DummyModule(SmartModule):
pass
dummy_device = await get_device_for_fixture_protocol(
"KS240(US)_1.0_1.0.5.json", "SMART"
)
mod = DummyModule(dummy_device, "dummy")
assert mod.query() == {}

9
tests/test_bulb.py
View File

@ -25,7 +25,7 @@ async def test_hsv(dev: Device, turn_on):
light = dev.modules.get(Module.Light)
assert light
await handle_turn_on(dev, turn_on)
assert light.is_color
assert light.has_feature("hsv")
hue, saturation, brightness = light.hsv
assert 0 <= hue <= 360
@ -106,7 +106,7 @@ async def test_invalid_hsv(
light = dev.modules.get(Module.Light)
assert light
await handle_turn_on(dev, turn_on)
assert light.is_color
assert light.has_feature("hsv")
with pytest.raises(exception_cls, match=error):
await light.set_hsv(hue, sat, brightness)
@ -124,7 +124,7 @@ async def test_color_state_information(dev: Device):
async def test_hsv_on_non_color(dev: Device):
light = dev.modules.get(Module.Light)
assert light
assert not light.is_color
assert not light.has_feature("hsv")
with pytest.raises(KasaException):
await light.set_hsv(0, 0, 0)
@ -173,9 +173,6 @@ async def test_non_variable_temp(dev: Device):
with pytest.raises(KasaException):
await light.set_color_temp(2700)
with pytest.raises(KasaException):
print(light.valid_temperature_range)
with pytest.raises(KasaException):
print(light.color_temp)

14
tests/test_cli.py
View File

@ -12,6 +12,7 @@ from pytest_mock import MockerFixture
from kasa import (
AuthenticationError,
ColorTempRange,
Credentials,
Device,
DeviceError,
@ -523,7 +524,9 @@ async def test_emeter(dev: Device, mocker, runner):
async def test_brightness(dev: Device, runner):
res = await runner.invoke(brightness, obj=dev)
if not (light := dev.modules.get(Module.Light)) or not light.is_dimmable:
if not (light := dev.modules.get(Module.Light)) or not light.has_feature(
"brightness"
):
assert "This device does not support brightness." in res.output
return
@ -540,13 +543,16 @@ async def test_brightness(dev: Device, runner):
async def test_color_temperature(dev: Device, runner):
res = await runner.invoke(temperature, obj=dev)
if not (light := dev.modules.get(Module.Light)) or not light.is_variable_color_temp:
if not (light := dev.modules.get(Module.Light)) or not (
color_temp_feat := light.get_feature("color_temp")
):
assert "Device does not support color temperature" in res.output
return
res = await runner.invoke(temperature, obj=dev)
assert f"Color temperature: {light.color_temp}" in res.output
valid_range = light.valid_temperature_range
valid_range = color_temp_feat.range
assert isinstance(valid_range, ColorTempRange)
assert f"(min: {valid_range.min}, max: {valid_range.max})" in res.output
val = int((valid_range.min + valid_range.max) / 2)
@ -572,7 +578,7 @@ async def test_color_temperature(dev: Device, runner):
async def test_color_hsv(dev: Device, runner: CliRunner):
res = await runner.invoke(hsv, obj=dev)
if not (light := dev.modules.get(Module.Light)) or not light.is_color:
if not (light := dev.modules.get(Module.Light)) or not light.has_feature("hsv"):
assert "Device does not support colors" in res.output
return

2
tests/test_common_modules.py
View File

@ -198,7 +198,7 @@ async def test_light_color_temp(dev: Device):
light = next(get_parent_and_child_modules(dev, Module.Light))
assert light
if not light.is_variable_color_temp:
if not light.has_feature("color_temp"):
pytest.skip(
"Some smart light strips have color_temperature"
" component but min and max are the same"

6
tests/test_device.py
View File

@ -280,19 +280,19 @@ async def test_deprecated_light_attributes(dev: Device):
await _test_attribute(dev, "is_color", bool(light), "Light")
await _test_attribute(dev, "is_variable_color_temp", bool(light), "Light")
exc = KasaException if light and not light.is_dimmable else None
exc = KasaException if light and not light.has_feature("brightness") else None
await _test_attribute(dev, "brightness", bool(light), "Light", will_raise=exc)
await _test_attribute(
dev, "set_brightness", bool(light), "Light", 50, will_raise=exc
)
exc = KasaException if light and not light.is_color else None
exc = KasaException if light and not light.has_feature("hsv") else None
await _test_attribute(dev, "hsv", bool(light), "Light", will_raise=exc)
await _test_attribute(
dev, "set_hsv", bool(light), "Light", 50, 50, 50, will_raise=exc
)
exc = KasaException if light and not light.is_variable_color_temp else None
exc = KasaException if light and not light.has_feature("color_temp") else None
await _test_attribute(dev, "color_temp", bool(light), "Light", will_raise=exc)
await _test_attribute(
dev, "set_color_temp", bool(light), "Light", 2700, will_raise=exc

11
tests/test_discovery.py
View File

@ -390,13 +390,12 @@ async def test_device_update_from_new_discovery_info(discovery_mock):
device_class = Discover._get_device_class(discovery_data)
device = device_class("127.0.0.1")
discover_info = DiscoveryResult.from_dict(discovery_data["result"])
discover_dump = discover_info.to_dict()
model, _, _ = discover_dump["device_model"].partition("(")
discover_dump["model"] = model
device.update_from_discover_info(discover_dump)
assert device.mac == discover_dump["mac"].replace("-", ":")
assert device.model == model
device.update_from_discover_info(discovery_data["result"])
assert device.mac == discover_info.mac.replace("-", ":")
no_region_model, _, _ = discover_info.device_model.partition("(")
assert device.model == no_region_model
# TODO implement requires_update for SmartDevice
if isinstance(device, IotDevice):

2
tests/test_readme_examples.py
View File

@ -19,7 +19,7 @@ def test_bulb_examples(mocker):
assert not res["failed"]
def test_smartdevice_examples(mocker):
def test_iotdevice_examples(mocker):
"""Use HS110 for emeter examples."""
p = asyncio.run(get_device_for_fixture_protocol("HS110(EU)_1.0_1.2.5.json", "IOT"))
asyncio.run(p.set_alias("Bedroom Lamp Plug"))