evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Speed up TP-Link lights (#33606)

Browse source 
* Speed up TP-Link lights

* Color temp kan be None

* hs as int, force color temp=0

* Fix color temp?

* Additional tplink cleanups to reduce api calls

* Update test to return state, remove Throttle

* Fix state restore on off/on

* Fix lights without hue/sat

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Paulus Schoutsen 2020-04-08 12:36:45 -07:00 • committed by GitHub
parent 292dcb7e37
commit df744c5944
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 173 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

286
homeassistant/components/tplink/light.py
View file

@ -15,18 +15,22 @@ from homeassistant.components.light import (
SUPPORT_COLOR_TEMP,
Light,
)
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.device_registry as dr
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util.color import (
color_temperature_kelvin_to_mired as kelvin_to_mired,
color_temperature_mired_to_kelvin as mired_to_kelvin,
)
import homeassistant.util.dt as dt_util
from . import CONF_LIGHT, DOMAIN as TPLINK_DOMAIN
from .common import async_add_entities_retry
PARALLEL_UPDATES = 0
SCAN_INTERVAL = timedelta(seconds=5)
CURRENT_POWER_UPDATE_INTERVAL = timedelta(seconds=60)
HISTORICAL_POWER_UPDATE_INTERVAL = timedelta(minutes=60)
_LOGGER = logging.getLogger(__name__)
@ -34,6 +38,21 @@ ATTR_CURRENT_POWER_W = "current_power_w"
ATTR_DAILY_ENERGY_KWH = "daily_energy_kwh"
ATTR_MONTHLY_ENERGY_KWH = "monthly_energy_kwh"
LIGHT_STATE_DFT_ON = "dft_on_state"
LIGHT_STATE_ON_OFF = "on_off"
LIGHT_STATE_BRIGHTNESS = "brightness"
LIGHT_STATE_COLOR_TEMP = "color_temp"
LIGHT_STATE_HUE = "hue"
LIGHT_STATE_SATURATION = "saturation"
LIGHT_STATE_ERROR_MSG = "err_msg"
LIGHT_SYSINFO_MAC = "mac"
LIGHT_SYSINFO_ALIAS = "alias"
LIGHT_SYSINFO_MODEL = "model"
LIGHT_SYSINFO_IS_DIMMABLE = "is_dimmable"
LIGHT_SYSINFO_IS_VARIABLE_COLOR_TEMP = "is_variable_color_temp"
LIGHT_SYSINFO_IS_COLOR = "is_color"
async def async_setup_entry(hass: HomeAssistantType, config_entry, async_add_entities):
"""Set up switches."""
@ -70,7 +89,21 @@ class LightState(NamedTuple):
brightness: int
color_temp: float
hs: Tuple[int, int]
emeter_params: dict
def to_param(self):
"""Return a version that we can send to the bulb."""
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,
LIGHT_STATE_BRIGHTNESS: brightness_to_percentage(self.brightness),
LIGHT_STATE_COLOR_TEMP: color_temp,
LIGHT_STATE_HUE: self.hs[0] if self.hs else 0,
LIGHT_STATE_SATURATION: self.hs[1] if self.hs else 0,
}
class LightFeatures(NamedTuple):
@ -95,6 +128,9 @@ class TPLinkSmartBulb(Light):
self._light_state = cast(LightState, None)
self._is_available = True
self._is_setting_light_state = False
self._last_current_power_update = None
self._last_historical_power_update = None
self._emeter_params = {}
@property
def unique_id(self):
@ -125,45 +161,42 @@ class TPLinkSmartBulb(Light):
@property
def device_state_attributes(self):
"""Return the state attributes of the device."""
return self._light_state.emeter_params
return self._emeter_params
async def async_turn_on(self, **kwargs):
"""Turn the light on."""
brightness = (
int(kwargs[ATTR_BRIGHTNESS])
if ATTR_BRIGHTNESS in kwargs
else self._light_state.brightness
if self._light_state.brightness is not None
else 255
)
color_tmp = (
int(kwargs[ATTR_COLOR_TEMP])
if ATTR_COLOR_TEMP in kwargs
else self._light_state.color_temp
)
if ATTR_BRIGHTNESS in kwargs:
brightness = int(kwargs[ATTR_BRIGHTNESS])
elif self._light_state.brightness is not None:
brightness = self._light_state.brightness
else:
brightness = 255
await self.async_set_light_state_retry(
if ATTR_COLOR_TEMP in kwargs:
color_tmp = int(kwargs[ATTR_COLOR_TEMP])
else:
color_tmp = self._light_state.color_temp
if ATTR_HS_COLOR in kwargs:
# TP-Link requires integers.
hue_sat = tuple(int(val) for val in kwargs[ATTR_HS_COLOR])
# TP-Link cannot have both color temp and hue_sat
color_tmp = 0
else:
hue_sat = self._light_state.hs
await self._async_set_light_state_retry(
self._light_state,
LightState(
state=True,
brightness=brightness,
color_temp=color_tmp,
hs=tuple(kwargs.get(ATTR_HS_COLOR, self._light_state.hs or ())),
emeter_params=self._light_state.emeter_params,
self._light_state._replace(
state=True, brightness=brightness, color_temp=color_tmp, hs=hue_sat,
),
)
async def async_turn_off(self, **kwargs):
"""Turn the light off."""
await self.async_set_light_state_retry(
self._light_state,
LightState(
state=False,
brightness=self._light_state.brightness,
color_temp=self._light_state.color_temp,
hs=self._light_state.hs,
emeter_params=self._light_state.emeter_params,
),
await self._async_set_light_state_retry(
self._light_state, self._light_state._replace(state=False),
)
@property
@ -202,21 +235,11 @@ class TPLinkSmartBulb(Light):
if self._is_setting_light_state:
return
# Initial run, perform call blocking.
if not self._light_features:
self.do_update_retry(False)
# Subsequent runs should not block.
else:
self.hass.add_job(self.do_update_retry, True)
def do_update_retry(self, update_state: bool) -> None:
"""Update state data with retry.""" ""
try:
# Update light features only once.
self._light_features = (
self._light_features or self.get_light_features_retry()
)
self._light_state = self.get_light_state_retry(self._light_features)
if not self._light_features:
self._light_features = self._get_light_features_retry()
self._light_state = self._get_light_state_retry()
self._is_available = True
except (SmartDeviceException, OSError) as ex:
if self._is_available:
@ -225,45 +248,42 @@ class TPLinkSmartBulb(Light):
)
self._is_available = False
# The local variables were updates asyncronousally,
# we need the entity registry to poll this object's properties for
# updated information. Calling schedule_update_ha_state will only
# cause a loop.
if update_state:
self.schedule_update_ha_state()
@property
def supported_features(self):
"""Flag supported features."""
return self._light_features.supported_features
def get_light_features_retry(self) -> LightFeatures:
def _get_light_features_retry(self) -> LightFeatures:
"""Retry the retrieval of the supported features."""
try:
return self.get_light_features()
return self._get_light_features()
except (SmartDeviceException, OSError):
pass
_LOGGER.debug("Retrying getting light features")
return self.get_light_features()
return self._get_light_features()
def get_light_features(self):
def _get_light_features(self):
"""Determine all supported features in one go."""
sysinfo = self.smartbulb.sys_info
supported_features = 0
# Calling api here as it reformats
mac = self.smartbulb.mac
alias = self.smartbulb.alias
model = self.smartbulb.model
alias = sysinfo[LIGHT_SYSINFO_ALIAS]
model = sysinfo[LIGHT_SYSINFO_MODEL]
min_mireds = None
max_mireds = None
if self.smartbulb.is_dimmable:
if sysinfo.get(LIGHT_SYSINFO_IS_DIMMABLE):
supported_features += SUPPORT_BRIGHTNESS
if getattr(self.smartbulb, "is_variable_color_temp", False):
if sysinfo.get(LIGHT_SYSINFO_IS_VARIABLE_COLOR_TEMP):
supported_features += SUPPORT_COLOR_TEMP
min_mireds = kelvin_to_mired(self.smartbulb.valid_temperature_range[1])
max_mireds = kelvin_to_mired(self.smartbulb.valid_temperature_range[0])
if getattr(self.smartbulb, "is_color", False):
# Have to make another api request here in
# order to not re-implement pyHS100 here
max_range, min_range = self.smartbulb.valid_temperature_range
min_mireds = kelvin_to_mired(min_range)
max_mireds = kelvin_to_mired(max_range)
if sysinfo.get(LIGHT_SYSINFO_IS_COLOR):
supported_features += SUPPORT_COLOR
return LightFeatures(
@ -276,110 +296,146 @@ class TPLinkSmartBulb(Light):
max_mireds=max_mireds,
)
def get_light_state_retry(self, light_features: LightFeatures) -> LightState:
def _get_light_state_retry(self) -> LightState:
"""Retry the retrieval of getting light states."""
try:
return self.get_light_state(light_features)
return self._get_light_state()
except (SmartDeviceException, OSError):
pass
_LOGGER.debug("Retrying getting light state")
return self.get_light_state(light_features)
return self._get_light_state()
def get_light_state(self, light_features: LightFeatures) -> LightState:
"""Get the light state."""
emeter_params = {}
def _light_state_from_params(self, light_state_params) -> LightState:
brightness = None
color_temp = None
hue_saturation = None
state = self.smartbulb.state == SmartBulb.BULB_STATE_ON
light_features = self._light_features
state = bool(light_state_params[LIGHT_STATE_ON_OFF])
if not state and LIGHT_STATE_DFT_ON in light_state_params:
light_state_params = light_state_params[LIGHT_STATE_DFT_ON]
if light_features.supported_features & SUPPORT_BRIGHTNESS:
brightness = brightness_from_percentage(self.smartbulb.brightness)
brightness = brightness_from_percentage(
light_state_params[LIGHT_STATE_BRIGHTNESS]
)
if light_features.supported_features & SUPPORT_COLOR_TEMP:
if self.smartbulb.color_temp is not None and self.smartbulb.color_temp != 0:
color_temp = kelvin_to_mired(self.smartbulb.color_temp)
if (
light_state_params.get(LIGHT_STATE_COLOR_TEMP) is not None
and light_state_params[LIGHT_STATE_COLOR_TEMP] != 0
):
color_temp = kelvin_to_mired(light_state_params[LIGHT_STATE_COLOR_TEMP])
if light_features.supported_features & SUPPORT_COLOR:
hue, sat, _ = self.smartbulb.hsv
hue_saturation = (hue, sat)
if self.smartbulb.has_emeter:
emeter_params[ATTR_CURRENT_POWER_W] = "{:.1f}".format(
self.smartbulb.current_consumption()
hue_saturation = (
light_state_params[LIGHT_STATE_HUE],
light_state_params[LIGHT_STATE_SATURATION],
)
daily_statistics = self.smartbulb.get_emeter_daily()
monthly_statistics = self.smartbulb.get_emeter_monthly()
try:
emeter_params[ATTR_DAILY_ENERGY_KWH] = "{:.3f}".format(
daily_statistics[int(time.strftime("%d"))]
)
emeter_params[ATTR_MONTHLY_ENERGY_KWH] = "{:.3f}".format(
monthly_statistics[int(time.strftime("%m"))]
)
except KeyError:
# device returned no daily/monthly history
pass
return LightState(
state=state,
brightness=brightness,
color_temp=color_temp,
hs=hue_saturation,
emeter_params=emeter_params,
)
async def async_set_light_state_retry(
def _get_light_state(self) -> LightState:
"""Get the light state."""
self._update_emeter()
return self._light_state_from_params(self.smartbulb.get_light_state())
def _update_emeter(self):
if not self.smartbulb.has_emeter:
return
now = dt_util.utcnow()
if (
not self._last_current_power_update
or self._last_current_power_update + CURRENT_POWER_UPDATE_INTERVAL < now
):
self._last_current_power_update = now
self._emeter_params[ATTR_CURRENT_POWER_W] = "{:.1f}".format(
self.smartbulb.current_consumption()
)
if (
not self._last_historical_power_update
or self._last_historical_power_update + HISTORICAL_POWER_UPDATE_INTERVAL
< now
):
self._last_historical_power_update = now
daily_statistics = self.smartbulb.get_emeter_daily()
monthly_statistics = self.smartbulb.get_emeter_monthly()
try:
self._emeter_params[ATTR_DAILY_ENERGY_KWH] = "{:.3f}".format(
daily_statistics[int(time.strftime("%d"))]
)
self._emeter_params[ATTR_MONTHLY_ENERGY_KWH] = "{:.3f}".format(
monthly_statistics[int(time.strftime("%m"))]
)
except KeyError:
# device returned no daily/monthly history
pass
async def _async_set_light_state_retry(
self, old_light_state: LightState, new_light_state: LightState
) -> None:
"""Set the light state with retry."""
# Optimistically setting the light state.
self._light_state = new_light_state
# Tell the device to set the states.
if not _light_state_diff(old_light_state, new_light_state):
# Nothing to do, avoid the executor
return
self._is_setting_light_state = True
try:
await self.hass.async_add_executor_job(
self.set_light_state, old_light_state, new_light_state
light_state_params = await self.hass.async_add_executor_job(
self._set_light_state, old_light_state, new_light_state
)
self._is_available = True
self._is_setting_light_state = False
if LIGHT_STATE_ERROR_MSG in light_state_params:
raise HomeAssistantError(light_state_params[LIGHT_STATE_ERROR_MSG])
self._light_state = self._light_state_from_params(light_state_params)
return
except (SmartDeviceException, OSError):
pass
try:
_LOGGER.debug("Retrying setting light state")
await self.hass.async_add_executor_job(
self.set_light_state, old_light_state, new_light_state
light_state_params = await self.hass.async_add_executor_job(
self._set_light_state, old_light_state, new_light_state
)
self._is_available = True
if LIGHT_STATE_ERROR_MSG in light_state_params:
raise HomeAssistantError(light_state_params[LIGHT_STATE_ERROR_MSG])
self._light_state = self._light_state_from_params(light_state_params)
except (SmartDeviceException, OSError) as ex:
self._is_available = False
_LOGGER.warning("Could not set data for %s: %s", self.smartbulb.host, ex)
self._is_setting_light_state = False
def set_light_state(
def _set_light_state(
self, old_light_state: LightState, new_light_state: LightState
) -> None:
"""Set the light state."""
# Calling the API with the new state information.
if new_light_state.state != old_light_state.state:
if new_light_state.state:
self.smartbulb.state = SmartBulb.BULB_STATE_ON
else:
self.smartbulb.state = SmartBulb.BULB_STATE_OFF
return
diff = _light_state_diff(old_light_state, new_light_state)
if new_light_state.color_temp != old_light_state.color_temp:
self.smartbulb.color_temp = mired_to_kelvin(new_light_state.color_temp)
if not diff:
return
brightness_pct = brightness_to_percentage(new_light_state.brightness)
if new_light_state.hs != old_light_state.hs and len(new_light_state.hs) > 1:
hue, sat = new_light_state.hs
hsv = (int(hue), int(sat), brightness_pct)
self.smartbulb.hsv = hsv
elif new_light_state.brightness != old_light_state.brightness:
self.smartbulb.brightness = brightness_pct
return self.smartbulb.set_light_state(diff)
def _light_state_diff(old_light_state: LightState, new_light_state: LightState):
old_state_param = old_light_state.to_param()
new_state_param = new_light_state.to_param()
return {
key: value
for key, value in new_state_param.items()
if new_state_param.get(key) != old_state_param.get(key)
}

3
tests/components/tplink/test_light.py
View file

@ -85,6 +85,7 @@ def light_mock_data_fixture() -> None:
light_state.update(state)
light_state["dft_on_state"] = drt_on_state
return light_state
set_light_state_patch = patch(
"homeassistant.components.tplink.common.SmartBulb.set_light_state",
@ -310,7 +311,7 @@ async def test_get_light_state_retry(
if set_state_call_count == 1:
raise SmartDeviceException()
light_mock_data.set_light_state(state_data)
return light_mock_data.set_light_state(state_data)
light_mock_data.set_light_state_mock.side_effect = set_light_state_side_effect