forked from home-assistant/core
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add cover platform to switchbot (home-assistant#56414)
Co-authored-by: J. Nick Koston <[email protected]>
- Loading branch information
Showing
10 changed files
with
323 additions
and
101 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,141 @@ | ||
| """Support for SwitchBot curtains.""" | ||
| from __future__ import annotations | ||
|
|
||
| import logging | ||
| from typing import Any | ||
|
|
||
| from switchbot import SwitchbotCurtain # pylint: disable=import-error | ||
|
|
||
| from homeassistant.components.cover import ( | ||
| ATTR_CURRENT_POSITION, | ||
| ATTR_POSITION, | ||
| DEVICE_CLASS_CURTAIN, | ||
| SUPPORT_CLOSE, | ||
| SUPPORT_OPEN, | ||
| SUPPORT_SET_POSITION, | ||
| SUPPORT_STOP, | ||
| CoverEntity, | ||
| ) | ||
| from homeassistant.config_entries import ConfigEntry | ||
| from homeassistant.const import CONF_MAC, CONF_NAME, CONF_PASSWORD | ||
| from homeassistant.core import HomeAssistant, callback | ||
| from homeassistant.helpers.entity_platform import AddEntitiesCallback | ||
| from homeassistant.helpers.restore_state import RestoreEntity | ||
|
|
||
| from .const import CONF_RETRY_COUNT, DATA_COORDINATOR, DOMAIN | ||
| from .coordinator import SwitchbotDataUpdateCoordinator | ||
| from .entity import SwitchbotEntity | ||
|
|
||
| # Initialize the logger | ||
| _LOGGER = logging.getLogger(__name__) | ||
| PARALLEL_UPDATES = 1 | ||
|
|
||
|
|
||
| async def async_setup_entry( | ||
| hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback | ||
| ) -> None: | ||
| """Set up Switchbot curtain based on a config entry.""" | ||
| coordinator: SwitchbotDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][ | ||
| DATA_COORDINATOR | ||
| ] | ||
|
|
||
| async_add_entities( | ||
| [ | ||
| SwitchBotCurtainEntity( | ||
| coordinator, | ||
| entry.unique_id, | ||
| entry.data[CONF_MAC], | ||
| entry.data[CONF_NAME], | ||
| coordinator.switchbot_api.SwitchbotCurtain( | ||
| mac=entry.data[CONF_MAC], | ||
| password=entry.data.get(CONF_PASSWORD), | ||
| retry_count=entry.options[CONF_RETRY_COUNT], | ||
| ), | ||
| ) | ||
| ] | ||
| ) | ||
|
|
||
|
|
||
| class SwitchBotCurtainEntity(SwitchbotEntity, CoverEntity, RestoreEntity): | ||
| """Representation of a Switchbot.""" | ||
|
|
||
| coordinator: SwitchbotDataUpdateCoordinator | ||
| _attr_device_class = DEVICE_CLASS_CURTAIN | ||
| _attr_supported_features = ( | ||
| SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP | SUPPORT_SET_POSITION | ||
| ) | ||
| _attr_assumed_state = True | ||
|
|
||
| def __init__( | ||
| self, | ||
| coordinator: SwitchbotDataUpdateCoordinator, | ||
| idx: str | None, | ||
| mac: str, | ||
| name: str, | ||
| device: SwitchbotCurtain, | ||
| ) -> None: | ||
| """Initialize the Switchbot.""" | ||
| super().__init__(coordinator, idx, mac, name) | ||
| self._attr_unique_id = idx | ||
| self._device = device | ||
|
|
||
| async def async_added_to_hass(self) -> None: | ||
| """Run when entity about to be added.""" | ||
| await super().async_added_to_hass() | ||
| last_state = await self.async_get_last_state() | ||
| if not last_state or ATTR_CURRENT_POSITION not in last_state.attributes: | ||
| return | ||
|
|
||
| self._attr_current_cover_position = last_state.attributes[ATTR_CURRENT_POSITION] | ||
| self._last_run_success = last_state.attributes["last_run_success"] | ||
| self._attr_is_closed = last_state.attributes[ATTR_CURRENT_POSITION] <= 20 | ||
|
|
||
| async def async_open_cover(self, **kwargs: Any) -> None: | ||
| """Open the curtain.""" | ||
|
|
||
| _LOGGER.debug("Switchbot to open curtain %s", self._mac) | ||
|
|
||
| async with self.coordinator.api_lock: | ||
| self._last_run_success = bool( | ||
| await self.hass.async_add_executor_job(self._device.open) | ||
| ) | ||
|
|
||
| async def async_close_cover(self, **kwargs: Any) -> None: | ||
| """Close the curtain.""" | ||
|
|
||
| _LOGGER.debug("Switchbot to close the curtain %s", self._mac) | ||
|
|
||
| async with self.coordinator.api_lock: | ||
| self._last_run_success = bool( | ||
| await self.hass.async_add_executor_job(self._device.close) | ||
| ) | ||
|
|
||
| async def async_stop_cover(self, **kwargs: Any) -> None: | ||
| """Stop the moving of this device.""" | ||
|
|
||
| _LOGGER.debug("Switchbot to stop %s", self._mac) | ||
|
|
||
| async with self.coordinator.api_lock: | ||
| self._last_run_success = bool( | ||
| await self.hass.async_add_executor_job(self._device.stop) | ||
| ) | ||
|
|
||
| async def async_set_cover_position(self, **kwargs: Any) -> None: | ||
| """Move the cover shutter to a specific position.""" | ||
| position = kwargs.get(ATTR_POSITION) | ||
|
|
||
| _LOGGER.debug("Switchbot to move at %d %s", position, self._mac) | ||
|
|
||
| async with self.coordinator.api_lock: | ||
| self._last_run_success = bool( | ||
| await self.hass.async_add_executor_job( | ||
| self._device.set_position, position | ||
| ) | ||
| ) | ||
|
|
||
| @callback | ||
| def _handle_coordinator_update(self) -> None: | ||
| """Handle updated data from the coordinator.""" | ||
| self._attr_current_cover_position = self.data["data"]["position"] | ||
| self._attr_is_closed = self.data["data"]["position"] <= 20 | ||
| self.async_write_ha_state() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| """An abstract class common to all Switchbot entities.""" | ||
| from __future__ import annotations | ||
|
|
||
| from collections.abc import Mapping | ||
| from typing import Any | ||
|
|
||
| from homeassistant.helpers import device_registry as dr | ||
| from homeassistant.helpers.entity import DeviceInfo, Entity | ||
| from homeassistant.helpers.update_coordinator import CoordinatorEntity | ||
|
|
||
| from .const import MANUFACTURER | ||
| from .coordinator import SwitchbotDataUpdateCoordinator | ||
|
|
||
|
|
||
| class SwitchbotEntity(CoordinatorEntity, Entity): | ||
| """Generic entity encapsulating common features of Switchbot device.""" | ||
|
|
||
| def __init__( | ||
| self, | ||
| coordinator: SwitchbotDataUpdateCoordinator, | ||
| idx: str | None, | ||
| mac: str, | ||
| name: str, | ||
| ) -> None: | ||
| """Initialize the entity.""" | ||
| super().__init__(coordinator) | ||
| self._last_run_success: bool | None = None | ||
| self._idx = idx | ||
| self._mac = mac | ||
| self._attr_name = name | ||
| self._attr_device_info: DeviceInfo = { | ||
| "connections": {(dr.CONNECTION_NETWORK_MAC, self._mac)}, | ||
| "name": self._attr_name, | ||
| "model": self.data["modelName"], | ||
| "manufacturer": MANUFACTURER, | ||
| } | ||
|
|
||
| @property | ||
| def data(self) -> dict[str, Any]: | ||
| """Return coordinator data for this entity.""" | ||
| return self.coordinator.data[self._idx] | ||
|
|
||
| @property | ||
| def extra_state_attributes(self) -> Mapping[Any, Any]: | ||
| """Return the state attributes.""" | ||
| return {"last_run_success": self._last_run_success, "mac_address": self._mac} |
Oops, something went wrong.