diff --git a/devtools/dump_devinfo.py b/devtools/dump_devinfo.py index 87c703e3..238522e6 100644 --- a/devtools/dump_devinfo.py +++ b/devtools/dump_devinfo.py @@ -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) diff --git a/devtools/helpers/smartrequests.py b/devtools/helpers/smartrequests.py index 1ece6c87..881488b5 100644 --- a/devtools/helpers/smartrequests.py +++ b/devtools/helpers/smartrequests.py @@ -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): diff --git a/kasa/aestransport.py b/kasa/aestransport.py index 3b8bfe5d..85624abc 100644 --- a/kasa/aestransport.py +++ b/kasa/aestransport.py @@ -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 diff --git a/kasa/bulb.py b/kasa/bulb.py index 5050e593..50c5d243 100644 --- a/kasa/bulb.py +++ b/kasa/bulb.py @@ -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.""" diff --git a/kasa/cli.py b/kasa/cli.py index d30c4630..41a7759e 100755 --- a/kasa/cli.py +++ b/kasa/cli.py @@ -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: diff --git a/kasa/device.py b/kasa/device.py index 3c5537b1..a4c2b5e3 100644 --- a/kasa/device.py +++ b/kasa/device.py @@ -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 diff --git a/kasa/device_factory.py b/kasa/device_factory.py index a40bc085..3c0ae716 100755 --- a/kasa/device_factory.py +++ b/kasa/device_factory.py @@ -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), diff --git a/kasa/device_type.py b/kasa/device_type.py index 34f0bd89..6a97867c 100755 --- a/kasa/device_type.py +++ b/kasa/device_type.py @@ -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: diff --git a/kasa/deviceconfig.py b/kasa/deviceconfig.py index 827fd03a..6ddff6ad 100644 --- a/kasa/deviceconfig.py +++ b/kasa/deviceconfig.py @@ -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 diff --git a/kasa/discover.py b/kasa/discover.py index a5d88b99..d727b2f8 100755 --- a/kasa/discover.py +++ b/kasa/discover.py @@ -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,17 +503,18 @@ class Discover: return device +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 # noqa: UP007 + + 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 - device_type: str device_model: str ip: 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. diff --git a/kasa/effects.py b/kasa/effects.py index cf72bb8d..8b3e7b32 100644 --- a/kasa/effects.py +++ b/kasa/effects.py @@ -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} diff --git a/kasa/emeterstatus.py b/kasa/emeterstatus.py index 54042499..41a43bc7 100644 --- a/kasa/emeterstatus.py +++ b/kasa/emeterstatus.py @@ -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"] diff --git a/kasa/exceptions.py b/kasa/exceptions.py index 9b91204a..567f01b4 100644 --- a/kasa/exceptions.py +++ b/kasa/exceptions.py @@ -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): diff --git a/kasa/feature.py b/kasa/feature.py index 60b43670..a04e1140 100644 --- a/kasa/feature.py +++ b/kasa/feature.py @@ -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.""" diff --git a/kasa/httpclient.py b/kasa/httpclient.py index 3240897c..55ac5a8e 100644 --- a/kasa/httpclient.py +++ b/kasa/httpclient.py @@ -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 diff --git a/kasa/iot/iotbulb.py b/kasa/iot/iotbulb.py index 1bf198af..26f40f06 100644 --- a/kasa/iot/iotbulb.py +++ b/kasa/iot/iotbulb.py @@ -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"]] diff --git a/kasa/iot/iotdevice.py b/kasa/iot/iotdevice.py index 8c93f016..32781a54 100755 --- a/kasa/iot/iotdevice.py +++ b/kasa/iot/iotdevice.py @@ -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): diff --git a/kasa/iot/iotdimmer.py b/kasa/iot/iotdimmer.py index fd0ff139..9c8c8f55 100644 --- a/kasa/iot/iotdimmer.py +++ b/kasa/iot/iotdimmer.py @@ -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 diff --git a/kasa/iot/iotlightstrip.py b/kasa/iot/iotlightstrip.py index 77b948f9..57b3282f 100644 --- a/kasa/iot/iotlightstrip.py +++ b/kasa/iot/iotlightstrip.py @@ -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. diff --git a/kasa/iot/iotplug.py b/kasa/iot/iotplug.py index 0a67debf..c584131d 100644 --- a/kasa/iot/iotplug.py +++ b/kasa/iot/iotplug.py @@ -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 diff --git a/kasa/iot/iotstrip.py b/kasa/iot/iotstrip.py index 1860c8fe..e1fdabae 100755 --- a/kasa/iot/iotstrip.py +++ b/kasa/iot/iotstrip.py @@ -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: diff --git a/kasa/iot/modules/emeter.py b/kasa/iot/modules/emeter.py index 178b92e4..52346ecc 100644 --- a/kasa/iot/modules/emeter.py +++ b/kasa/iot/modules/emeter.py @@ -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:: diff --git a/kasa/iot/modules/motion.py b/kasa/iot/modules/motion.py index 59fe4299..fe59748e 100644 --- a/kasa/iot/modules/motion.py +++ b/kasa/iot/modules/motion.py @@ -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. diff --git a/kasa/iot/modules/rulemodule.py b/kasa/iot/modules/rulemodule.py index 0739058d..1feaf456 100644 --- a/kasa/iot/modules/rulemodule.py +++ b/kasa/iot/modules/rulemodule.py @@ -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 [ diff --git a/kasa/iot/modules/usage.py b/kasa/iot/modules/usage.py index faffb5d8..5acf1dbe 100644 --- a/kasa/iot/modules/usage.py +++ b/kasa/iot/modules/usage.py @@ -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:: diff --git a/kasa/iotprotocol.py b/kasa/iotprotocol.py index a0a28612..1795566e 100755 --- a/kasa/iotprotocol.py +++ b/kasa/iotprotocol.py @@ -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: diff --git a/kasa/klaptransport.py b/kasa/klaptransport.py index 8feae98c..3a1eb336 100644 --- a/kasa/klaptransport.py +++ b/kasa/klaptransport.py @@ -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) diff --git a/kasa/module.py b/kasa/module.py index 3aa973fc..ad0b5562 100644 --- a/kasa/module.py +++ b/kasa/module.py @@ -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): diff --git a/kasa/protocol.py b/kasa/protocol.py index a62bf4de..c7d505b8 100755 --- a/kasa/protocol.py +++ b/kasa/protocol.py @@ -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() diff --git a/kasa/smart/modules/alarmmodule.py b/kasa/smart/modules/alarmmodule.py index a05fde35..66790326 100644 --- a/kasa/smart/modules/alarmmodule.py +++ b/kasa/smart/modules/alarmmodule.py @@ -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 diff --git a/kasa/smart/modules/autooffmodule.py b/kasa/smart/modules/autooffmodule.py index d72b6290..1d31bfb9 100644 --- a/kasa/smart/modules/autooffmodule.py +++ b/kasa/smart/modules/autooffmodule.py @@ -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 diff --git a/kasa/smart/modules/battery.py b/kasa/smart/modules/battery.py index 13d35f6f..982f9c6a 100644 --- a/kasa/smart/modules/battery.py +++ b/kasa/smart/modules/battery.py @@ -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( diff --git a/kasa/smart/modules/brightness.py b/kasa/smart/modules/brightness.py index 0d9f035b..a783ec3a 100644 --- a/kasa/smart/modules/brightness.py +++ b/kasa/smart/modules/brightness.py @@ -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 {} diff --git a/kasa/smart/modules/cloudmodule.py b/kasa/smart/modules/cloudmodule.py index 4027a25b..d53633f2 100644 --- a/kasa/smart/modules/cloudmodule.py +++ b/kasa/smart/modules/cloudmodule.py @@ -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( diff --git a/kasa/smart/modules/colortemp.py b/kasa/smart/modules/colortemp.py index a1338a34..3fda9c8a 100644 --- a/kasa/smart/modules/colortemp.py +++ b/kasa/smart/modules/colortemp.py @@ -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 {} diff --git a/kasa/smart/modules/devicemodule.py b/kasa/smart/modules/devicemodule.py index 050a864b..6a846d54 100644 --- a/kasa/smart/modules/devicemodule.py +++ b/kasa/smart/modules/devicemodule.py @@ -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, diff --git a/kasa/smart/modules/energymodule.py b/kasa/smart/modules/energymodule.py index 7645d125..a3e0b4a1 100644 --- a/kasa/smart/modules/energymodule.py +++ b/kasa/smart/modules/energymodule.py @@ -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 diff --git a/kasa/smart/modules/fanmodule.py b/kasa/smart/modules/fanmodule.py index 4734aa91..1d79cdea 100644 --- a/kasa/smart/modules/fanmodule.py +++ b/kasa/smart/modules/fanmodule.py @@ -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 {} diff --git a/kasa/smart/modules/firmware.py b/kasa/smart/modules/firmware.py index abe5dc39..88effe07 100644 --- a/kasa/smart/modules/firmware.py +++ b/kasa/smart/modules/firmware.py @@ -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, diff --git a/kasa/smart/modules/humidity.py b/kasa/smart/modules/humidity.py index 668bde2d..8f829b26 100644 --- a/kasa/smart/modules/humidity.py +++ b/kasa/smart/modules/humidity.py @@ -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( diff --git a/kasa/smart/modules/ledmodule.py b/kasa/smart/modules/ledmodule.py index 34f87710..cac447b5 100644 --- a/kasa/smart/modules/ledmodule.py +++ b/kasa/smart/modules/ledmodule.py @@ -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}} diff --git a/kasa/smart/modules/lighttransitionmodule.py b/kasa/smart/modules/lighttransitionmodule.py index bf824823..229dea57 100644 --- a/kasa/smart/modules/lighttransitionmodule.py +++ b/kasa/smart/modules/lighttransitionmodule.py @@ -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() diff --git a/kasa/smart/modules/reportmodule.py b/kasa/smart/modules/reportmodule.py index 5bae299c..0f3987bd 100644 --- a/kasa/smart/modules/reportmodule.py +++ b/kasa/smart/modules/reportmodule.py @@ -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( diff --git a/kasa/smart/modules/temperature.py b/kasa/smart/modules/temperature.py index 0817e941..2a5d73ba 100644 --- a/kasa/smart/modules/temperature.py +++ b/kasa/smart/modules/temperature.py @@ -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( diff --git a/kasa/smart/modules/timemodule.py b/kasa/smart/modules/timemodule.py index fd48f43b..7a0eb51b 100644 --- a/kasa/smart/modules/timemodule.py +++ b/kasa/smart/modules/timemodule.py @@ -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( diff --git a/kasa/smart/smartbulb.py b/kasa/smart/smartbulb.py index 365130c7..082035e7 100644 --- a/kasa/smart/smartbulb.py +++ b/kasa/smart/smartbulb.py @@ -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 [] diff --git a/kasa/smart/smartchilddevice.py b/kasa/smart/smartchilddevice.py index 6289dbc0..ecff7cfe 100644 --- a/kasa/smart/smartchilddevice.py +++ b/kasa/smart/smartchilddevice.py @@ -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 diff --git a/kasa/smart/smartdevice.py b/kasa/smart/smartdevice.py index 331cf66e..f921fda9 100644 --- a/kasa/smart/smartdevice.py +++ b/kasa/smart/smartdevice.py @@ -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: diff --git a/kasa/smart/smartmodule.py b/kasa/smart/smartmodule.py index 20580975..a0f3c105 100644 --- a/kasa/smart/smartmodule.py +++ b/kasa/smart/smartmodule.py @@ -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. diff --git a/kasa/smartprotocol.py b/kasa/smartprotocol.py index 0b07be5f..3020a575 100644 --- a/kasa/smartprotocol.py +++ b/kasa/smartprotocol.py @@ -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 = { diff --git a/kasa/tests/conftest.py b/kasa/tests/conftest.py index a3bd6df2..7829eac1 100644 --- a/kasa/tests/conftest.py +++ b/kasa/tests/conftest.py @@ -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: diff --git a/kasa/tests/device_fixtures.py b/kasa/tests/device_fixtures.py index 9d01a830..7fe40f48 100644 --- a/kasa/tests/device_fixtures.py +++ b/kasa/tests/device_fixtures.py @@ -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: diff --git a/kasa/tests/discovery_fixtures.py b/kasa/tests/discovery_fixtures.py index 653f9970..957dc007 100644 --- a/kasa/tests/discovery_fixtures.py +++ b/kasa/tests/discovery_fixtures.py @@ -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"]} diff --git a/kasa/tests/fixtureinfo.py b/kasa/tests/fixtureinfo.py index c0b4b506..153d6cc3 100644 --- a/kasa/tests/fixtureinfo.py +++ b/kasa/tests/fixtureinfo.py @@ -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. diff --git a/kasa/tests/smart/modules/test_fan.py b/kasa/tests/smart/modules/test_fan.py index 7c7ad9d8..260fcf1a 100644 --- a/kasa/tests/smart/modules/test_fan.py +++ b/kasa/tests/smart/modules/test_fan.py @@ -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) diff --git a/kasa/tests/test_aestransport.py b/kasa/tests/test_aestransport.py index 859c35be..ffd32cb1 100644 --- a/kasa/tests/test_aestransport.py +++ b/kasa/tests/test_aestransport.py @@ -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 diff --git a/kasa/tests/test_smartdevice.py b/kasa/tests/test_smartdevice.py index 584897b8..ffcd57ae 100644 --- a/kasa/tests/test_smartdevice.py +++ b/kasa/tests/test_smartdevice.py @@ -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()} diff --git a/kasa/xortransport.py b/kasa/xortransport.py index 085a6d64..0bca0321 100644 --- a/kasa/xortransport.py +++ b/kasa/xortransport.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 533abd2b..fa01911a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,6 +110,7 @@ select = [ "UP", # pyupgrade "B", # flake8-bugbear "SIM", # flake8-simplify + "FA", # flake8-future-annotations "I", # isort "S", # bandit ]