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

Enable and convert to future annotations (#838)

Browse Source
This commit is contained in:
Steven B 2024-04-17 14:39:24 +01:00 committed by GitHub
parent 82d92aeea5
commit 203bd79253
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
59 changed files with 562 additions and 462 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
devtools/dump_devinfo.py
View File

@ -8,6 +8,8 @@ Executing this script will several modules and methods one by one,
and finally execute a query to query all of them at once.
"""
from __future__ import annotations
import base64
import collections.abc
import json
@ -17,7 +19,6 @@ import traceback
from collections import defaultdict, namedtuple
from pathlib import Path
from pprint import pprint
from typing import Dict, List, Union
import asyncclick as click
@ -143,7 +144,7 @@ def default_to_regular(d):
async def handle_device(basedir, autosave, device: Device, batch_size: int):
"""Create a fixture for a single device instance."""
if isinstance(device, SmartDevice):
fixture_results: List[FixtureResult] = await get_smart_fixtures(
fixture_results: list[FixtureResult] = await get_smart_fixtures(
device, batch_size
)
else:
@ -344,12 +345,12 @@ def _echo_error(msg: str):
async def _make_requests_or_exit(
device: SmartDevice,
requests: List[SmartRequest],
requests: list[SmartRequest],
name: str,
batch_size: int,
*,
child_device_id: str,
) -> Dict[str, Dict]:
) -> dict[str, dict]:
final = {}
protocol = (
device.protocol
@ -362,7 +363,7 @@ async def _make_requests_or_exit(
for i in range(0, end, step):
x = i
requests_step = requests[x : x + step]
request: Union[List[SmartRequest], SmartRequest] = (
request: list[SmartRequest] | SmartRequest = (
requests_step[0] if len(requests_step) == 1 else requests_step
)
responses = await protocol.query(SmartRequest._create_request_dict(request))
@ -586,7 +587,7 @@ async def get_smart_fixtures(device: SmartDevice, batch_size: int):
finally:
await device.protocol.close()
device_requests: Dict[str, List[SmartRequest]] = {}
device_requests: dict[str, list[SmartRequest]] = {}
for success in successes:
device_request = device_requests.setdefault(success.child_device_id, [])
device_request.append(success.request)

91
devtools/helpers/smartrequests.py
View File

@ -25,9 +25,10 @@ heart_beat
"""
from __future__ import annotations
import logging
from dataclasses import asdict, dataclass
from typing import List, Optional, Union
_LOGGER = logging.getLogger(__name__)
@ -35,7 +36,7 @@ _LOGGER = logging.getLogger(__name__)
class SmartRequest:
"""Class to represent a smart protocol request."""
def __init__(self, method_name: str, params: Optional["SmartRequestParams"] = None):
def __init__(self, method_name: str, params: SmartRequestParams | None = None):
self.method_name = method_name
if params:
self.params = params.to_dict()
@ -93,7 +94,7 @@ class SmartRequest:
class LedStatusParams(SmartRequestParams):
"""LED Status params."""
led_rule: Optional[str] = None
led_rule: str | None = None
@staticmethod
def from_bool(state: bool):
@ -105,42 +106,42 @@ class SmartRequest:
class LightInfoParams(SmartRequestParams):
"""LightInfo params."""
brightness: Optional[int] = None
color_temp: Optional[int] = None
hue: Optional[int] = None
saturation: Optional[int] = None
brightness: int | None = None
color_temp: int | None = None
hue: int | None = None
saturation: int | None = None
@dataclass
class DynamicLightEffectParams(SmartRequestParams):
"""LightInfo params."""
enable: bool
id: Optional[str] = None
id: str | None = None
@staticmethod
def get_raw_request(
method: str, params: Optional[SmartRequestParams] = None
) -> "SmartRequest":
method: str, params: SmartRequestParams | None = None
) -> SmartRequest:
"""Send a raw request to the device."""
return SmartRequest(method, params)
@staticmethod
def component_nego() -> "SmartRequest":
def component_nego() -> SmartRequest:
"""Get quick setup component info."""
return SmartRequest("component_nego")
@staticmethod
def get_device_info() -> "SmartRequest":
def get_device_info() -> SmartRequest:
"""Get device info."""
return SmartRequest("get_device_info")
@staticmethod
def get_device_usage() -> "SmartRequest":
def get_device_usage() -> SmartRequest:
"""Get device usage."""
return SmartRequest("get_device_usage")
@staticmethod
def device_info_list(ver_code) -> List["SmartRequest"]:
def device_info_list(ver_code) -> list[SmartRequest]:
"""Get device info list."""
if ver_code == 1:
return [SmartRequest.get_device_info()]
@ -151,12 +152,12 @@ class SmartRequest:
]
@staticmethod
def get_auto_update_info() -> "SmartRequest":
def get_auto_update_info() -> SmartRequest:
"""Get auto update info."""
return SmartRequest("get_auto_update_info")
@staticmethod
def firmware_info_list() -> List["SmartRequest"]:
def firmware_info_list() -> list[SmartRequest]:
"""Get info list."""
return [
SmartRequest.get_raw_request("get_fw_download_state"),
@ -164,48 +165,48 @@ class SmartRequest:
]
@staticmethod
def qs_component_nego() -> "SmartRequest":
def qs_component_nego() -> SmartRequest:
"""Get quick setup component info."""
return SmartRequest("qs_component_nego")
@staticmethod
def get_device_time() -> "SmartRequest":
def get_device_time() -> SmartRequest:
"""Get device time."""
return SmartRequest("get_device_time")
@staticmethod
def get_child_device_list() -> "SmartRequest":
def get_child_device_list() -> SmartRequest:
"""Get child device list."""
return SmartRequest("get_child_device_list")
@staticmethod
def get_child_device_component_list() -> "SmartRequest":
def get_child_device_component_list() -> SmartRequest:
"""Get child device component list."""
return SmartRequest("get_child_device_component_list")
@staticmethod
def get_wireless_scan_info(
params: Optional[GetRulesParams] = None,
) -> "SmartRequest":
params: GetRulesParams | None = None,
) -> SmartRequest:
"""Get wireless scan info."""
return SmartRequest(
"get_wireless_scan_info", params or SmartRequest.GetRulesParams()
)
@staticmethod
def get_schedule_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_schedule_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get schedule rules."""
return SmartRequest(
"get_schedule_rules", params or SmartRequest.GetScheduleRulesParams()
)
@staticmethod
def get_next_event(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_next_event(params: GetRulesParams | None = None) -> SmartRequest:
"""Get next scheduled event."""
return SmartRequest("get_next_event", params or SmartRequest.GetRulesParams())
@staticmethod
def schedule_info_list() -> List["SmartRequest"]:
def schedule_info_list() -> list[SmartRequest]:
"""Get schedule info list."""
return [
SmartRequest.get_schedule_rules(),
@ -213,38 +214,38 @@ class SmartRequest:
]
@staticmethod
def get_countdown_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_countdown_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get countdown rules."""
return SmartRequest(
"get_countdown_rules", params or SmartRequest.GetRulesParams()
)
@staticmethod
def get_antitheft_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_antitheft_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get antitheft rules."""
return SmartRequest(
"get_antitheft_rules", params or SmartRequest.GetRulesParams()
)
@staticmethod
def get_led_info(params: Optional[LedStatusParams] = None) -> "SmartRequest":
def get_led_info(params: LedStatusParams | None = None) -> SmartRequest:
"""Get led info."""
return SmartRequest("get_led_info", params or SmartRequest.LedStatusParams())
@staticmethod
def get_auto_off_config(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_auto_off_config(params: GetRulesParams | None = None) -> SmartRequest:
"""Get auto off config."""
return SmartRequest(
"get_auto_off_config", params or SmartRequest.GetRulesParams()
)
@staticmethod
def get_delay_action_info() -> "SmartRequest":
def get_delay_action_info() -> SmartRequest:
"""Get delay action info."""
return SmartRequest("get_delay_action_info")
@staticmethod
def auto_off_list() -> List["SmartRequest"]:
def auto_off_list() -> list[SmartRequest]:
"""Get energy usage."""
return [
SmartRequest.get_auto_off_config(),
@ -252,12 +253,12 @@ class SmartRequest:
]
@staticmethod
def get_energy_usage() -> "SmartRequest":
def get_energy_usage() -> SmartRequest:
"""Get energy usage."""
return SmartRequest("get_energy_usage")
@staticmethod
def energy_monitoring_list() -> List["SmartRequest"]:
def energy_monitoring_list() -> list[SmartRequest]:
"""Get energy usage."""
return [
SmartRequest("get_energy_usage"),
@ -265,12 +266,12 @@ class SmartRequest:
]
@staticmethod
def get_current_power() -> "SmartRequest":
def get_current_power() -> SmartRequest:
"""Get current power."""
return SmartRequest("get_current_power")
@staticmethod
def power_protection_list() -> List["SmartRequest"]:
def power_protection_list() -> list[SmartRequest]:
"""Get power protection info list."""
return [
SmartRequest.get_current_power(),
@ -279,45 +280,45 @@ class SmartRequest:
]
@staticmethod
def get_preset_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_preset_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get preset rules."""
return SmartRequest("get_preset_rules", params or SmartRequest.GetRulesParams())
@staticmethod
def get_auto_light_info() -> "SmartRequest":
def get_auto_light_info() -> SmartRequest:
"""Get auto light info."""
return SmartRequest("get_auto_light_info")
@staticmethod
def get_dynamic_light_effect_rules(
params: Optional[GetRulesParams] = None,
) -> "SmartRequest":
params: GetRulesParams | None = None,
) -> SmartRequest:
"""Get dynamic light effect rules."""
return SmartRequest(
"get_dynamic_light_effect_rules", params or SmartRequest.GetRulesParams()
)
@staticmethod
def set_device_on(params: DeviceOnParams) -> "SmartRequest":
def set_device_on(params: DeviceOnParams) -> SmartRequest:
"""Set device on state."""
return SmartRequest("set_device_info", params)
@staticmethod
def set_light_info(params: LightInfoParams) -> "SmartRequest":
def set_light_info(params: LightInfoParams) -> SmartRequest:
"""Set color temperature."""
return SmartRequest("set_device_info", params)
@staticmethod
def set_dynamic_light_effect_rule_enable(
params: DynamicLightEffectParams,
) -> "SmartRequest":
) -> SmartRequest:
"""Enable dynamic light effect rule."""
return SmartRequest("set_dynamic_light_effect_rule_enable", params)
@staticmethod
def get_component_info_requests(component_nego_response) -> List["SmartRequest"]:
def get_component_info_requests(component_nego_response) -> list[SmartRequest]:
"""Get a list of requests based on the component info response."""
request_list: List["SmartRequest"] = []
request_list: list[SmartRequest] = []
for component in component_nego_response["component_list"]:
if (
requests := get_component_requests(
@ -329,7 +330,7 @@ class SmartRequest:
@staticmethod
def _create_request_dict(
smart_request: Union["SmartRequest", List["SmartRequest"]],
smart_request: SmartRequest | list[SmartRequest],
) -> dict:
"""Create request dict to be passed to SmartProtocol.query()."""
if isinstance(smart_request, list):

26
kasa/aestransport.py
View File

@ -4,13 +4,15 @@ Based on the work of https://github.com/petretiandrea/plugp100
under compatible GNU GPL3 license.
"""
from __future__ import annotations
import asyncio
import base64
import hashlib
import logging
import time
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, Optional, Tuple, cast
from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, cast
from cryptography.hazmat.primitives import padding, serialization
from cryptography.hazmat.primitives.asymmetric import padding as asymmetric_padding
@ -92,19 +94,19 @@ class AesTransport(BaseTransport):
self._login_params = json_loads(
base64.b64decode(self._credentials_hash.encode()).decode() # type: ignore[union-attr]
)
self._default_credentials: Optional[Credentials] = None
self._default_credentials: Credentials | None = None
self._http_client: HttpClient = HttpClient(config)
self._state = TransportState.HANDSHAKE_REQUIRED
self._encryption_session: Optional[AesEncyptionSession] = None
self._session_expire_at: Optional[float] = None
self._encryption_session: AesEncyptionSession | None = None
self._session_expire_at: float | None = None
self._session_cookie: Optional[Dict[str, str]] = None
self._session_cookie: dict[str, str] | None = None
self._key_pair: Optional[KeyPair] = None
self._key_pair: KeyPair | None = None
self._app_url = URL(f"http://{self._host}:{self._port}/app")
self._token_url: Optional[URL] = None
self._token_url: URL | None = None
_LOGGER.debug("Created AES transport for %s", self._host)
@ -118,14 +120,14 @@ class AesTransport(BaseTransport):
"""The hashed credentials used by the transport."""
return base64.b64encode(json_dumps(self._login_params).encode()).decode()
def _get_login_params(self, credentials: Credentials) -> Dict[str, str]:
def _get_login_params(self, credentials: Credentials) -> dict[str, str]:
"""Get the login parameters based on the login_version."""
un, pw = self.hash_credentials(self._login_version == 2, credentials)
password_field_name = "password2" if self._login_version == 2 else "password"
return {password_field_name: pw, "username": un}
@staticmethod
def hash_credentials(login_v2: bool, credentials: Credentials) -> Tuple[str, str]:
def hash_credentials(login_v2: bool, credentials: Credentials) -> tuple[str, str]:
"""Hash the credentials."""
un = base64.b64encode(_sha1(credentials.username.encode()).encode()).decode()
if login_v2:
@ -148,7 +150,7 @@ class AesTransport(BaseTransport):
raise AuthenticationError(msg, error_code=error_code)
raise DeviceError(msg, error_code=error_code)
async def send_secure_passthrough(self, request: str) -> Dict[str, Any]:
async def send_secure_passthrough(self, request: str) -> dict[str, Any]:
"""Send encrypted message as passthrough."""
if self._state is TransportState.ESTABLISHED and self._token_url:
url = self._token_url
@ -230,7 +232,7 @@ class AesTransport(BaseTransport):
ex,
) from ex
async def try_login(self, login_params: Dict[str, Any]) -> None:
async def try_login(self, login_params: dict[str, Any]) -> None:
"""Try to login with supplied login_params."""
login_request = {
"method": "login_device",
@ -333,7 +335,7 @@ class AesTransport(BaseTransport):
or self._session_expire_at - time.time() <= 0
)
async def send(self, request: str) -> Dict[str, Any]:
async def send(self, request: str) -> dict[str, Any]:
"""Send the request."""
if (
self._state is TransportState.HANDSHAKE_REQUIRED

32
kasa/bulb.py
View File

@ -1,7 +1,9 @@
"""Module for Device base class."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Dict, List, NamedTuple, Optional
from typing import NamedTuple, Optional
from .device import Device
@ -33,14 +35,14 @@ class BulbPreset(BaseModel):
brightness: int
# These are not available for effect mode presets on light strips
hue: Optional[int]
saturation: Optional[int]
color_temp: Optional[int]
hue: Optional[int] # noqa: UP007
saturation: Optional[int] # noqa: UP007
color_temp: Optional[int] # noqa: UP007
# Variables for effect mode presets
custom: Optional[int]
id: Optional[str]
mode: Optional[int]
custom: Optional[int] # noqa: UP007
id: Optional[str] # noqa: UP007
mode: Optional[int] # noqa: UP007
class Bulb(Device, ABC):
@ -101,10 +103,10 @@ class Bulb(Device, ABC):
self,
hue: int,
saturation: int,
value: Optional[int] = None,
value: int | None = None,
*,
transition: Optional[int] = None,
) -> Dict:
transition: int | None = None,
) -> dict:
"""Set new HSV.
Note, transition is not supported and will be ignored.
@ -117,8 +119,8 @@ class Bulb(Device, ABC):
@abstractmethod
async def set_color_temp(
self, temp: int, *, brightness=None, transition: Optional[int] = None
) -> Dict:
self, temp: int, *, brightness=None, transition: int | None = None
) -> dict:
"""Set the color temperature of the device in kelvin.
Note, transition is not supported and will be ignored.
@ -129,8 +131,8 @@ class Bulb(Device, ABC):
@abstractmethod
async def set_brightness(
self, brightness: int, *, transition: Optional[int] = None
) -> Dict:
self, brightness: int, *, transition: int | None = None
) -> dict:
"""Set the brightness in percentage.
Note, transition is not supported and will be ignored.
@ -141,5 +143,5 @@ class Bulb(Device, ABC):
@property
@abstractmethod
def presets(self) -> List[BulbPreset]:
def presets(self) -> list[BulbPreset]:
"""Return a list of available bulb setting presets."""

6
kasa/cli.py
View File

@ -1,5 +1,7 @@
"""python-kasa cli tool."""
from __future__ import annotations
import ast
import asyncio
import json
@ -9,7 +11,7 @@ import sys
from contextlib import asynccontextmanager
from functools import singledispatch, wraps
from pprint import pformat as pf
from typing import Any, Dict, cast
from typing import Any, cast
import asyncclick as click
@ -320,7 +322,7 @@ async def cli(
global _do_echo
echo = _do_echo
logging_config: Dict[str, Any] = {
logging_config: dict[str, Any] = {
"level": logging.DEBUG if debug > 0 else logging.INFO
}
try:

76
kasa/device.py
View File

@ -1,10 +1,12 @@
"""Module for Device base class."""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Dict, List, Mapping, Optional, Sequence, Union
from typing import Any, Mapping, Sequence
from .credentials import Credentials
from .device_type import DeviceType
@ -24,13 +26,13 @@ class WifiNetwork:
ssid: str
key_type: int
# These are available only on softaponboarding
cipher_type: Optional[int] = None
bssid: Optional[str] = None
channel: Optional[int] = None
rssi: Optional[int] = None
cipher_type: int | None = None
bssid: str | None = None
channel: int | None = None
rssi: int | None = None
# For SMART devices
signal_level: Optional[int] = None
signal_level: int | None = None
_LOGGER = logging.getLogger(__name__)
@ -48,8 +50,8 @@ class Device(ABC):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
"""Create a new Device instance.
@ -68,19 +70,19 @@ class Device(ABC):
# checks in accessors. the @updated_required decorator does not ensure
# mypy that these are not accessed incorrectly.
self._last_update: Any = None
self._discovery_info: Optional[Dict[str, Any]] = None
self._discovery_info: dict[str, Any] | None = None
self.modules: Dict[str, Any] = {}
self._features: Dict[str, Feature] = {}
self._parent: Optional["Device"] = None
self._children: Mapping[str, "Device"] = {}
self.modules: dict[str, Any] = {}
self._features: dict[str, Feature] = {}
self._parent: Device | None = None
self._children: Mapping[str, Device] = {}
@staticmethod
async def connect(
*,
host: Optional[str] = None,
config: Optional[DeviceConfig] = None,
) -> "Device":
host: str | None = None,
config: DeviceConfig | None = None,
) -> Device:
"""Connect to a single device by the given hostname or device configuration.
This method avoids the UDP based discovery process and
@ -120,11 +122,11 @@ class Device(ABC):
return not self.is_on
@abstractmethod
async def turn_on(self, **kwargs) -> Optional[Dict]:
async def turn_on(self, **kwargs) -> dict | None:
"""Turn on the device."""
@abstractmethod
async def turn_off(self, **kwargs) -> Optional[Dict]:
async def turn_off(self, **kwargs) -> dict | None:
"""Turn off the device."""
@property
@ -147,12 +149,12 @@ class Device(ABC):
return self.protocol._transport._port
@property
def credentials(self) -> Optional[Credentials]:
def credentials(self) -> Credentials | None:
"""The device credentials."""
return self.protocol._transport._credentials
@property
def credentials_hash(self) -> Optional[str]:
def credentials_hash(self) -> str | None:
"""The protocol specific hash of the credentials the device is using."""
return self.protocol._transport.credentials_hash
@ -177,25 +179,25 @@ class Device(ABC):
@property
@abstractmethod
def alias(self) -> Optional[str]:
def alias(self) -> str | None:
"""Returns the device alias or nickname."""
async def _raw_query(self, request: Union[str, Dict]) -> Any:
async def _raw_query(self, request: str | dict) -> Any:
"""Send a raw query to the device."""
return await self.protocol.query(request=request)
@property
def children(self) -> Sequence["Device"]:
def children(self) -> Sequence[Device]:
"""Returns the child devices."""
return list(self._children.values())
def get_child_device(self, id_: str) -> "Device":
def get_child_device(self, id_: str) -> Device:
"""Return child device by its ID."""
return self._children[id_]
@property
@abstractmethod
def sys_info(self) -> Dict[str, Any]:
def sys_info(self) -> dict[str, Any]:
"""Returns the device info."""
@property
@ -248,7 +250,7 @@ class Device(ABC):
"""Return True if the device supports color changes."""
return False
def get_plug_by_name(self, name: str) -> "Device":
def get_plug_by_name(self, name: str) -> Device:
"""Return child device for the given name."""
for p in self.children:
if p.alias == name:
@ -256,7 +258,7 @@ class Device(ABC):
raise KasaException(f"Device has no child with {name}")
def get_plug_by_index(self, index: int) -> "Device":
def get_plug_by_index(self, index: int) -> Device:
"""Return child device for the given index."""
if index + 1 > len(self.children) or index < 0:
raise KasaException(
@ -271,22 +273,22 @@ class Device(ABC):
@property
@abstractmethod
def timezone(self) -> Dict:
def timezone(self) -> dict:
"""Return the timezone and time_difference."""
@property
@abstractmethod
def hw_info(self) -> Dict:
def hw_info(self) -> dict:
"""Return hardware info for the device."""
@property
@abstractmethod
def location(self) -> Dict:
def location(self) -> dict:
"""Return the device location."""
@property
@abstractmethod
def rssi(self) -> Optional[int]:
def rssi(self) -> int | None:
"""Return the rssi."""
@property
@ -305,12 +307,12 @@ class Device(ABC):
"""Return all the internal state data."""
@property
def state_information(self) -> Dict[str, Any]:
def state_information(self) -> dict[str, Any]:
"""Return available features and their values."""
return {feat.name: feat.value for feat in self._features.values()}
@property
def features(self) -> Dict[str, Feature]:
def features(self) -> dict[str, Feature]:
"""Return the list of supported features."""
return self._features
@ -328,7 +330,7 @@ class Device(ABC):
@property
@abstractmethod
def on_since(self) -> Optional[datetime]:
def on_since(self) -> datetime | None:
"""Return the time that the device was turned on or None if turned off."""
@abstractmethod
@ -342,18 +344,18 @@ class Device(ABC):
@property
@abstractmethod
def emeter_this_month(self) -> Optional[float]:
def emeter_this_month(self) -> float | None:
"""Get the emeter value for this month."""
@property
@abstractmethod
def emeter_today(self) -> Union[Optional[float], Any]:
def emeter_today(self) -> float | None | Any:
"""Get the emeter value for today."""
# Return type of Any ensures consumers being shielded from the return
# type by @update_required are not affected.
@abstractmethod
async def wifi_scan(self) -> List[WifiNetwork]:
async def wifi_scan(self) -> list[WifiNetwork]:
"""Scan for available wifi networks."""
@abstractmethod

30
kasa/device_factory.py
View File

@ -1,8 +1,10 @@
"""Device creation via DeviceConfig."""
from __future__ import annotations
import logging
import time
from typing import Any, Dict, Optional, Tuple, Type
from typing import Any
from .aestransport import AesTransport
from .device import Device
@ -35,7 +37,7 @@ GET_SYSINFO_QUERY = {
}
async def connect(*, host: Optional[str] = None, config: DeviceConfig) -> "Device":
async def connect(*, host: str | None = None, config: DeviceConfig) -> Device:
"""Connect to a single device by the given hostname or device configuration.
This method avoids the UDP based discovery process and
@ -72,7 +74,7 @@ async def connect(*, host: Optional[str] = None, config: DeviceConfig) -> "Devic
raise
async def _connect(config: DeviceConfig, protocol: BaseProtocol) -> "Device":
async def _connect(config: DeviceConfig, protocol: BaseProtocol) -> Device:
debug_enabled = _LOGGER.isEnabledFor(logging.DEBUG)
if debug_enabled:
start_time = time.perf_counter()
@ -87,8 +89,8 @@ async def _connect(config: DeviceConfig, protocol: BaseProtocol) -> "Device":
)
start_time = time.perf_counter()
device_class: Optional[Type[Device]]
device: Optional[Device] = None
device_class: type[Device] | None
device: Device | None = None
if isinstance(protocol, IotProtocol) and isinstance(
protocol._transport, XorTransport
@ -115,13 +117,13 @@ async def _connect(config: DeviceConfig, protocol: BaseProtocol) -> "Device":
)
def _get_device_type_from_sys_info(info: Dict[str, Any]) -> DeviceType:
def _get_device_type_from_sys_info(info: dict[str, Any]) -> DeviceType:
"""Find SmartDevice subclass for device described by passed data."""
if "system" not in info or "get_sysinfo" not in info["system"]:
raise KasaException("No 'system' or 'get_sysinfo' in response")
sysinfo: Dict[str, Any] = info["system"]["get_sysinfo"]
type_: Optional[str] = sysinfo.get("type", sysinfo.get("mic_type"))
sysinfo: dict[str, Any] = info["system"]["get_sysinfo"]
type_: str | None = sysinfo.get("type", sysinfo.get("mic_type"))
if type_ is None:
raise KasaException("Unable to find the device type field!")
@ -143,7 +145,7 @@ def _get_device_type_from_sys_info(info: Dict[str, Any]) -> DeviceType:
raise UnsupportedDeviceError("Unknown device type: %s" % type_)
def get_device_class_from_sys_info(sysinfo: Dict[str, Any]) -> Type[IotDevice]:
def get_device_class_from_sys_info(sysinfo: dict[str, Any]) -> type[IotDevice]:
"""Find SmartDevice subclass for device described by passed data."""
TYPE_TO_CLASS = {
DeviceType.Bulb: IotBulb,
@ -156,9 +158,9 @@ def get_device_class_from_sys_info(sysinfo: Dict[str, Any]) -> Type[IotDevice]:
return TYPE_TO_CLASS[_get_device_type_from_sys_info(sysinfo)]
def get_device_class_from_family(device_type: str) -> Optional[Type[Device]]:
def get_device_class_from_family(device_type: str) -> type[Device] | None:
"""Return the device class from the type name."""
supported_device_types: Dict[str, Type[Device]] = {
supported_device_types: dict[str, type[Device]] = {
"SMART.TAPOPLUG": SmartDevice,
"SMART.TAPOBULB": SmartBulb,
"SMART.TAPOSWITCH": SmartBulb,
@ -173,14 +175,14 @@ def get_device_class_from_family(device_type: str) -> Optional[Type[Device]]:
def get_protocol(
config: DeviceConfig,
) -> Optional[BaseProtocol]:
) -> BaseProtocol | None:
"""Return the protocol from the connection name."""
protocol_name = config.connection_type.device_family.value.split(".")[0]
protocol_transport_key = (
protocol_name + "." + config.connection_type.encryption_type.value
)
supported_device_protocols: Dict[
str, Tuple[Type[BaseProtocol], Type[BaseTransport]]
supported_device_protocols: dict[
str, tuple[type[BaseProtocol], type[BaseTransport]]
] = {
"IOT.XOR": (IotProtocol, XorTransport),
"IOT.KLAP": (IotProtocol, KlapTransport),

4
kasa/device_type.py
View File

@ -1,5 +1,7 @@
"""TP-Link device types."""
from __future__ import annotations
from enum import Enum
@ -20,7 +22,7 @@ class DeviceType(Enum):
Unknown = "unknown"
@staticmethod
def from_value(name: str) -> "DeviceType":
def from_value(name: str) -> DeviceType:
"""Return device type from string value."""
for device_type in DeviceType:
if device_type.value == name:

8
kasa/deviceconfig.py
View File

@ -1,5 +1,11 @@
"""Module for holding connection parameters."""
"""Module for holding connection parameters.
Note that this module does not work with from __future__ import annotations
due to it's use of type returned by fields() which becomes a string with the import.
https://bugs.python.org/issue39442
"""
# ruff: noqa: FA100
import logging
from dataclasses import asdict, dataclass, field, fields, is_dataclass
from enum import Enum

58
kasa/discover.py
View File

@ -1,11 +1,13 @@
"""Discovery module for TP-Link Smart Home devices."""
from __future__ import annotations
import asyncio
import binascii
import ipaddress
import logging
import socket
from typing import Awaitable, Callable, Dict, List, Optional, Set, Type, cast
from typing import Awaitable, Callable, Dict, Optional, Type, cast
# When support for cpython older than 3.11 is dropped
# async_timeout can be replaced with asyncio.timeout
@ -38,6 +40,7 @@ _LOGGER = logging.getLogger(__name__)
OnDiscoveredCallable = Callable[[Device], Awaitable[None]]
OnUnsupportedCallable = Callable[[UnsupportedDeviceError], Awaitable[None]]
DeviceDict = Dict[str, Device]
@ -54,17 +57,15 @@ class _DiscoverProtocol(asyncio.DatagramProtocol):
def __init__(
self,
*,
on_discovered: Optional[OnDiscoveredCallable] = None,
on_discovered: OnDiscoveredCallable | None = None,
target: str = "255.255.255.255",
discovery_packets: int = 3,
discovery_timeout: int = 5,
interface: Optional[str] = None,
on_unsupported: Optional[
Callable[[UnsupportedDeviceError], Awaitable[None]]
] = None,
port: Optional[int] = None,
credentials: Optional[Credentials] = None,
timeout: Optional[int] = None,
interface: str | None = None,
on_unsupported: OnUnsupportedCallable | None = None,
port: int | None = None,
credentials: Credentials | None = None,
timeout: int | None = None,
) -> None:
self.transport = None
self.discovery_packets = discovery_packets
@ -78,15 +79,15 @@ class _DiscoverProtocol(asyncio.DatagramProtocol):
self.target_2 = (target, Discover.DISCOVERY_PORT_2)
self.discovered_devices = {}
self.unsupported_device_exceptions: Dict = {}
self.invalid_device_exceptions: Dict = {}
self.unsupported_device_exceptions: dict = {}
self.invalid_device_exceptions: dict = {}
self.on_unsupported = on_unsupported
self.credentials = credentials
self.timeout = timeout
self.discovery_timeout = discovery_timeout
self.seen_hosts: Set[str] = set()
self.discover_task: Optional[asyncio.Task] = None
self.callback_tasks: List[asyncio.Task] = []
self.seen_hosts: set[str] = set()
self.discover_task: asyncio.Task | None = None
self.callback_tasks: list[asyncio.Task] = []
self.target_discovered: bool = False
self._started_event = asyncio.Event()
@ -148,7 +149,7 @@ class _DiscoverProtocol(asyncio.DatagramProtocol):
return
self.seen_hosts.add(ip)
device: Optional[Device] = None
device: Device | None = None
config = DeviceConfig(host=ip, port_override=self.port)
if self.credentials:
@ -328,9 +329,9 @@ class Discover:
host: str,
*,
discovery_timeout: int = 5,
port: Optional[int] = None,
timeout: Optional[int] = None,
credentials: Optional[Credentials] = None,
port: int | None = None,
timeout: int | None = None,
credentials: Credentials | None = None,
) -> Device:
"""Discover a single device by the given IP address.
@ -403,7 +404,7 @@ class Discover:
raise TimeoutError(f"Timed out getting discovery response for {host}")
@staticmethod
def _get_device_class(info: dict) -> Type[Device]:
def _get_device_class(info: dict) -> type[Device]:
"""Find SmartDevice subclass for device described by passed data."""
if "result" in info:
discovery_result = DiscoveryResult(**info["result"])
@ -502,16 +503,17 @@ class Discover:
return device
class DiscoveryResult(BaseModel):
"""Base model for discovery result."""
class EncryptionScheme(BaseModel):
"""Base model for encryption scheme of discovery result."""
is_support_https: bool
encrypt_type: str
http_port: int
lv: Optional[int] = None
lv: Optional[int] = None # noqa: UP007
class DiscoveryResult(BaseModel):
"""Base model for discovery result."""
device_type: str
device_model: str
@ -520,11 +522,11 @@ class DiscoveryResult(BaseModel):
mgt_encrypt_schm: EncryptionScheme
device_id: str
hw_ver: Optional[str] = None
owner: Optional[str] = None
is_support_iot_cloud: Optional[bool] = None
obd_src: Optional[str] = None
factory_default: Optional[bool] = None
hw_ver: Optional[str] = None # noqa: UP007
owner: Optional[str] = None # noqa: UP007
is_support_iot_cloud: Optional[bool] = None # noqa: UP007
obd_src: Optional[str] = None # noqa: UP007
factory_default: Optional[bool] = None # noqa: UP007
def get_dict(self) -> dict:
"""Return a dict for this discovery result.

6
kasa/effects.py
View File

@ -1,6 +1,8 @@
"""Module for light strip effects (LB*, KL*, KB*)."""
from typing import List, cast
from __future__ import annotations
from typing import cast
EFFECT_AURORA = {
"custom": 0,
@ -292,5 +294,5 @@ EFFECTS_LIST_V1 = [
EFFECT_VALENTINES,
]
EFFECT_NAMES_V1: List[str] = [cast(str, effect["name"]) for effect in EFFECTS_LIST_V1]
EFFECT_NAMES_V1: list[str] = [cast(str, effect["name"]) for effect in EFFECTS_LIST_V1]
EFFECT_MAPPING_V1 = {effect["name"]: effect for effect in EFFECTS_LIST_V1}

11
kasa/emeterstatus.py
View File

@ -1,7 +1,8 @@
"""Module for emeter container."""
from __future__ import annotations
import logging
from typing import Optional
_LOGGER = logging.getLogger(__name__)
@ -17,7 +18,7 @@ class EmeterStatus(dict):
"""
@property
def voltage(self) -> Optional[float]:
def voltage(self) -> float | None:
"""Return voltage in V."""
try:
return self["voltage"]
@ -25,7 +26,7 @@ class EmeterStatus(dict):
return None
@property
def power(self) -> Optional[float]:
def power(self) -> float | None:
"""Return power in W."""
try:
return self["power"]
@ -33,7 +34,7 @@ class EmeterStatus(dict):
return None
@property
def current(self) -> Optional[float]:
def current(self) -> float | None:
"""Return current in A."""
try:
return self["current"]
@ -41,7 +42,7 @@ class EmeterStatus(dict):
return None
@property
def total(self) -> Optional[float]:
def total(self) -> float | None:
"""Return total in kWh."""
try:
return self["total"]

6
kasa/exceptions.py
View File

@ -1,8 +1,10 @@
"""python-kasa exceptions."""
from __future__ import annotations
from asyncio import TimeoutError as _asyncioTimeoutError
from enum import IntEnum
from typing import Any, Optional
from typing import Any
class KasaException(Exception):
@ -35,7 +37,7 @@ class DeviceError(KasaException):
"""Base exception for device errors."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self.error_code: Optional["SmartErrorCode"] = kwargs.get("error_code", None)
self.error_code: SmartErrorCode | None = kwargs.get("error_code", None)
super().__init__(*args)
def __repr__(self):

14
kasa/feature.py
View File

@ -1,8 +1,10 @@
"""Generic interface for defining device features."""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Callable, Optional, Union
from typing import TYPE_CHECKING, Any, Callable
if TYPE_CHECKING:
from .device import Device
@ -23,17 +25,17 @@ class Feature:
"""Feature defines a generic interface for device features."""
#: Device instance required for getting and setting values
device: "Device"
device: Device
#: User-friendly short description
name: str
#: Name of the property that allows accessing the value
attribute_getter: Union[str, Callable]
attribute_getter: str | Callable
#: Name of the method that allows changing the value
attribute_setter: Optional[str] = None
attribute_setter: str | None = None
#: Container storing the data, this overrides 'device' for getters
container: Any = None
#: Icon suggestion
icon: Optional[str] = None
icon: str | None = None
#: Type of the feature
type: FeatureType = FeatureType.Sensor
@ -44,7 +46,7 @@ class Feature:
maximum_value: int = 2**16 # Arbitrary max
#: Attribute containing the name of the range getter property.
#: If set, this property will be used to set *minimum_value* and *maximum_value*.
range_getter: Optional[str] = None
range_getter: str | None = None
def __post_init__(self):
"""Handle late-binding of members."""

18
kasa/httpclient.py
View File

@ -1,8 +1,10 @@
"""Module for HttpClientSession class."""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Optional, Tuple, Union
from typing import Any, Dict
import aiohttp
from yarl import URL
@ -48,12 +50,12 @@ class HttpClient:
self,
url: URL,
*,
params: Optional[Dict[str, Any]] = None,
data: Optional[bytes] = None,
json: Optional[Union[Dict, Any]] = None,
headers: Optional[Dict[str, str]] = None,
cookies_dict: Optional[Dict[str, str]] = None,
) -> Tuple[int, Optional[Union[Dict, bytes]]]:
params: dict[str, Any] | None = None,
data: bytes | None = None,
json: dict | Any | None = None,
headers: dict[str, str] | None = None,
cookies_dict: dict[str, str] | None = None,
) -> tuple[int, dict | bytes | None]:
"""Send an http post request to the device.
If the request is provided via the json parameter json will be returned.
@ -103,7 +105,7 @@ class HttpClient:
return resp.status, response_data
def get_cookie(self, cookie_name: str) -> Optional[str]:
def get_cookie(self, cookie_name: str) -> str | None:
"""Return the cookie with cookie_name."""
if cookie := self.client.cookie_jar.filter_cookies(self._last_url).get(
cookie_name

40
kasa/iot/iotbulb.py
View File

@ -1,9 +1,11 @@
"""Module for bulbs (LB*, KL*, KB*)."""
from __future__ import annotations
import logging
import re
from enum import Enum
from typing import Dict, List, Optional, cast
from typing import Optional, cast
try:
from pydantic.v1 import BaseModel, Field, root_validator
@ -40,7 +42,7 @@ class TurnOnBehavior(BaseModel):
"""
#: Index of preset to use, or ``None`` for the last known state.
preset: Optional[int] = Field(alias="index", default=None)
preset: Optional[int] = Field(alias="index", default=None) # noqa: UP007
#: Wanted behavior
mode: BehaviorMode
@ -193,8 +195,8 @@ class IotBulb(IotDevice, Bulb):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Bulb
@ -275,7 +277,7 @@ class IotBulb(IotDevice, Bulb):
@property # type: ignore
@requires_update
def light_state(self) -> Dict[str, str]:
def light_state(self) -> dict[str, str]:
"""Query the light state."""
light_state = self.sys_info["light_state"]
if light_state is None:
@ -298,7 +300,7 @@ class IotBulb(IotDevice, Bulb):
"""Return True if the device supports effects."""
return "lighting_effect_state" in self.sys_info
async def get_light_details(self) -> Dict[str, int]:
async def get_light_details(self) -> dict[str, int]:
"""Return light details.
Example::
@ -325,14 +327,14 @@ class IotBulb(IotDevice, Bulb):
self.LIGHT_SERVICE, "set_default_behavior", behavior.dict(by_alias=True)
)
async def get_light_state(self) -> Dict[str, Dict]:
async def get_light_state(self) -> dict[str, dict]:
"""Query the light state."""
# TODO: add warning and refer to use light.state?
return await self._query_helper(self.LIGHT_SERVICE, "get_light_state")
async def set_light_state(
self, state: Dict, *, transition: Optional[int] = None
) -> Dict:
self, state: dict, *, transition: int | None = None
) -> dict:
"""Set the light state."""
if transition is not None:
state["transition_period"] = transition
@ -378,10 +380,10 @@ class IotBulb(IotDevice, Bulb):
self,
hue: int,
saturation: int,
value: Optional[int] = None,
value: int | None = None,
*,
transition: Optional[int] = None,
) -> Dict:
transition: int | None = None,
) -> dict:
"""Set new HSV.
:param int hue: hue in degrees
@ -424,8 +426,8 @@ class IotBulb(IotDevice, Bulb):
@requires_update
async def set_color_temp(
self, temp: int, *, brightness=None, transition: Optional[int] = None
) -> Dict:
self, temp: int, *, brightness=None, transition: int | None = None
) -> dict:
"""Set the color temperature of the device in kelvin.
:param int temp: The new color temperature, in Kelvin
@ -460,8 +462,8 @@ class IotBulb(IotDevice, Bulb):
@requires_update
async def set_brightness(
self, brightness: int, *, transition: Optional[int] = None
) -> Dict:
self, brightness: int, *, transition: int | None = None
) -> dict:
"""Set the brightness in percentage.
:param int brightness: brightness in percent
@ -482,14 +484,14 @@ class IotBulb(IotDevice, Bulb):
light_state = self.light_state
return bool(light_state["on_off"])
async def turn_off(self, *, transition: Optional[int] = None, **kwargs) -> Dict:
async def turn_off(self, *, transition: int | None = None, **kwargs) -> dict:
"""Turn the bulb off.
:param int transition: transition in milliseconds.
"""
return await self.set_light_state({"on_off": 0}, transition=transition)
async def turn_on(self, *, transition: Optional[int] = None, **kwargs) -> Dict:
async def turn_on(self, *, transition: int | None = None, **kwargs) -> dict:
"""Turn the bulb on.
:param int transition: transition in milliseconds.
@ -513,7 +515,7 @@ class IotBulb(IotDevice, Bulb):
@property # type: ignore
@requires_update
def presets(self) -> List[BulbPreset]:
def presets(self) -> list[BulbPreset]:
"""Return a list of available bulb setting presets."""
return [BulbPreset(**vals) for vals in self.sys_info["preferred_state"]]

74
kasa/iot/iotdevice.py
View File

@ -12,12 +12,14 @@ You may obtain a copy of the license at
http://www.apache.org/licenses/LICENSE-2.0
"""
from __future__ import annotations
import collections.abc
import functools
import inspect
import logging
from datetime import datetime, timedelta
from typing import Any, Dict, List, Mapping, Optional, Sequence, Set
from typing import Any, Mapping, Sequence
from ..device import Device, WifiNetwork
from ..deviceconfig import DeviceConfig
@ -66,7 +68,7 @@ def requires_update(f):
@functools.lru_cache
def _parse_features(features: str) -> Set[str]:
def _parse_features(features: str) -> set[str]:
"""Parse features string."""
return set(features.split(":"))
@ -177,19 +179,19 @@ class IotDevice(Device):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
"""Create a new IotDevice instance."""
super().__init__(host=host, config=config, protocol=protocol)
self._sys_info: Any = None # TODO: this is here to avoid changing tests
self._supported_modules: Optional[Dict[str, IotModule]] = None
self._legacy_features: Set[str] = set()
self._children: Mapping[str, "IotDevice"] = {}
self._supported_modules: dict[str, IotModule] | None = None
self._legacy_features: set[str] = set()
self._children: Mapping[str, IotDevice] = {}
@property
def children(self) -> Sequence["IotDevice"]:
def children(self) -> Sequence[IotDevice]:
"""Return list of children."""
return list(self._children.values())
@ -203,9 +205,9 @@ class IotDevice(Device):
self.modules[name] = module
def _create_request(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
):
request: Dict[str, Any] = {target: {cmd: arg}}
request: dict[str, Any] = {target: {cmd: arg}}
if child_ids is not None:
request = {"context": {"child_ids": child_ids}, target: {cmd: arg}}
@ -219,7 +221,7 @@ class IotDevice(Device):
raise KasaException("update() required prior accessing emeter")
async def _query_helper(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
) -> Any:
"""Query device, return results or raise an exception.
@ -256,13 +258,13 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def features(self) -> Dict[str, Feature]:
def features(self) -> dict[str, Feature]:
"""Return a set of features that the device supports."""
return self._features
@property # type: ignore
@requires_update
def supported_modules(self) -> List[str]:
def supported_modules(self) -> list[str]:
"""Return a set of modules supported by the device."""
# TODO: this should rather be called `features`, but we don't want to break
# the API now. Maybe just deprecate it and point the users to use this?
@ -274,7 +276,7 @@ class IotDevice(Device):
"""Return True if device has an energy meter."""
return "ENE" in self._legacy_features
async def get_sys_info(self) -> Dict[str, Any]:
async def get_sys_info(self) -> dict[str, Any]:
"""Retrieve system information."""
return await self._query_helper("system", "get_sysinfo")
@ -363,12 +365,12 @@ class IotDevice(Device):
# responses on top of it so we remember
# which modules are not supported, otherwise
# every other update will query for them
update: Dict = self._last_update.copy() if self._last_update else {}
update: dict = self._last_update.copy() if self._last_update else {}
for response in responses:
update = {**update, **response}
self._last_update = update
def update_from_discover_info(self, info: Dict[str, Any]) -> None:
def update_from_discover_info(self, info: dict[str, Any]) -> None:
"""Update state from info from the discover call."""
self._discovery_info = info
if "system" in info and (sys_info := info["system"].get("get_sysinfo")):
@ -380,7 +382,7 @@ class IotDevice(Device):
# by the requires_update decorator
self._set_sys_info(info)
def _set_sys_info(self, sys_info: Dict[str, Any]) -> None:
def _set_sys_info(self, sys_info: dict[str, Any]) -> None:
"""Set sys_info."""
self._sys_info = sys_info
if features := sys_info.get("feature"):
@ -388,7 +390,7 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def sys_info(self) -> Dict[str, Any]:
def sys_info(self) -> dict[str, Any]:
"""
Return system information.
@ -405,7 +407,7 @@ class IotDevice(Device):
return str(sys_info["model"])
@property # type: ignore
def alias(self) -> Optional[str]:
def alias(self) -> str | None:
"""Return device name (alias)."""
sys_info = self._sys_info
return sys_info.get("alias") if sys_info else None
@ -422,18 +424,18 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def timezone(self) -> Dict:
def timezone(self) -> dict:
"""Return the current timezone."""
return self.modules["time"].timezone
async def get_time(self) -> Optional[datetime]:
async def get_time(self) -> datetime | None:
"""Return current time from the device, if available."""
_LOGGER.warning(
"Use `time` property instead, this call will be removed in the future."
)
return await self.modules["time"].get_time()
async def get_timezone(self) -> Dict:
async def get_timezone(self) -> dict:
"""Return timezone information."""
_LOGGER.warning(
"Use `timezone` property instead, this call will be removed in the future."
@ -442,7 +444,7 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def hw_info(self) -> Dict:
def hw_info(self) -> dict:
"""Return hardware information.
This returns just a selection of sysinfo keys that are related to hardware.
@ -464,7 +466,7 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def location(self) -> Dict:
def location(self) -> dict:
"""Return geographical location."""
sys_info = self._sys_info
loc = {"latitude": None, "longitude": None}
@ -482,7 +484,7 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def rssi(self) -> Optional[int]:
def rssi(self) -> int | None:
"""Return WiFi signal strength (rssi)."""
rssi = self._sys_info.get("rssi")
return None if rssi is None else int(rssi)
@ -528,21 +530,21 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def emeter_today(self) -> Optional[float]:
def emeter_today(self) -> float | None:
"""Return today's energy consumption in kWh."""
self._verify_emeter()
return self.modules["emeter"].emeter_today
@property # type: ignore
@requires_update
def emeter_this_month(self) -> Optional[float]:
def emeter_this_month(self) -> float | None:
"""Return this month's energy consumption in kWh."""
self._verify_emeter()
return self.modules["emeter"].emeter_this_month
async def get_emeter_daily(
self, year: Optional[int] = None, month: Optional[int] = None, kwh: bool = True
) -> Dict:
self, year: int | None = None, month: int | None = None, kwh: bool = True
) -> dict:
"""Retrieve daily statistics for a given month.
:param year: year for which to retrieve statistics (default: this year)
@ -556,8 +558,8 @@ class IotDevice(Device):
@requires_update
async def get_emeter_monthly(
self, year: Optional[int] = None, kwh: bool = True
) -> Dict:
self, year: int | None = None, kwh: bool = True
) -> dict:
"""Retrieve monthly statistics for a given year.
:param year: year for which to retrieve statistics (default: this year)
@ -568,7 +570,7 @@ class IotDevice(Device):
return await self.modules["emeter"].get_monthstat(year=year, kwh=kwh)
@requires_update
async def erase_emeter_stats(self) -> Dict:
async def erase_emeter_stats(self) -> dict:
"""Erase energy meter statistics."""
self._verify_emeter()
return await self.modules["emeter"].erase_stats()
@ -588,11 +590,11 @@ class IotDevice(Device):
"""
await self._query_helper("system", "reboot", {"delay": delay})
async def turn_off(self, **kwargs) -> Dict:
async def turn_off(self, **kwargs) -> dict:
"""Turn off the device."""
raise NotImplementedError("Device subclass needs to implement this.")
async def turn_on(self, **kwargs) -> Optional[Dict]:
async def turn_on(self, **kwargs) -> dict | None:
"""Turn device on."""
raise NotImplementedError("Device subclass needs to implement this.")
@ -604,7 +606,7 @@ class IotDevice(Device):
@property # type: ignore
@requires_update
def on_since(self) -> Optional[datetime]:
def on_since(self) -> datetime | None:
"""Return pretty-printed on-time, or None if not available."""
if "on_time" not in self._sys_info:
return None
@ -626,7 +628,7 @@ class IotDevice(Device):
"""
return self.mac
async def wifi_scan(self) -> List[WifiNetwork]: # noqa: D202
async def wifi_scan(self) -> list[WifiNetwork]: # noqa: D202
"""Scan for available wifi networks."""
async def _scan(target):

20
kasa/iot/iotdimmer.py
View File

@ -1,7 +1,9 @@
"""Module for dimmers (currently only HS220)."""
from __future__ import annotations
from enum import Enum
from typing import Any, Dict, Optional
from typing import Any
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
@ -72,8 +74,8 @@ class IotDimmer(IotPlug):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Dimmer
@ -112,9 +114,7 @@ class IotDimmer(IotPlug):
return int(sys_info["brightness"])
@requires_update
async def set_brightness(
self, brightness: int, *, transition: Optional[int] = None
):
async def set_brightness(self, brightness: int, *, transition: int | None = None):
"""Set the new dimmer brightness level in percentage.
:param int transition: transition duration in milliseconds.
@ -143,7 +143,7 @@ class IotDimmer(IotPlug):
self.DIMMER_SERVICE, "set_brightness", {"brightness": brightness}
)
async def turn_off(self, *, transition: Optional[int] = None, **kwargs):
async def turn_off(self, *, transition: int | None = None, **kwargs):
"""Turn the bulb off.
:param int transition: transition duration in milliseconds.
@ -154,7 +154,7 @@ class IotDimmer(IotPlug):
return await super().turn_off()
@requires_update
async def turn_on(self, *, transition: Optional[int] = None, **kwargs):
async def turn_on(self, *, transition: int | None = None, **kwargs):
"""Turn the bulb on.
:param int transition: transition duration in milliseconds.
@ -202,7 +202,7 @@ class IotDimmer(IotPlug):
@requires_update
async def set_button_action(
self, action_type: ActionType, action: ButtonAction, index: Optional[int] = None
self, action_type: ActionType, action: ButtonAction, index: int | None = None
):
"""Set action to perform on button click/hold.
@ -213,7 +213,7 @@ class IotDimmer(IotPlug):
"""
action_type_setter = f"set_{action_type}"
payload: Dict[str, Any] = {"mode": str(action)}
payload: dict[str, Any] = {"mode": str(action)}
if index is not None:
payload["index"] = index

16
kasa/iot/iotlightstrip.py
View File

@ -1,6 +1,6 @@
"""Module for light strips (KL430)."""
from typing import Dict, List, Optional
from __future__ import annotations
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
@ -49,8 +49,8 @@ class IotLightStrip(IotBulb):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.LightStrip
@ -63,7 +63,7 @@ class IotLightStrip(IotBulb):
@property # type: ignore
@requires_update
def effect(self) -> Dict:
def effect(self) -> dict:
"""Return effect state.
Example:
@ -77,7 +77,7 @@ class IotLightStrip(IotBulb):
@property # type: ignore
@requires_update
def effect_list(self) -> Optional[List[str]]:
def effect_list(self) -> list[str] | None:
"""Return built-in effects list.
Example:
@ -90,8 +90,8 @@ class IotLightStrip(IotBulb):
self,
effect: str,
*,
brightness: Optional[int] = None,
transition: Optional[int] = None,
brightness: int | None = None,
transition: int | None = None,
) -> None:
"""Set an effect on the device.
@ -118,7 +118,7 @@ class IotLightStrip(IotBulb):
@requires_update
async def set_custom_effect(
self,
effect_dict: Dict,
effect_dict: dict,
) -> None:
"""Set a custom effect on the device.

11
kasa/iot/iotplug.py
View File

@ -1,7 +1,8 @@
"""Module for smart plugs (HS100, HS110, ..)."""
from __future__ import annotations
import logging
from typing import Optional
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
@ -47,8 +48,8 @@ class IotPlug(IotDevice):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.Plug
@ -108,8 +109,8 @@ class IotWallSwitch(IotPlug):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self._device_type = DeviceType.WallSwitch

46
kasa/iot/iotstrip.py
View File

@ -1,9 +1,11 @@
"""Module for multi-socket devices (HS300, HS107, KP303, ..)."""
from __future__ import annotations
import logging
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any, DefaultDict, Dict, Optional
from typing import Any
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
@ -23,7 +25,7 @@ _LOGGER = logging.getLogger(__name__)
def merge_sums(dicts):
"""Merge the sum of dicts."""
total_dict: DefaultDict[int, float] = defaultdict(lambda: 0.0)
total_dict: defaultdict[int, float] = defaultdict(lambda: 0.0)
for sum_dict in dicts:
for day, value in sum_dict.items():
total_dict[day] += value
@ -86,8 +88,8 @@ class IotStrip(IotDevice):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[BaseProtocol] = None,
config: DeviceConfig | None = None,
protocol: BaseProtocol | None = None,
) -> None:
super().__init__(host=host, config=config, protocol=protocol)
self.emeter_type = "emeter"
@ -137,7 +139,7 @@ class IotStrip(IotDevice):
@property # type: ignore
@requires_update
def on_since(self) -> Optional[datetime]:
def on_since(self) -> datetime | None:
"""Return the maximum on-time of all outlets."""
if self.is_off:
return None
@ -170,8 +172,8 @@ class IotStrip(IotDevice):
@requires_update
async def get_emeter_daily(
self, year: Optional[int] = None, month: Optional[int] = None, kwh: bool = True
) -> Dict:
self, year: int | None = None, month: int | None = None, kwh: bool = True
) -> dict:
"""Retrieve daily statistics for a given month.
:param year: year for which to retrieve statistics (default: this year)
@ -186,8 +188,8 @@ class IotStrip(IotDevice):
@requires_update
async def get_emeter_monthly(
self, year: Optional[int] = None, kwh: bool = True
) -> Dict:
self, year: int | None = None, kwh: bool = True
) -> dict:
"""Retrieve monthly statistics for a given year.
:param year: year for which to retrieve statistics (default: this year)
@ -197,7 +199,7 @@ class IotStrip(IotDevice):
"get_emeter_monthly", {"year": year, "kwh": kwh}
)
async def _async_get_emeter_sum(self, func: str, kwargs: Dict[str, Any]) -> Dict:
async def _async_get_emeter_sum(self, func: str, kwargs: dict[str, Any]) -> dict:
"""Retreive emeter stats for a time period from children."""
self._verify_emeter()
return merge_sums(
@ -212,13 +214,13 @@ class IotStrip(IotDevice):
@property # type: ignore
@requires_update
def emeter_this_month(self) -> Optional[float]:
def emeter_this_month(self) -> float | None:
"""Return this month's energy consumption in kWh."""
return sum(plug.emeter_this_month for plug in self.children)
@property # type: ignore
@requires_update
def emeter_today(self) -> Optional[float]:
def emeter_today(self) -> float | None:
"""Return this month's energy consumption in kWh."""
return sum(plug.emeter_today for plug in self.children)
@ -243,7 +245,7 @@ class IotStripPlug(IotPlug):
The plug inherits (most of) the system information from the parent.
"""
def __init__(self, host: str, parent: "IotStrip", child_id: str) -> None:
def __init__(self, host: str, parent: IotStrip, child_id: str) -> None:
super().__init__(host)
self.parent = parent
@ -262,16 +264,14 @@ class IotStripPlug(IotPlug):
"""
await self._modular_update({})
def _create_emeter_request(
self, year: Optional[int] = None, month: Optional[int] = None
):
def _create_emeter_request(self, year: int | None = None, month: int | None = None):
"""Create a request for requesting all emeter statistics at once."""
if year is None:
year = datetime.now().year
if month is None:
month = datetime.now().month
req: Dict[str, Any] = {}
req: dict[str, Any] = {}
merge(req, self._create_request("emeter", "get_realtime"))
merge(req, self._create_request("emeter", "get_monthstat", {"year": year}))
@ -285,16 +285,16 @@ class IotStripPlug(IotPlug):
return req
def _create_request(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
):
request: Dict[str, Any] = {
request: dict[str, Any] = {
"context": {"child_ids": [self.child_id]},
target: {cmd: arg},
}
return request
async def _query_helper(
self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None
self, target: str, cmd: str, arg: dict | None = None, child_ids=None
) -> Any:
"""Override query helper to include the child_ids."""
return await self.parent._query_helper(
@ -335,14 +335,14 @@ class IotStripPlug(IotPlug):
@property # type: ignore
@requires_update
def next_action(self) -> Dict:
def next_action(self) -> dict:
"""Return next scheduled(?) action."""
info = self._get_child_info()
return info["next_action"]
@property # type: ignore
@requires_update
def on_since(self) -> Optional[datetime]:
def on_since(self) -> datetime | None:
"""Return on-time, if available."""
if self.is_off:
return None
@ -359,7 +359,7 @@ class IotStripPlug(IotPlug):
sys_info = self.parent.sys_info
return f"Socket for {sys_info['model']}"
def _get_child_info(self) -> Dict:
def _get_child_info(self) -> dict:
"""Return the subdevice information for this device."""
for plug in self.parent.sys_info["children"]:
if plug["id"] == self.child_id:

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

@ -1,7 +1,8 @@
"""Implementation of the emeter module."""
from __future__ import annotations
from datetime import datetime
from typing import Dict, List, Optional, Union
from ...emeterstatus import EmeterStatus
from .usage import Usage
@ -16,7 +17,7 @@ class Emeter(Usage):
return EmeterStatus(self.data["get_realtime"])
@property
def emeter_today(self) -> Optional[float]:
def emeter_today(self) -> float | None:
"""Return today's energy consumption in kWh."""
raw_data = self.daily_data
today = datetime.now().day
@ -24,7 +25,7 @@ class Emeter(Usage):
return data.get(today)
@property
def emeter_this_month(self) -> Optional[float]:
def emeter_this_month(self) -> float | None:
"""Return this month's energy consumption in kWh."""
raw_data = self.monthly_data
current_month = datetime.now().month
@ -42,7 +43,7 @@ class Emeter(Usage):
"""Return real-time statistics."""
return await self.call("get_realtime")
async def get_daystat(self, *, year=None, month=None, kwh=True) -> Dict:
async def get_daystat(self, *, year=None, month=None, kwh=True) -> dict:
"""Return daily stats for the given year & month.
The return value is a dictionary of {day: energy, ...}.
@ -51,7 +52,7 @@ class Emeter(Usage):
data = self._convert_stat_data(data["day_list"], entry_key="day", kwh=kwh)
return data
async def get_monthstat(self, *, year=None, kwh=True) -> Dict:
async def get_monthstat(self, *, year=None, kwh=True) -> dict:
"""Return monthly stats for the given year.
The return value is a dictionary of {month: energy, ...}.
@ -62,11 +63,11 @@ class Emeter(Usage):
def _convert_stat_data(
self,
data: List[Dict[str, Union[int, float]]],
data: list[dict[str, int | float]],
entry_key: str,
kwh: bool = True,
key: Optional[int] = None,
) -> Dict[Union[int, float], Union[int, float]]:
key: int | None = None,
) -> dict[int | float, int | float]:
"""Return emeter information keyed with the day/month.
The incoming data is a list of dictionaries::

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

@ -1,7 +1,8 @@
"""Implementation of the motion detection (PIR) module found in some dimmers."""
from __future__ import annotations
from enum import Enum
from typing import Optional
from ...exceptions import KasaException
from ..iotmodule import IotModule
@ -43,7 +44,7 @@ class Motion(IotModule):
return await self.call("set_enable", {"enable": int(state)})
async def set_range(
self, *, range: Optional[Range] = None, custom_range: Optional[int] = None
self, *, range: Range | None = None, custom_range: int | None = None
):
"""Set the range for the sensor.

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

@ -1,5 +1,7 @@
"""Base implementation for all rule-based modules."""
from __future__ import annotations
import logging
from enum import Enum
from typing import Dict, List, Optional
@ -37,20 +39,20 @@ class Rule(BaseModel):
id: str
name: str
enable: bool
wday: List[int]
wday: List[int] # noqa: UP006
repeat: bool
# start action
sact: Optional[Action]
sact: Optional[Action] # noqa: UP007
stime_opt: TimeOption
smin: int
eact: Optional[Action]
eact: Optional[Action] # noqa: UP007
etime_opt: TimeOption
emin: int
# Only on bulbs
s_light: Optional[Dict]
s_light: Optional[Dict] # noqa: UP006,UP007
_LOGGER = logging.getLogger(__name__)
@ -65,7 +67,7 @@ class RuleModule(IotModule):
return merge(q, self.query_for_command("get_next_action"))
@property
def rules(self) -> List[Rule]:
def rules(self) -> list[Rule]:
"""Return the list of rules for the service."""
try:
return [

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

@ -1,7 +1,8 @@
"""Implementation of the usage interface."""
from __future__ import annotations
from datetime import datetime
from typing import Dict
from ..iotmodule import IotModule, merge
@ -58,7 +59,7 @@ 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=None, month=None) -> dict:
"""Return raw daily stats for the given year & month."""
if year is None:
year = datetime.now().year
@ -67,14 +68,14 @@ 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=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=None, month=None) -> dict:
"""Return daily stats for the given year & month.
The return value is a dictionary of {day: time, ...}.
@ -83,7 +84,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=None) -> dict:
"""Return monthly stats for the given year.
The return value is a dictionary of {month: time, ...}.
@ -96,7 +97,7 @@ class Usage(IotModule):
"""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, entry_key) -> dict:
"""Return usage information keyed with the day/month.
The incoming data is a list of dictionaries::

