From b4a536ca099c52d7a4ac01e70941f71c94bdbc32 Mon Sep 17 00:00:00 2001 From: Thibaut Date: Wed, 20 Apr 2022 14:53:20 +0200 Subject: [PATCH] Fix opening/closing for awning in Overkiz integration (#68890) --- .../overkiz/cover_entities/awning.py | 39 ++++++++++++- .../overkiz/cover_entities/generic_cover.py | 55 +++---------------- .../overkiz/cover_entities/vertical_cover.py | 41 +++++++++++++- 3 files changed, 84 insertions(+), 51 deletions(-) diff --git a/homeassistant/components/overkiz/cover_entities/awning.py b/homeassistant/components/overkiz/cover_entities/awning.py index fe6bfeb73d6..07665c2a5c8 100644 --- a/homeassistant/components/overkiz/cover_entities/awning.py +++ b/homeassistant/components/overkiz/cover_entities/awning.py @@ -11,7 +11,12 @@ from homeassistant.components.cover import ( CoverEntityFeature, ) -from .generic_cover import COMMANDS_STOP, OverkizGenericCover +from .generic_cover import ( + COMMANDS_CLOSE, + COMMANDS_OPEN, + COMMANDS_STOP, + OverkizGenericCover, +) class Awning(OverkizGenericCover): @@ -64,3 +69,35 @@ class Awning(OverkizGenericCover): async def async_close_cover(self, **kwargs: Any) -> None: """Close the cover.""" await self.executor.async_execute_command(OverkizCommand.UNDEPLOY) + + @property + def is_opening(self) -> bool | None: + """Return if the cover is opening or not.""" + if self.is_running(COMMANDS_OPEN): + 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_DEPLOYMENT) + 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.is_running(COMMANDS_CLOSE): + 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_DEPLOYMENT) + 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) diff --git a/homeassistant/components/overkiz/cover_entities/generic_cover.py b/homeassistant/components/overkiz/cover_entities/generic_cover.py index 4a0cbe3b0cb..e36d7a680b1 100644 --- a/homeassistant/components/overkiz/cover_entities/generic_cover.py +++ b/homeassistant/components/overkiz/cover_entities/generic_cover.py @@ -11,7 +11,8 @@ from homeassistant.components.cover import ( CoverEntity, CoverEntityFeature, ) -from homeassistant.components.overkiz.entity import OverkizEntity + +from ..entity import OverkizEntity ATTR_OBSTRUCTION_DETECTED = "obstruction-detected" @@ -108,55 +109,13 @@ class OverkizGenericCover(OverkizEntity, CoverEntity): 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( + def is_running(self, commands: list[OverkizCommand]) -> bool: + """Return if the given commands are currently running.""" + return any( execution.get("device_url") == self.device.device_url - and execution.get("command_name") in COMMANDS_OPEN + COMMANDS_OPEN_TILT + and execution.get("command_name") in commands 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: diff --git a/homeassistant/components/overkiz/cover_entities/vertical_cover.py b/homeassistant/components/overkiz/cover_entities/vertical_cover.py index df3ba813ecb..f76f3849e83 100644 --- a/homeassistant/components/overkiz/cover_entities/vertical_cover.py +++ b/homeassistant/components/overkiz/cover_entities/vertical_cover.py @@ -16,9 +16,14 @@ from homeassistant.components.cover import ( CoverDeviceClass, CoverEntityFeature, ) -from homeassistant.components.overkiz.coordinator import OverkizDataUpdateCoordinator -from .generic_cover import COMMANDS_STOP, OverkizGenericCover +from ..coordinator import OverkizDataUpdateCoordinator +from .generic_cover import ( + COMMANDS_CLOSE_TILT, + COMMANDS_OPEN_TILT, + COMMANDS_STOP, + OverkizGenericCover, +) COMMANDS_OPEN = [OverkizCommand.OPEN, OverkizCommand.UP, OverkizCommand.CYCLE] COMMANDS_CLOSE = [OverkizCommand.CLOSE, OverkizCommand.DOWN, OverkizCommand.CYCLE] @@ -104,6 +109,38 @@ class VerticalCover(OverkizGenericCover): if command := self.executor.select_command(*COMMANDS_CLOSE): await self.executor.async_execute_command(command) + @property + def is_opening(self) -> bool | None: + """Return if the cover is opening or not.""" + if self.is_running(COMMANDS_OPEN + COMMANDS_OPEN_TILT): + 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.is_running(COMMANDS_CLOSE + COMMANDS_CLOSE_TILT): + 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) + class LowSpeedCover(VerticalCover): """Representation of an Overkiz Low Speed cover."""