Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2025-08-06 10:44:04 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Enable ruff check for ANN (#1139)

Browse Source
This commit is contained in:
Teemu R.
2024-11-10 19:55:13 +01:00
committed by GitHub
parent 6b44fe6242
commit 66eb17057e
89 changed files with 596 additions and 452 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
kasa/iot/iotbulb.py
View File

@@ -54,7 +54,7 @@ class TurnOnBehavior(BaseModel):
mode: BehaviorMode
@root_validator
def _mode_based_on_preset(cls, values):
def _mode_based_on_preset(cls, values: dict) -> dict:
"""Set the mode based on the preset value."""
if values["preset"] is not None:
values["mode"] = BehaviorMode.Preset
@@ -209,7 +209,7 @@ class IotBulb(IotDevice):
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Bulb
async def _initialize_modules(self):
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
await super()._initialize_modules()
self.add_module(
@@ -307,7 +307,7 @@ class IotBulb(IotDevice):
await self._query_helper(self.LIGHT_SERVICE, "get_default_behavior")
)
async def set_turn_on_behavior(self, behavior: TurnOnBehaviors):
async def set_turn_on_behavior(self, behavior: TurnOnBehaviors) -> dict:
"""Set the behavior for turning the bulb on.
If you do not want to manually construct the behavior object,
@@ -426,7 +426,7 @@ class IotBulb(IotDevice):
@requires_update
async def _set_color_temp(
self, temp: int, *, brightness=None, transition: int | None = None
self, temp: int, *, brightness: int | None = None, transition: int | None = None
) -> dict:
"""Set the color temperature of the device in kelvin.
@@ -450,7 +450,7 @@ class IotBulb(IotDevice):
return await self._set_light_state(light_state, transition=transition)
def _raise_for_invalid_brightness(self, value):
def _raise_for_invalid_brightness(self, value: int) -> None:
if not isinstance(value, int):
raise TypeError("Brightness must be an integer")
if not (0 <= value <= 100):
@@ -517,7 +517,7 @@ class IotBulb(IotDevice):
"""Return that the bulb has an emeter."""
return True
async def set_alias(self, alias: str) -> None:
async def set_alias(self, alias: str) -> dict:
"""Set the device name (alias).
Overridden to use a different module name.

52
kasa/iot/iotdevice.py
View File

@@ -19,7 +19,7 @@ import inspect
import logging
from collections.abc import Mapping, Sequence
from datetime import datetime, timedelta, tzinfo
from typing import TYPE_CHECKING, Any, cast
from typing import TYPE_CHECKING, Any, Callable, cast
from warnings import warn
from ..device import Device, WifiNetwork
@@ -35,12 +35,12 @@ from .modules import Emeter
_LOGGER = logging.getLogger(__name__)
def requires_update(f):
def requires_update(f: Callable) -> Any:
"""Indicate that `update` should be called before accessing this method.""" # noqa: D202
if inspect.iscoroutinefunction(f):
@functools.wraps(f)
async def wrapped(*args, **kwargs):
async def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if self._last_update is None and f.__name__ not in self._sys_info:
raise KasaException("You need to await update() to access the data")
@@ -49,13 +49,13 @@ def requires_update(f):
else:
@functools.wraps(f)
def wrapped(*args, **kwargs):
def wrapped(*args: Any, **kwargs: Any) -> Any:
self = args[0]
if self._last_update is None and f.__name__ not in self._sys_info:
raise KasaException("You need to await update() to access the data")
return f(*args, **kwargs)
f.requires_update = True
f.requires_update = True # type: ignore[attr-defined]
return wrapped
@@ -197,7 +197,7 @@ class IotDevice(Device):
return cast(ModuleMapping[IotModule], self._supported_modules)
return self._supported_modules
def add_module(self, name: str | ModuleName[Module], module: IotModule):
def add_module(self, name: str | ModuleName[Module], module: IotModule) -> None:
"""Register a module."""
if name in self._modules:
_LOGGER.debug("Module %s already registered, ignoring...", name)
@@ -207,8 +207,12 @@ class IotDevice(Device):
self._modules[name] = module
def _create_request(
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
):
self,
target: str,
cmd: str,
arg: dict | None = None,
child_ids: list | None = None,
) -> dict:
if arg is None:
arg = {}
request: dict[str, Any] = {target: {cmd: arg}}
@@ -225,8 +229,12 @@ class IotDevice(Device):
raise KasaException("update() required prior accessing emeter")
async def _query_helper(
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
) -> Any:
self,
target: str,
cmd: str,
arg: dict | None = None,
child_ids: list | None = None,
) -> dict:
"""Query device, return results or raise an exception.
:param target: Target system {system, time, emeter, ..}
@@ -276,7 +284,7 @@ class IotDevice(Device):
"""Retrieve system information."""
return await self._query_helper("system", "get_sysinfo")
async def update(self, update_children: bool = True):
async def update(self, update_children: bool = True) -> None:
"""Query the device to update the data.
Needed for properties that are decorated with `requires_update`.
@@ -305,7 +313,7 @@ class IotDevice(Device):
if not self._features:
await self._initialize_features()
async def _initialize_modules(self):
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
if self.has_emeter:
_LOGGER.debug(
@@ -313,7 +321,7 @@ class IotDevice(Device):
)
self.add_module(Module.Energy, Emeter(self, self.emeter_type))
async def _initialize_features(self):
async def _initialize_features(self) -> None:
"""Initialize common features."""
self._add_feature(
Feature(
@@ -364,7 +372,7 @@ class IotDevice(Device):
)
)
for module in self._supported_modules.values():
for module in self.modules.values():
module._initialize_features()
for module_feat in module._module_features.values():
self._add_feature(module_feat)
@@ -453,7 +461,7 @@ class IotDevice(Device):
sys_info = self._sys_info
return sys_info.get("alias") if sys_info else None
async def set_alias(self, alias: str) -> None:
async def set_alias(self, alias: str) -> dict:
"""Set the device name (alias)."""
return await self._query_helper("system", "set_dev_alias", {"alias": alias})
@@ -550,7 +558,7 @@ class IotDevice(Device):
return mac
async def set_mac(self, mac):
async def set_mac(self, mac: str) -> dict:
"""Set the mac address.
:param str mac: mac in hexadecimal with colons, e.g. 01:23:45:67:89:ab
@@ -576,7 +584,7 @@ class IotDevice(Device):
"""Turn off the device."""
raise NotImplementedError("Device subclass needs to implement this.")
async def turn_on(self, **kwargs) -> dict | None:
async def turn_on(self, **kwargs) -> dict:
"""Turn device on."""
raise NotImplementedError("Device subclass needs to implement this.")
@@ -586,7 +594,7 @@ class IotDevice(Device):
"""Return True if the device is on."""
raise NotImplementedError("Device subclass needs to implement this.")
async def set_state(self, on: bool):
async def set_state(self, on: bool) -> dict:
"""Set the device state."""
if on:
return await self.turn_on()
@@ -627,7 +635,7 @@ class IotDevice(Device):
async def wifi_scan(self) -> list[WifiNetwork]: # noqa: D202
"""Scan for available wifi networks."""
async def _scan(target):
async def _scan(target: str) -> dict:
return await self._query_helper(target, "get_scaninfo", {"refresh": 1})
try:
@@ -639,17 +647,17 @@ class IotDevice(Device):
info = await _scan("smartlife.iot.common.softaponboarding")
if "ap_list" not in info:
raise KasaException("Invalid response for wifi scan: %s" % info)
raise KasaException(f"Invalid response for wifi scan: {info}")
return [WifiNetwork(**x) for x in info["ap_list"]]
async def wifi_join(self, ssid: str, password: str, keytype: str = "3"): # noqa: D202
async def wifi_join(self, ssid: str, password: str, keytype: str = "3") -> dict: # noqa: D202
"""Join the given wifi network.
If joining the network fails, the device will return to AP mode after a while.
"""
async def _join(target, payload):
async def _join(target: str, payload: dict) -> dict:
return await self._query_helper(target, "set_stainfo", payload)
payload = {"ssid": ssid, "password": password, "key_type": int(keytype)}

26
kasa/iot/iotdimmer.py
View File

@@ -80,7 +80,7 @@ class IotDimmer(IotPlug):
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Dimmer
async def _initialize_modules(self):
async def _initialize_modules(self) -> None:
"""Initialize modules."""
await super()._initialize_modules()
# TODO: need to be verified if it's okay to call these on HS220 w/o these
@@ -103,7 +103,9 @@ class IotDimmer(IotPlug):
return int(sys_info["brightness"])
@requires_update
async def _set_brightness(self, brightness: int, *, transition: int | None = None):
async def _set_brightness(
self, brightness: int, *, transition: int | None = None
) -> dict:
"""Set the new dimmer brightness level in percentage.
:param int transition: transition duration in milliseconds.
@@ -134,7 +136,7 @@ class IotDimmer(IotPlug):
self.DIMMER_SERVICE, "set_brightness", {"brightness": brightness}
)
async def turn_off(self, *, transition: int | None = None, **kwargs):
async def turn_off(self, *, transition: int | None = None, **kwargs) -> dict:
"""Turn the bulb off.
:param int transition: transition duration in milliseconds.
@@ -145,7 +147,7 @@ class IotDimmer(IotPlug):
return await super().turn_off()
@requires_update
async def turn_on(self, *, transition: int | None = None, **kwargs):
async def turn_on(self, *, transition: int | None = None, **kwargs) -> dict:
"""Turn the bulb on.
:param int transition: transition duration in milliseconds.
@@ -157,7 +159,7 @@ class IotDimmer(IotPlug):
return await super().turn_on()
async def set_dimmer_transition(self, brightness: int, transition: int):
async def set_dimmer_transition(self, brightness: int, transition: int) -> dict:
"""Turn the bulb on to brightness percentage over transition milliseconds.
A brightness value of 0 will turn off the dimmer.
@@ -176,7 +178,7 @@ class IotDimmer(IotPlug):
if not isinstance(transition, int):
raise TypeError(f"Transition must be integer, not of {type(transition)}.")
if transition <= 0:
raise ValueError("Transition value %s is not valid." % transition)
raise ValueError(f"Transition value {transition} is not valid.")
return await self._query_helper(
self.DIMMER_SERVICE,
@@ -185,7 +187,7 @@ class IotDimmer(IotPlug):
)
@requires_update
async def get_behaviors(self):
async def get_behaviors(self) -> dict:
"""Return button behavior settings."""
behaviors = await self._query_helper(
self.DIMMER_SERVICE, "get_default_behavior", {}
@@ -195,7 +197,7 @@ class IotDimmer(IotPlug):
@requires_update
async def set_button_action(
self, action_type: ActionType, action: ButtonAction, index: int | None = None
):
) -> dict:
"""Set action to perform on button click/hold.
:param action_type ActionType: whether to control double click or hold action.
@@ -209,15 +211,17 @@ class IotDimmer(IotPlug):
if index is not None:
payload["index"] = index
await self._query_helper(self.DIMMER_SERVICE, action_type_setter, payload)
return await self._query_helper(
self.DIMMER_SERVICE, action_type_setter, payload
)
@requires_update
async def set_fade_time(self, fade_type: FadeType, time: int):
async def set_fade_time(self, fade_type: FadeType, time: int) -> dict:
"""Set time for fade in / fade out."""
fade_type_setter = f"set_{fade_type}_time"
payload = {"fadeTime": time}
await self._query_helper(self.DIMMER_SERVICE, fade_type_setter, payload)
return await self._query_helper(self.DIMMER_SERVICE, fade_type_setter, payload)
@property # type: ignore
@requires_update

2
kasa/iot/iotlightstrip.py
View File

@@ -57,7 +57,7 @@ class IotLightStrip(IotBulb):
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.LightStrip
async def _initialize_modules(self):
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
await super()._initialize_modules()
self.add_module(

13
kasa/iot/iotmodule.py
View File

@@ -1,6 +1,9 @@
"""Base class for IOT module implementations."""
from __future__ import annotations
import logging
from typing import Any
from ..exceptions import KasaException
from ..module import Module
@@ -24,16 +27,16 @@ merge = _merge_dict
class IotModule(Module):
"""Base class implemention for all IOT modules."""
def call(self, method, params=None):
async def call(self, method: str, params: dict | None = None) -> dict:
"""Call the given method with the given parameters."""
return self._device._query_helper(self._module, method, params)
return await self._device._query_helper(self._module, method, params)
def query_for_command(self, query, params=None):
def query_for_command(self, query: str, params: dict | None = None) -> dict:
"""Create a request object for the given parameters."""
return self._device._create_request(self._module, query, params)
@property
def estimated_query_response_size(self):
def estimated_query_response_size(self) -> int:
"""Estimated maximum size of query response.
The inheriting modules implement this to estimate how large a query response
@@ -42,7 +45,7 @@ class IotModule(Module):
return 256 # Estimate for modules that don't specify
@property
def data(self):
def data(self) -> dict[str, Any]:
"""Return the module specific raw data from the last update."""
dev = self._device
q = self.query()

7
kasa/iot/iotplug.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
import logging
from typing import Any
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
@@ -54,7 +55,7 @@ class IotPlug(IotDevice):
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Plug
async def _initialize_modules(self):
async def _initialize_modules(self) -> None:
"""Initialize modules."""
await super()._initialize_modules()
self.add_module(Module.IotSchedule, Schedule(self, "schedule"))
@@ -71,11 +72,11 @@ class IotPlug(IotDevice):
sys_info = self.sys_info
return bool(sys_info["relay_state"])
async def turn_on(self, **kwargs):
async def turn_on(self, **kwargs: Any) -> dict:
"""Turn the switch on."""
return await self._query_helper("system", "set_relay_state", {"state": 1})
async def turn_off(self, **kwargs):
async def turn_off(self, **kwargs: Any) -> dict:
"""Turn the switch off."""
return await self._query_helper("system", "set_relay_state", {"state": 0})

49
kasa/iot/iotstrip.py
View File

@@ -26,7 +26,7 @@ from .modules import Antitheft, Cloud, Countdown, Emeter, Led, Schedule, Time, U
_LOGGER = logging.getLogger(__name__)
def merge_sums(dicts):
def merge_sums(dicts: list[dict]) -> dict:
"""Merge the sum of dicts."""
total_dict: defaultdict[int, float] = defaultdict(lambda: 0.0)
for sum_dict in dicts:
@@ -99,7 +99,7 @@ class IotStrip(IotDevice):
self.emeter_type = "emeter"
self._device_type = DeviceType.Strip
async def _initialize_modules(self):
async def _initialize_modules(self) -> None:
"""Initialize modules."""
# Strip has different modules to plug so do not call super
self.add_module(Module.IotAntitheft, Antitheft(self, "anti_theft"))
@@ -121,7 +121,7 @@ class IotStrip(IotDevice):
"""Return if any of the outlets are on."""
return any(plug.is_on for plug in self.children)
async def update(self, update_children: bool = True):
async def update(self, update_children: bool = True) -> None:
"""Update some of the attributes.
Needed for methods that are decorated with `requires_update`.
@@ -150,20 +150,20 @@ class IotStrip(IotDevice):
if not self.features:
await self._initialize_features()
async def _initialize_features(self):
async def _initialize_features(self) -> None:
"""Initialize common features."""
# Do not initialize features until children are created
if not self.children:
return
await super()._initialize_features()
async def turn_on(self, **kwargs):
async def turn_on(self, **kwargs) -> dict:
"""Turn the strip on."""
await self._query_helper("system", "set_relay_state", {"state": 1})
return await self._query_helper("system", "set_relay_state", {"state": 1})
async def turn_off(self, **kwargs):
async def turn_off(self, **kwargs) -> dict:
"""Turn the strip off."""
await self._query_helper("system", "set_relay_state", {"state": 0})
return await self._query_helper("system", "set_relay_state", {"state": 0})
@property # type: ignore
@requires_update
@@ -188,7 +188,7 @@ class StripEmeter(IotModule, Energy):
"""Return True if module supports the feature."""
return module_feature in self._supported
def query(self):
def query(self) -> dict:
"""Return the base query."""
return {}
@@ -246,11 +246,13 @@ class StripEmeter(IotModule, Energy):
]
)
async def erase_stats(self):
async def erase_stats(self) -> dict:
"""Erase energy meter statistics for all plugs."""
for plug in self._device.children:
await plug.modules[Module.Energy].erase_stats()
return {}
@property # type: ignore
def consumption_this_month(self) -> float | None:
"""Return this month's energy consumption in kWh."""
@@ -320,7 +322,7 @@ class IotStripPlug(IotPlug):
self.protocol = parent.protocol # Must use the same connection as the parent
self._on_since: datetime | None = None
async def _initialize_modules(self):
async def _initialize_modules(self) -> None:
"""Initialize modules not added in init."""
if self.has_emeter:
self.add_module(Module.Energy, Emeter(self, self.emeter_type))
@@ -329,7 +331,7 @@ class IotStripPlug(IotPlug):
self.add_module(Module.IotSchedule, Schedule(self, "schedule"))
self.add_module(Module.IotCountdown, Countdown(self, "countdown"))
async def _initialize_features(self):
async def _initialize_features(self) -> None:
"""Initialize common features."""
self._add_feature(
Feature(
@@ -353,19 +355,20 @@ class IotStripPlug(IotPlug):
type=Feature.Type.Sensor,
)
)
for module in self._supported_modules.values():
for module in self.modules.values():
module._initialize_features()
for module_feat in module._module_features.values():
self._add_feature(module_feat)
async def update(self, update_children: bool = True):
async def update(self, update_children: bool = True) -> None:
"""Query the device to update the data.
Needed for properties that are decorated with `requires_update`.
"""
await self._update(update_children)
async def _update(self, update_children: bool = True):
async def _update(self, update_children: bool = True) -> None:
"""Query the device to update the data.
Internal implementation to allow patching of public update in the cli
@@ -379,8 +382,12 @@ class IotStripPlug(IotPlug):
await self._initialize_features()
def _create_request(
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
):
self,
target: str,
cmd: str,
arg: dict | None = None,
child_ids: list | None = None,
) -> dict:
request: dict[str, Any] = {
"context": {"child_ids": [self.child_id]},
target: {cmd: arg},
@@ -388,8 +395,12 @@ class IotStripPlug(IotPlug):
return request
async def _query_helper(
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
) -> Any:
self,
target: str,
cmd: str,
arg: dict | None = None,
child_ids: list | None = None,
) -> dict:
"""Override query helper to include the child_ids."""
return await self._parent._query_helper(
target, cmd, arg, child_ids=[self.child_id]

10
kasa/iot/modules/ambientlight.py
View File

@@ -11,7 +11,7 @@ _LOGGER = logging.getLogger(__name__)
class AmbientLight(IotModule):
"""Implements ambient light controls for the motion sensor."""
def _initialize_features(self):
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
@@ -40,7 +40,7 @@ class AmbientLight(IotModule):
)
)
def query(self):
def query(self) -> dict:
"""Request configuration."""
req = merge(
self.query_for_command("get_config"),
@@ -74,18 +74,18 @@ class AmbientLight(IotModule):
"""Return True if the module is enabled."""
return int(self.data["get_current_brt"]["value"])
async def set_enabled(self, state: bool):
async def set_enabled(self, state: bool) -> dict:
"""Enable/disable LAS."""
return await self.call("set_enable", {"enable": int(state)})
async def current_brightness(self) -> int:
async def current_brightness(self) -> dict:
"""Return current brightness.
Return value units.
"""
return await self.call("get_current_brt")
async def set_brightness_limit(self, value: int):
async def set_brightness_limit(self, value: int) -> dict:
"""Set the limit when the motion sensor is inactive.
See `presets` for preset values. Custom values are also likely allowed.

12
kasa/iot/modules/cloud.py
View File

@@ -24,7 +24,7 @@ class CloudInfo(BaseModel):
class Cloud(IotModule):
"""Module implementing support for cloud services."""
def _initialize_features(self):
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
self._add_feature(
Feature(
@@ -44,7 +44,7 @@ class Cloud(IotModule):
"""Return true if device is connected to the cloud."""
return self.info.binded
def query(self):
def query(self) -> dict:
"""Request cloud connectivity info."""
return self.query_for_command("get_info")
@@ -53,20 +53,20 @@ class Cloud(IotModule):
"""Return information about the cloud connectivity."""
return CloudInfo.parse_obj(self.data["get_info"])
def get_available_firmwares(self):
def get_available_firmwares(self) -> dict:
"""Return list of available firmwares."""
return self.query_for_command("get_intl_fw_list")
def set_server(self, url: str):
def set_server(self, url: str) -> dict:
"""Set the update server URL."""
return self.query_for_command("set_server_url", {"server": url})
def connect(self, username: str, password: str):
def connect(self, username: str, password: str) -> dict:
"""Login to the cloud using given information."""
return self.query_for_command(
"bind", {"username": username, "password": password}
)
def disconnect(self):
def disconnect(self) -> dict:
"""Disconnect from the cloud."""
return self.query_for_command("unbind")

10
kasa/iot/modules/emeter.py
View File

@@ -70,7 +70,7 @@ class Emeter(Usage, EnergyInterface):
"""Get the current voltage in V."""
return self.status.voltage
async def erase_stats(self):
async def erase_stats(self) -> dict:
"""Erase all stats.
Uses different query than usage meter.
@@ -81,7 +81,9 @@ class Emeter(Usage, EnergyInterface):
"""Return real-time statistics."""
return EmeterStatus(await self.call("get_realtime"))
async def get_daily_stats(self, *, year=None, month=None, kwh=True) -> dict:
async def get_daily_stats(
self, *, year: int | None = None, month: int | None = None, kwh: bool = True
) -> dict:
"""Return daily stats for the given year & month.
The return value is a dictionary of {day: energy, ...}.
@@ -90,7 +92,9 @@ class Emeter(Usage, EnergyInterface):
data = self._convert_stat_data(data["day_list"], entry_key="day", kwh=kwh)
return data
async def get_monthly_stats(self, *, year=None, kwh=True) -> dict:
async def get_monthly_stats(
self, *, year: int | None = None, kwh: bool = True
) -> dict:
"""Return monthly stats for the given year.
The return value is a dictionary of {month: energy, ...}.

4
kasa/iot/modules/led.py
View File

@@ -14,7 +14,7 @@ class Led(IotModule, LedInterface):
return {}
@property
def mode(self):
def mode(self) -> str:
"""LED mode setting.
"always", "never"
@@ -27,7 +27,7 @@ class Led(IotModule, LedInterface):
sys_info = self.data
return bool(1 - sys_info["led_off"])
async def set_led(self, state: bool):
async def set_led(self, state: bool) -> dict:
"""Set the state of the led (night mode)."""
return await self.call("set_led_off", {"off": int(not state)})

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

@@ -27,7 +27,7 @@ class Light(IotModule, LightInterface):
_device: IotBulb | IotDimmer
_light_state: LightState
def _initialize_features(self):
def _initialize_features(self) -> None:
"""Initialize features."""
super()._initialize_features()
device = self._device
@@ -185,7 +185,7 @@ class Light(IotModule, LightInterface):
return bulb._color_temp
async def set_color_temp(
self, temp: int, *, brightness=None, transition: int | None = None
self, temp: int, *, brightness: int | None = None, transition: int | None = None
) -> dict:
"""Set the color temperature of the device in kelvin.

10
kasa/iot/modules/lighteffect.py
View File

@@ -50,7 +50,7 @@ class LightEffect(IotModule, LightEffectInterface):
*,
brightness: int | None = None,
transition: int | None = None,
) -> None:
) -> dict:
"""Set an effect on the device.
If brightness or transition is defined,
@@ -73,7 +73,7 @@ class LightEffect(IotModule, LightEffectInterface):
effect_dict = EFFECT_MAPPING_V1["Aurora"]
effect_dict = {**effect_dict}
effect_dict["enable"] = 0
await self.set_custom_effect(effect_dict)
return await self.set_custom_effect(effect_dict)
elif effect not in EFFECT_MAPPING_V1:
raise ValueError(f"The effect {effect} is not a built in effect.")
else:
@@ -84,12 +84,12 @@ class LightEffect(IotModule, LightEffectInterface):
if transition is not None:
effect_dict["transition"] = transition
await self.set_custom_effect(effect_dict)
return await self.set_custom_effect(effect_dict)
async def set_custom_effect(
self,
effect_dict: dict,
) -> None:
) -> dict:
"""Set a custom effect on the device.
:param str effect_dict: The custom effect dict to set
@@ -104,7 +104,7 @@ class LightEffect(IotModule, LightEffectInterface):
"""Return True if the device supports setting custom effects."""
return True
def query(self):
def query(self) -> dict:
"""Return the base query."""
return {}

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

@@ -41,7 +41,7 @@ class LightPreset(IotModule, LightPresetInterface):
_presets: dict[str, IotLightPreset]
_preset_list: list[str]
async def _post_update_hook(self):
async def _post_update_hook(self) -> None:
"""Update the internal presets."""
self._presets = {
f"Light preset {index+1}": IotLightPreset(**vals)
@@ -93,7 +93,7 @@ class LightPreset(IotModule, LightPresetInterface):
async def set_preset(
self,
preset_name: str,
) -> None:
) -> dict:
"""Set a light preset for the device."""
light = self._device.modules[Module.Light]
if preset_name == self.PRESET_NOT_SET:
@@ -104,7 +104,7 @@ class LightPreset(IotModule, LightPresetInterface):
elif (preset := self._presets.get(preset_name)) is None: # type: ignore[assignment]
raise ValueError(f"{preset_name} is not a valid preset: {self.preset_list}")
await light.set_state(preset)
return await light.set_state(preset)
@property
def has_save_preset(self) -> bool:
@@ -115,7 +115,7 @@ class LightPreset(IotModule, LightPresetInterface):
self,
preset_name: str,
preset_state: LightState,
) -> None:
) -> dict:
"""Update the preset with preset_name with the new preset_info."""
if len(self._presets) == 0:
raise KasaException("Device does not supported saving presets")
@@ -129,7 +129,7 @@ class LightPreset(IotModule, LightPresetInterface):
return await self.call("set_preferred_state", state)
def query(self):
def query(self) -> dict:
"""Return the base query."""
return {}
@@ -142,7 +142,7 @@ class LightPreset(IotModule, LightPresetInterface):
if "id" not in vals
]
async def _deprecated_save_preset(self, preset: IotLightPreset):
async def _deprecated_save_preset(self, preset: IotLightPreset) -> dict:
"""Save a setting preset.
You can either construct a preset object manually, or pass an existing one

10
kasa/iot/modules/motion.py
View File

@@ -24,7 +24,7 @@ class Range(Enum):
class Motion(IotModule):
"""Implements the motion detection (PIR) module."""
def _initialize_features(self):
def _initialize_features(self) -> None:
"""Initialize features after the initial update."""
# Only add features if the device supports the module
if "get_config" not in self.data:
@@ -48,7 +48,7 @@ class Motion(IotModule):
)
)
def query(self):
def query(self) -> dict:
"""Request PIR configuration."""
return self.query_for_command("get_config")
@@ -67,13 +67,13 @@ class Motion(IotModule):
"""Return True if module is enabled."""
return bool(self.config["enable"])
async def set_enabled(self, state: bool):
async def set_enabled(self, state: bool) -> dict:
"""Enable/disable PIR."""
return await self.call("set_enable", {"enable": int(state)})
async def set_range(
self, *, range: Range | None = None, custom_range: int | None = None
):
) -> dict:
"""Set the range for the sensor.
:param range: for using standard ranges
@@ -93,7 +93,7 @@ class Motion(IotModule):
"""Return inactivity timeout in milliseconds."""
return self.config["cold_time"]
async def set_inactivity_timeout(self, timeout: int):
async def set_inactivity_timeout(self, timeout: int) -> dict:
"""Set inactivity timeout in milliseconds.
Note, that you need to delete the default "Smart Control" rule in the app