17
kasa/iotprotocol.py
View File

@ -1,8 +1,9 @@
"""Module for the IOT legacy IOT KASA protocol."""
from __future__ import annotations
import asyncio
import logging
from typing import Dict, Optional, Union
from .deviceconfig import DeviceConfig
from .exceptions import (
@ -34,7 +35,7 @@ class IotProtocol(BaseProtocol):
self._query_lock = asyncio.Lock()
async def query(self, request: Union[str, Dict], retry_count: int = 3) -> Dict:
async def query(self, request: str | dict, retry_count: int = 3) -> dict:
"""Query the device retrying for retry_count on failure."""
if isinstance(request, dict):
request = json_dumps(request)
@ -43,7 +44,7 @@ class IotProtocol(BaseProtocol):
async with self._query_lock:
return await self._query(request, retry_count)
async def _query(self, request: str, retry_count: int = 3) -> Dict:
async def _query(self, request: str, retry_count: int = 3) -> dict:
for retry in range(retry_count + 1):
try:
return await self._execute_query(request, retry)
@ -83,7 +84,7 @@ class IotProtocol(BaseProtocol):
# make mypy happy, this should never be reached..
raise KasaException("Query reached somehow to unreachable")
async def _execute_query(self, request: str, retry_count: int) -> Dict:
async def _execute_query(self, request: str, retry_count: int) -> dict:
return await self._transport.send(request)
async def close(self) -> None:
@ -94,11 +95,11 @@ class IotProtocol(BaseProtocol):
class _deprecated_TPLinkSmartHomeProtocol(IotProtocol):
def __init__(
self,
host: Optional[str] = None,
host: str | None = None,
*,
port: Optional[int] = None,
timeout: Optional[int] = None,
transport: Optional[BaseTransport] = None,
port: int | None = None,
timeout: int | None = None,
transport: BaseTransport | None = None,
) -> None:
"""Create a protocol object."""
if not host and not transport:

18
kasa/klaptransport.py
View File

@ -40,6 +40,8 @@ https://github.com/python-kasa/python-kasa/pull/117
"""
from __future__ import annotations
import asyncio
import base64
import datetime
@ -49,7 +51,7 @@ import secrets
import struct
import time
from pprint import pformat as pf
from typing import Any, Dict, Optional, Tuple, cast
from typing import Any, cast
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
@ -99,7 +101,7 @@ class KlapTransport(BaseTransport):
super().__init__(config=config)
self._http_client = HttpClient(config)
self._local_seed: Optional[bytes] = None
self._local_seed: bytes | None = None
if (
not self._credentials or self._credentials.username is None
) and not self._credentials_hash:
@ -109,16 +111,16 @@ class KlapTransport(BaseTransport):
self._local_auth_owner = self.generate_owner_hash(self._credentials).hex()
else:
self._local_auth_hash = base64.b64decode(self._credentials_hash.encode()) # type: ignore[union-attr]
self._default_credentials_auth_hash: Dict[str, bytes] = {}
self._default_credentials_auth_hash: dict[str, bytes] = {}
self._blank_auth_hash = None
self._handshake_lock = asyncio.Lock()
self._query_lock = asyncio.Lock()
self._handshake_done = False
self._encryption_session: Optional[KlapEncryptionSession] = None
self._session_expire_at: Optional[float] = None
self._encryption_session: KlapEncryptionSession | None = None
self._session_expire_at: float | None = None
self._session_cookie: Optional[Dict[str, Any]] = None
self._session_cookie: dict[str, Any] | None = None
_LOGGER.debug("Created KLAP transport for %s", self._host)
self._app_url = URL(f"http://{self._host}:{self._port}/app")
@ -134,7 +136,7 @@ class KlapTransport(BaseTransport):
"""The hashed credentials used by the transport."""
return base64.b64encode(self._local_auth_hash).decode()
async def perform_handshake1(self) -> Tuple[bytes, bytes, bytes]:
async def perform_handshake1(self) -> tuple[bytes, bytes, bytes]:
"""Perform handshake1."""
local_seed: bytes = secrets.token_bytes(16)
@ -240,7 +242,7 @@ class KlapTransport(BaseTransport):
async def perform_handshake2(
self, local_seed, remote_seed, auth_hash
) -> "KlapEncryptionSession":
) -> KlapEncryptionSession:
"""Perform handshake2."""
# Handshake 2 has the following payload:
# sha256(serverBytes | authenticator)

7
kasa/module.py
View File

@ -1,8 +1,9 @@
"""Base class for all module implementations."""
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from typing import Dict
from .device import Device
from .exceptions import KasaException
@ -18,10 +19,10 @@ class Module(ABC):
executed during the regular update cycle.
"""
def __init__(self, device: "Device", module: str):
def __init__(self, device: Device, module: str):
self._device = device
self._module = module
self._module_features: Dict[str, Feature] = {}
self._module_features: dict[str, Feature] = {}
@abstractmethod
def query(self):

9
kasa/protocol.py
View File

@ -10,13 +10,14 @@ which are licensed under the Apache License, Version 2.0
http://www.apache.org/licenses/LICENSE-2.0
"""
from __future__ import annotations
import base64
import errno
import hashlib
import logging
import struct
from abc import ABC, abstractmethod
from typing import Dict, Tuple, Union
# When support for cpython older than 3.11 is dropped
# async_timeout can be replaced with asyncio.timeout
@ -62,7 +63,7 @@ class BaseTransport(ABC):
"""The hashed credentials used by the transport."""
@abstractmethod
async def send(self, request: str) -> Dict:
async def send(self, request: str) -> dict:
"""Send a message to the device and return a response."""
@abstractmethod
@ -95,7 +96,7 @@ class BaseProtocol(ABC):
return self._transport._config
@abstractmethod
async def query(self, request: Union[str, Dict], retry_count: int = 3) -> Dict:
async def query(self, request: str | dict, retry_count: int = 3) -> dict:
"""Query the device for the protocol. Abstract method to be overriden."""
@abstractmethod
@ -103,7 +104,7 @@ class BaseProtocol(ABC):
"""Close the protocol. Abstract method to be overriden."""
def get_default_credentials(tuple: Tuple[str, str]) -> Credentials:
def get_default_credentials(tuple: tuple[str, str]) -> Credentials:
"""Return decoded default credentials."""
un = base64.b64decode(tuple[0].encode()).decode()
pw = base64.b64decode(tuple[1].encode()).decode()

12
kasa/smart/modules/alarmmodule.py
View File

@ -1,6 +1,8 @@
"""Implementation of alarm module."""
from typing import TYPE_CHECKING, Dict, List, Optional
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature, FeatureType
from ..smartmodule import SmartModule
@ -14,14 +16,14 @@ class AlarmModule(SmartModule):
REQUIRED_COMPONENT = "alarm"
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {
"get_alarm_configure": None,
"get_support_alarm_type_list": None, # This should be needed only once
}
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
@ -59,7 +61,7 @@ class AlarmModule(SmartModule):
return self.data["get_alarm_configure"]["type"]
@property
def alarm_sounds(self) -> List[str]:
def alarm_sounds(self) -> list[str]:
"""Return list of available alarm sounds."""
return self.data["get_support_alarm_type_list"]["alarm_type_list"]
@ -74,7 +76,7 @@ class AlarmModule(SmartModule):
return self._device.sys_info["in_alarm"]
@property
def source(self) -> Optional[str]:
def source(self) -> str | None:
"""Return the alarm cause."""
src = self._device.sys_info["in_alarm_source"]
return src if src else None

10
kasa/smart/modules/autooffmodule.py
View File

@ -1,7 +1,9 @@
"""Implementation of auto off module."""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Dict, Optional
from typing import TYPE_CHECKING
from ...feature import Feature
from ..smartmodule import SmartModule
@ -16,7 +18,7 @@ class AutoOffModule(SmartModule):
REQUIRED_COMPONENT = "auto_off"
QUERY_GETTER_NAME = "get_auto_off_config"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
@ -42,7 +44,7 @@ class AutoOffModule(SmartModule):
)
)
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {self.QUERY_GETTER_NAME: {"start_index": 0}}
@ -75,7 +77,7 @@ class AutoOffModule(SmartModule):
return self._device.sys_info["auto_off_status"] == "on"
@property
def auto_off_at(self) -> Optional[datetime]:
def auto_off_at(self) -> datetime | None:
"""Return when the device will be turned off automatically."""
if not self.is_timer_active:
return None

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

@ -1,5 +1,7 @@
"""Implementation of battery module."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature, FeatureType
@ -15,7 +17,7 @@ class BatterySensor(SmartModule):
REQUIRED_COMPONENT = "battery_detect"
QUERY_GETTER_NAME = "get_battery_detect_info"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(

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

@ -1,6 +1,8 @@
"""Implementation of brightness module."""
from typing import TYPE_CHECKING, Dict
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature, FeatureType
from ..smartmodule import SmartModule
@ -14,7 +16,7 @@ class Brightness(SmartModule):
REQUIRED_COMPONENT = "brightness"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
@ -29,7 +31,7 @@ class Brightness(SmartModule):
)
)
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
# Brightness is contained in the main device info response.
return {}

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

