Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve typing of Tasmota (3/3) (#52748)

pull/52348/head^2
Erik Montnemery 2021-07-12 19:17:44 +02:00 committed by GitHub
parent 1a74fd7a14
commit 2e44e256f0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 142 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
homeassistant/components/tasmota/light.py
View File

@ -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"}

39
homeassistant/components/tasmota/mixins.py
View File

@ -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'",

50
homeassistant/components/tasmota/sensor.py
View File

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

31
homeassistant/components/tasmota/switch.py
View File

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

24
tests/components/tasmota/test_light.py
View File

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