10
kasa/iot/modules/rulemodule.py
View File

@@ -57,7 +57,7 @@ _LOGGER = logging.getLogger(__name__)
class RuleModule(IotModule):
"""Base class for rule-based modules, such as countdown and antitheft."""
def query(self):
def query(self) -> dict:
"""Prepare the query for rules."""
q = self.query_for_command("get_rules")
return merge(q, self.query_for_command("get_next_action"))
@@ -73,14 +73,14 @@ class RuleModule(IotModule):
_LOGGER.error("Unable to read rule list: %s (data: %s)", ex, self.data)
return []
async def set_enabled(self, state: bool):
async def set_enabled(self, state: bool) -> dict:
"""Enable or disable the service."""
return await self.call("set_overall_enable", state)
return await self.call("set_overall_enable", {"enable": state})
async def delete_rule(self, rule: Rule):
async def delete_rule(self, rule: Rule) -> dict:
"""Delete the given rule."""
return await self.call("delete_rule", {"id": rule.id})
async def delete_all_rules(self):
async def delete_all_rules(self) -> dict:
"""Delete all rules."""
return await self.call("delete_all_rules")

8
kasa/iot/modules/time.py
View File

@@ -15,14 +15,14 @@ class Time(IotModule, TimeInterface):
_timezone: tzinfo = timezone.utc
def query(self):
def query(self) -> dict:
"""Request time and timezone."""
q = self.query_for_command("get_time")
merge(q, self.query_for_command("get_timezone"))
return q
async def _post_update_hook(self):
async def _post_update_hook(self) -> None:
"""Perform actions after a device update."""
if res := self.data.get("get_timezone"):
self._timezone = await get_timezone(res.get("index"))
@@ -47,7 +47,7 @@ class Time(IotModule, TimeInterface):
"""Return current timezone."""
return self._timezone
async def get_time(self):
async def get_time(self) -> datetime | None:
"""Return current device time."""
try:
res = await self.call("get_time")
@@ -88,6 +88,6 @@ class Time(IotModule, TimeInterface):
except Exception as ex:
raise KasaException(ex) from ex
async def get_timezone(self):
async def get_timezone(self) -> dict:
"""Request timezone information from the device."""
return await self.call("get_timezone")