@ -1,5 +1,7 @@
"""Implementation of cloud module."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature, FeatureType
@ -15,7 +17,7 @@ class CloudModule(SmartModule):
QUERY_GETTER_NAME = "get_connect_cloud_state"
REQUIRED_COMPONENT = "cloud_connect"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(

8
kasa/smart/modules/colortemp.py
View File

@ -1,6 +1,8 @@
"""Implementation of color temp module."""
from typing import TYPE_CHECKING, Dict
from __future__ import annotations
from typing import TYPE_CHECKING
from ...bulb import ColorTempRange
from ...feature import Feature
@ -15,7 +17,7 @@ class ColorTemperatureModule(SmartModule):
REQUIRED_COMPONENT = "color_temperature"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
@ -28,7 +30,7 @@ class ColorTemperatureModule(SmartModule):
)
)
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
# Color temp is contained in the main device info response.
return {}

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

@ -1,6 +1,6 @@
"""Implementation of device module."""
from typing import Dict
from __future__ import annotations
from ..smartmodule import SmartModule
@ -10,7 +10,7 @@ class DeviceModule(SmartModule):
REQUIRED_COMPONENT = "device"
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
query = {
"get_device_info": None,

14
kasa/smart/modules/energymodule.py
View File

@ -1,6 +1,8 @@
"""Implementation of energy monitoring module."""
from typing import TYPE_CHECKING, Dict, Optional
from __future__ import annotations
from typing import TYPE_CHECKING
from ...emeterstatus import EmeterStatus
from ...feature import Feature
@ -15,7 +17,7 @@ class EnergyModule(SmartModule):
REQUIRED_COMPONENT = "energy_monitoring"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
@ -42,7 +44,7 @@ class EnergyModule(SmartModule):
)
) # Wh or kWH?
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
req = {
"get_energy_usage": None,
@ -77,15 +79,15 @@ class EnergyModule(SmartModule):
)
@property
def emeter_this_month(self) -> Optional[float]:
def emeter_this_month(self) -> float | None:
"""Get the emeter value for this month."""
return self._convert_energy_data(self.energy.get("month_energy"), 1 / 1000)
@property
def emeter_today(self) -> Optional[float]:
def emeter_today(self) -> float | None:
"""Get the emeter value for today."""
return self._convert_energy_data(self.energy.get("today_energy"), 1 / 1000)
def _convert_energy_data(self, data, scale) -> Optional[float]:
def _convert_energy_data(self, data, scale) -> float | None:
"""Return adjusted emeter information."""
return data if not data else data * scale

11
kasa/smart/modules/fanmodule.py
View File

@ -1,5 +1,8 @@
"""Implementation of fan_control module."""
from typing import TYPE_CHECKING, Dict
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature, FeatureType
from ..smartmodule import SmartModule
@ -13,7 +16,7 @@ class FanModule(SmartModule):
REQUIRED_COMPONENT = "fan_control"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
@ -37,11 +40,11 @@ class FanModule(SmartModule):
attribute_getter="sleep_mode",
attribute_setter="set_sleep_mode",
icon="mdi:sleep",
type=FeatureType.Switch
type=FeatureType.Switch,
)
)
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {}

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

@ -1,6 +1,8 @@
"""Implementation of firmware module."""
from typing import TYPE_CHECKING, Dict, Optional
from __future__ import annotations
from typing import TYPE_CHECKING, Optional
from ...exceptions import SmartErrorCode
from ...feature import Feature, FeatureType
@ -20,11 +22,11 @@ class UpdateInfo(BaseModel):
"""Update info status object."""
status: int = Field(alias="type")
fw_ver: Optional[str] = None
release_date: Optional[date] = None
release_notes: Optional[str] = Field(alias="release_note", default=None)
fw_size: Optional[int] = None
oem_id: Optional[str] = None
fw_ver: Optional[str] = None # noqa: UP007
release_date: Optional[date] = None # noqa: UP007
release_notes: Optional[str] = Field(alias="release_note", default=None) # noqa: UP007
fw_size: Optional[int] = None # noqa: UP007
oem_id: Optional[str] = None # noqa: UP007
needs_upgrade: bool = Field(alias="need_to_upgrade")
@validator("release_date", pre=True)
@ -47,7 +49,7 @@ class Firmware(SmartModule):
REQUIRED_COMPONENT = "firmware"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
if self.supported_version > 1:
self._add_feature(
@ -70,7 +72,7 @@ class Firmware(SmartModule):
)
)
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
req = {
"get_latest_fw": None,

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

@ -1,5 +1,7 @@
"""Implementation of humidity module."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature, FeatureType
@ -15,7 +17,7 @@ class HumiditySensor(SmartModule):
REQUIRED_COMPONENT = "humidity"
QUERY_GETTER_NAME = "get_comfort_humidity_config"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(

8
kasa/smart/modules/ledmodule.py
View File

@ -1,6 +1,8 @@
"""Module for led controls."""
from typing import TYPE_CHECKING, Dict
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature, FeatureType
from ..smartmodule import SmartModule
@ -15,7 +17,7 @@ class LedModule(SmartModule):
REQUIRED_COMPONENT = "led"
QUERY_GETTER_NAME = "get_led_info"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(
@ -29,7 +31,7 @@ class LedModule(SmartModule):
)
)
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle."""
return {self.QUERY_GETTER_NAME: {"led_rule": None}}

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

@ -1,5 +1,7 @@
"""Module for smooth light transitions."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...exceptions import KasaException
@ -17,7 +19,7 @@ class LightTransitionModule(SmartModule):
QUERY_GETTER_NAME = "get_on_off_gradually_info"
MAXIMUM_DURATION = 60
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._create_features()

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

@ -1,5 +1,7 @@
"""Implementation of report module."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...feature import Feature
@ -15,7 +17,7 @@ class ReportModule(SmartModule):
REQUIRED_COMPONENT = "report_mode"
QUERY_GETTER_NAME = "get_report_mode"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(

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

@ -1,5 +1,7 @@
"""Implementation of temperature module."""
from __future__ import annotations
from typing import TYPE_CHECKING, Literal
from ...feature import Feature, FeatureType
@ -15,7 +17,7 @@ class TemperatureSensor(SmartModule):
REQUIRED_COMPONENT = "temperature"
QUERY_GETTER_NAME = "get_comfort_temp_config"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(
Feature(

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

@ -1,5 +1,7 @@
"""Implementation of time module."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from time import mktime
from typing import TYPE_CHECKING, cast
@ -17,7 +19,7 @@ class TimeModule(SmartModule):
REQUIRED_COMPONENT = "time"
QUERY_GETTER_NAME = "get_device_time"
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
super().__init__(device, module)
self._add_feature(

26
kasa/smart/smartbulb.py
View File

@ -1,6 +1,6 @@
"""Module for tapo-branded smart bulbs (L5**)."""
from typing import Dict, List, Optional
from __future__ import annotations
from ..bulb import Bulb
from ..exceptions import KasaException
@ -55,7 +55,7 @@ class SmartBulb(SmartDevice, Bulb):
return "dynamic_light_effect_enable" in self._info
@property
def effect(self) -> Dict:
def effect(self) -> dict:
"""Return effect state.
This follows the format used by SmartLightStrip.
@ -79,7 +79,7 @@ class SmartBulb(SmartDevice, Bulb):
return data
@property
def effect_list(self) -> Optional[List[str]]:
def effect_list(self) -> list[str] | None:
"""Return built-in effects list.
Example:
@ -124,10 +124,10 @@ class SmartBulb(SmartDevice, Bulb):
self,
hue: int,
saturation: int,
value: Optional[int] = None,
value: int | None = None,
*,
transition: Optional[int] = None,
) -> Dict:
transition: int | None = None,
) -> dict:
"""Set new HSV.
Note, transition is not supported and will be ignored.
@ -163,8 +163,8 @@ class SmartBulb(SmartDevice, Bulb):
return await self.protocol.query({"set_device_info": {**request_payload}})
async def set_color_temp(
self, temp: int, *, brightness=None, transition: Optional[int] = None
) -> Dict:
self, temp: int, *, brightness=None, transition: int | None = None
) -> dict:
"""Set the color temperature of the device in kelvin.
Note, transition is not supported and will be ignored.
@ -193,8 +193,8 @@ class SmartBulb(SmartDevice, Bulb):
raise ValueError(f"Invalid brightness value: {value} (valid range: 1-100%)")
async def set_brightness(
self, brightness: int, *, transition: Optional[int] = None
) -> Dict:
self, brightness: int, *, transition: int | None = None
) -> dict:
"""Set the brightness in percentage.
Note, transition is not supported and will be ignored.
@ -215,13 +215,13 @@ class SmartBulb(SmartDevice, Bulb):
self,
effect: str,
*,
brightness: Optional[int] = None,
transition: Optional[int] = None,
brightness: int | None = None,
transition: int | None = None,
) -> None:
"""Set an effect on the device."""
raise NotImplementedError()
@property
def presets(self) -> List[BulbPreset]:
def presets(self) -> list[BulbPreset]:
"""Return a list of available bulb setting presets."""
return []

9
kasa/smart/smartchilddevice.py
View File

@ -1,7 +1,8 @@
"""Child device implementation."""
from __future__ import annotations
import logging
from typing import Optional
from ..device_type import DeviceType
from ..deviceconfig import DeviceConfig
@ -22,8 +23,8 @@ class SmartChildDevice(SmartDevice):
parent: SmartDevice,
info,
component_info,
config: Optional[DeviceConfig] = None,
protocol: Optional[SmartProtocol] = None,
config: DeviceConfig | None = None,
protocol: SmartProtocol | None = None,
) -> None:
super().__init__(parent.host, config=parent.config, protocol=parent.protocol)
self._parent = parent
@ -38,7 +39,7 @@ class SmartChildDevice(SmartDevice):
@classmethod
async def create(cls, parent: SmartDevice, child_info, child_components):
"""Create a child device based on device info and component listing."""
child: "SmartChildDevice" = cls(parent, child_info, child_components)
child: SmartChildDevice = cls(parent, child_info, child_components)
await child._initialize_modules()
await child._initialize_features()
return child

48
kasa/smart/smartdevice.py
View File

@ -1,9 +1,11 @@
"""Module for a SMART device."""
from __future__ import annotations
import base64
import logging
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, cast
from typing import TYPE_CHECKING, Any, Mapping, Sequence, cast
from ..aestransport import AesTransport
from ..device import Device, WifiNetwork
@ -28,20 +30,20 @@ class SmartDevice(Device):
self,
host: str,
*,
config: Optional[DeviceConfig] = None,
protocol: Optional[SmartProtocol] = None,
config: DeviceConfig | None = None,
protocol: SmartProtocol | None = None,
) -> None:
_protocol = protocol or SmartProtocol(
transport=AesTransport(config=config or DeviceConfig(host=host)),
)
super().__init__(host=host, config=config, protocol=_protocol)
self.protocol: SmartProtocol
self._components_raw: Optional[Dict[str, Any]] = None
self._components: Dict[str, int] = {}
self._state_information: Dict[str, Any] = {}
self.modules: Dict[str, "SmartModule"] = {}
self._parent: Optional["SmartDevice"] = None
self._children: Mapping[str, "SmartDevice"] = {}
self._components_raw: dict[str, Any] | None = None
self._components: dict[str, int] = {}
self._state_information: dict[str, Any] = {}
self.modules: dict[str, SmartModule] = {}
self._parent: SmartDevice | None = None
self._children: Mapping[str, SmartDevice] = {}
self._last_update = {}
async def _initialize_children(self):
@ -74,7 +76,7 @@ class SmartDevice(Device):
}
@property
def children(self) -> Sequence["SmartDevice"]:
def children(self) -> Sequence[SmartDevice]:
"""Return list of children."""
return list(self._children.values())
@ -130,7 +132,7 @@ class SmartDevice(Device):
await self._negotiate()
await self._initialize_modules()
req: Dict[str, Any] = {}
req: dict[str, Any] = {}
# TODO: this could be optimized by constructing the query only once
for module in self.modules.values():
@ -236,7 +238,7 @@ class SmartDevice(Device):
self._add_feature(feat)
@property
def sys_info(self) -> Dict[str, Any]:
def sys_info(self) -> dict[str, Any]:
"""Returns the device info."""
return self._info # type: ignore
@ -246,7 +248,7 @@ class SmartDevice(Device):
return str(self._info.get("model"))
@property
def alias(self) -> Optional[str]:
def alias(self) -> str | None:
"""Returns the device alias or nickname."""
if self._info and (nickname := self._info.get("nickname")):
return base64.b64decode(nickname).decode()
@ -265,13 +267,13 @@ class SmartDevice(Device):
return _timemod.time
@property
def timezone(self) -> Dict:
def timezone(self) -> dict:
"""Return the timezone and time_difference."""
ti = self.time
return {"timezone": ti.tzname()}
@property
def hw_info(self) -> Dict:
def hw_info(self) -> dict:
"""Return hardware info for the device."""
return {
"sw_ver": self._info.get("fw_ver"),
@ -284,7 +286,7 @@ class SmartDevice(Device):
}
@property
def location(self) -> Dict:
def location(self) -> dict:
"""Return the device location."""
loc = {
"latitude": cast(float, self._info.get("latitude", 0)) / 10_000,
@ -293,7 +295,7 @@ class SmartDevice(Device):
return loc
@property
def rssi(self) -> Optional[int]:
def rssi(self) -> int | None:
"""Return the rssi."""
rssi = self._info.get("rssi")
return int(rssi) if rssi else None
@ -321,7 +323,7 @@ class SmartDevice(Device):
self._info = info
async def _query_helper(
self, method: str, params: Optional[Dict] = None, child_ids=None
self, method: str, params: dict | None = None, child_ids=None
) -> Any:
res = await self.protocol.query({method: params})
@ -378,19 +380,19 @@ class SmartDevice(Device):
return energy.emeter_realtime
@property
def emeter_this_month(self) -> Optional[float]:
def emeter_this_month(self) -> float | None:
"""Get the emeter value for this month."""
energy = cast(EnergyModule, self.modules["EnergyModule"]) # noqa: F405
return energy.emeter_this_month
@property
def emeter_today(self) -> Optional[float]:
def emeter_today(self) -> float | None:
"""Get the emeter value for today."""
energy = cast(EnergyModule, self.modules["EnergyModule"]) # noqa: F405
return energy.emeter_today
@property
def on_since(self) -> Optional[datetime]:
def on_since(self) -> datetime | None:
"""Return the time that the device was turned on or None if turned off."""
if (
not self._info.get("device_on")
@ -404,7 +406,7 @@ class SmartDevice(Device):
else: # We have no device time, use current local time.
return datetime.now().replace(microsecond=0) - timedelta(seconds=on_time)
async def wifi_scan(self) -> List[WifiNetwork]:
async def wifi_scan(self) -> list[WifiNetwork]:
"""Scan for available wifi networks."""
def _net_for_scan_info(res):
@ -527,7 +529,7 @@ class SmartDevice(Device):
@staticmethod
def _get_device_type_from_components(
components: List[str], device_type: str
components: list[str], device_type: str
) -> DeviceType:
"""Find type to be displayed as a supported device category."""
if "HUB" in device_type:

10
kasa/smart/smartmodule.py
View File

@ -1,7 +1,9 @@
"""Base implementation for SMART modules."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Dict, Type
from typing import TYPE_CHECKING
from ..exceptions import KasaException
from ..module import Module
@ -18,9 +20,9 @@ class SmartModule(Module):
NAME: str
REQUIRED_COMPONENT: str
QUERY_GETTER_NAME: str
REGISTERED_MODULES: Dict[str, Type["SmartModule"]] = {}
REGISTERED_MODULES: dict[str, type[SmartModule]] = {}
def __init__(self, device: "SmartDevice", module: str):
def __init__(self, device: SmartDevice, module: str):
self._device: SmartDevice
super().__init__(device, module)
@ -36,7 +38,7 @@ class SmartModule(Module):
"""Name of the module."""
return getattr(self, "NAME", self.__class__.__name__)
def query(self) -> Dict:
def query(self) -> dict:
"""Query to execute during the update cycle.
Default implementation uses the raw query getter w/o parameters.

16
kasa/smartprotocol.py
View File

@ -4,13 +4,15 @@ Based on the work of https://github.com/petretiandrea/plugp100
under compatible GNU GPL3 license.
"""
from __future__ import annotations
import asyncio
import base64
import logging
import time
import uuid
from pprint import pformat as pf
from typing import Any, Dict, Union
from typing import Any
from .exceptions import (
SMART_AUTHENTICATION_ERRORS,
@ -57,12 +59,12 @@ class SmartProtocol(BaseProtocol):
}
return json_dumps(request)
async def query(self, request: Union[str, Dict], retry_count: int = 3) -> Dict:
async def query(self, request: str | dict, retry_count: int = 3) -> dict:
"""Query the device retrying for retry_count on failure."""
async with self._query_lock:
return await self._query(request, retry_count)
async def _query(self, request: Union[str, Dict], retry_count: int = 3) -> Dict:
async def _query(self, request: str | dict, retry_count: int = 3) -> dict:
for retry in range(retry_count + 1):
try:
return await self._execute_query(request, retry)
@ -103,9 +105,9 @@ class SmartProtocol(BaseProtocol):
# make mypy happy, this should never be reached..
raise KasaException("Query reached somehow to unreachable")
async def _execute_multiple_query(self, request: Dict, retry_count: int) -> Dict:
async def _execute_multiple_query(self, request: dict, retry_count: int) -> dict:
debug_enabled = _LOGGER.isEnabledFor(logging.DEBUG)
multi_result: Dict[str, Any] = {}
multi_result: dict[str, Any] = {}
smart_method = "multipleRequest"
requests = [
{"method": method, "params": params} for method, params in request.items()
@ -146,7 +148,7 @@ class SmartProtocol(BaseProtocol):
multi_result[method] = result
return multi_result
async def _execute_query(self, request: Union[str, Dict], retry_count: int) -> Dict:
async def _execute_query(self, request: str | dict, retry_count: int) -> dict:
debug_enabled = _LOGGER.isEnabledFor(logging.DEBUG)
if isinstance(request, dict):
@ -322,7 +324,7 @@ class _ChildProtocolWrapper(SmartProtocol):
return smart_method, smart_params
async def query(self, request: Union[str, Dict], retry_count: int = 3) -> Dict:
async def query(self, request: str | dict, retry_count: int = 3) -> dict:
"""Wrap request inside control_child envelope."""
method, params = self._get_method_and_params_for_request(request)
request_data = {

5
kasa/tests/conftest.py
View File

@ -1,5 +1,6 @@
from __future__ import annotations
import warnings
from typing import Dict
from unittest.mock import MagicMock, patch
import pytest
@ -37,7 +38,7 @@ def dummy_protocol():
def credentials_hash(self) -> str:
return "dummy hash"
async def send(self, request: str) -> Dict:
async def send(self, request: str) -> dict:
return {}
async def close(self) -> None:

9
kasa/tests/device_fixtures.py
View File

@ -1,5 +1,6 @@
from __future__ import annotations
from itertools import chain
from typing import Dict, List, Set
import pytest
@ -128,10 +129,10 @@ ALL_DEVICES_SMART = (
)
ALL_DEVICES = ALL_DEVICES_IOT.union(ALL_DEVICES_SMART)
IP_MODEL_CACHE: Dict[str, str] = {}
IP_MODEL_CACHE: dict[str, str] = {}
def parametrize_combine(parametrized: List[pytest.MarkDecorator]):
def parametrize_combine(parametrized: list[pytest.MarkDecorator]):
"""Combine multiple pytest parametrize dev marks into one set of fixtures."""
fixtures = set()
for param in parametrized:
@ -291,7 +292,7 @@ def check_categories():
+ hubs_smart.args[1]
+ sensors_smart.args[1]
)
diffs: Set[FixtureInfo] = set(FIXTURE_DATA) - set(categorized_fixtures)
diffs: set[FixtureInfo] = set(FIXTURE_DATA) - set(categorized_fixtures)
if diffs:
print(diffs)
for diff in diffs:

7
kasa/tests/discovery_fixtures.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from json import dumps as json_dumps
from typing import Optional
import pytest
@ -76,8 +77,8 @@ def discovery_mock(request, mocker):
query_data: dict
device_type: str
encrypt_type: str
login_version: Optional[int] = None
port_override: Optional[int] = None
login_version: int | None = None
port_override: int | None = None
if "discovery_result" in fixture_data:
discovery_data = {"result": fixture_data["discovery_result"]}

20
kasa/tests/fixtureinfo.py
View File

@ -1,8 +1,10 @@
from __future__ import annotations
import glob
import json
import os
from pathlib import Path
from typing import Dict, List, NamedTuple, Optional, Set
from typing import NamedTuple
from kasa.device_factory import _get_device_type_from_sys_info
from kasa.device_type import DeviceType
@ -12,7 +14,7 @@ from kasa.smart.smartdevice import SmartDevice
class FixtureInfo(NamedTuple):
name: str
protocol: str
data: Dict
data: dict
FixtureInfo.__hash__ = lambda self: hash((self.name, self.protocol)) # type: ignore[attr-defined, method-assign]
@ -55,7 +57,7 @@ def idgenerator(paramtuple: FixtureInfo):
return None
def get_fixture_info() -> List[FixtureInfo]:
def get_fixture_info() -> list[FixtureInfo]:
"""Return raw discovery file contents as JSON. Used for discovery tests."""
fixture_data = []
for file, protocol in SUPPORTED_DEVICES:
@ -77,17 +79,17 @@ def get_fixture_info() -> List[FixtureInfo]:
return fixture_data
FIXTURE_DATA: List[FixtureInfo] = get_fixture_info()
FIXTURE_DATA: list[FixtureInfo] = get_fixture_info()
def filter_fixtures(
desc,
*,
data_root_filter: Optional[str] = None,
protocol_filter: Optional[Set[str]] = None,
model_filter: Optional[Set[str]] = None,
component_filter: Optional[str] = None,
device_type_filter: Optional[List[DeviceType]] = None,
data_root_filter: str | None = None,
protocol_filter: set[str] | None = None,
model_filter: set[str] | None = None,
component_filter: str | None = None,
device_type_filter: list[DeviceType] | None = None,
):
"""Filter the fixtures based on supplied parameters.

6
kasa/tests/smart/modules/test_fan.py
View File

@ -14,7 +14,11 @@ async def test_fan_speed(dev: SmartDevice, mocker: MockerFixture):
"""Test fan speed feature."""
fan: FanModule = dev.modules["FanModule"]
level_feature = fan._module_features["fan_speed_level"]
assert level_feature.minimum_value <= level_feature.value <= level_feature.maximum_value
assert (
level_feature.minimum_value
<= level_feature.value
<= level_feature.maximum_value
)
call = mocker.spy(fan, "call")
await fan.set_fan_speed_level(3)

14
kasa/tests/test_aestransport.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
import base64
import json
import logging
@ -7,7 +9,7 @@ import time
from contextlib import nullcontext as does_not_raise
from json import dumps as json_dumps
from json import loads as json_loads
from typing import Any, Dict
from typing import Any
import aiohttp
import pytest
@ -335,7 +337,7 @@ class MockAesDevice:
json = json_loads(item.decode())
return await self._post(url, json)
async def _post(self, url: URL, json: Dict[str, Any]):
async def _post(self, url: URL, json: dict[str, Any]):
if json["method"] == "handshake":
return await self._return_handshake_response(url, json)
elif json["method"] == "securePassthrough":
@ -346,7 +348,7 @@ class MockAesDevice:
assert str(url) == f"http://{self.host}:80/app?token={self.token}"
return await self._return_send_response(url, json)
async def _return_handshake_response(self, url: URL, json: Dict[str, Any]):
async def _return_handshake_response(self, url: URL, json: dict[str, Any]):
start = len("-----BEGIN PUBLIC KEY-----\n")
end = len("\n-----END PUBLIC KEY-----\n")
client_pub_key = json["params"]["key"][start:-end]
@ -359,7 +361,7 @@ class MockAesDevice:
self.status_code, {"result": {"key": key_64}, "error_code": self.error_code}
)
async def _return_secure_passthrough_response(self, url: URL, json: Dict[str, Any]):
async def _return_secure_passthrough_response(self, url: URL, json: dict[str, Any]):
encrypted_request = json["params"]["request"]
decrypted_request = self.encryption_session.decrypt(encrypted_request.encode())
decrypted_request_dict = json_loads(decrypted_request)
@ -378,7 +380,7 @@ class MockAesDevice:
}
return self._mock_response(self.status_code, result)
async def _return_login_response(self, url: URL, json: Dict[str, Any]):
async def _return_login_response(self, url: URL, json: dict[str, Any]):
if "token=" in str(url):
raise Exception("token should not be in url for a login request")
self.token = "".join(random.choices(string.ascii_uppercase, k=32)) # noqa: S311
@ -386,7 +388,7 @@ class MockAesDevice:
self.inner_call_count += 1
return self._mock_response(self.status_code, result)
async def _return_send_response(self, url: URL, json: Dict[str, Any]):
async def _return_send_response(self, url: URL, json: dict[str, Any]):
result = {"result": {"method": None}, "error_code": self.inner_error_code}
response = self.send_response if self.send_response else result
self.inner_call_count += 1

6
kasa/tests/test_smartdevice.py
View File

@ -1,7 +1,9 @@
"""Tests for SMART devices."""
from __future__ import annotations
import logging
from typing import Any, Dict
from typing import Any
import pytest
from pytest_mock import MockerFixture
@ -99,7 +101,7 @@ async def test_update_module_queries(dev: SmartDevice, mocker: MockerFixture):
assert dev.modules
await dev.update()
full_query: Dict[str, Any] = {}
full_query: dict[str, Any] = {}
for mod in dev.modules.values():
full_query = {**full_query, **mod.query()}

14
kasa/xortransport.py
View File

@ -10,6 +10,8 @@ which are licensed under the Apache License, Version 2.0
http://www.apache.org/licenses/LICENSE-2.0
"""
from __future__ import annotations
import asyncio
import contextlib
import errno
@ -17,7 +19,7 @@ import logging
import socket
import struct
from pprint import pformat as pf
from typing import Dict, Generator, Optional
from typing import Generator
# When support for cpython older than 3.11 is dropped
# async_timeout can be replaced with asyncio.timeout
@ -41,10 +43,10 @@ class XorTransport(BaseTransport):
def __init__(self, *, config: DeviceConfig) -> None:
super().__init__(config=config)
self.reader: Optional[asyncio.StreamReader] = None
self.writer: Optional[asyncio.StreamWriter] = None
self.reader: asyncio.StreamReader | None = None
self.writer: asyncio.StreamWriter | None = None
self.query_lock = asyncio.Lock()
self.loop: Optional[asyncio.AbstractEventLoop] = None
self.loop: asyncio.AbstractEventLoop | None = None
@property
def default_port(self):
@ -72,7 +74,7 @@ class XorTransport(BaseTransport):
# the buffer on the device
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
async def _execute_send(self, request: str) -> Dict:
async def _execute_send(self, request: str) -> dict:
"""Execute a query on the device and wait for the response."""
assert self.writer is not None # noqa: S101
assert self.reader is not None # noqa: S101
@ -115,7 +117,7 @@ class XorTransport(BaseTransport):
"""
await self.close()
async def send(self, request: str) -> Dict:
async def send(self, request: str) -> dict:
"""Send a message to the device and return a response."""
#
# Most of the time we will already be connected if the device is online

1
pyproject.toml
View File

@ -110,6 +110,7 @@ select = [
"UP", # pyupgrade
"B", # flake8-bugbear
"SIM", # flake8-simplify
"FA", # flake8-future-annotations
"I", # isort
"S", # bandit
]