Add cover platform for switchbee integration (#78383)

* Added Platform cover for switchbee integration

* added cover to .coveragerc

* Applied code review feedback from other PR

* Addressed comments from other PRs

* rebased

* Re-add carriage return

* Update homeassistant/components/switchbee/cover.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/switchbee/cover.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/switchbee/cover.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/switchbee/cover.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* addressed CR comments

* fixes

* fixes

* more fixes

* more fixes

* separate entities for cover and somfy cover

* fixed isort

* more fixes

* more fixes

* Update homeassistant/components/switchbee/cover.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Update homeassistant/components/switchbee/cover.py

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* more fixes

* more fixes

* more

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
This commit is contained in:
Jafar Atili 2022-09-29 16:03:39 +03:00 committed by GitHub
parent 0b5289f748
commit 75510b8e90
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 171 additions and 10 deletions

View File

@ -1231,6 +1231,7 @@ omit =
homeassistant/components/switchbee/__init__.py
homeassistant/components/switchbee/button.py
homeassistant/components/switchbee/coordinator.py
homeassistant/components/switchbee/cover.py
homeassistant/components/switchbee/entity.py
homeassistant/components/switchbee/light.py
homeassistant/components/switchbee/switch.py

View File

@ -13,7 +13,12 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .coordinator import SwitchBeeCoordinator
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.LIGHT, Platform.SWITCH]
PLATFORMS: list[Platform] = [
Platform.BUTTON,
Platform.COVER,
Platform.LIGHT,
Platform.SWITCH,
]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:

View File

