Electric-Special/ha-core
1
0
Fork 0
mirror of https://github.com/Electric-Special/ha-core.git synced 2026-03-20 23:03:18 +01:00
Code Issues Packages Projects Releases Wiki Activity

Use opening/closing state for Z-Wave covers (#163368)

Browse Source 
Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
AlCalzone
2026-02-19 10:41:52 +01:00
committed by GitHub
parent 7914ebe54e
commit e229ba591a
2 changed files with 573 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
homeassistant/components/zwave_js/cover.py
View File

@@ -6,8 +6,10 @@ from typing import Any, cast
from zwave_js_server.const import (
CURRENT_VALUE_PROPERTY,
SET_VALUE_SUCCESS,
TARGET_STATE_PROPERTY,
TARGET_VALUE_PROPERTY,
SetValueStatus,
)
from zwave_js_server.const.command_class.barrier_operator import BarrierState
from zwave_js_server.const.command_class.multilevel_switch import (
@@ -145,6 +147,21 @@ class CoverPositionMixin(ZWaveBaseEntity, CoverEntity):
return None
return bool(value.value == self._fully_closed_position)
@callback
def on_value_update(self) -> None:
"""Clear moving state when current position reaches target."""
if not self._attr_is_opening and not self._attr_is_closing:
return
if (
(current := self._current_position_value) is not None
and (target := self._target_position_value) is not None
and current.value is not None
and current.value == target.value
):
self._attr_is_opening = False
self._attr_is_closing = False
@property
def current_cover_position(self) -> int | None:
"""Return the current position of cover where 0 means closed and 100 is fully open."""
@@ -156,33 +173,69 @@ class CoverPositionMixin(ZWaveBaseEntity, CoverEntity):
return None
return self.zwave_to_percent_position(self._current_position_value.value)
async def _async_set_position_and_update_moving_state(
self, target_position: int
) -> None:
"""Set the target position and update the moving state if applicable."""
assert self._target_position_value
result = await self._async_set_value(
self._target_position_value, target_position
)
if (
# If the command is unsupervised, or the device reported that it started
# working, we can assume the cover is moving in the desired direction.
result is None
or result.status
not in (SetValueStatus.WORKING, SetValueStatus.SUCCESS_UNSUPERVISED)
# If we don't know the current position, we don't know which direction
# the cover is moving, so we can't update the moving state.
or (current_value := self._current_position_value) is None
or (current := current_value.value) is None
):
return
if target_position > current:
self._attr_is_opening = True
self._attr_is_closing = False
elif target_position < current:
self._attr_is_opening = False
self._attr_is_closing = True
else:
return
self.async_write_ha_state()
async def async_set_cover_position(self, **kwargs: Any) -> None:
"""Move the cover to a specific position."""
assert self._target_position_value
await self._async_set_value(
self._target_position_value,
self.percent_to_zwave_position(kwargs[ATTR_POSITION]),
await self._async_set_position_and_update_moving_state(
self.percent_to_zwave_position(kwargs[ATTR_POSITION])
)
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
assert self._target_position_value
await self._async_set_value(
self._target_position_value, self._fully_open_position
await self._async_set_position_and_update_moving_state(
self._fully_open_position
)
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close cover."""
assert self._target_position_value
await self._async_set_value(
self._target_position_value, self._fully_closed_position
await self._async_set_position_and_update_moving_state(
self._fully_closed_position
)
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop cover."""
assert self._stop_position_value
# Stop the cover, will stop regardless of the actual direction of travel.
await self._async_set_value(self._stop_position_value, False)
result = await self._async_set_value(self._stop_position_value, False)
# When stopping is successful (or unsupervised), we can assume the cover has stopped moving.
if result is not None and result.status in (
SetValueStatus.SUCCESS,
SetValueStatus.SUCCESS_UNSUPERVISED,
):
self._attr_is_opening = False
self._attr_is_closing = False
self.async_write_ha_state()
class CoverTiltMixin(ZWaveBaseEntity, CoverEntity):
@@ -425,15 +478,33 @@ class ZWaveWindowCovering(CoverPositionMixin, CoverTiltMixin):
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
await self._async_set_value(self._up_value, True)
result = await self._async_set_value(self._up_value, True)
# StartLevelChange: SUCCESS means the device started moving in the desired direction
if result is not None and result.status in SET_VALUE_SUCCESS:
self._attr_is_opening = True
self._attr_is_closing = False
self.async_write_ha_state()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
await self._async_set_value(self._down_value, True)
result = await self._async_set_value(self._down_value, True)
# StartLevelChange: SUCCESS means the device started moving in the desired direction
if result is not None and result.status in SET_VALUE_SUCCESS:
self._attr_is_opening = False
self._attr_is_closing = True
self.async_write_ha_state()
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
await self._async_set_value(self._up_value, False)
result = await self._async_set_value(self._up_value, False)
# When stopping is successful (or unsupervised), we can assume the cover has stopped moving.
if result is not None and result.status in (
SetValueStatus.SUCCESS,
SetValueStatus.SUCCESS_UNSUPERVISED,
):
self._attr_is_opening = False
self._attr_is_closing = False
self.async_write_ha_state()
class ZwaveMotorizedBarrier(ZWaveBaseEntity, CoverEntity):

502
tests/components/zwave_js/test_cover.py
View File

@@ -1,12 +1,17 @@
"""Test the Z-Wave JS cover platform."""
from __future__ import annotations
import logging
from typing import Any
from unittest.mock import MagicMock
import pytest
from zwave_js_server.const import (
CURRENT_STATE_PROPERTY,
CURRENT_VALUE_PROPERTY,
CommandClass,
SetValueStatus,
)
from zwave_js_server.event import Event
from zwave_js_server.model.node import Node
@@ -42,6 +47,8 @@ from homeassistant.core import HomeAssistant
from .common import replace_value_of_zwave_value
from tests.common import MockConfigEntry
WINDOW_COVER_ENTITY = "cover.zws_12"
GDC_COVER_ENTITY = "cover.aeon_labs_garage_door_controller_gen5"
BLIND_COVER_ENTITY = "cover.window_blind_controller"
@@ -60,7 +67,10 @@ def platforms() -> list[str]:
async def test_window_cover(
hass: HomeAssistant, client, chain_actuator_zws12, integration
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test the cover entity."""
node = chain_actuator_zws12
@@ -243,7 +253,10 @@ async def test_window_cover(
async def test_fibaro_fgr222_shutter_cover(
hass: HomeAssistant, client, fibaro_fgr222_shutter, integration
hass: HomeAssistant,
client: MagicMock,
fibaro_fgr222_shutter: Node,
integration: MockConfigEntry,
) -> None:
"""Test tilt function of the Fibaro Shutter devices."""
state = hass.states.get(FIBARO_FGR_222_SHUTTER_COVER_ENTITY)
@@ -344,7 +357,10 @@ async def test_fibaro_fgr222_shutter_cover(
async def test_fibaro_fgr223_shutter_cover(
hass: HomeAssistant, client, fibaro_fgr223_shutter, integration
hass: HomeAssistant,
client: MagicMock,
fibaro_fgr223_shutter: Node,
integration: MockConfigEntry,
) -> None:
"""Test tilt function of the Fibaro Shutter devices."""
state = hass.states.get(FIBARO_FGR_223_SHUTTER_COVER_ENTITY)
@@ -438,7 +454,10 @@ async def test_fibaro_fgr223_shutter_cover(
async def test_shelly_wave_shutter_cover_with_tilt(
hass: HomeAssistant, client, qubino_shutter_firmware_14_2_0, integration
hass: HomeAssistant,
client: MagicMock,
qubino_shutter_firmware_14_2_0: Node,
integration: MockConfigEntry,
) -> None:
"""Test tilt function of the Shelly Wave Shutter with firmware 14.2.0.
@@ -537,7 +556,10 @@ async def test_shelly_wave_shutter_cover_with_tilt(
async def test_aeotec_nano_shutter_cover(
hass: HomeAssistant, client, aeotec_nano_shutter, integration
hass: HomeAssistant,
client: MagicMock,
aeotec_nano_shutter: Node,
integration: MockConfigEntry,
) -> None:
"""Test movement of an Aeotec Nano Shutter cover entity. Useful to make sure the stop command logic is handled properly."""
node = aeotec_nano_shutter
@@ -655,7 +677,10 @@ async def test_aeotec_nano_shutter_cover(
async def test_blind_cover(
hass: HomeAssistant, client, iblinds_v2, integration
hass: HomeAssistant,
client: MagicMock,
iblinds_v2: Node,
integration: MockConfigEntry,
) -> None:
"""Test a blind cover entity."""
state = hass.states.get(BLIND_COVER_ENTITY)
@@ -665,7 +690,10 @@ async def test_blind_cover(
async def test_shutter_cover(
hass: HomeAssistant, client, qubino_shutter, integration
hass: HomeAssistant,
client: MagicMock,
qubino_shutter: Node,
integration: MockConfigEntry,
) -> None:
"""Test a shutter cover entity."""
state = hass.states.get(SHUTTER_COVER_ENTITY)
@@ -675,7 +703,10 @@ async def test_shutter_cover(
async def test_motor_barrier_cover(
hass: HomeAssistant, client, gdc_zw062, integration
hass: HomeAssistant,
client: MagicMock,
gdc_zw062: Node,
integration: MockConfigEntry,
) -> None:
"""Test the cover entity."""
node = gdc_zw062
@@ -853,7 +884,10 @@ async def test_motor_barrier_cover(
async def test_motor_barrier_cover_no_primary_value(
hass: HomeAssistant, client, gdc_zw062_state, integration
hass: HomeAssistant,
client: MagicMock,
gdc_zw062_state: dict[str, Any],
integration: MockConfigEntry,
) -> None:
"""Test the cover entity where primary value value is None."""
node_state = replace_value_of_zwave_value(
@@ -879,7 +913,10 @@ async def test_motor_barrier_cover_no_primary_value(
async def test_fibaro_fgr222_shutter_cover_no_tilt(
hass: HomeAssistant, client, fibaro_fgr222_shutter_state, integration
hass: HomeAssistant,
client: MagicMock,
fibaro_fgr222_shutter_state: dict[str, Any],
integration: MockConfigEntry,
) -> None:
"""Test tilt function of the Fibaro Shutter devices with tilt value is None."""
node_state = replace_value_of_zwave_value(
@@ -909,7 +946,10 @@ async def test_fibaro_fgr222_shutter_cover_no_tilt(
async def test_fibaro_fgr223_shutter_cover_no_tilt(
hass: HomeAssistant, client, fibaro_fgr223_shutter_state, integration
hass: HomeAssistant,
client: MagicMock,
fibaro_fgr223_shutter_state: dict[str, Any],
integration: MockConfigEntry,
) -> None:
"""Test absence of tilt function for Fibaro Shutter roller blind.
@@ -938,7 +978,10 @@ async def test_fibaro_fgr223_shutter_cover_no_tilt(
async def test_iblinds_v3_cover(
hass: HomeAssistant, client, iblinds_v3, integration
hass: HomeAssistant,
client: MagicMock,
iblinds_v3: Node,
integration: MockConfigEntry,
) -> None:
"""Test iBlinds v3 cover which uses Window Covering CC."""
entity_id = "cover.blind_west_bed_1_horizontal_slats_angle"
@@ -1042,7 +1085,10 @@ async def test_iblinds_v3_cover(
async def test_nice_ibt4zwave_cover(
hass: HomeAssistant, client, nice_ibt4zwave, integration
hass: HomeAssistant,
client: MagicMock,
nice_ibt4zwave: Node,
integration: MockConfigEntry,
) -> None:
"""Test Nice IBT4ZWAVE cover."""
entity_id = "cover.portail"
@@ -1102,7 +1148,10 @@ async def test_nice_ibt4zwave_cover(
async def test_window_covering_open_close(
hass: HomeAssistant, client, window_covering_outbound_bottom, integration
hass: HomeAssistant,
client: MagicMock,
window_covering_outbound_bottom: Node,
integration: MockConfigEntry,
) -> None:
"""Test Window Covering device open and close commands.
@@ -1202,3 +1251,428 @@ async def test_window_covering_open_close(
assert args["value"] is False
client.async_send_command.reset_mock()
async def test_multilevel_switch_cover_moving_state_working(
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test opening state with Supervision WORKING on Multilevel Switch cover."""
node = chain_actuator_zws12
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state
assert state.state == CoverState.CLOSED
# Simulate Supervision WORKING response
client.async_send_command.return_value = {
"result": {"status": SetValueStatus.WORKING}
}
# Open cover - should set OPENING state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.OPENING
# Simulate intermediate position update (still moving)
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Multilevel Switch",
"commandClass": 38,
"endpoint": 0,
"property": "currentValue",
"newValue": 50,
"prevValue": 0,
"propertyName": "currentValue",
},
},
)
node.receive_event(event)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.OPENING
# Simulate targetValue update (driver sets this when command is sent)
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Multilevel Switch",
"commandClass": 38,
"endpoint": 0,
"property": "targetValue",
"newValue": 99,
"prevValue": 0,
"propertyName": "targetValue",
},
},
)
node.receive_event(event)
# Simulate reaching target position
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Multilevel Switch",
"commandClass": 38,
"endpoint": 0,
"property": "currentValue",
"newValue": 99,
"prevValue": 50,
"propertyName": "currentValue",
},
},
)
node.receive_event(event)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.OPEN
async def test_multilevel_switch_cover_moving_state_closing(
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test closing state with Supervision WORKING on Multilevel Switch cover."""
node = chain_actuator_zws12
# First set position to open
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Multilevel Switch",
"commandClass": 38,
"endpoint": 0,
"property": "currentValue",
"newValue": 99,
"prevValue": 0,
"propertyName": "currentValue",
},
},
)
node.receive_event(event)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.OPEN
# Simulate Supervision WORKING response
client.async_send_command.return_value = {
"result": {"status": SetValueStatus.WORKING}
}
# Close cover - should set CLOSING state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.CLOSING
async def test_multilevel_switch_cover_moving_state_success_no_moving(
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test that SUCCESS does not set moving state on Multilevel Switch cover."""
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state
assert state.state == CoverState.CLOSED
# Default mock already returns status 255 (SUCCESS)
# Open cover - SUCCESS means device already at target, no moving state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
# State should still be CLOSED since no value update has been received
# and SUCCESS means the command completed immediately
assert state.state == CoverState.CLOSED
async def test_multilevel_switch_cover_moving_state_unsupervised(
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test SUCCESS_UNSUPERVISED sets moving state on Multilevel Switch cover."""
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state
assert state.state == CoverState.CLOSED
# Simulate SUCCESS_UNSUPERVISED response
client.async_send_command.return_value = {
"result": {"status": SetValueStatus.SUCCESS_UNSUPERVISED}
}
# Open cover - should set OPENING state optimistically
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.OPENING
async def test_multilevel_switch_cover_moving_state_stop_clears(
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test stop_cover clears moving state on Multilevel Switch cover."""
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state
assert state.state == CoverState.CLOSED
# Simulate WORKING response
client.async_send_command.return_value = {
"result": {"status": SetValueStatus.WORKING}
}
# Open cover to set OPENING state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.OPENING
# Reset to SUCCESS for stop command
client.async_send_command.return_value = {
"result": {"status": SetValueStatus.SUCCESS}
}
# Stop cover - should clear opening state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
# Cover is still at position 0 (closed), so is_closed returns True
assert state.state == CoverState.CLOSED
async def test_multilevel_switch_cover_moving_state_set_position(
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test moving state direction with set_cover_position on Multilevel Switch cover."""
node = chain_actuator_zws12
# First set position to 50 (open)
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Multilevel Switch",
"commandClass": 38,
"endpoint": 0,
"property": "currentValue",
"newValue": 50,
"prevValue": 0,
"propertyName": "currentValue",
},
},
)
node.receive_event(event)
# Simulate WORKING response
client.async_send_command.return_value = {
"result": {"status": SetValueStatus.WORKING}
}
# Set position to 20 (closing direction)
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY, ATTR_POSITION: 20},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.CLOSING
# Set position to 80 (opening direction)
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_SET_COVER_POSITION,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY, ATTR_POSITION: 80},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.OPENING
async def test_window_covering_cover_moving_state(
hass: HomeAssistant,
client: MagicMock,
window_covering_outbound_bottom: Node,
integration: MockConfigEntry,
) -> None:
"""Test moving state for Window Covering CC (StartLevelChange commands)."""
node = window_covering_outbound_bottom
entity_id = "cover.node_2_outbound_bottom"
state = hass.states.get(entity_id)
assert state
# Default mock returns SUCCESS (255). For StartLevelChange,
# SUCCESS means the device started moving.
# Open cover - should set OPENING state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.state == CoverState.OPENING
client.async_send_command.reset_mock()
# Stop cover - should clear moving state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_STOP_COVER,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.state not in (CoverState.OPENING, CoverState.CLOSING)
client.async_send_command.reset_mock()
# Close cover - should set CLOSING state
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_CLOSE_COVER,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.state == CoverState.CLOSING
# Simulate reaching target: currentValue matches targetValue
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Window Covering",
"commandClass": 106,
"endpoint": 0,
"property": "targetValue",
"propertyKey": 13,
"newValue": 0,
"prevValue": 52,
"propertyName": "targetValue",
},
},
)
node.receive_event(event)
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Window Covering",
"commandClass": 106,
"endpoint": 0,
"property": "currentValue",
"propertyKey": 13,
"newValue": 0,
"prevValue": 52,
"propertyName": "currentValue",
},
},
)
node.receive_event(event)
state = hass.states.get(entity_id)
assert state.state == CoverState.CLOSED
async def test_multilevel_switch_cover_moving_state_none_result(
hass: HomeAssistant,
client: MagicMock,
chain_actuator_zws12: Node,
integration: MockConfigEntry,
) -> None:
"""Test None result (node asleep) does not set moving state on Multilevel Switch cover."""
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state
assert state.state == CoverState.CLOSED
# Simulate None result (node asleep/command queued).
# When node.async_send_command returns None, async_set_value returns None.
client.async_send_command.return_value = None
# Open cover - should NOT set OPENING state since result is None
await hass.services.async_call(
COVER_DOMAIN,
SERVICE_OPEN_COVER,
{ATTR_ENTITY_ID: WINDOW_COVER_ENTITY},
blocking=True,
)
state = hass.states.get(WINDOW_COVER_ENTITY)
assert state.state == CoverState.CLOSED