Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-10 07:12:54 +01:00
Code Activity

Restore KNX cover state and mark uninitialized covers as assumed (#177216)

Browse Source
This commit is contained in:
Matthias Alphart
2026-07-26 01:10:22 +02:00
committed by GitHub
parent e75a891d65
commit c1eaade9d9
2 changed files with 162 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/knx/cover.py
+40 -2
View File
@@ -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:
tests/components/knx/test_cover.py
+122 -6
View File
@@ -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",