@ -62,6 +62,8 @@ class SwitchBeeCoordinator(DataUpdateCoordinator[Mapping[int, SwitchBeeBaseDevic
DeviceType.TimedPowerSwitch,
DeviceType.Scenario,
DeviceType.Dimmer,
DeviceType.Shutter,
DeviceType.Somfy,
]
)
except SwitchBeeError as exp:

View File

@ -0,0 +1,152 @@
"""Support for SwitchBee cover."""
from __future__ import annotations
from typing import Any
from switchbee.api import SwitchBeeError, SwitchBeeTokenError
from switchbee.const import SomfyCommand
from switchbee.device import SwitchBeeShutter, SwitchBeeSomfy
from homeassistant.components.cover import (
ATTR_POSITION,
CoverDeviceClass,
CoverEntity,
CoverEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .coordinator import SwitchBeeCoordinator
from .entity import SwitchBeeDeviceEntity
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up SwitchBee switch."""
coordinator: SwitchBeeCoordinator = hass.data[DOMAIN][entry.entry_id]
entities: list[CoverEntity] = []
for device in coordinator.data.values():
if isinstance(device, SwitchBeeShutter):
entities.append(SwitchBeeCoverEntity(device, coordinator))
elif isinstance(device, SwitchBeeSomfy):
entities.append(SwitchBeeSomfyEntity(device, coordinator))
async_add_entities(entities)
class SwitchBeeSomfyEntity(SwitchBeeDeviceEntity[SwitchBeeSomfy], CoverEntity):
"""Representation of a SwitchBee Somfy cover."""
_attr_device_class = CoverDeviceClass.SHUTTER
_attr_supported_features = (
CoverEntityFeature.CLOSE | CoverEntityFeature.OPEN | CoverEntityFeature.STOP
)
_attr_is_closed = None
async def _fire_somfy_command(self, command: str) -> None:
"""Async function to fire Somfy device command."""
try:
await self.coordinator.api.set_state(self._device.id, command)
except (SwitchBeeError, SwitchBeeTokenError) as exp:
raise HomeAssistantError(
f"Failed to fire {command} for {self.name}, {str(exp)}"
) from exp
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
return await self._fire_somfy_command(SomfyCommand.UP)
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
return await self._fire_somfy_command(SomfyCommand.DOWN)
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop a moving cover."""
return await self._fire_somfy_command(SomfyCommand.MY)
class SwitchBeeCoverEntity(SwitchBeeDeviceEntity[SwitchBeeShutter], CoverEntity):
"""Representation of a SwitchBee cover."""
_attr_device_class = CoverDeviceClass.SHUTTER
_attr_supported_features = (
CoverEntityFeature.CLOSE
| CoverEntityFeature.OPEN
| CoverEntityFeature.SET_POSITION
| CoverEntityFeature.STOP
)
_attr_is_closed = None
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._update_from_coordinator()
super()._handle_coordinator_update()
def _update_from_coordinator(self) -> None:
"""Update the entity attributes from the coordinator data."""
coordinator_device = self._get_coordinator_device()
if coordinator_device.position == -1:
self._check_if_became_offline()
return
# check if the device was offline (now online) and bring it back
self._check_if_became_online()
self._attr_current_cover_position = coordinator_device.position
if self.current_cover_position == 0:
self._attr_is_closed = True
else:
self._attr_is_closed = False
super()._handle_coordinator_update()
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
if self.current_cover_position == 100:
return
await self.async_set_cover_position(position=100)
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
if self.current_cover_position == 0:
return
await self.async_set_cover_position(position=0)
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop a moving cover."""
# to stop the shutter, we just interrupt it with any state during operation
await self.async_set_cover_position(
position=self.current_cover_position, force=True
)
# fetch data from the Central Unit to get the new position
await self.coordinator.async_request_refresh()
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Async function to set position to cover."""
if (
self.current_cover_position == kwargs[ATTR_POSITION]
and "force" not in kwargs
):
return
try:
await self.coordinator.api.set_state(self._device.id, kwargs[ATTR_POSITION])
except (SwitchBeeError, SwitchBeeTokenError) as exp:
raise HomeAssistantError(
f"Failed to set {self.name} position to {kwargs[ATTR_POSITION]}, error: {str(exp)}"
) from exp
self._get_coordinator_device().position = kwargs[ATTR_POSITION]
self.coordinator.async_set_updated_data(self.coordinator.data)
self.async_write_ha_state()

View File

@ -1,6 +1,6 @@
"""Support for SwitchBee entity."""
import logging
from typing import Generic, TypeVar
from typing import Generic, TypeVar, cast
from switchbee import SWITCHBEE_BRAND
from switchbee.api import SwitchBeeDeviceOfflineError, SwitchBeeError
@ -108,3 +108,6 @@ class SwitchBeeDeviceEntity(SwitchBeeEntity[_DeviceTypeT]):
self.name,
)
self._is_online = True
def _get_coordinator_device(self) -> _DeviceTypeT:
return cast(_DeviceTypeT, self.coordinator.data[self._device.id])

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Any, cast
from typing import Any
from switchbee.api import SwitchBeeDeviceOfflineError, SwitchBeeError
from switchbee.device import ApiStateCommand, DeviceType, SwitchBeeDimmer
@ -72,9 +72,7 @@ class SwitchBeeLightEntity(SwitchBeeDeviceEntity[SwitchBeeDimmer], LightEntity):
def _update_attrs_from_coordinator(self) -> None:
coordinator_device = cast(
SwitchBeeDimmer, self.coordinator.data[self._device.id]
)
coordinator_device = self._get_coordinator_device()
brightness = coordinator_device.brightness
# module is offline
@ -112,7 +110,7 @@ class SwitchBeeLightEntity(SwitchBeeDeviceEntity[SwitchBeeDimmer], LightEntity):
return
# update the coordinator data manually we already know the Central Unit brightness data for this light
cast(SwitchBeeDimmer, self.coordinator.data[self._device.id]).brightness = state
self._get_coordinator_device().brightness = state
self.coordinator.async_set_updated_data(self.coordinator.data)
async def async_turn_off(self, **kwargs: Any) -> None:
@ -125,5 +123,5 @@ class SwitchBeeLightEntity(SwitchBeeDeviceEntity[SwitchBeeDimmer], LightEntity):
) from exp
# update the coordinator manually
cast(SwitchBeeDimmer, self.coordinator.data[self._device.id]).brightness = 0
self._get_coordinator_device().brightness = 0
self.coordinator.async_set_updated_data(self.coordinator.data)

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Any, TypeVar, Union, cast
from typing import Any, TypeVar, Union
from switchbee.api import SwitchBeeDeviceOfflineError, SwitchBeeError
from switchbee.device import (
@ -76,7 +76,7 @@ class SwitchBeeSwitchEntity(SwitchBeeDeviceEntity[_DeviceTypeT], SwitchEntity):
def _update_from_coordinator(self) -> None:
"""Update the entity attributes from the coordinator data."""
coordinator_device = cast(_DeviceTypeT, self.coordinator.data[self._device.id])
coordinator_device = self._get_coordinator_device()
if coordinator_device.state == -1: