mirror of https://github.com/home-assistant/core
Add cover platform to Overkiz integration (#64564)
This commit is contained in:
Mick Vleeshouwer
GitHub
parent
78e92d1662
commit
d6c547e9a3
|
|
@ -820,6 +820,8 @@ omit =
|
|||
homeassistant/components/overkiz/__init__.py
|
||||
homeassistant/components/overkiz/binary_sensor.py
|
||||
homeassistant/components/overkiz/button.py
|
||||
homeassistant/components/overkiz/cover.py
|
||||
homeassistant/components/overkiz/cover_entities/*
|
||||
homeassistant/components/overkiz/coordinator.py
|
||||
homeassistant/components/overkiz/diagnostics.py
|
||||
homeassistant/components/overkiz/entity.py
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ UPDATE_INTERVAL_ALL_ASSUMED_STATE: Final = timedelta(minutes=60)
|
|||
PLATFORMS: list[Platform] = [
|
||||
Platform.BINARY_SENSOR,
|
||||
Platform.BUTTON,
|
||||
Platform.COVER,
|
||||
Platform.LIGHT,
|
||||
Platform.LOCK,
|
||||
Platform.NUMBER,
|
||||
|
|
@ -37,13 +38,29 @@ IGNORED_OVERKIZ_DEVICES: list[UIClass | UIWidget] = [
|
|||
|
||||
# Used to map the Somfy widget and ui_class to the Home Assistant platform
|
||||
OVERKIZ_DEVICE_TO_PLATFORM: dict[UIClass | UIWidget, Platform] = {
|
||||
UIClass.ADJUSTABLE_SLATS_ROLLER_SHUTTER: Platform.COVER,
|
||||
UIClass.AWNING: Platform.COVER,
|
||||
UIClass.CURTAIN: Platform.COVER,
|
||||
UIClass.DOOR_LOCK: Platform.LOCK,
|
||||
UIWidget.DOMESTIC_HOT_WATER_TANK: Platform.SWITCH, # widgetName, uiClass is WaterHeatingSystem (not supported)
|
||||
UIClass.EXTERIOR_SCREEN: Platform.COVER,
|
||||
UIClass.EXTERIOR_VENETIAN_BLIND: Platform.COVER,
|
||||
UIClass.GARAGE_DOOR: Platform.COVER,
|
||||
UIClass.GATE: Platform.COVER,
|
||||
UIClass.LIGHT: Platform.LIGHT,
|
||||
UIClass.ON_OFF: Platform.SWITCH,
|
||||
UIClass.PERGOLA: Platform.COVER,
|
||||
UIClass.ROLLER_SHUTTER: Platform.COVER,
|
||||
UIClass.SCREEN: Platform.COVER,
|
||||
UIClass.SHUTTER: Platform.COVER,
|
||||
UIClass.SWIMMING_POOL: Platform.SWITCH,
|
||||
UIClass.SWINGING_SHUTTER: Platform.COVER,
|
||||
UIClass.VENETIAN_BLIND: Platform.COVER,
|
||||
UIClass.WINDOW: Platform.COVER,
|
||||
UIWidget.DOMESTIC_HOT_WATER_TANK: Platform.SWITCH, # widgetName, uiClass is WaterHeatingSystem (not supported)
|
||||
UIWidget.MY_FOX_SECURITY_CAMERA: Platform.COVER, # widgetName, uiClass is Camera (not supported)
|
||||
UIWidget.RTD_INDOOR_SIREN: Platform.SWITCH, # widgetName, uiClass is Siren (not supported)
|
||||
UIWidget.RTD_OUTDOOR_SIREN: Platform.SWITCH, # widgetName, uiClass is Siren (not supported)
|
||||
UIClass.SWIMMING_POOL: Platform.SWITCH,
|
||||
UIWidget.RTS_GENERIC: Platform.COVER, # widgetName, uiClass is Generic (not supported)
|
||||
}
|
||||
|
||||
# Map Overkiz camelCase to Home Assistant snake_case for translation
|
||||
|
|
|
|||
|
|
@ -0,0 +1,34 @@
|
|||
"""Support for Overkiz covers - shutters etc."""
|
||||
from pyoverkiz.enums import UIClass
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import HomeAssistantOverkizData
|
||||
from .const import DOMAIN
|
||||
from .cover_entities.awning import Awning
|
||||
from .cover_entities.generic_cover import OverkizGenericCover
|
||||
from .cover_entities.vertical_cover import VerticalCover
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up the Overkiz covers from a config entry."""
|
||||
data: HomeAssistantOverkizData = hass.data[DOMAIN][entry.entry_id]
|
||||
|
||||
entities: list[OverkizGenericCover] = [
|
||||
Awning(device.device_url, data.coordinator)
|
||||
for device in data.platforms[Platform.COVER]
|
||||
if device.ui_class == UIClass.AWNING
|
||||
]
|
||||
|
||||
entities += [
|
||||
VerticalCover(device.device_url, data.coordinator)
|
||||
for device in data.platforms[Platform.COVER]
|
||||
if device.ui_class != UIClass.AWNING
|
||||
]
|
||||
|
||||
async_add_entities(entities)
|
||||
|
|
@ -0,0 +1 @@
|
|||
"""Cover entities for the Overkiz (by Somfy) integration."""
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
"""Support for Overkiz awnings."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, cast
|
||||
|
||||
from pyoverkiz.enums import OverkizCommand, OverkizState
|
||||
|
||||
from homeassistant.components.cover import (
|
||||
ATTR_POSITION,
|
||||
DEVICE_CLASS_AWNING,
|
||||
SUPPORT_CLOSE,
|
||||
SUPPORT_OPEN,
|
||||
SUPPORT_SET_POSITION,
|
||||
SUPPORT_STOP,
|
||||
)
|
||||
|
||||
from .generic_cover import COMMANDS_STOP, OverkizGenericCover
|
||||
|
||||
|
||||
class Awning(OverkizGenericCover):
|
||||
"""Representation of an Overkiz awning."""
|
||||
|
||||
_attr_device_class = DEVICE_CLASS_AWNING
|
||||
|
||||
@property
|
||||
def supported_features(self) -> int:
|
||||
"""Flag supported features."""
|
||||
supported_features: int = super().supported_features
|
||||
|
||||
if self.executor.has_command(OverkizCommand.SET_DEPLOYMENT):
|
||||
supported_features |= SUPPORT_SET_POSITION
|
||||
|
||||
if self.executor.has_command(OverkizCommand.DEPLOY):
|
||||
supported_features |= SUPPORT_OPEN
|
||||
|
||||
if self.executor.has_command(*COMMANDS_STOP):
|
||||
supported_features |= SUPPORT_STOP
|
||||
|
||||
if self.executor.has_command(OverkizCommand.UNDEPLOY):
|
||||
supported_features |= SUPPORT_CLOSE
|
||||
|
||||
return supported_features
|
||||
|
||||
@property
|
||||
def current_cover_position(self) -> int | None:
|
||||
"""
|
||||
Return current position of cover.
|
||||
|
||||
None is unknown, 0 is closed, 100 is fully open.
|
||||
"""
|
||||
if current_position := self.executor.select_state(OverkizState.CORE_DEPLOYMENT):
|
||||
return cast(int, current_position)
|
||||
|
||||
return None
|
||||
|
||||
async def async_set_cover_position(self, **kwargs: Any) -> None:
|
||||
"""Move the cover to a specific position."""
|
||||
position = kwargs.get(ATTR_POSITION, 0)
|
||||
await self.executor.async_execute_command(
|
||||
OverkizCommand.SET_DEPLOYMENT, position
|
||||
)
|
||||
|
||||
async def async_open_cover(self, **kwargs: Any) -> None:
|
||||
"""Open the cover."""
|
||||
await self.executor.async_execute_command(OverkizCommand.DEPLOY)
|
||||
|
||||
async def async_close_cover(self, **kwargs: Any) -> None:
|
||||
"""Close the cover."""
|
||||
await self.executor.async_execute_command(OverkizCommand.UNDEPLOY)
|
||||
|
|
@ -0,0 +1,191 @@
|
|||
"""Base class for Overkiz covers, shutters, awnings, etc."""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, cast
|
||||
|
||||
from pyoverkiz.enums import OverkizCommand, OverkizCommandParam, OverkizState
|
||||
|
||||
from homeassistant.components.cover import (
|
||||
ATTR_TILT_POSITION,
|
||||
SUPPORT_CLOSE_TILT,
|
||||
SUPPORT_OPEN_TILT,
|
||||
SUPPORT_SET_TILT_POSITION,
|
||||
SUPPORT_STOP_TILT,
|
||||
CoverEntity,
|
||||
)
|
||||
from homeassistant.components.overkiz.entity import OverkizEntity
|
||||
|
||||
ATTR_OBSTRUCTION_DETECTED = "obstruction-detected"
|
||||
|
||||
COMMANDS_STOP: list[OverkizCommand] = [
|
||||
OverkizCommand.STOP,
|
||||
OverkizCommand.MY,
|
||||
]
|
||||
COMMANDS_STOP_TILT: list[OverkizCommand] = [
|
||||
OverkizCommand.STOP,
|
||||
OverkizCommand.MY,
|
||||
]
|
||||
COMMANDS_OPEN: list[OverkizCommand] = [
|
||||
OverkizCommand.OPEN,
|
||||
OverkizCommand.UP,
|
||||
OverkizCommand.CYCLE,
|
||||
]
|
||||
COMMANDS_OPEN_TILT: list[OverkizCommand] = [OverkizCommand.OPEN_SLATS]
|
||||
COMMANDS_CLOSE: list[OverkizCommand] = [
|
||||
OverkizCommand.CLOSE,
|
||||
OverkizCommand.DOWN,
|
||||
OverkizCommand.CYCLE,
|
||||
]
|
||||
COMMANDS_CLOSE_TILT: list[OverkizCommand] = [OverkizCommand.CLOSE_SLATS]
|
||||
|
||||
COMMANDS_SET_TILT_POSITION: list[OverkizCommand] = [OverkizCommand.SET_ORIENTATION]
|
||||
|
||||
|
||||
class OverkizGenericCover(OverkizEntity, CoverEntity):
|
||||
"""Representation of an Overkiz Cover."""
|
||||
|
||||
@property
|
||||
def current_cover_tilt_position(self) -> int | None:
|
||||
"""Return current position of cover tilt.
|
||||
|
||||
None is unknown, 0 is closed, 100 is fully open.
|
||||
"""
|
||||
if position := self.executor.select_state(
|
||||
OverkizState.CORE_SLATS_ORIENTATION, OverkizState.CORE_SLATE_ORIENTATION
|
||||
):
|
||||
return 100 - cast(int, position)
|
||||
|
||||
return None
|
||||
|
||||
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
|
||||
"""Move the cover tilt to a specific position."""
|
||||
if command := self.executor.select_command(*COMMANDS_SET_TILT_POSITION):
|
||||
await self.executor.async_execute_command(
|
||||
command,
|
||||
100 - kwargs.get(ATTR_TILT_POSITION, 0),
|
||||
)
|
||||
|
||||
@property
|
||||
def is_closed(self) -> bool | None:
|
||||
"""Return if the cover is closed."""
|
||||
|
||||
state = self.executor.select_state(
|
||||
OverkizState.CORE_OPEN_CLOSED,
|
||||
OverkizState.CORE_SLATS_OPEN_CLOSED,
|
||||
OverkizState.CORE_OPEN_CLOSED_PARTIAL,
|
||||
OverkizState.CORE_OPEN_CLOSED_PEDESTRIAN,
|
||||
OverkizState.CORE_OPEN_CLOSED_UNKNOWN,
|
||||
OverkizState.MYFOX_SHUTTER_STATUS,
|
||||
)
|
||||
if state is not None:
|
||||
return state == OverkizCommandParam.CLOSED
|
||||
|
||||
# Keep this condition after the previous one. Some device like the pedestrian gate, always return 50 as position.
|
||||
if self.current_cover_position is not None:
|
||||
return self.current_cover_position == 0
|
||||
|
||||
if self.current_cover_tilt_position is not None:
|
||||
return self.current_cover_tilt_position == 0
|
||||
|
||||
return None
|
||||
|
||||
async def async_open_cover_tilt(self, **kwargs: Any) -> None:
|
||||
"""Open the cover tilt."""
|
||||
if command := self.executor.select_command(*COMMANDS_OPEN_TILT):
|
||||
await self.executor.async_execute_command(command)
|
||||
|
||||
async def async_close_cover_tilt(self, **kwargs: Any) -> None:
|
||||
"""Close the cover tilt."""
|
||||
if command := self.executor.select_command(*COMMANDS_CLOSE_TILT):
|
||||
await self.executor.async_execute_command(command)
|
||||
|
||||
async def async_stop_cover(self, **kwargs: Any) -> None:
|
||||
"""Stop the cover."""
|
||||
if command := self.executor.select_command(*COMMANDS_STOP):
|
||||
await self.executor.async_execute_command(command)
|
||||
|
||||
async def async_stop_cover_tilt(self, **kwargs: Any) -> None:
|
||||
"""Stop the cover tilt."""
|
||||
if command := self.executor.select_command(*COMMANDS_STOP_TILT):
|
||||
await self.executor.async_execute_command(command)
|
||||
|
||||
@property
|
||||
def is_opening(self) -> bool | None:
|
||||
"""Return if the cover is opening or not."""
|
||||
|
||||
if self.assumed_state:
|
||||
return None
|
||||
|
||||
# Check if cover movement execution is currently running
|
||||
if any(
|
||||
execution.get("device_url") == self.device.device_url
|
||||
and execution.get("command_name") in COMMANDS_OPEN + COMMANDS_OPEN_TILT
|
||||
for execution in self.coordinator.executions.values()
|
||||
):
|
||||
return True
|
||||
|
||||
# Check if cover is moving based on current state
|
||||
is_moving = self.device.states.get(OverkizState.CORE_MOVING)
|
||||
current_closure = self.device.states.get(OverkizState.CORE_CLOSURE)
|
||||
target_closure = self.device.states.get(OverkizState.CORE_TARGET_CLOSURE)
|
||||
|
||||
if not is_moving or not current_closure or not target_closure:
|
||||
return None
|
||||
|
||||
return cast(int, current_closure.value) > cast(int, target_closure.value)
|
||||
|
||||
@property
|
||||
def is_closing(self) -> bool | None:
|
||||
"""Return if the cover is closing or not."""
|
||||
|
||||
if self.assumed_state:
|
||||
return None
|
||||
|
||||
# Check if cover movement execution is currently running
|
||||
if any(
|
||||
execution.get("device_url") == self.device.device_url
|
||||
and execution.get("command_name") in COMMANDS_CLOSE + COMMANDS_CLOSE_TILT
|
||||
for execution in self.coordinator.executions.values()
|
||||
):
|
||||
return True
|
||||
|
||||
# Check if cover is moving based on current state
|
||||
is_moving = self.device.states.get(OverkizState.CORE_MOVING)
|
||||
current_closure = self.device.states.get(OverkizState.CORE_CLOSURE)
|
||||
target_closure = self.device.states.get(OverkizState.CORE_TARGET_CLOSURE)
|
||||
|
||||
if not is_moving or not current_closure or not target_closure:
|
||||
return None
|
||||
|
||||
return cast(int, current_closure.value) < cast(int, target_closure.value)
|
||||
|
||||
@property
|
||||
def extra_state_attributes(self) -> Mapping[str, Any] | None:
|
||||
"""Return the device state attributes."""
|
||||
attr = super().extra_state_attributes or {}
|
||||
|
||||
# Obstruction Detected attribute is used by HomeKit
|
||||
if self.executor.has_state(OverkizState.IO_PRIORITY_LOCK_LEVEL):
|
||||
return {**attr, **{ATTR_OBSTRUCTION_DETECTED: True}}
|
||||
|
||||
return attr
|
||||
|
||||
@property
|
||||
def supported_features(self) -> int:
|
||||
"""Flag supported features."""
|
||||
supported_features = 0
|
||||
|
||||
if self.executor.has_command(*COMMANDS_OPEN_TILT):
|
||||
supported_features |= SUPPORT_OPEN_TILT
|
||||
|
||||
if self.executor.has_command(*COMMANDS_STOP_TILT):
|
||||
supported_features |= SUPPORT_STOP_TILT
|
||||
|
||||
if self.executor.has_command(*COMMANDS_CLOSE_TILT):
|
||||
supported_features |= SUPPORT_CLOSE_TILT
|
||||
|
||||
if self.executor.has_command(*COMMANDS_SET_TILT_POSITION):
|
||||
supported_features |= SUPPORT_SET_TILT_POSITION
|
||||
|
||||
return supported_features
|
||||
|
|
@ -0,0 +1,111 @@
|
|||
"""Support for Overkiz Vertical Covers."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Union, cast
|
||||
|
||||
from pyoverkiz.enums import OverkizCommand, OverkizState, UIClass, UIWidget
|
||||
|
||||
from homeassistant.components.cover import (
|
||||
ATTR_POSITION,
|
||||
DEVICE_CLASS_AWNING,
|
||||
DEVICE_CLASS_BLIND,
|
||||
DEVICE_CLASS_CURTAIN,
|
||||
DEVICE_CLASS_GARAGE,
|
||||
DEVICE_CLASS_GATE,
|
||||
DEVICE_CLASS_SHUTTER,
|
||||
DEVICE_CLASS_WINDOW,
|
||||
SUPPORT_CLOSE,
|
||||
SUPPORT_OPEN,
|
||||
SUPPORT_SET_POSITION,
|
||||
SUPPORT_STOP,
|
||||
)
|
||||
|
||||
from .generic_cover import COMMANDS_STOP, OverkizGenericCover
|
||||
|
||||
COMMANDS_OPEN = [OverkizCommand.OPEN, OverkizCommand.UP, OverkizCommand.CYCLE]
|
||||
COMMANDS_CLOSE = [OverkizCommand.CLOSE, OverkizCommand.DOWN, OverkizCommand.CYCLE]
|
||||
|
||||
OVERKIZ_DEVICE_TO_DEVICE_CLASS = {
|
||||
UIClass.CURTAIN: DEVICE_CLASS_CURTAIN,
|
||||
UIClass.EXTERIOR_SCREEN: DEVICE_CLASS_BLIND,
|
||||
UIClass.EXTERIOR_VENETIAN_BLIND: DEVICE_CLASS_BLIND,
|
||||
UIClass.GARAGE_DOOR: DEVICE_CLASS_GARAGE,
|
||||
UIClass.GATE: DEVICE_CLASS_GATE,
|
||||
UIWidget.MY_FOX_SECURITY_CAMERA: DEVICE_CLASS_SHUTTER,
|
||||
UIClass.PERGOLA: DEVICE_CLASS_AWNING,
|
||||
UIClass.ROLLER_SHUTTER: DEVICE_CLASS_SHUTTER,
|
||||
UIClass.SWINGING_SHUTTER: DEVICE_CLASS_SHUTTER,
|
||||
UIClass.WINDOW: DEVICE_CLASS_WINDOW,
|
||||
}
|
||||
|
||||
|
||||
class VerticalCover(OverkizGenericCover):
|
||||
"""Representation of an Overkiz vertical cover."""
|
||||
|
||||
@property
|
||||
def supported_features(self) -> int:
|
||||
"""Flag supported features."""
|
||||
supported_features: int = super().supported_features
|
||||
|
||||
if self.executor.has_command(OverkizCommand.SET_CLOSURE):
|
||||
supported_features |= SUPPORT_SET_POSITION
|
||||
|
||||
if self.executor.has_command(*COMMANDS_OPEN):
|
||||
supported_features |= SUPPORT_OPEN
|
||||
|
||||
if self.executor.has_command(*COMMANDS_STOP):
|
||||
supported_features |= SUPPORT_STOP
|
||||
|
||||
if self.executor.has_command(*COMMANDS_CLOSE):
|
||||
supported_features |= SUPPORT_CLOSE
|
||||
|
||||
return supported_features
|
||||
|
||||
@property
|
||||
def device_class(self) -> str:
|
||||
"""Return the class of the device."""
|
||||
return cast(
|
||||
str,
|
||||
(
|
||||
OVERKIZ_DEVICE_TO_DEVICE_CLASS.get(self.device.widget)
|
||||
or OVERKIZ_DEVICE_TO_DEVICE_CLASS.get(self.device.ui_class)
|
||||
or DEVICE_CLASS_BLIND
|
||||
),
|
||||
)
|
||||
|
||||
@property
|
||||
def current_cover_position(self) -> int | None:
|
||||
"""
|
||||
Return current position of cover.
|
||||
|
||||
None is unknown, 0 is closed, 100 is fully open.
|
||||
"""
|
||||
position = cast(
|
||||
Union[int, None],
|
||||
self.executor.select_state(
|
||||
OverkizState.CORE_CLOSURE,
|
||||
OverkizState.CORE_CLOSURE_OR_ROCKER_POSITION,
|
||||
OverkizState.CORE_PEDESTRIAN_POSITION,
|
||||
),
|
||||
)
|
||||
|
||||
# Uno devices can have a position not in 0 to 100 range when unknown
|
||||
if position is None or position < 0 or position > 100:
|
||||
return None
|
||||
|
||||
return 100 - position
|
||||
|
||||
async def async_set_cover_position(self, **kwargs: Any) -> None:
|
||||
"""Move the cover to a specific position."""
|
||||
position = 100 - kwargs.get(ATTR_POSITION, 0)
|
||||
await self.executor.async_execute_command(OverkizCommand.SET_CLOSURE, position)
|
||||
|
||||
async def async_open_cover(self, **kwargs: Any) -> None:
|
||||
"""Open the cover."""
|
||||
if command := self.executor.select_command(*COMMANDS_OPEN):
|
||||
await self.executor.async_execute_command(command)
|
||||
|
||||
async def async_close_cover(self, **kwargs: Any) -> None:
|
||||
"""Close the cover."""
|
||||
if command := self.executor.select_command(*COMMANDS_CLOSE):
|
||||
await self.executor.async_execute_command(command)
|
||||
|
|
@ -4,6 +4,7 @@ from __future__ import annotations
|
|||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pyoverkiz.enums.command import OverkizCommand
|
||||
from pyoverkiz.models import Command, Device
|
||||
from pyoverkiz.types import StateType as OverkizStateType
|
||||
|
||||
|
|
@ -76,7 +77,9 @@ class OverkizExecutor:
|
|||
|
||||
await self.coordinator.async_refresh()
|
||||
|
||||
async def async_cancel_command(self, commands_to_cancel: list[str]) -> bool:
|
||||
async def async_cancel_command(
|
||||
self, commands_to_cancel: list[OverkizCommand]
|
||||
) -> bool:
|
||||
"""Cancel running execution by command."""
|
||||
|
||||
# Cancel a running execution
|
||||
|
|
|
|||
Loading…
Reference in New Issue