32
kasa/iot/modules/usage.py
View File

@@ -10,7 +10,7 @@ from ..iotmodule import IotModule, merge
class Usage(IotModule):
"""Baseclass for emeter/usage interfaces."""
def query(self):
def query(self) -> dict:
"""Return the base query."""
now = datetime.now()
year = now.year
@@ -25,22 +25,22 @@ class Usage(IotModule):
return req
@property
def estimated_query_response_size(self):
def estimated_query_response_size(self) -> int:
"""Estimated maximum query response size."""
return 2048
@property
def daily_data(self):
def daily_data(self) -> list[dict]:
"""Return statistics on daily basis."""
return self.data["get_daystat"]["day_list"]
@property
def monthly_data(self):
def monthly_data(self) -> list[dict]:
"""Return statistics on monthly basis."""
return self.data["get_monthstat"]["month_list"]
@property
def usage_today(self):
def usage_today(self) -> int | None:
"""Return today's usage in minutes."""
today = datetime.now().day
# Traverse the list in reverse order to find the latest entry.
@@ -50,7 +50,7 @@ class Usage(IotModule):
return None
@property
def usage_this_month(self):
def usage_this_month(self) -> int | None:
"""Return usage in this month in minutes."""
this_month = datetime.now().month
# Traverse the list in reverse order to find the latest entry.
@@ -59,7 +59,9 @@ class Usage(IotModule):
return entry["time"]
return None
async def get_raw_daystat(self, *, year=None, month=None) -> dict:
async def get_raw_daystat(
self, *, year: int | None = None, month: int | None = None
) -> dict:
"""Return raw daily stats for the given year & month."""
if year is None:
year = datetime.now().year
@@ -68,14 +70,16 @@ class Usage(IotModule):
return await self.call("get_daystat", {"year": year, "month": month})
async def get_raw_monthstat(self, *, year=None) -> dict:
async def get_raw_monthstat(self, *, year: int | None = None) -> dict:
"""Return raw monthly stats for the given year."""
if year is None:
year = datetime.now().year
return await self.call("get_monthstat", {"year": year})
async def get_daystat(self, *, year=None, month=None) -> dict:
async def get_daystat(
self, *, year: int | None = None, month: int | None = None
) -> dict:
"""Return daily stats for the given year & month.
The return value is a dictionary of {day: time, ...}.
@@ -84,7 +88,7 @@ class Usage(IotModule):
data = self._convert_stat_data(data["day_list"], entry_key="day")
return data
async def get_monthstat(self, *, year=None) -> dict:
async def get_monthstat(self, *, year: int | None = None) -> dict:
"""Return monthly stats for the given year.
The return value is a dictionary of {month: time, ...}.
@@ -93,11 +97,11 @@ class Usage(IotModule):
data = self._convert_stat_data(data["month_list"], entry_key="month")
return data
async def erase_stats(self):
async def erase_stats(self) -> dict:
"""Erase all stats."""
return await self.call("erase_runtime_stat")
def _convert_stat_data(self, data, entry_key) -> dict:
def _convert_stat_data(self, data: list[dict], entry_key: str) -> dict:
"""Return usage information keyed with the day/month.
The incoming data is a list of dictionaries::
@@ -113,6 +117,6 @@ class Usage(IotModule):
if not data:
return {}
data = {entry[entry_key]: entry["time"] for entry in data}
res = {entry[entry_key]: entry["time"] for entry in data}
return data
return res