Typing improvements for TPLink (#50947)

* Typing improvements for TPLink

* Update homeassistant/components/tplink/common.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Franck Nijhof 2021-05-22 14:47:26 +02:00 committed by GitHub
parent afb372a680
commit 560dd0a0cc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 78 additions and 50 deletions

View File

@ -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)

View File

@ -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]

View File

@ -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()

View File

@ -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(