core/homeassistant/components/switcher_kis/button.py

146 lines
5.1 KiB
Python

"""Switcher integration Button platform."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
from typing import Any, cast
from aioswitcher.api import (
DeviceState,
SwitcherApi,
SwitcherBaseResponse,
ThermostatSwing,
)
from aioswitcher.api.remotes import SwitcherBreezeRemote
from aioswitcher.device import DeviceCategory
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SwitcherConfigEntry
from .const import SIGNAL_DEVICE_ADD
from .coordinator import SwitcherDataUpdateCoordinator
from .entity import SwitcherEntity
from .utils import get_breeze_remote_manager
@dataclass(frozen=True, kw_only=True)
class SwitcherThermostatButtonEntityDescription(ButtonEntityDescription):
"""Class to describe a Switcher Thermostat Button entity."""
press_fn: Callable[
[SwitcherApi, SwitcherBreezeRemote],
Coroutine[Any, Any, SwitcherBaseResponse],
]
supported: Callable[[SwitcherBreezeRemote], bool]
THERMOSTAT_BUTTONS = [
SwitcherThermostatButtonEntityDescription(
key="assume_on",
translation_key="assume_on",
entity_category=EntityCategory.CONFIG,
press_fn=lambda api, remote: api.control_breeze_device(
remote, state=DeviceState.ON, update_state=True
),
supported=lambda _: True,
),
SwitcherThermostatButtonEntityDescription(
key="assume_off",
translation_key="assume_off",
entity_category=EntityCategory.CONFIG,
press_fn=lambda api, remote: api.control_breeze_device(
remote, state=DeviceState.OFF, update_state=True
),
supported=lambda _: True,
),
SwitcherThermostatButtonEntityDescription(
key="vertical_swing_on",
translation_key="vertical_swing_on",
press_fn=lambda api, remote: api.control_breeze_device(
remote, swing=ThermostatSwing.ON
),
supported=lambda remote: bool(remote.separated_swing_command),
),
SwitcherThermostatButtonEntityDescription(
key="vertical_swing_off",
translation_key="vertical_swing_off",
press_fn=lambda api, remote: api.control_breeze_device(
remote, swing=ThermostatSwing.OFF
),
supported=lambda remote: bool(remote.separated_swing_command),
),
]
async def async_setup_entry(
hass: HomeAssistant,
config_entry: SwitcherConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Switcher button from config entry."""
async def async_add_buttons(coordinator: SwitcherDataUpdateCoordinator) -> None:
"""Get remote and add button from Switcher device."""
data = cast(SwitcherBreezeRemote, coordinator.data)
if coordinator.data.device_type.category == DeviceCategory.THERMOSTAT:
remote: SwitcherBreezeRemote = await hass.async_add_executor_job(
get_breeze_remote_manager(hass).get_remote, data.remote_id
)
async_add_entities(
SwitcherThermostatButtonEntity(coordinator, description, remote)
for description in THERMOSTAT_BUTTONS
if description.supported(remote)
)
config_entry.async_on_unload(
async_dispatcher_connect(hass, SIGNAL_DEVICE_ADD, async_add_buttons)
)
class SwitcherThermostatButtonEntity(SwitcherEntity, ButtonEntity):
"""Representation of a Switcher climate entity."""
entity_description: SwitcherThermostatButtonEntityDescription
def __init__(
self,
coordinator: SwitcherDataUpdateCoordinator,
description: SwitcherThermostatButtonEntityDescription,
remote: SwitcherBreezeRemote,
) -> None:
"""Initialize the entity."""
super().__init__(coordinator)
self.entity_description = description
self._remote = remote
self._attr_unique_id = f"{coordinator.mac_address}-{description.key}"
async def async_press(self) -> None:
"""Press the button."""
response: SwitcherBaseResponse | None = None
error = None
try:
async with SwitcherApi(
self.coordinator.data.device_type,
self.coordinator.data.ip_address,
self.coordinator.data.device_id,
self.coordinator.data.device_key,
) as swapi:
response = await self.entity_description.press_fn(swapi, self._remote)
except (TimeoutError, OSError, RuntimeError) as err:
error = repr(err)
if error or not response or not response.successful:
self.coordinator.last_update_success = False
self.async_write_ha_state()
raise HomeAssistantError(
f"Call api for {self.name} failed, response/error: {response or error}"
)