From 2e44e256f0eb2853d7b6b7f9ed3a257bfeccad76 Mon Sep 17 00:00:00 2001 From: Erik Montnemery Date: Mon, 12 Jul 2021 19:17:44 +0200 Subject: [PATCH] Improve typing of Tasmota (3/3) (#52748) --- homeassistant/components/tasmota/light.py | 88 +++++++++++++--------- homeassistant/components/tasmota/mixins.py | 39 ++++++---- homeassistant/components/tasmota/sensor.py | 50 +++++++----- homeassistant/components/tasmota/switch.py | 31 +++++--- tests/components/tasmota/test_light.py | 24 +++--- 5 files changed, 142 insertions(+), 90 deletions(-) diff --git a/homeassistant/components/tasmota/light.py b/homeassistant/components/tasmota/light.py index 675a3d175c3..de25a25fd4f 100644 --- a/homeassistant/components/tasmota/light.py +++ b/homeassistant/components/tasmota/light.py @@ -1,4 +1,10 @@ """Support for Tasmota lights.""" +from __future__ import annotations + +from typing import Any + +from hatasmota import light as tasmota_light +from hatasmota.entity import TasmotaEntity as HATasmotaEntity, TasmotaEntityConfig from hatasmota.light import ( LIGHT_TYPE_COLDWARM, LIGHT_TYPE_NONE, @@ -6,6 +12,7 @@ from hatasmota.light import ( LIGHT_TYPE_RGBCW, LIGHT_TYPE_RGBW, ) +from hatasmota.models import DiscoveryHashType from homeassistant.components import light from homeassistant.components.light import ( @@ -25,8 +32,10 @@ from homeassistant.components.light import ( LightEntity, brightness_supported, ) -from homeassistant.core import callback +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import DATA_REMOVE_DISCOVER_COMPONENT from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW @@ -36,11 +45,17 @@ DEFAULT_BRIGHTNESS_MAX = 255 TASMOTA_BRIGHTNESS_MAX = 100 -async def async_setup_entry(hass, config_entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Set up Tasmota light dynamically through discovery.""" @callback - def async_discover(tasmota_entity, discovery_hash): + def async_discover( + tasmota_entity: HATasmotaEntity, discovery_hash: DiscoveryHashType + ) -> None: """Discover and add a Tasmota light.""" async_add_entities( [TasmotaLight(tasmota_entity=tasmota_entity, discovery_hash=discovery_hash)] @@ -55,12 +70,12 @@ async def async_setup_entry(hass, config_entry, async_add_entities): ) -def clamp(value): +def clamp(value: float) -> float: """Clamp value to the range 0..255.""" return min(max(value, 0), 255) -def scale_brightness(brightness): +def scale_brightness(brightness: float) -> float: """Scale brightness from 0..255 to 1..100.""" brightness_normalized = brightness / DEFAULT_BRIGHTNESS_MAX device_brightness = min( @@ -79,19 +94,20 @@ class TasmotaLight( ): """Representation of a Tasmota light.""" - def __init__(self, **kwds): + _tasmota_entity: tasmota_light.TasmotaLight + + def __init__(self, **kwds: Any) -> None: """Initialize Tasmota light.""" - self._state = False - self._supported_color_modes = None + self._supported_color_modes: set[str] | None = None self._supported_features = 0 - self._brightness = None - self._color_mode = None - self._color_temp = None - self._effect = None - self._white_value = None + self._brightness: int | None = None + self._color_mode: str | None = None + self._color_temp: int | None = None + self._effect: str | None = None + self._white_value: int | None = None self._flash_times = None - self._hs = None + self._hs: tuple[float, float] | None = None super().__init__( **kwds, @@ -99,13 +115,15 @@ class TasmotaLight( self._setup_from_entity() - async def discovery_update(self, update, write_state=True): + async def discovery_update( + self, update: TasmotaEntityConfig, write_state: bool = True + ) -> None: """Handle updated discovery message.""" await super().discovery_update(update, write_state=False) self._setup_from_entity() self.async_write_ha_state() - def _setup_from_entity(self): + def _setup_from_entity(self) -> None: """(Re)Setup the entity.""" self._supported_color_modes = set() supported_features = 0 @@ -141,7 +159,7 @@ class TasmotaLight( self._supported_features = supported_features @callback - def state_updated(self, state, **kwargs): + def state_updated(self, state: bool, **kwargs: Any) -> None: """Handle state updates.""" self._on_off_state = state attributes = kwargs.get("attributes") @@ -149,7 +167,7 @@ class TasmotaLight( if "brightness" in attributes: brightness = float(attributes["brightness"]) percent_bright = brightness / TASMOTA_BRIGHTNESS_MAX - self._brightness = percent_bright * 255 + self._brightness = round(percent_bright * 255) if "color_hs" in attributes: self._hs = attributes["color_hs"] if "color_temp" in attributes: @@ -159,7 +177,7 @@ class TasmotaLight( if "white_value" in attributes: white_value = float(attributes["white_value"]) percent_white = white_value / TASMOTA_BRIGHTNESS_MAX - self._white_value = percent_white * 255 + self._white_value = round(percent_white * 255) if self._tasmota_entity.light_type == LIGHT_TYPE_RGBW: # Tasmota does not support RGBW mode, set mode to white or hs if self._white_value == 0: @@ -176,68 +194,68 @@ class TasmotaLight( self.async_write_ha_state() @property - def brightness(self): + def brightness(self) -> int | None: """Return the brightness of this light between 0..255.""" return self._brightness @property - def color_mode(self): + def color_mode(self) -> str | None: """Return the color mode of the light.""" return self._color_mode @property - def color_temp(self): + def color_temp(self) -> int | None: """Return the color temperature in mired.""" return self._color_temp @property - def min_mireds(self): + def min_mireds(self) -> int: """Return the coldest color_temp that this light supports.""" return self._tasmota_entity.min_mireds @property - def max_mireds(self): + def max_mireds(self) -> int: """Return the warmest color_temp that this light supports.""" return self._tasmota_entity.max_mireds @property - def effect(self): + def effect(self) -> str | None: """Return the current effect.""" return self._effect @property - def effect_list(self): + def effect_list(self) -> list[str] | None: """Return the list of supported effects.""" return self._tasmota_entity.effect_list @property - def hs_color(self): + def hs_color(self) -> tuple[float, float] | None: """Return the hs color value.""" if self._hs is None: return None hs_color = self._hs - return [hs_color[0], hs_color[1]] + return (hs_color[0], hs_color[1]) @property - def force_update(self): + def force_update(self) -> bool: """Force update.""" return False @property - def supported_color_modes(self): + def supported_color_modes(self) -> set[str] | None: """Flag supported color modes.""" return self._supported_color_modes @property - def supported_features(self): + def supported_features(self) -> int: """Flag supported features.""" return self._supported_features - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the entity on.""" - supported_color_modes = self._supported_color_modes + supported_color_modes = self._supported_color_modes or set() - attributes = {} + attributes: dict[str, Any] = {} if ATTR_HS_COLOR in kwargs and COLOR_MODE_HS in supported_color_modes: hs_color = kwargs[ATTR_HS_COLOR] @@ -260,7 +278,7 @@ class TasmotaLight( self._tasmota_entity.set_state(True, attributes) - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the entity off.""" attributes = {"state": "OFF"} diff --git a/homeassistant/components/tasmota/mixins.py b/homeassistant/components/tasmota/mixins.py index f1b0554957e..a07e48b53a7 100644 --- a/homeassistant/components/tasmota/mixins.py +++ b/homeassistant/components/tasmota/mixins.py @@ -4,6 +4,13 @@ from __future__ import annotations import logging from typing import Any +from hatasmota.entity import ( + TasmotaAvailability as HATasmotaAvailability, + TasmotaEntity as HATasmotaEntity, + TasmotaEntityConfig, +) +from hatasmota.models import DiscoveryHashType + from homeassistant.components.mqtt import ( async_subscribe_connection_status, is_connected as mqtt_connected, @@ -11,7 +18,7 @@ from homeassistant.components.mqtt import ( from homeassistant.core import callback from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC from homeassistant.helpers.dispatcher import async_dispatcher_connect -from homeassistant.helpers.entity import Entity +from homeassistant.helpers.entity import DeviceInfo, Entity from .discovery import ( TASMOTA_DISCOVERY_ENTITY_UPDATED, @@ -25,48 +32,50 @@ _LOGGER = logging.getLogger(__name__) class TasmotaEntity(Entity): """Base class for Tasmota entities.""" - def __init__(self, tasmota_entity) -> None: + def __init__(self, tasmota_entity: HATasmotaEntity) -> None: """Initialize.""" self._tasmota_entity = tasmota_entity self._unique_id = tasmota_entity.unique_id - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Subscribe to MQTT events.""" await self._subscribe_topics() - async def async_will_remove_from_hass(self): + async def async_will_remove_from_hass(self) -> None: """Unsubscribe when removed.""" await self._tasmota_entity.unsubscribe_topics() await super().async_will_remove_from_hass() - async def discovery_update(self, update, write_state=True): + async def discovery_update( + self, update: TasmotaEntityConfig, write_state: bool = True + ) -> None: """Handle updated discovery message.""" self._tasmota_entity.config_update(update) await self._subscribe_topics() if write_state: self.async_write_ha_state() - async def _subscribe_topics(self): + async def _subscribe_topics(self) -> None: """(Re)Subscribe to topics.""" await self._tasmota_entity.subscribe_topics() @property - def device_info(self): + def device_info(self) -> DeviceInfo: """Return a device description for device registry.""" return {"connections": {(CONNECTION_NETWORK_MAC, self._tasmota_entity.mac)}} @property - def name(self): + def name(self) -> str | None: """Return the name of the binary sensor.""" return self._tasmota_entity.name @property - def should_poll(self): + def should_poll(self) -> bool: """Return the polling state.""" return False @property - def unique_id(self): + def unique_id(self) -> str: """Return a unique ID.""" return self._unique_id @@ -99,7 +108,9 @@ class TasmotaOnOffEntity(TasmotaEntity): class TasmotaAvailability(TasmotaEntity): """Mixin used for platforms that report availability.""" - def __init__(self, **kwds) -> None: + _tasmota_entity: HATasmotaAvailability + + def __init__(self, **kwds: Any) -> None: """Initialize the availability mixin.""" self._available = False super().__init__(**kwds) @@ -120,7 +131,7 @@ class TasmotaAvailability(TasmotaEntity): self.async_write_ha_state() @callback - def async_mqtt_connected(self, _): + def async_mqtt_connected(self, _: bool) -> None: """Update state on connection/disconnection to MQTT broker.""" if not self.hass.is_stopping: if not mqtt_connected(self.hass): @@ -136,7 +147,7 @@ class TasmotaAvailability(TasmotaEntity): class TasmotaDiscoveryUpdate(TasmotaEntity): """Mixin used to handle updated discovery message.""" - def __init__(self, discovery_hash, **kwds) -> None: + def __init__(self, discovery_hash: DiscoveryHashType, **kwds: Any) -> None: """Initialize the discovery update mixin.""" self._discovery_hash = discovery_hash self._removed_from_hass = False @@ -147,7 +158,7 @@ class TasmotaDiscoveryUpdate(TasmotaEntity): self._removed_from_hass = False await super().async_added_to_hass() - async def discovery_callback(config): + async def discovery_callback(config: TasmotaEntityConfig) -> None: """Handle discovery update.""" _LOGGER.debug( "Got update for entity with hash: %s '%s'", diff --git a/homeassistant/components/tasmota/sensor.py b/homeassistant/components/tasmota/sensor.py index fa4a8270cc7..87b81322799 100644 --- a/homeassistant/components/tasmota/sensor.py +++ b/homeassistant/components/tasmota/sensor.py @@ -1,12 +1,17 @@ """Support for Tasmota sensors.""" from __future__ import annotations +from datetime import datetime import logging +from typing import Any from hatasmota import const as hc, sensor as tasmota_sensor, status_sensor +from hatasmota.entity import TasmotaEntity as HATasmotaEntity +from hatasmota.models import DiscoveryHashType from homeassistant.components import sensor from homeassistant.components.sensor import STATE_CLASS_MEASUREMENT, SensorEntity +from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( CONCENTRATION_MICROGRAMS_PER_CUBIC_METER, CONCENTRATION_PARTS_PER_BILLION, @@ -41,8 +46,9 @@ from homeassistant.const import ( TEMP_KELVIN, VOLT, ) -from homeassistant.core import callback +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util import dt as dt_util from .const import DATA_REMOVE_DISCOVER_COMPONENT @@ -150,10 +156,17 @@ SENSOR_UNIT_MAP = { } -async def async_setup_entry(hass, config_entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Set up Tasmota sensor dynamically through discovery.""" - async def async_discover_sensor(tasmota_entity, discovery_hash): + @callback + def async_discover( + tasmota_entity: HATasmotaEntity, discovery_hash: DiscoveryHashType + ) -> None: """Discover and add a Tasmota sensor.""" async_add_entities( [ @@ -168,7 +181,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): ] = async_dispatcher_connect( hass, TASMOTA_DISCOVERY_ENTITY_NEW.format(sensor.DOMAIN), - async_discover_sensor, + async_discover, ) @@ -178,9 +191,10 @@ class TasmotaSensor(TasmotaAvailability, TasmotaDiscoveryUpdate, SensorEntity): _attr_last_reset = None _tasmota_entity: tasmota_sensor.TasmotaSensor - def __init__(self, **kwds): + def __init__(self, **kwds: Any) -> None: """Initialize the Tasmota sensor.""" - self._state = None + self._state: Any | None = None + self._state_timestamp: datetime | None = None super().__init__( **kwds, @@ -192,14 +206,16 @@ class TasmotaSensor(TasmotaAvailability, TasmotaDiscoveryUpdate, SensorEntity): await super().async_added_to_hass() @callback - def sensor_state_updated(self, state, **kwargs): + def sensor_state_updated(self, state: Any, **kwargs: Any) -> None: """Handle state updates.""" - self._state = state + if self.device_class == DEVICE_CLASS_TIMESTAMP: + self._state_timestamp = state + else: + self._state = state if "last_reset" in kwargs: try: - last_reset = dt_util.as_utc( - dt_util.parse_datetime(kwargs["last_reset"]) - ) + last_reset_dt = dt_util.parse_datetime(kwargs["last_reset"]) + last_reset = dt_util.as_utc(last_reset_dt) if last_reset_dt else None if last_reset is None: raise ValueError self._attr_last_reset = last_reset @@ -234,7 +250,7 @@ class TasmotaSensor(TasmotaAvailability, TasmotaDiscoveryUpdate, SensorEntity): return True @property - def icon(self): + def icon(self) -> str | None: """Return the icon.""" class_or_icon = SENSOR_DEVICE_CLASS_ICON_MAP.get( self._tasmota_entity.quantity, {} @@ -242,18 +258,18 @@ class TasmotaSensor(TasmotaAvailability, TasmotaDiscoveryUpdate, SensorEntity): return class_or_icon.get(ICON) @property - def state(self): + def state(self) -> str | None: """Return the state of the entity.""" - if self._state and self.device_class == DEVICE_CLASS_TIMESTAMP: - return self._state.isoformat() + if self._state_timestamp and self.device_class == DEVICE_CLASS_TIMESTAMP: + return self._state_timestamp.isoformat() return self._state @property - def force_update(self): + def force_update(self) -> bool: """Force update.""" return True @property - def unit_of_measurement(self): + def unit_of_measurement(self) -> str | None: """Return the unit this state is expressed in.""" return SENSOR_UNIT_MAP.get(self._tasmota_entity.unit, self._tasmota_entity.unit) diff --git a/homeassistant/components/tasmota/switch.py b/homeassistant/components/tasmota/switch.py index f7fa67bed22..50319abac56 100644 --- a/homeassistant/components/tasmota/switch.py +++ b/homeassistant/components/tasmota/switch.py @@ -1,20 +1,33 @@ """Support for Tasmota switches.""" +from typing import Any + +from hatasmota import relay as tasmota_relay +from hatasmota.entity import TasmotaEntity as HATasmotaEntity +from hatasmota.models import DiscoveryHashType from homeassistant.components import switch from homeassistant.components.switch import SwitchEntity -from homeassistant.core import callback +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.dispatcher import async_dispatcher_connect +from homeassistant.helpers.entity_platform import AddEntitiesCallback from .const import DATA_REMOVE_DISCOVER_COMPONENT from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW from .mixins import TasmotaAvailability, TasmotaDiscoveryUpdate, TasmotaOnOffEntity -async def async_setup_entry(hass, config_entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Set up Tasmota switch dynamically through discovery.""" @callback - def async_discover(tasmota_entity, discovery_hash): + def async_discover( + tasmota_entity: HATasmotaEntity, discovery_hash: DiscoveryHashType + ) -> None: """Discover and add a Tasmota switch.""" async_add_entities( [ @@ -41,18 +54,12 @@ class TasmotaSwitch( ): """Representation of a Tasmota switch.""" - def __init__(self, **kwds): - """Initialize the Tasmota switch.""" - self._state = False + _tasmota_entity: tasmota_relay.TasmotaRelay - super().__init__( - **kwds, - ) - - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Turn the device on.""" self._tasmota_entity.set_state(True) - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Turn the device off.""" self._tasmota_entity.set_state(False) diff --git a/tests/components/tasmota/test_light.py b/tests/components/tasmota/test_light.py index e1ba2615742..411567208db 100644 --- a/tests/components/tasmota/test_light.py +++ b/tests/components/tasmota/test_light.py @@ -384,7 +384,7 @@ async def test_controlling_state_via_mqtt_ct(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("color_mode") == "color_temp" async_fire_mqtt_message( @@ -402,7 +402,7 @@ async def test_controlling_state_via_mqtt_ct(hass, mqtt_mock, setup_tasmota): state = hass.states.get("light.test") assert state.state == STATE_ON assert state.attributes.get("color_temp") == 300 - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("color_mode") == "color_temp" @@ -446,7 +446,7 @@ async def test_controlling_state_via_mqtt_rgbw(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("color_mode") == "hs" async_fire_mqtt_message( @@ -454,7 +454,7 @@ async def test_controlling_state_via_mqtt_rgbw(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 191.25 + assert state.attributes.get("brightness") == 191 assert state.attributes.get("color_mode") == "white" async_fire_mqtt_message( @@ -464,7 +464,7 @@ async def test_controlling_state_via_mqtt_rgbw(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("hs_color") == (30, 100) assert state.attributes.get("color_mode") == "hs" @@ -473,7 +473,7 @@ async def test_controlling_state_via_mqtt_rgbw(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("rgb_color") is None assert state.attributes.get("color_mode") == "white" @@ -544,7 +544,7 @@ async def test_controlling_state_via_mqtt_rgbww(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("color_mode") == "color_temp" async_fire_mqtt_message( @@ -645,7 +645,7 @@ async def test_controlling_state_via_mqtt_rgbww_tuya(hass, mqtt_mock, setup_tasm ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("color_mode") == "color_temp" async_fire_mqtt_message( @@ -1216,7 +1216,7 @@ async def test_transition(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 # Dim the light from 50->0: Speed should be 6*2*2=24 await common.async_turn_off(hass, "light.test", transition=6) @@ -1254,7 +1254,7 @@ async def test_transition(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("rgb_color") == (0, 255, 0) # Set color of the light from 0,255,0 to 255,0,0 @ 50%: Speed should be 6*2*2=24 @@ -1296,7 +1296,7 @@ async def test_transition(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("color_temp") == 153 # Set color_temp of the light from 153 to 500 @ 50%: Speed should be 6*2*2=24 @@ -1315,7 +1315,7 @@ async def test_transition(hass, mqtt_mock, setup_tasmota): ) state = hass.states.get("light.test") assert state.state == STATE_ON - assert state.attributes.get("brightness") == 127.5 + assert state.attributes.get("brightness") == 128 assert state.attributes.get("color_temp") == 500 # Set color_temp of the light from 500 to 326 @ 50%: Speed should be 6*2*2*2=48->40