From c1eaade9d97141a9269c19af0d98041568a8d038 Mon Sep 17 00:00:00 2001 From: Matthias Alphart Date: Sun, 26 Jul 2026 01:10:22 +0200 Subject: [PATCH] Restore KNX cover state and mark uninitialized covers as assumed (#177216) --- homeassistant/components/knx/cover.py | 42 ++++++++- tests/components/knx/test_cover.py | 128 ++++++++++++++++++++++++-- 2 files changed, 162 insertions(+), 8 deletions(-) diff --git a/homeassistant/components/knx/cover.py b/homeassistant/components/knx/cover.py index 1bb22b527559..b17c0c1ca9f1 100644 --- a/homeassistant/components/knx/cover.py +++ b/homeassistant/components/knx/cover.py @@ -7,6 +7,8 @@ from xknx.devices import Cover as XknxCover from homeassistant import config_entries from homeassistant.components.cover import ( + ATTR_CURRENT_POSITION, + ATTR_CURRENT_TILT_POSITION, ATTR_POSITION, ATTR_TILT_POSITION, CoverDeviceClass, @@ -17,6 +19,8 @@ from homeassistant.const import ( CONF_DEVICE_CLASS, CONF_ENTITY_CATEGORY, CONF_NAME, + STATE_UNAVAILABLE, + STATE_UNKNOWN, Platform, ) from homeassistant.core import HomeAssistant @@ -24,6 +28,7 @@ from homeassistant.helpers.entity_platform import ( AddConfigEntryEntitiesCallback, async_get_current_platform, ) +from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.typing import ConfigType from .const import CONF_SYNC_STATE, DOMAIN, KNX_MODULE_KEY, CoverConf @@ -74,20 +79,20 @@ async def async_setup_entry( async_add_entities(entities) -class _KnxCover(CoverEntity): +class _KnxCover(CoverEntity, RestoreEntity): """Representation of a KNX cover.""" _device: XknxCover def init_base(self) -> None: """Initialize common attributes - may be based on xknx device instance.""" - _supports_tilt = False self._attr_supported_features = ( CoverEntityFeature.CLOSE | CoverEntityFeature.OPEN ) if self._device.supports_position or self._device.supports_stop: # when stop is supported, xknx travelcalculator can set position self._attr_supported_features |= CoverEntityFeature.SET_POSITION + _supports_tilt = False if self._device.step.writable: _supports_tilt = True self._attr_supported_features |= ( @@ -105,6 +110,39 @@ class _KnxCover(CoverEntity): self._attr_device_class = CoverDeviceClass.BLIND if _supports_tilt else None + @override + async def async_added_to_hass(self) -> None: + """Restore last known position and tilt. + + For readable group addresses this bridges the gap until the bus read + response arrives; for non-readable ones it is the only source of state. + """ + await super().async_added_to_hass() + if (last_state := await self.async_get_last_state()) is None or ( + last_state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE) + ): + return + if (position := last_state.attributes.get(ATTR_CURRENT_POSITION)) is not None: + # In KNX 0 is open, 100 is closed. + self._device.travelcalculator.set_position(100 - position) + if ( + tilt_position := last_state.attributes.get(ATTR_CURRENT_TILT_POSITION) + ) is not None: + self._device.angle.value = 100 - tilt_position + + @property + @override + def assumed_state(self) -> bool: + """Return True if unable to access real state of the entity.""" + # Without a known position or movement value, the position is only + # read from the restored state in the travelcalculator. This prevents + # out-of-sync positions from disabling controls in the UI. + return ( + self._device.position_current.value is None + and self._device.position_target.value is None + and self._device.updown.value is None + ) + @property @override def current_cover_position(self) -> int | None: diff --git a/tests/components/knx/test_cover.py b/tests/components/knx/test_cover.py index 1e91cb9dcba6..a8452f4ad7d5 100644 --- a/tests/components/knx/test_cover.py +++ b/tests/components/knx/test_cover.py @@ -4,15 +4,20 @@ from typing import Any import pytest -from homeassistant.components.cover import CoverEntityFeature, CoverState +from homeassistant.components.cover import ( + ATTR_CURRENT_POSITION, + ATTR_CURRENT_TILT_POSITION, + CoverEntityFeature, + CoverState, +) from homeassistant.components.knx.schema import CoverSchema -from homeassistant.const import CONF_NAME, STATE_UNKNOWN, Platform -from homeassistant.core import HomeAssistant +from homeassistant.const import CONF_NAME, STATE_UNAVAILABLE, STATE_UNKNOWN, Platform +from homeassistant.core import HomeAssistant, State from . import KnxEntityGenerator from .conftest import KNXTestKit -from tests.common import async_capture_events +from tests.common import async_capture_events, mock_restore_cache async def test_cover_basic(hass: HomeAssistant, knx: KNXTestKit) -> None: @@ -80,6 +85,7 @@ async def test_cover_basic(hass: HomeAssistant, knx: KNXTestKit) -> None: knx.assert_state( "cover.test", CoverState.CLOSING, + assumed_state=None, ) assert len(events) == 1 @@ -169,8 +175,108 @@ async def test_cover_tilt_move_short(hass: HomeAssistant, knx: KNXTestKit) -> No await knx.assert_write("1/0/1", 0) +async def test_cover_restore_assumed_state( + hass: HomeAssistant, knx: KNXTestKit +) -> None: + """Test a cover without position feedback restores its state and is assumed.""" + mock_restore_cache( + hass, + ( + State( + "cover.test", + CoverState.OPEN, + {ATTR_CURRENT_POSITION: 40, ATTR_CURRENT_TILT_POSITION: 60}, + ), + ), + ) + await knx.setup_integration( + { + CoverSchema.PLATFORM: { + CONF_NAME: "test", + CoverSchema.CONF_MOVE_LONG_ADDRESS: "1/0/0", + CoverSchema.CONF_STOP_ADDRESS: "1/0/1", + CoverSchema.CONF_ANGLE_ADDRESS: "1/0/2", + } + } + ) + # position and tilt can't be read from the bus - no telegrams are sent + await knx.assert_no_telegram() + knx.assert_state( + "cover.test", + CoverState.OPEN, + current_position=40, + current_tilt_position=60, + assumed_state=True, + ) + + @pytest.mark.parametrize( - ("knx_data", "read_responses", "initial_state", "supported_features"), + "restored_state", + [STATE_UNKNOWN, STATE_UNAVAILABLE], +) +async def test_cover_restore_unknown_state( + hass: HomeAssistant, knx: KNXTestKit, restored_state: str +) -> None: + """Test a cover doesn't restore a non-numeric position.""" + mock_restore_cache( + hass, + (State("cover.test", restored_state, {ATTR_CURRENT_POSITION: 40}),), + ) + await knx.setup_integration( + { + CoverSchema.PLATFORM: { + CONF_NAME: "test", + CoverSchema.CONF_MOVE_LONG_ADDRESS: "1/0/0", + CoverSchema.CONF_STOP_ADDRESS: "1/0/1", + } + } + ) + await knx.assert_no_telegram() + knx.assert_state( + "cover.test", + STATE_UNKNOWN, + current_position=None, + assumed_state=True, + ) + + +async def test_cover_restore_readable(hass: HomeAssistant, knx: KNXTestKit) -> None: + """Test a readable cover shows the restored state until the bus read completes.""" + mock_restore_cache( + hass, + (State("cover.test", CoverState.CLOSED, {ATTR_CURRENT_POSITION: 0}),), + ) + await knx.setup_integration( + { + CoverSchema.PLATFORM: { + CONF_NAME: "test", + CoverSchema.CONF_MOVE_LONG_ADDRESS: "1/0/0", + CoverSchema.CONF_POSITION_STATE_ADDRESS: "1/0/2", + CoverSchema.CONF_POSITION_ADDRESS: "1/0/3", + } + } + ) + # restored value bridges the gap until the bus read completes - it is + # assumed as long as it hasn't been confirmed by the bus + knx.assert_state( + "cover.test", + CoverState.CLOSED, + current_position=0, + assumed_state=True, + ) + # bus reports a different position - the confirmed value overwrites the + # restored one and the state is no longer assumed + await knx.assert_read("1/0/2", response=(0x00,)) + knx.assert_state( + "cover.test", + CoverState.OPEN, + current_position=100, + assumed_state=None, + ) + + +@pytest.mark.parametrize( + ("knx_data", "read_responses", "initial_state", "supported_features", "assumed"), [ ( { @@ -180,6 +286,7 @@ async def test_cover_tilt_move_short(hass: HomeAssistant, knx: KNXTestKit) -> No {}, STATE_UNKNOWN, CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE, + True, ), ( { @@ -192,6 +299,7 @@ async def test_cover_tilt_move_short(hass: HomeAssistant, knx: KNXTestKit) -> No CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.SET_POSITION, + None, ), ( { @@ -213,6 +321,7 @@ async def test_cover_tilt_move_short(hass: HomeAssistant, knx: KNXTestKit) -> No | CoverEntityFeature.SET_TILT_POSITION | CoverEntityFeature.STOP | CoverEntityFeature.STOP_TILT, + None, ), ], ) @@ -223,6 +332,7 @@ async def test_cover_ui_create( read_responses: dict[str, int | tuple[int]], initial_state: str, supported_features: int, + assumed: bool | None, ) -> None: """Test creating a cover.""" await knx.setup_integration() @@ -234,7 +344,12 @@ async def test_cover_ui_create( # created entity sends read-request to KNX bus for ga, value in read_responses.items(): await knx.assert_read(ga, response=value, ignore_order=True) - knx.assert_state("cover.test", initial_state, supported_features=supported_features) + knx.assert_state( + "cover.test", + initial_state, + supported_features=supported_features, + assumed_state=assumed, + ) async def test_cover_ui_load(knx: KNXTestKit) -> None: @@ -249,6 +364,7 @@ async def test_cover_ui_load(knx: KNXTestKit) -> None: "cover.minimal", STATE_UNKNOWN, supported_features=CoverEntityFeature.CLOSE | CoverEntityFeature.OPEN, + assumed_state=True, ) knx.assert_state( "cover.position_only",