mirror of https://github.com/home-assistant/core
154 lines
4.7 KiB
Python
154 lines
4.7 KiB
Python
"""Support for fan entities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from thinqconnect import DeviceType
|
|
from thinqconnect.integration import ExtendedProperty
|
|
|
|
from homeassistant.components.fan import (
|
|
FanEntity,
|
|
FanEntityDescription,
|
|
FanEntityFeature,
|
|
)
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
from homeassistant.util.percentage import (
|
|
ordered_list_item_to_percentage,
|
|
percentage_to_ordered_list_item,
|
|
)
|
|
|
|
from . import ThinqConfigEntry
|
|
from .coordinator import DeviceDataUpdateCoordinator
|
|
from .entity import ThinQEntity
|
|
|
|
DEVICE_TYPE_FAN_MAP: dict[DeviceType, tuple[FanEntityDescription, ...]] = {
|
|
DeviceType.CEILING_FAN: (
|
|
FanEntityDescription(
|
|
key=ExtendedProperty.FAN,
|
|
name=None,
|
|
),
|
|
),
|
|
}
|
|
|
|
FOUR_STEP_SPEEDS = ["low", "mid", "high", "turbo"]
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: ThinqConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up an entry for fan platform."""
|
|
entities: list[ThinQFanEntity] = []
|
|
for coordinator in entry.runtime_data.coordinators.values():
|
|
if (
|
|
descriptions := DEVICE_TYPE_FAN_MAP.get(coordinator.api.device.device_type)
|
|
) is not None:
|
|
for description in descriptions:
|
|
entities.extend(
|
|
ThinQFanEntity(coordinator, description, property_id)
|
|
for property_id in coordinator.api.get_active_idx(description.key)
|
|
)
|
|
|
|
if entities:
|
|
async_add_entities(entities)
|
|
|
|
|
|
class ThinQFanEntity(ThinQEntity, FanEntity):
|
|
"""Represent a thinq fan platform."""
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: DeviceDataUpdateCoordinator,
|
|
entity_description: FanEntityDescription,
|
|
property_id: str,
|
|
) -> None:
|
|
"""Initialize fan platform."""
|
|
super().__init__(coordinator, entity_description, property_id)
|
|
|
|
self._ordered_named_fan_speeds = []
|
|
self._attr_supported_features = (
|
|
FanEntityFeature.SET_SPEED
|
|
| FanEntityFeature.TURN_ON
|
|
| FanEntityFeature.TURN_OFF
|
|
)
|
|
if (fan_modes := self.data.fan_modes) is not None:
|
|
self._attr_speed_count = len(fan_modes)
|
|
if self.speed_count == 4:
|
|
self._ordered_named_fan_speeds = FOUR_STEP_SPEEDS
|
|
|
|
def _update_status(self) -> None:
|
|
"""Update status itself."""
|
|
super()._update_status()
|
|
|
|
# Update power on state.
|
|
self._attr_is_on = self.data.is_on
|
|
|
|
# Update fan speed.
|
|
if (
|
|
self.data.is_on
|
|
and (mode := self.data.fan_mode) in self._ordered_named_fan_speeds
|
|
):
|
|
self._attr_percentage = ordered_list_item_to_percentage(
|
|
self._ordered_named_fan_speeds, mode
|
|
)
|
|
else:
|
|
self._attr_percentage = 0
|
|
|
|
_LOGGER.debug(
|
|
"[%s:%s] update status: %s -> %s (percentage=%s)",
|
|
self.coordinator.device_name,
|
|
self.property_id,
|
|
self.data.is_on,
|
|
self.is_on,
|
|
self.percentage,
|
|
)
|
|
|
|
async def async_set_percentage(self, percentage: int) -> None:
|
|
"""Set the speed percentage of the fan."""
|
|
if percentage == 0:
|
|
await self.async_turn_off()
|
|
return
|
|
try:
|
|
value = percentage_to_ordered_list_item(
|
|
self._ordered_named_fan_speeds, percentage
|
|
)
|
|
except ValueError:
|
|
_LOGGER.exception("Failed to async_set_percentage")
|
|
return
|
|
|
|
_LOGGER.debug(
|
|
"[%s:%s] async_set_percentage. percentage=%s, value=%s",
|
|
self.coordinator.device_name,
|
|
self.property_id,
|
|
percentage,
|
|
value,
|
|
)
|
|
await self.async_call_api(
|
|
self.coordinator.api.async_set_fan_mode(self.property_id, value)
|
|
)
|
|
|
|
async def async_turn_on(
|
|
self,
|
|
percentage: int | None = None,
|
|
preset_mode: str | None = None,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""Turn on the fan."""
|
|
_LOGGER.debug(
|
|
"[%s:%s] async_turn_on", self.coordinator.device_name, self.property_id
|
|
)
|
|
await self.async_call_api(self.coordinator.api.async_turn_on(self.property_id))
|
|
|
|
async def async_turn_off(self, **kwargs: Any) -> None:
|
|
"""Turn the fan off."""
|
|
_LOGGER.debug(
|
|
"[%s:%s] async_turn_off", self.coordinator.device_name, self.property_id
|
|
)
|
|
await self.async_call_api(self.coordinator.api.async_turn_off(self.property_id))
|