Fix opening/closing for awning in Overkiz integration (#68890)
parent
a22f36178f
commit
b4a536ca09
|
@ -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)
|
||||
|
|
|
@ -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:
|
||||
|
|
|
@ -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."""
|
||||
|
|
Loading…
Reference in New Issue