Files

112 lines
3.7 KiB
Python

"""Support for AirGradient buttons."""
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import override
from airgradient import AirGradientClient, ConfigurationControl
from homeassistant.components.button import (
DOMAIN as BUTTON_DOMAIN,
ButtonEntity,
ButtonEntityDescription,
)
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import AirGradientConfigEntry
from .const import DOMAIN, supports_action
from .coordinator import AirGradientCoordinator
from .entity import AirGradientEntity, exception_handler
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class AirGradientButtonEntityDescription(ButtonEntityDescription):
"""Describes AirGradient button entity."""
press_fn: Callable[[AirGradientClient], Awaitable[None]]
CO2_CALIBRATION = AirGradientButtonEntityDescription(
key="co2_calibration",
translation_key="co2_calibration",
entity_category=EntityCategory.CONFIG,
press_fn=lambda client: client.request_co2_calibration(),
)
LED_BAR_TEST = AirGradientButtonEntityDescription(
key="led_bar_test",
translation_key="led_bar_test",
entity_category=EntityCategory.CONFIG,
press_fn=lambda client: client.request_led_bar_test(),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: AirGradientConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up AirGradient button entities based on a config entry."""
coordinator = entry.runtime_data
model = coordinator.data.measures.model
descriptions = (CO2_CALIBRATION, LED_BAR_TEST)
descriptions_by_key = {description.key: description for description in descriptions}
added_entities: set[str] = set()
@callback
def _check_entities() -> None:
nonlocal added_entities
desired_entities = {
description.key
for description in descriptions
if coordinator.data.config.configuration_control
is ConfigurationControl.LOCAL
and supports_action(model, description.key)
}
if entities_to_add := desired_entities - added_entities:
async_add_entities(
[
AirGradientButton(coordinator, descriptions_by_key[key])
for key in entities_to_add
]
)
if entities_to_remove := added_entities - desired_entities:
entity_registry = er.async_get(hass)
for key in entities_to_remove:
unique_id = f"{coordinator.serial_number}-{key}"
if entity_id := entity_registry.async_get_entity_id(
BUTTON_DOMAIN, DOMAIN, unique_id
):
entity_registry.async_remove(entity_id)
added_entities = desired_entities
coordinator.async_add_listener(_check_entities)
_check_entities()
class AirGradientButton(AirGradientEntity, ButtonEntity):
"""Defines an AirGradient button."""
entity_description: AirGradientButtonEntityDescription
def __init__(
self,
coordinator: AirGradientCoordinator,
description: AirGradientButtonEntityDescription,
) -> None:
"""Initialize airgradient button."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.serial_number}-{description.key}"
@exception_handler
@override
async def async_press(self) -> None:
"""Press the button."""
await self.entity_description.press_fn(self.coordinator.client)