From 560dd0a0cccebb45dc23a026ca438e4e250b99c3 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Sat, 22 May 2021 14:47:26 +0200 Subject: [PATCH] Typing improvements for TPLink (#50947) * Typing improvements for TPLink * Update homeassistant/components/tplink/common.py Co-authored-by: Martin Hjelmare Co-authored-by: Martin Hjelmare --- homeassistant/components/tplink/__init__.py | 16 ++++-- homeassistant/components/tplink/common.py | 20 ++++---- homeassistant/components/tplink/light.py | 57 ++++++++++++--------- homeassistant/components/tplink/switch.py | 35 ++++++++----- 4 files changed, 78 insertions(+), 50 deletions(-) diff --git a/homeassistant/components/tplink/__init__.py b/homeassistant/components/tplink/__init__.py index f424f90d6d3..69241f1cb44 100644 --- a/homeassistant/components/tplink/__init__.py +++ b/homeassistant/components/tplink/__init__.py @@ -4,6 +4,7 @@ import logging import voluptuous as vol from homeassistant import config_entries +from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_HOST from homeassistant.core import HomeAssistant import homeassistant.helpers.config_validation as cv @@ -54,7 +55,7 @@ CONFIG_SCHEMA = vol.Schema( ) -async def async_setup(hass, config): +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the TP-Link component.""" conf = config.get(DOMAIN) @@ -71,7 +72,7 @@ async def async_setup(hass, config): return True -async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigType): +async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: """Set up TPLink from a config entry.""" config_data = hass.data[DOMAIN].get(ATTR_CONFIG) @@ -97,19 +98,24 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigType): forward_setup = hass.config_entries.async_forward_entry_setup if lights: _LOGGER.debug( - "Got %s lights: %s", len(lights), ", ".join([d.host for d in lights]) + "Got %s lights: %s", len(lights), ", ".join(d.host for d in lights) ) + hass.async_create_task(forward_setup(config_entry, "light")) + if switches: _LOGGER.debug( - "Got %s switches: %s", len(switches), ", ".join([d.host for d in switches]) + "Got %s switches: %s", + len(switches), + ", ".join(d.host for d in switches), ) + hass.async_create_task(forward_setup(config_entry, "switch")) return True -async def async_unload_entry(hass, entry): +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" platforms = [platform for platform in PLATFORMS if hass.data[DOMAIN].get(platform)] unload_ok = await hass.config_entries.async_unload_platforms(entry, platforms) diff --git a/homeassistant/components/tplink/common.py b/homeassistant/components/tplink/common.py index f9ed57d26fb..8b1ee4a44b1 100644 --- a/homeassistant/components/tplink/common.py +++ b/homeassistant/components/tplink/common.py @@ -2,6 +2,7 @@ from __future__ import annotations import logging +from typing import Callable from pyHS100 import ( Discover, @@ -38,16 +39,16 @@ class SmartDevices: self._switches = switches or [] @property - def lights(self): + def lights(self) -> list[SmartDevice]: """Get the lights.""" return self._lights @property - def switches(self): + def switches(self) -> list[SmartDevice]: """Get the switches.""" return self._switches - def has_device_with_host(self, host): + def has_device_with_host(self, host: str) -> bool: """Check if a devices exists with a specific host.""" for device in self.lights + self.switches: if device.host == host: @@ -56,12 +57,11 @@ class SmartDevices: return False -async def async_get_discoverable_devices(hass): +async def async_get_discoverable_devices(hass: HomeAssistant) -> dict[str, SmartDevice]: """Return if there are devices that can be discovered.""" - def discover(): - devs = Discover.discover() - return devs + def discover() -> dict[str, SmartDevice]: + return Discover.discover() return await hass.async_add_executor_job(discover) @@ -77,7 +77,7 @@ async def async_discover_devices( lights = [] switches = [] - def process_devices(): + def process_devices() -> None: for dev in devices.values(): # If this device already exists, ignore dynamic setup. if existing_devices.has_device_with_host(dev.host): @@ -132,7 +132,9 @@ def get_static_devices(config_data) -> SmartDevices: return SmartDevices(lights, switches) -def add_available_devices(hass, device_type, device_class): +def add_available_devices( + hass: HomeAssistant, device_type: str, device_class: Callable +) -> list: """Get sysinfo for all devices.""" devices = hass.data[TPLINK_DOMAIN][device_type] diff --git a/homeassistant/components/tplink/light.py b/homeassistant/components/tplink/light.py index 5984698a796..e5217cbc143 100644 --- a/homeassistant/components/tplink/light.py +++ b/homeassistant/components/tplink/light.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio +from collections.abc import Mapping from datetime import timedelta import logging import re @@ -19,9 +20,12 @@ from homeassistant.components.light import ( SUPPORT_COLOR_TEMP, LightEntity, ) +from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError, PlatformNotReady import homeassistant.helpers.device_registry as dr +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util.color import ( color_temperature_kelvin_to_mired as kelvin_to_mired, color_temperature_mired_to_kelvin as mired_to_kelvin, @@ -78,7 +82,11 @@ FALLBACK_MIN_COLOR = 2700 FALLBACK_MAX_COLOR = 5000 -async def async_setup_entry(hass: HomeAssistant, config_entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Set up lights.""" entities = await hass.async_add_executor_job( add_available_devices, hass, CONF_LIGHT, TPLinkSmartBulb @@ -111,10 +119,9 @@ class LightState(NamedTuple): def to_param(self): """Return a version that we can send to the bulb.""" + color_temp = None if self.color_temp: color_temp = mired_to_kelvin(self.color_temp) - else: - color_temp = None return { LIGHT_STATE_ON_OFF: 1 if self.state else 0, @@ -157,17 +164,17 @@ class TPLinkSmartBulb(LightEntity): self._alias = None @property - def unique_id(self): + def unique_id(self) -> str | None: """Return a unique ID.""" return self._light_features.mac @property - def name(self): + def name(self) -> str | None: """Return the name of the Smart Bulb.""" return self._light_features.alias @property - def device_info(self): + def device_info(self) -> DeviceInfo: """Return information about the device.""" return { "name": self._light_features.alias, @@ -183,11 +190,11 @@ class TPLinkSmartBulb(LightEntity): return self._is_available @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> Mapping[str, Any] | None: """Return the state attributes of the device.""" return self._emeter_params - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the light on.""" if ATTR_BRIGHTNESS in kwargs: brightness = int(kwargs[ATTR_BRIGHTNESS]) @@ -220,7 +227,7 @@ class TPLinkSmartBulb(LightEntity): ), ) - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the light off.""" await self._async_set_light_state_retry( self._light_state, @@ -228,36 +235,36 @@ class TPLinkSmartBulb(LightEntity): ) @property - def min_mireds(self): + def min_mireds(self) -> int: """Return minimum supported color temperature.""" return self._light_features.min_mireds @property - def max_mireds(self): + def max_mireds(self) -> int: """Return maximum supported color temperature.""" return self._light_features.max_mireds @property - def color_temp(self): + def color_temp(self) -> int | None: """Return the color temperature of this light in mireds for HA.""" return self._light_state.color_temp @property - def brightness(self): + def brightness(self) -> int | None: """Return the brightness of this light between 0..255.""" return self._light_state.brightness @property - def hs_color(self): + def hs_color(self) -> tuple[float, float] | None: """Return the color.""" return self._light_state.hs @property - def is_on(self): + def is_on(self) -> bool: """Return True if device is on.""" return self._light_state.state - def attempt_update(self, update_attempt): + def attempt_update(self, update_attempt: int) -> bool: """Attempt to get details the TP-Link bulb.""" # State is currently being set, ignore. if self._is_setting_light_state: @@ -283,11 +290,11 @@ class TPLinkSmartBulb(LightEntity): return False @property - def supported_features(self): + def supported_features(self) -> int: """Flag supported features.""" return self._light_features.supported_features - def _get_valid_temperature_range(self): + def _get_valid_temperature_range(self) -> tuple[int, int]: """Return the device-specific white temperature range (in Kelvin). :return: White temperature range in Kelvin (minimum, maximum) @@ -300,7 +307,7 @@ class TPLinkSmartBulb(LightEntity): # use "safe" values for something that advertises color temperature return FALLBACK_MIN_COLOR, FALLBACK_MAX_COLOR - def _get_light_features(self): + def _get_light_features(self) -> LightFeatures: """Determine all supported features in one go.""" sysinfo = self.smartbulb.sys_info supported_features = 0 @@ -333,7 +340,7 @@ class TPLinkSmartBulb(LightEntity): has_emeter=has_emeter, ) - def _light_state_from_params(self, light_state_params) -> LightState: + def _light_state_from_params(self, light_state_params: Any) -> LightState: brightness = None color_temp = None hue_saturation = None @@ -374,7 +381,7 @@ class TPLinkSmartBulb(LightEntity): self._update_emeter() return self._light_state_from_params(self._get_device_state()) - def _update_emeter(self): + def _update_emeter(self) -> None: if not self._light_features.has_emeter: return @@ -456,7 +463,7 @@ class TPLinkSmartBulb(LightEntity): return self._set_device_state(diff) - def _get_device_state(self): + def _get_device_state(self) -> dict: """State of the bulb or smart dimmer switch.""" if isinstance(self.smartbulb, SmartBulb): return self.smartbulb.get_light_state() @@ -493,7 +500,7 @@ class TPLinkSmartBulb(LightEntity): return self._get_device_state() - async def async_update(self): + async def async_update(self) -> None: """Update the TP-Link bulb's state.""" for update_attempt in range(MAX_ATTEMPTS): is_ready = await self.hass.async_add_executor_job( @@ -521,7 +528,9 @@ class TPLinkSmartBulb(LightEntity): self._is_available = False -def _light_state_diff(old_light_state: LightState, new_light_state: LightState): +def _light_state_diff( + old_light_state: LightState, new_light_state: LightState +) -> dict[str, Any]: old_state_param = old_light_state.to_param() new_state_param = new_light_state.to_param() diff --git a/homeassistant/components/tplink/switch.py b/homeassistant/components/tplink/switch.py index 011caa463b2..d088584c4ad 100644 --- a/homeassistant/components/tplink/switch.py +++ b/homeassistant/components/tplink/switch.py @@ -1,8 +1,12 @@ """Support for TPLink HS100/HS110/HS200 smart switch.""" +from __future__ import annotations + import asyncio +from collections.abc import Mapping from contextlib import suppress import logging import time +from typing import Any from pyHS100 import SmartDeviceException, SmartPlug @@ -11,10 +15,13 @@ from homeassistant.components.switch import ( ATTR_TODAY_ENERGY_KWH, SwitchEntity, ) +from homeassistant.config_entries import ConfigEntry from homeassistant.const import ATTR_VOLTAGE from homeassistant.core import HomeAssistant from homeassistant.exceptions import PlatformNotReady import homeassistant.helpers.device_registry as dr +from homeassistant.helpers.entity import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback from . import CONF_SWITCH, DOMAIN as TPLINK_DOMAIN from .common import add_available_devices @@ -30,7 +37,11 @@ MAX_ATTEMPTS = 300 SLEEP_TIME = 2 -async def async_setup_entry(hass: HomeAssistant, config_entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Set up switches.""" entities = await hass.async_add_executor_job( add_available_devices, hass, CONF_SWITCH, SmartPlugSwitch @@ -62,17 +73,17 @@ class SmartPlugSwitch(SwitchEntity): self._host = None @property - def unique_id(self): + def unique_id(self) -> str | None: """Return a unique ID.""" return self._device_id @property - def name(self): + def name(self) -> str | None: """Return the name of the Smart Plug.""" return self._alias @property - def device_info(self): + def device_info(self) -> DeviceInfo: """Return information about the device.""" return { "name": self._alias, @@ -88,37 +99,37 @@ class SmartPlugSwitch(SwitchEntity): return self._is_available @property - def is_on(self): + def is_on(self) -> bool | None: """Return true if switch is on.""" return self._state - def turn_on(self, **kwargs): + def turn_on(self, **kwargs: Any) -> None: """Turn the switch on.""" self.smartplug.turn_on() - def turn_off(self, **kwargs): + def turn_off(self, **kwargs: Any) -> None: """Turn the switch off.""" self.smartplug.turn_off() @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> Mapping[str, Any] | None: """Return the state attributes of the device.""" return self._emeter_params @property - def _plug_from_context(self): + def _plug_from_context(self) -> Any: """Return the plug from the context.""" children = self.smartplug.sys_info["children"] return next(c for c in children if c["id"] == self.smartplug.context) - def update_state(self): + def update_state(self) -> None: """Update the TP-Link switch's state.""" if self.smartplug.context is None: self._state = self.smartplug.state == self.smartplug.SWITCH_STATE_ON else: self._state = self._plug_from_context["state"] == 1 - def attempt_update(self, update_attempt): + def attempt_update(self, update_attempt: int) -> bool: """Attempt to get details from the TP-Link switch.""" try: if not self._sysinfo: @@ -168,7 +179,7 @@ class SmartPlugSwitch(SwitchEntity): ) return False - async def async_update(self): + async def async_update(self) -> None: """Update the TP-Link switch's state.""" for update_attempt in range(MAX_ATTEMPTS): is_ready = await self.hass.async_add_executor_job(