Improve typing of Tasmota (2/3) (#52747)

* Improve typing of Tasmota (2/3)

* Add more typing, add TasmotaOnOffEntity

* Address review comments
This commit is contained in:
Erik Montnemery 2021-07-12 18:27:11 +02:00 committed by GitHub
parent 98109caee9
commit 8c812bc25c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 241 additions and 95 deletions

View file

@ -1,9 +1,19 @@
"""Support for Tasmota binary sensors."""
from __future__ import annotations
from datetime import datetime
from typing import Any, Callable
from hatasmota import switch as tasmota_switch
from hatasmota.entity import TasmotaEntity as HATasmotaEntity
from hatasmota.models import DiscoveryHashType
from homeassistant.components import binary_sensor
from homeassistant.components.binary_sensor import BinarySensorEntity
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
import homeassistant.helpers.event as evt
from .const import DATA_REMOVE_DISCOVER_COMPONENT
@ -11,11 +21,17 @@ from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW
from .mixins import TasmotaAvailability, TasmotaDiscoveryUpdate
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 binary sensor 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 binary sensor."""
async_add_entities(
[
@ -41,33 +57,40 @@ class TasmotaBinarySensor(
):
"""Representation a Tasmota binary sensor."""
def __init__(self, **kwds):
_tasmota_entity: tasmota_switch.TasmotaSwitch
def __init__(self, **kwds: Any) -> None:
"""Initialize the Tasmota binary sensor."""
self._delay_listener = None
self._state = None
self._delay_listener: Callable | None = None
self._on_off_state: bool | None = None
super().__init__(
**kwds,
)
async def async_added_to_hass(self) -> None:
"""Subscribe to MQTT events."""
self._tasmota_entity.set_on_state_callback(self.on_off_state_updated)
await super().async_added_to_hass()
@callback
def off_delay_listener(self, now):
def off_delay_listener(self, now: datetime) -> None:
"""Switch device off after a delay."""
self._delay_listener = None
self._state = False
self._on_off_state = False
self.async_write_ha_state()
@callback
def state_updated(self, state, **kwargs):
def on_off_state_updated(self, state: bool, **kwargs: Any) -> None:
"""Handle state updates."""
self._state = state
self._on_off_state = state
if self._delay_listener is not None:
self._delay_listener()
self._delay_listener = None
off_delay = self._tasmota_entity.off_delay
if self._state and off_delay is not None:
if self._on_off_state and off_delay is not None:
self._delay_listener = evt.async_call_later(
self.hass, off_delay, self.off_delay_listener
)
@ -75,11 +98,11 @@ class TasmotaBinarySensor(
self.async_write_ha_state()
@property
def force_update(self):
def force_update(self) -> bool:
"""Force update."""
return True
@property
def is_on(self):
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
return self._state
return self._on_off_state

View file

@ -1,22 +1,35 @@
"""Support for Tasmota covers."""
from __future__ import annotations
from hatasmota import const as tasmota_const
from typing import Any
from hatasmota import const as tasmota_const, shutter as tasmota_shutter
from hatasmota.entity import TasmotaEntity as HATasmotaEntity
from hatasmota.models import DiscoveryHashType
from homeassistant.components import cover
from homeassistant.components.cover import CoverEntity
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
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 cover 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 cover."""
async_add_entities(
[TasmotaCover(tasmota_entity=tasmota_entity, discovery_hash=discovery_hash)]
@ -38,24 +51,31 @@ class TasmotaCover(
):
"""Representation of a Tasmota cover."""
def __init__(self, **kwds):
_tasmota_entity: tasmota_shutter.TasmotaShutter
def __init__(self, **kwds: Any) -> None:
"""Initialize the Tasmota cover."""
self._direction = None
self._position = None
self._direction: int | None = None
self._position: int | None = None
super().__init__(
**kwds,
)
async def async_added_to_hass(self) -> None:
"""Subscribe to MQTT events."""
self._tasmota_entity.set_on_state_callback(self.cover_state_updated)
await super().async_added_to_hass()
@callback
def state_updated(self, state, **kwargs):
def cover_state_updated(self, state: bool, **kwargs: Any) -> None:
"""Handle state updates."""
self._direction = kwargs["direction"]
self._position = kwargs["position"]
self.async_write_ha_state()
@property
def current_cover_position(self):
def current_cover_position(self) -> int | None:
"""Return current position of cover.
None is unknown, 0 is closed, 100 is fully open.
@ -63,7 +83,7 @@ class TasmotaCover(
return self._position
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag supported features."""
return (
cover.SUPPORT_OPEN
@ -73,35 +93,35 @@ class TasmotaCover(
)
@property
def is_opening(self):
def is_opening(self) -> bool:
"""Return if the cover is opening or not."""
return self._direction == tasmota_const.SHUTTER_DIRECTION_UP
@property
def is_closing(self):
def is_closing(self) -> bool:
"""Return if the cover is closing or not."""
return self._direction == tasmota_const.SHUTTER_DIRECTION_DOWN
@property
def is_closed(self):
def is_closed(self) -> bool | None:
"""Return if the cover is closed or not."""
if self._position is None:
return None
return self._position == 0
async def async_open_cover(self, **kwargs):
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
self._tasmota_entity.open()
async def async_close_cover(self, **kwargs):
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close cover."""
self._tasmota_entity.close()
async def async_set_cover_position(self, **kwargs):
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""
position = kwargs[cover.ATTR_POSITION]
self._tasmota_entity.set_position(position)
async def async_stop_cover(self, **kwargs):
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
self._tasmota_entity.stop()

View file

@ -1,7 +1,11 @@
"""Provides device automations for Tasmota."""
from hatasmota.const import AUTOMATION_TYPE_TRIGGER
from hatasmota.models import DiscoveryHashType
from hatasmota.trigger import TasmotaTrigger
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import Event, HomeAssistant
from homeassistant.helpers.device_registry import EVENT_DEVICE_REGISTRY_UPDATED
from homeassistant.helpers.dispatcher import async_dispatcher_connect
@ -10,21 +14,23 @@ from .const import DATA_REMOVE_DISCOVER_COMPONENT, DATA_UNSUB
from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW
async def async_remove_automations(hass, device_id):
async def async_remove_automations(hass: HomeAssistant, device_id: str) -> None:
"""Remove automations for a Tasmota device."""
await device_trigger.async_remove_triggers(hass, device_id)
async def async_setup_entry(hass, config_entry):
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Set up Tasmota device automation dynamically through discovery."""
async def async_device_removed(event):
async def async_device_removed(event: Event) -> None:
"""Handle the removal of a device."""
if event.data["action"] != "remove":
return
await async_remove_automations(hass, event.data["device_id"])
async def async_discover(tasmota_automation, discovery_hash):
async def async_discover(
tasmota_automation: TasmotaTrigger, discovery_hash: DiscoveryHashType
) -> None:
"""Discover and add a Tasmota device automation."""
if tasmota_automation.automation_type == AUTOMATION_TYPE_TRIGGER:
await device_trigger.async_setup_trigger(

View file

@ -5,12 +5,14 @@ import logging
from typing import Callable
import attr
from hatasmota.trigger import TasmotaTrigger
from hatasmota.models import DiscoveryHashType
from hatasmota.trigger import TasmotaTrigger, TasmotaTriggerConfig
import voluptuous as vol
from homeassistant.components.automation import AutomationActionType
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
from homeassistant.components.homeassistant.triggers import event as event_trigger
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
@ -51,8 +53,9 @@ class TriggerInstance:
trigger: Trigger = attr.ib()
remove: CALLBACK_TYPE | None = attr.ib(default=None)
async def async_attach_trigger(self):
async def async_attach_trigger(self) -> None:
"""Attach event trigger."""
assert self.trigger.tasmota_trigger is not None
event_config = {
event_trigger.CONF_PLATFORM: "event",
event_trigger.CONF_EVENT_TYPE: TASMOTA_EVENT,
@ -81,15 +84,17 @@ class Trigger:
"""Device trigger settings."""
device_id: str = attr.ib()
discovery_hash: dict | None = attr.ib()
discovery_hash: DiscoveryHashType | None = attr.ib()
hass: HomeAssistant = attr.ib()
remove_update_signal: Callable[[], None] | None = attr.ib()
subtype: str = attr.ib()
tasmota_trigger: TasmotaTrigger = attr.ib()
tasmota_trigger: TasmotaTrigger | None = attr.ib()
type: str = attr.ib()
trigger_instances: list[TriggerInstance] = attr.ib(factory=list)
async def add_trigger(self, action, automation_info):
async def add_trigger(
self, action: AutomationActionType, automation_info: dict
) -> Callable[[], None]:
"""Add Tasmota trigger."""
instance = TriggerInstance(action, automation_info, self)
self.trigger_instances.append(instance)
@ -110,7 +115,7 @@ class Trigger:
return async_remove
def detach_trigger(self):
def detach_trigger(self) -> None:
"""Remove Tasmota device trigger."""
# Mark trigger as unknown
self.tasmota_trigger = None
@ -121,11 +126,12 @@ class Trigger:
trig.remove()
trig.remove = None
async def arm_tasmota_trigger(self):
async def arm_tasmota_trigger(self) -> None:
"""Arm Tasmota trigger: subscribe to MQTT topics and fire events."""
@callback
def _on_trigger():
def _on_trigger() -> None:
assert self.tasmota_trigger is not None
data = {
"mac": self.tasmota_trigger.cfg.mac,
"source": self.tasmota_trigger.cfg.subtype,
@ -136,10 +142,13 @@ class Trigger:
data,
)
assert self.tasmota_trigger is not None
self.tasmota_trigger.set_on_trigger_callback(_on_trigger)
await self.tasmota_trigger.subscribe_topics()
async def set_tasmota_trigger(self, tasmota_trigger, remove_update_signal):
async def set_tasmota_trigger(
self, tasmota_trigger: TasmotaTrigger, remove_update_signal: Callable[[], None]
) -> None:
"""Set Tasmota trigger."""
await self.update_tasmota_trigger(tasmota_trigger.cfg, remove_update_signal)
self.tasmota_trigger = tasmota_trigger
@ -147,22 +156,31 @@ class Trigger:
for trig in self.trigger_instances:
await trig.async_attach_trigger()
async def update_tasmota_trigger(self, tasmota_trigger_cfg, remove_update_signal):
async def update_tasmota_trigger(
self,
tasmota_trigger_cfg: TasmotaTriggerConfig,
remove_update_signal: Callable[[], None],
) -> None:
"""Update Tasmota trigger."""
self.remove_update_signal = remove_update_signal
self.type = tasmota_trigger_cfg.type
self.subtype = tasmota_trigger_cfg.subtype
async def async_setup_trigger(hass, tasmota_trigger, config_entry, discovery_hash):
async def async_setup_trigger(
hass: HomeAssistant,
tasmota_trigger: TasmotaTrigger,
config_entry: ConfigEntry,
discovery_hash: DiscoveryHashType,
) -> None:
"""Set up a discovered Tasmota device trigger."""
discovery_id = tasmota_trigger.cfg.trigger_id
remove_update_signal = None
remove_update_signal: Callable[[], None] | None = None
_LOGGER.debug(
"Discovered trigger with ID: %s '%s'", discovery_id, tasmota_trigger.cfg
)
async def discovery_update(trigger_config):
async def discovery_update(trigger_config: TasmotaTriggerConfig) -> None:
"""Handle discovery update."""
_LOGGER.debug(
"Got update for trigger with hash: %s '%s'", discovery_hash, trigger_config
@ -175,7 +193,8 @@ async def async_setup_trigger(hass, tasmota_trigger, config_entry, discovery_has
await device_trigger.tasmota_trigger.unsubscribe_topics()
device_trigger.detach_trigger()
clear_discovery_hash(hass, discovery_hash)
remove_update_signal()
if remove_update_signal is not None:
remove_update_signal()
return
device_trigger = hass.data[DEVICE_TRIGGERS][discovery_id]
@ -226,7 +245,7 @@ async def async_setup_trigger(hass, tasmota_trigger, config_entry, discovery_has
await device_trigger.arm_tasmota_trigger()
async def async_remove_triggers(hass: HomeAssistant, device_id: str):
async def async_remove_triggers(hass: HomeAssistant, device_id: str) -> None:
"""Cleanup any device triggers for a Tasmota device."""
triggers = await async_get_triggers(hass, device_id)
for trig in triggers:
@ -287,6 +306,5 @@ async def async_attach_trigger(
subtype=config[CONF_SUBTYPE],
tasmota_trigger=None,
)
return await hass.data[DEVICE_TRIGGERS][discovery_id].add_trigger(
action, automation_info
)
trigger: Trigger = hass.data[DEVICE_TRIGGERS][discovery_id]
return await trigger.add_trigger(action, automation_info)

View file

@ -1,5 +1,8 @@
"""Support for Tasmota device discovery."""
from __future__ import annotations
import logging
from typing import Callable
from hatasmota.discovery import (
TasmotaDiscovery,
@ -10,8 +13,13 @@ from hatasmota.discovery import (
get_triggers as tasmota_get_triggers,
unique_id_from_hash,
)
from hatasmota.entity import TasmotaEntityConfig
from hatasmota.models import DiscoveryHashType, TasmotaDeviceConfig
from hatasmota.mqtt import TasmotaMQTTClient
from hatasmota.sensor import TasmotaBaseSensorConfig
import homeassistant.components.sensor as sensor
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dev_reg
from homeassistant.helpers.dispatcher import async_dispatcher_send
@ -26,8 +34,12 @@ TASMOTA_DISCOVERY_ENTITY_NEW = "tasmota_discovery_entity_new_{}"
TASMOTA_DISCOVERY_ENTITY_UPDATED = "tasmota_discovery_entity_updated_{}_{}_{}_{}"
TASMOTA_DISCOVERY_INSTANCE = "tasmota_discovery_instance"
SetupDeviceCallback = Callable[[TasmotaDeviceConfig, str], None]
def clear_discovery_hash(hass, discovery_hash):
def clear_discovery_hash(
hass: HomeAssistant, discovery_hash: DiscoveryHashType
) -> None:
"""Clear entry in ALREADY_DISCOVERED list."""
if ALREADY_DISCOVERED not in hass.data:
# Discovery is shutting down
@ -35,17 +47,25 @@ def clear_discovery_hash(hass, discovery_hash):
del hass.data[ALREADY_DISCOVERED][discovery_hash]
def set_discovery_hash(hass, discovery_hash):
def set_discovery_hash(hass: HomeAssistant, discovery_hash: DiscoveryHashType) -> None:
"""Set entry in ALREADY_DISCOVERED list."""
hass.data[ALREADY_DISCOVERED][discovery_hash] = {}
async def async_start(
hass: HomeAssistant, discovery_topic, config_entry, tasmota_mqtt, setup_device
hass: HomeAssistant,
discovery_topic: str,
config_entry: ConfigEntry,
tasmota_mqtt: TasmotaMQTTClient,
setup_device: SetupDeviceCallback,
) -> None:
"""Start Tasmota device discovery."""
async def _discover_entity(tasmota_entity_config, discovery_hash, platform):
async def _discover_entity(
tasmota_entity_config: TasmotaEntityConfig | None,
discovery_hash: DiscoveryHashType,
platform: str,
) -> None:
"""Handle adding or updating a discovered entity."""
if not tasmota_entity_config:
# Entity disabled, clean up entity registry
@ -70,6 +90,10 @@ async def async_start(
)
else:
tasmota_entity = tasmota_get_entity(tasmota_entity_config, tasmota_mqtt)
if not tasmota_entity:
_LOGGER.error("Failed to create entity %s %s", platform, discovery_hash)
return
_LOGGER.debug(
"Adding new entity: %s %s %s",
platform,
@ -86,7 +110,7 @@ async def async_start(
discovery_hash,
)
async def async_device_discovered(payload, mac):
async def async_device_discovered(payload: dict, mac: str) -> None:
"""Process the received message."""
if ALREADY_DISCOVERED not in hass.data:
@ -102,7 +126,12 @@ async def async_start(
tasmota_triggers = tasmota_get_triggers(payload)
for trigger_config in tasmota_triggers:
discovery_hash = (mac, "automation", "trigger", trigger_config.trigger_id)
discovery_hash: DiscoveryHashType = (
mac,
"automation",
"trigger",
trigger_config.trigger_id,
)
if discovery_hash in hass.data[ALREADY_DISCOVERED]:
_LOGGER.debug(
"Trigger already added, sending update: %s",
@ -131,7 +160,9 @@ async def async_start(
for (tasmota_entity_config, discovery_hash) in tasmota_entities:
await _discover_entity(tasmota_entity_config, discovery_hash, platform)
async def async_sensors_discovered(sensors, mac):
async def async_sensors_discovered(
sensors: list[tuple[TasmotaBaseSensorConfig, DiscoveryHashType]], mac: str
) -> None:
"""Handle discovery of (additional) sensors."""
platform = sensor.DOMAIN

View file

@ -1,11 +1,18 @@
"""Support for Tasmota fans."""
from __future__ import annotations
from hatasmota import const as tasmota_const
from typing import Any
from hatasmota import const as tasmota_const, fan as tasmota_fan
from hatasmota.entity import TasmotaEntity as HATasmotaEntity
from hatasmota.models import DiscoveryHashType
from homeassistant.components import fan
from homeassistant.components.fan import FanEntity
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 homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item,
@ -22,11 +29,17 @@ ORDERED_NAMED_FAN_SPEEDS = [
] # off is not included
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 fan 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 fan."""
async_add_entities(
[TasmotaFan(tasmota_entity=tasmota_entity, discovery_hash=discovery_hash)]
@ -48,21 +61,34 @@ class TasmotaFan(
):
"""Representation of a Tasmota fan."""
def __init__(self, **kwds):
_tasmota_entity: tasmota_fan.TasmotaFan
def __init__(self, **kwds: Any) -> None:
"""Initialize the Tasmota fan."""
self._state = None
self._state: int | None = None
super().__init__(
**kwds,
)
async def async_added_to_hass(self) -> None:
"""Subscribe to MQTT events."""
self._tasmota_entity.set_on_state_callback(self.fan_state_updated)
await super().async_added_to_hass()
@callback
def fan_state_updated(self, state: int, **kwargs: Any) -> None:
"""Handle state updates."""
self._state = state
self.async_write_ha_state()
@property
def speed_count(self) -> int:
"""Return the number of speeds the fan supports."""
return len(ORDERED_NAMED_FAN_SPEEDS)
@property
def percentage(self):
def percentage(self) -> int | None:
"""Return the current speed percentage."""
if self._state is None:
return None
@ -71,11 +97,11 @@ class TasmotaFan(
return ordered_list_item_to_percentage(ORDERED_NAMED_FAN_SPEEDS, self._state)
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag supported features."""
return fan.SUPPORT_SET_SPEED
async def async_set_percentage(self, percentage):
async def async_set_percentage(self, percentage: int) -> None:
"""Set the speed of the fan."""
if percentage == 0:
await self.async_turn_off()
@ -86,8 +112,12 @@ class TasmotaFan(
self._tasmota_entity.set_speed(tasmota_speed)
async def async_turn_on(
self, speed=None, percentage=None, preset_mode=None, **kwargs
):
self,
speed: str | None = None,
percentage: int | None = None,
preset_mode: str | None = None,
**kwargs: Any,
) -> None:
"""Turn the fan on."""
# Tasmota does not support turning a fan on with implicit speed
await self.async_set_percentage(
@ -97,6 +127,6 @@ class TasmotaFan(
)
)
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the fan off."""
self._tasmota_entity.set_speed(tasmota_const.FAN_SPEED_OFF)

View file

@ -30,7 +30,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import DATA_REMOVE_DISCOVER_COMPONENT
from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW
from .mixins import TasmotaAvailability, TasmotaDiscoveryUpdate
from .mixins import TasmotaAvailability, TasmotaDiscoveryUpdate, TasmotaOnOffEntity
DEFAULT_BRIGHTNESS_MAX = 255
TASMOTA_BRIGHTNESS_MAX = 100
@ -74,6 +74,7 @@ def scale_brightness(brightness):
class TasmotaLight(
TasmotaAvailability,
TasmotaDiscoveryUpdate,
TasmotaOnOffEntity,
LightEntity,
):
"""Representation of a Tasmota light."""
@ -142,7 +143,7 @@ class TasmotaLight(
@callback
def state_updated(self, state, **kwargs):
"""Handle state updates."""
self._state = state
self._on_off_state = state
attributes = kwargs.get("attributes")
if attributes:
if "brightness" in attributes:
@ -222,11 +223,6 @@ class TasmotaLight(
"""Force update."""
return False
@property
def is_on(self):
"""Return true if device is on."""
return self._state
@property
def supported_color_modes(self):
"""Flag supported color modes."""

View file

@ -1,5 +1,8 @@
"""Tasmota entity mixins."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.mqtt import (
async_subscribe_connection_status,
@ -24,13 +27,11 @@ class TasmotaEntity(Entity):
def __init__(self, tasmota_entity) -> None:
"""Initialize."""
self._state = None
self._tasmota_entity = tasmota_entity
self._unique_id = tasmota_entity.unique_id
async def async_added_to_hass(self):
"""Subscribe to MQTT events."""
self._tasmota_entity.set_on_state_callback(self.state_updated)
await self._subscribe_topics()
async def async_will_remove_from_hass(self):
@ -49,12 +50,6 @@ class TasmotaEntity(Entity):
"""(Re)Subscribe to topics."""
await self._tasmota_entity.subscribe_topics()
@callback
def state_updated(self, state, **kwargs):
"""Handle state updates."""
self._state = state
self.async_write_ha_state()
@property
def device_info(self):
"""Return a device description for device registry."""
@ -76,6 +71,31 @@ class TasmotaEntity(Entity):
return self._unique_id
class TasmotaOnOffEntity(TasmotaEntity):
"""Base class for Tasmota entities which can be on or off."""
def __init__(self, **kwds: Any) -> None:
"""Initialize."""
self._on_off_state: bool = False
super().__init__(**kwds)
async def async_added_to_hass(self) -> None:
"""Subscribe to MQTT events."""
self._tasmota_entity.set_on_state_callback(self.state_updated)
await super().async_added_to_hass()
@callback
def state_updated(self, state: bool, **kwargs: Any) -> None:
"""Handle state updates."""
self._on_off_state = state
self.async_write_ha_state()
@property
def is_on(self) -> bool:
"""Return true if device is on."""
return self._on_off_state
class TasmotaAvailability(TasmotaEntity):
"""Mixin used for platforms that report availability."""

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import logging
from hatasmota import const as hc, status_sensor
from hatasmota import const as hc, sensor as tasmota_sensor, status_sensor
from homeassistant.components import sensor
from homeassistant.components.sensor import STATE_CLASS_MEASUREMENT, SensorEntity
@ -176,6 +176,7 @@ class TasmotaSensor(TasmotaAvailability, TasmotaDiscoveryUpdate, SensorEntity):
"""Representation of a Tasmota sensor."""
_attr_last_reset = None
_tasmota_entity: tasmota_sensor.TasmotaSensor
def __init__(self, **kwds):
"""Initialize the Tasmota sensor."""
@ -185,8 +186,13 @@ class TasmotaSensor(TasmotaAvailability, TasmotaDiscoveryUpdate, SensorEntity):
**kwds,
)
async def async_added_to_hass(self) -> None:
"""Subscribe to MQTT events."""
self._tasmota_entity.set_on_state_callback(self.sensor_state_updated)
await super().async_added_to_hass()
@callback
def state_updated(self, state, **kwargs):
def sensor_state_updated(self, state, **kwargs):
"""Handle state updates."""
self._state = state
if "last_reset" in kwargs:

View file

@ -7,7 +7,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import DATA_REMOVE_DISCOVER_COMPONENT
from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW
from .mixins import TasmotaAvailability, TasmotaDiscoveryUpdate
from .mixins import TasmotaAvailability, TasmotaDiscoveryUpdate, TasmotaOnOffEntity
async def async_setup_entry(hass, config_entry, async_add_entities):
@ -36,6 +36,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class TasmotaSwitch(
TasmotaAvailability,
TasmotaDiscoveryUpdate,
TasmotaOnOffEntity,
SwitchEntity,
):
"""Representation of a Tasmota switch."""
@ -48,11 +49,6 @@ class TasmotaSwitch(
**kwds,
)
@property
def is_on(self):
"""Return true if device is on."""
return self._state
async def async_turn_on(self, **kwargs):
"""Turn the device on."""
self._tasmota_entity.set_state(True)