Add support to create KNX Cover entities from UI (#141944)

* Add UI to create KNX Cover entities

* Use common constants source for UI and YAML config keys
This commit is contained in:
Matthias Alphart 2025-05-09 11:28:25 +02:00 committed by GitHub
parent b4ae08f83d
commit 307bb05653
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 506 additions and 90 deletions

View File

@ -160,6 +160,7 @@ SUPPORTED_PLATFORMS_YAML: Final = {
SUPPORTED_PLATFORMS_UI: Final = {
Platform.BINARY_SENSOR,
Platform.COVER,
Platform.LIGHT,
Platform.SWITCH,
}
@ -182,3 +183,13 @@ CURRENT_HVAC_ACTIONS: Final = {
HVACMode.FAN_ONLY: HVACAction.FAN,
HVACMode.DRY: HVACAction.DRYING,
}
class CoverConf:
"""Common config keys for cover."""
TRAVELLING_TIME_DOWN: Final = "travelling_time_down"
TRAVELLING_TIME_UP: Final = "travelling_time_up"
INVERT_UPDOWN: Final = "invert_updown"
INVERT_POSITION: Final = "invert_position"
INVERT_ANGLE: Final = "invert_angle"

View File

@ -2,9 +2,9 @@
from __future__ import annotations
from collections.abc import Callable
from typing import Any
from typing import Any, Literal
from xknx import XKNX
from xknx.devices import Cover as XknxCover
from homeassistant import config_entries
@ -22,13 +22,28 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.entity_platform import (
AddConfigEntryEntitiesCallback,
async_get_current_platform,
)
from homeassistant.helpers.typing import ConfigType
from . import KNXModule
from .const import KNX_MODULE_KEY
from .entity import KnxYamlEntity
from .const import CONF_SYNC_STATE, DOMAIN, KNX_MODULE_KEY, CoverConf
from .entity import KnxUiEntity, KnxUiEntityPlatformController, KnxYamlEntity
from .schema import CoverSchema
from .storage.const import (
CONF_ENTITY,
CONF_GA_ANGLE,
CONF_GA_PASSIVE,
CONF_GA_POSITION_SET,
CONF_GA_POSITION_STATE,
CONF_GA_STATE,
CONF_GA_STEP,
CONF_GA_STOP,
CONF_GA_UP_DOWN,
CONF_GA_WRITE,
)
async def async_setup_entry(
@ -36,52 +51,47 @@ async def async_setup_entry(
config_entry: config_entries.ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up cover(s) for KNX platform."""
"""Set up the KNX cover platform."""
knx_module = hass.data[KNX_MODULE_KEY]
config: list[ConfigType] = knx_module.config_yaml[Platform.COVER]
platform = async_get_current_platform()
knx_module.config_store.add_platform(
platform=Platform.COVER,
controller=KnxUiEntityPlatformController(
knx_module=knx_module,
entity_platform=platform,
entity_class=KnxUiCover,
),
)
async_add_entities(KNXCover(knx_module, entity_config) for entity_config in config)
entities: list[KnxYamlEntity | KnxUiEntity] = []
if yaml_platform_config := knx_module.config_yaml.get(Platform.COVER):
entities.extend(
KnxYamlCover(knx_module, entity_config)
for entity_config in yaml_platform_config
)
if ui_config := knx_module.config_store.data["entities"].get(Platform.COVER):
entities.extend(
KnxUiCover(knx_module, unique_id, config)
for unique_id, config in ui_config.items()
)
if entities:
async_add_entities(entities)
class KNXCover(KnxYamlEntity, CoverEntity):
class _KnxCover(CoverEntity):
"""Representation of a KNX cover."""
_device: XknxCover
def __init__(self, knx_module: KNXModule, config: ConfigType) -> None:
"""Initialize the cover."""
super().__init__(
knx_module=knx_module,
device=XknxCover(
xknx=knx_module.xknx,
name=config[CONF_NAME],
group_address_long=config.get(CoverSchema.CONF_MOVE_LONG_ADDRESS),
group_address_short=config.get(CoverSchema.CONF_MOVE_SHORT_ADDRESS),
group_address_stop=config.get(CoverSchema.CONF_STOP_ADDRESS),
group_address_position_state=config.get(
CoverSchema.CONF_POSITION_STATE_ADDRESS
),
group_address_angle=config.get(CoverSchema.CONF_ANGLE_ADDRESS),
group_address_angle_state=config.get(
CoverSchema.CONF_ANGLE_STATE_ADDRESS
),
group_address_position=config.get(CoverSchema.CONF_POSITION_ADDRESS),
travel_time_down=config[CoverSchema.CONF_TRAVELLING_TIME_DOWN],
travel_time_up=config[CoverSchema.CONF_TRAVELLING_TIME_UP],
invert_updown=config[CoverSchema.CONF_INVERT_UPDOWN],
invert_position=config[CoverSchema.CONF_INVERT_POSITION],
invert_angle=config[CoverSchema.CONF_INVERT_ANGLE],
),
)
self._unsubscribe_auto_updater: Callable[[], None] | None = None
self._attr_entity_category = config.get(CONF_ENTITY_CATEGORY)
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
| CoverEntityFeature.SET_POSITION
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
if self._device.step.writable:
_supports_tilt = True
self._attr_supported_features |= (
@ -97,13 +107,7 @@ class KNXCover(KnxYamlEntity, CoverEntity):
if _supports_tilt:
self._attr_supported_features |= CoverEntityFeature.STOP_TILT
self._attr_device_class = config.get(CONF_DEVICE_CLASS) or (
CoverDeviceClass.BLIND if _supports_tilt else None
)
self._attr_unique_id = (
f"{self._device.updown.group_address}_"
f"{self._device.position_target.group_address}"
)
self._attr_device_class = CoverDeviceClass.BLIND if _supports_tilt else None
@property
def current_cover_position(self) -> int | None:
@ -180,3 +184,102 @@ class KNXCover(KnxYamlEntity, CoverEntity):
async def async_stop_cover_tilt(self, **kwargs: Any) -> None:
"""Stop the cover tilt."""
await self._device.stop()
class KnxYamlCover(_KnxCover, KnxYamlEntity):
"""Representation of a KNX cover configured from YAML."""
_device: XknxCover
def __init__(self, knx_module: KNXModule, config: ConfigType) -> None:
"""Initialize the cover."""
super().__init__(
knx_module=knx_module,
device=XknxCover(
xknx=knx_module.xknx,
name=config[CONF_NAME],
group_address_long=config.get(CoverSchema.CONF_MOVE_LONG_ADDRESS),
group_address_short=config.get(CoverSchema.CONF_MOVE_SHORT_ADDRESS),
group_address_stop=config.get(CoverSchema.CONF_STOP_ADDRESS),
group_address_position_state=config.get(
CoverSchema.CONF_POSITION_STATE_ADDRESS
),
group_address_angle=config.get(CoverSchema.CONF_ANGLE_ADDRESS),
group_address_angle_state=config.get(
CoverSchema.CONF_ANGLE_STATE_ADDRESS
),
group_address_position=config.get(CoverSchema.CONF_POSITION_ADDRESS),
travel_time_down=config[CoverConf.TRAVELLING_TIME_DOWN],
travel_time_up=config[CoverConf.TRAVELLING_TIME_UP],
invert_updown=config[CoverConf.INVERT_UPDOWN],
invert_position=config[CoverConf.INVERT_POSITION],
invert_angle=config[CoverConf.INVERT_ANGLE],
),
)
self.init_base()
self._attr_entity_category = config.get(CONF_ENTITY_CATEGORY)
self._attr_unique_id = (
f"{self._device.updown.group_address}_"
f"{self._device.position_target.group_address}"
)
if custom_device_class := config.get(CONF_DEVICE_CLASS):
self._attr_device_class = custom_device_class
def _create_ui_cover(xknx: XKNX, knx_config: ConfigType, name: str) -> XknxCover:
"""Return a KNX Light device to be used within XKNX."""
def get_address(
key: str, address_type: Literal["write", "state"] = CONF_GA_WRITE
) -> str | None:
"""Get a single group address for given key."""
return knx_config[key][address_type] if key in knx_config else None
def get_addresses(
key: str, address_type: Literal["write", "state"] = CONF_GA_STATE
) -> list[Any] | None:
"""Get group address including passive addresses as list."""
return (
[knx_config[key][address_type], *knx_config[key][CONF_GA_PASSIVE]]
if key in knx_config
else None
)
return XknxCover(
xknx=xknx,
name=name,
group_address_long=get_addresses(CONF_GA_UP_DOWN, CONF_GA_WRITE),
group_address_short=get_addresses(CONF_GA_STEP, CONF_GA_WRITE),
group_address_stop=get_addresses(CONF_GA_STOP, CONF_GA_WRITE),
group_address_position=get_addresses(CONF_GA_POSITION_SET, CONF_GA_WRITE),
group_address_position_state=get_addresses(CONF_GA_POSITION_STATE),
group_address_angle=get_address(CONF_GA_ANGLE),
group_address_angle_state=get_addresses(CONF_GA_ANGLE),
travel_time_down=knx_config[CoverConf.TRAVELLING_TIME_DOWN],
travel_time_up=knx_config[CoverConf.TRAVELLING_TIME_UP],
invert_updown=knx_config.get(CoverConf.INVERT_UPDOWN, False),
invert_position=knx_config.get(CoverConf.INVERT_POSITION, False),
invert_angle=knx_config.get(CoverConf.INVERT_ANGLE, False),
sync_state=knx_config[CONF_SYNC_STATE],
)
class KnxUiCover(_KnxCover, KnxUiEntity):
"""Representation of a KNX cover configured from the UI."""
_device: XknxCover
def __init__(
self, knx_module: KNXModule, unique_id: str, config: dict[str, Any]
) -> None:
"""Initialize KNX cover."""
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
)
self._device = _create_ui_cover(
knx_module.xknx, config[DOMAIN], config[CONF_ENTITY][CONF_NAME]
)
self.init_base()

View File

@ -56,6 +56,7 @@ from .const import (
CONF_SYNC_STATE,
KNX_ADDRESS,
ColorTempModes,
CoverConf,
FanZeroMode,
)
from .validation import (
@ -453,11 +454,6 @@ class CoverSchema(KNXPlatformSchema):
CONF_POSITION_STATE_ADDRESS = "position_state_address"
CONF_ANGLE_ADDRESS = "angle_address"
CONF_ANGLE_STATE_ADDRESS = "angle_state_address"
CONF_TRAVELLING_TIME_DOWN = "travelling_time_down"
CONF_TRAVELLING_TIME_UP = "travelling_time_up"
CONF_INVERT_UPDOWN = "invert_updown"
CONF_INVERT_POSITION = "invert_position"
CONF_INVERT_ANGLE = "invert_angle"
DEFAULT_TRAVEL_TIME = 25
DEFAULT_NAME = "KNX Cover"
@ -474,14 +470,14 @@ class CoverSchema(KNXPlatformSchema):
vol.Optional(CONF_ANGLE_ADDRESS): ga_list_validator,
vol.Optional(CONF_ANGLE_STATE_ADDRESS): ga_list_validator,
vol.Optional(
CONF_TRAVELLING_TIME_DOWN, default=DEFAULT_TRAVEL_TIME
CoverConf.TRAVELLING_TIME_DOWN, default=DEFAULT_TRAVEL_TIME
): cv.positive_float,
vol.Optional(
CONF_TRAVELLING_TIME_UP, default=DEFAULT_TRAVEL_TIME
CoverConf.TRAVELLING_TIME_UP, default=DEFAULT_TRAVEL_TIME
): cv.positive_float,
vol.Optional(CONF_INVERT_UPDOWN, default=False): cv.boolean,
vol.Optional(CONF_INVERT_POSITION, default=False): cv.boolean,
vol.Optional(CONF_INVERT_ANGLE, default=False): cv.boolean,
vol.Optional(CoverConf.INVERT_UPDOWN, default=False): cv.boolean,
vol.Optional(CoverConf.INVERT_POSITION, default=False): cv.boolean,
vol.Optional(CoverConf.INVERT_ANGLE, default=False): cv.boolean,
vol.Optional(CONF_DEVICE_CLASS): COVER_DEVICE_CLASSES_SCHEMA,
vol.Optional(CONF_ENTITY_CATEGORY): ENTITY_CATEGORIES_SCHEMA,
}

View File

@ -27,3 +27,9 @@ CONF_GA_WHITE_BRIGHTNESS: Final = "ga_white_brightness"
CONF_GA_WHITE_SWITCH: Final = "ga_white_switch"
CONF_GA_HUE: Final = "ga_hue"
CONF_GA_SATURATION: Final = "ga_saturation"
CONF_GA_UP_DOWN: Final = "ga_up_down"
CONF_GA_STOP: Final = "ga_stop"
CONF_GA_STEP: Final = "ga_step"
CONF_GA_POSITION_SET: Final = "ga_position_set"
CONF_GA_POSITION_STATE: Final = "ga_position_state"
CONF_GA_ANGLE: Final = "ga_angle"

View File

@ -25,6 +25,7 @@ from ..const import (
DOMAIN,
SUPPORTED_PLATFORMS_UI,
ColorTempModes,
CoverConf,
)
from ..validation import sync_state_validator
from .const import (
@ -33,6 +34,7 @@ from .const import (
CONF_DATA,
CONF_DEVICE_INFO,
CONF_ENTITY,
CONF_GA_ANGLE,
CONF_GA_BLUE_BRIGHTNESS,
CONF_GA_BLUE_SWITCH,
CONF_GA_BRIGHTNESS,
@ -42,12 +44,17 @@ from .const import (
CONF_GA_GREEN_SWITCH,
CONF_GA_HUE,
CONF_GA_PASSIVE,
CONF_GA_POSITION_SET,
CONF_GA_POSITION_STATE,
CONF_GA_RED_BRIGHTNESS,
CONF_GA_RED_SWITCH,
CONF_GA_SATURATION,
CONF_GA_SENSOR,
CONF_GA_STATE,
CONF_GA_STEP,
CONF_GA_STOP,
CONF_GA_SWITCH,
CONF_GA_UP_DOWN,
CONF_GA_WHITE_BRIGHTNESS,
CONF_GA_WHITE_SWITCH,
CONF_GA_WRITE,
@ -121,15 +128,64 @@ BINARY_SENSOR_SCHEMA = vol.Schema(
}
)
SWITCH_SCHEMA = vol.Schema(
COVER_SCHEMA = vol.Schema(
{
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA,
vol.Required(DOMAIN): {
vol.Optional(CONF_INVERT, default=False): bool,
vol.Required(CONF_GA_SWITCH): GASelector(write_required=True),
vol.Optional(CONF_RESPOND_TO_READ, default=False): bool,
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
},
vol.Required(DOMAIN): vol.All(
vol.Schema(
{
**optional_ga_schema(CONF_GA_UP_DOWN, GASelector(state=False)),
vol.Optional(CoverConf.INVERT_UPDOWN): selector.BooleanSelector(),
**optional_ga_schema(CONF_GA_STOP, GASelector(state=False)),
**optional_ga_schema(CONF_GA_STEP, GASelector(state=False)),
**optional_ga_schema(CONF_GA_POSITION_SET, GASelector(state=False)),
**optional_ga_schema(
CONF_GA_POSITION_STATE, GASelector(write=False)
),
vol.Optional(CoverConf.INVERT_POSITION): selector.BooleanSelector(),
**optional_ga_schema(CONF_GA_ANGLE, GASelector()),
vol.Optional(CoverConf.INVERT_ANGLE): selector.BooleanSelector(),
vol.Optional(
CoverConf.TRAVELLING_TIME_DOWN, default=25
): selector.NumberSelector(
selector.NumberSelectorConfig(
min=0, max=1000, step=0.1, unit_of_measurement="s"
)
),
vol.Optional(
CoverConf.TRAVELLING_TIME_UP, default=25
): selector.NumberSelector(
selector.NumberSelectorConfig(
min=0, max=1000, step=0.1, unit_of_measurement="s"
)
),
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
},
extra=vol.REMOVE_EXTRA,
),
vol.Any(
vol.Schema(
{
vol.Required(CONF_GA_UP_DOWN): GASelector(
state=False, write_required=True
)
},
extra=vol.ALLOW_EXTRA,
),
vol.Schema(
{
vol.Required(CONF_GA_POSITION_SET): GASelector(
state=False, write_required=True
)
},
extra=vol.ALLOW_EXTRA,
),
msg=(
"At least one of 'Up/Down control' or"
" 'Position - Set position' is required."
),
),
),
}
)
@ -226,6 +282,19 @@ LIGHT_SCHEMA = vol.Schema(
}
)
SWITCH_SCHEMA = vol.Schema(
{
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA,
vol.Required(DOMAIN): {
vol.Optional(CONF_INVERT, default=False): bool,
vol.Required(CONF_GA_SWITCH): GASelector(write_required=True),
vol.Optional(CONF_RESPOND_TO_READ, default=False): bool,
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
},
}
)
ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All(
vol.Schema(
{
@ -243,11 +312,14 @@ ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All(
Platform.BINARY_SENSOR: vol.Schema(
{vol.Required(CONF_DATA): BINARY_SENSOR_SCHEMA}, extra=vol.ALLOW_EXTRA
),
Platform.SWITCH: vol.Schema(
{vol.Required(CONF_DATA): SWITCH_SCHEMA}, extra=vol.ALLOW_EXTRA
Platform.COVER: vol.Schema(
{vol.Required(CONF_DATA): COVER_SCHEMA}, extra=vol.ALLOW_EXTRA
),
Platform.LIGHT: vol.Schema(
{vol.Required("data"): LIGHT_SCHEMA}, extra=vol.ALLOW_EXTRA
{vol.Required(CONF_DATA): LIGHT_SCHEMA}, extra=vol.ALLOW_EXTRA
),
Platform.SWITCH: vol.Schema(
{vol.Required(CONF_DATA): SWITCH_SCHEMA}, extra=vol.ALLOW_EXTRA
),
},
),

View File

@ -43,7 +43,20 @@ class GASelector:
self._add_group_addresses(schema)
self._add_passive(schema)
self._add_dpt(schema)
return vol.Schema(schema)
return vol.Schema(
vol.All(
schema,
vol.Schema( # one group address shall be included
vol.Any(
{vol.Required(CONF_GA_WRITE): vol.IsTrue()},
{vol.Required(CONF_GA_STATE): vol.IsTrue()},
{vol.Required(CONF_GA_PASSIVE): vol.IsTrue()},
msg="At least one group address must be set",
),
extra=vol.ALLOW_EXTRA,
),
)
)
def _add_group_addresses(self, schema: dict[vol.Marker, Any]) -> None:
"""Add basic group address items to the schema."""

View File

@ -0,0 +1,82 @@
{
"version": 1,
"minor_version": 1,
"key": "knx/config_store.json",
"data": {
"entities": {
"cover": {
"knx_es_01JQNM9A9G03952ZH0GDF51HB6": {
"entity": {
"name": "minimal",
"entity_category": null,
"device_info": null
},
"knx": {
"ga_up_down": {
"write": "1/0/1",
"passive": []
},
"travelling_time_down": 25.0,
"travelling_time_up": 25.0,
"sync_state": true
}
},
"knx_es_01JQNQVEB7WT3MYCX61RK361F8": {
"entity": {
"name": "position_only",
"entity_category": null,
"device_info": null
},
"knx": {
"ga_position_set": {
"write": "2/0/1",
"passive": []
},
"ga_position_state": {
"state": "2/0/0",
"passive": []
},
"invert_position": true,
"travelling_time_up": 25.0,
"travelling_time_down": 25.0,
"sync_state": true
}
},
"knx_es_01JQNQSDS4ZW96TX27S2NT3FYQ": {
"entity": {
"name": "tiltable",
"entity_category": null,
"device_info": null
},
"knx": {
"ga_up_down": {
"write": "3/0/1",
"passive": []
},
"ga_stop": {
"write": "3/0/2",
"passive": []
},
"ga_position_set": {
"write": "3/1/1",
"passive": []
},
"ga_position_state": {
"state": "3/1/0",
"passive": []
},
"ga_angle": {
"write": "3/2/1",
"state": "3/2/0",
"passive": []
},
"travelling_time_down": 16.0,
"travelling_time_up": 16.0,
"invert_angle": true,
"sync_state": true
}
}
}
}
}
}

View File

@ -1,10 +1,15 @@
"""Test KNX cover."""
from homeassistant.components.cover import CoverState
from typing import Any
import pytest
from homeassistant.components.cover import CoverEntityFeature, CoverState
from homeassistant.components.knx.schema import CoverSchema
from homeassistant.const import CONF_NAME
from homeassistant.const import CONF_NAME, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant
from . import KnxEntityGenerator
from .conftest import KNXTestKit
from tests.common import async_capture_events
@ -160,3 +165,103 @@ async def test_cover_tilt_move_short(hass: HomeAssistant, knx: KNXTestKit) -> No
"cover", "open_cover_tilt", target={"entity_id": "cover.test"}, blocking=True
)
await knx.assert_write("1/0/1", 0)
@pytest.mark.parametrize(
("knx_data", "read_responses", "initial_state", "supported_features"),
[
(
{
"ga_up_down": {"write": "1/0/1"},
"sync_state": True,
},
{},
STATE_UNKNOWN,
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE,
),
(
{
"ga_position_set": {"write": "2/0/1"},
"ga_position_state": {"state": "2/0/0"},
"sync_state": True,
},
{"2/0/0": (0x00,)},
CoverState.OPEN,
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.SET_POSITION,
),
(
{
"ga_up_down": {"write": "3/0/1", "passive": []},
"ga_stop": {"write": "3/0/2", "passive": []},
"ga_position_set": {"write": "3/1/1", "passive": []},
"ga_position_state": {"state": "3/1/0", "passive": []},
"ga_angle": {"write": "3/2/1", "state": "3/2/0", "passive": []},
"travelling_time_down": 16.0,
"travelling_time_up": 16.0,
"invert_angle": True,
"sync_state": True,
},
{"3/1/0": (0x00,), "3/2/0": (0x00,)},
CoverState.OPEN,
CoverEntityFeature.CLOSE
| CoverEntityFeature.OPEN
| CoverEntityFeature.SET_POSITION
| CoverEntityFeature.SET_TILT_POSITION
| CoverEntityFeature.STOP
| CoverEntityFeature.STOP_TILT,
),
],
)
async def test_cover_ui_create(
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
knx_data: dict[str, Any],
read_responses: dict[str, int | tuple[int]],
initial_state: str,
supported_features: int,
) -> None:
"""Test creating a cover."""
await knx.setup_integration()
await create_ui_entity(
platform=Platform.COVER,
entity_data={"name": "test"},
knx_data=knx_data,
)
# 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)
async def test_cover_ui_load(knx: KNXTestKit) -> None:
"""Test loading a cover from storage."""
await knx.setup_integration(config_store_fixture="config_store_cover.json")
await knx.assert_read("2/0/0", response=(0xFF,), ignore_order=True)
await knx.assert_read("3/1/0", response=(0xFF,), ignore_order=True)
await knx.assert_read("3/2/0", response=(0xFF,), ignore_order=True)
knx.assert_state(
"cover.minimal",
STATE_UNKNOWN,
supported_features=CoverEntityFeature.CLOSE | CoverEntityFeature.OPEN,
)
knx.assert_state(
"cover.position_only",
CoverState.OPEN,
supported_features=CoverEntityFeature.CLOSE
| CoverEntityFeature.OPEN
| CoverEntityFeature.SET_POSITION,
)
knx.assert_state(
"cover.tiltable",
CoverState.CLOSED,
supported_features=CoverEntityFeature.CLOSE
| CoverEntityFeature.OPEN
| CoverEntityFeature.SET_POSITION
| CoverEntityFeature.SET_TILT_POSITION
| CoverEntityFeature.STOP
| CoverEntityFeature.STOP_TILT,
)

View File

@ -14,11 +14,49 @@ INVALID = "invalid"
@pytest.mark.parametrize(
("selector_config", "data", "expected"),
[
# empty data is invalid
(
{},
{},
{"write": None, "state": None, "passive": []},
{INVALID: "At least one group address must be set"},
),
(
{"write": False},
{},
{INVALID: "At least one group address must be set"},
),
(
{"passive": False},
{},
{INVALID: "At least one group address must be set"},
),
(
{"write": False, "state": False, "passive": False},
{},
{INVALID: "At least one group address must be set"},
),
# stale data is invalid
(
{"write": False},
{"write": "1/2/3"},
{INVALID: "At least one group address must be set"},
),
(
{"write": False},
{"passive": []},
{INVALID: "At least one group address must be set"},
),
(
{"state": False},
{"write": None},
{INVALID: "At least one group address must be set"},
),
(
{"passive": False},
{"passive": ["1/2/3"]},
{INVALID: "At least one group address must be set"},
),
# valid data
(
{},
{"write": "1/2/3"},
@ -39,11 +77,6 @@ INVALID = "invalid"
{"write": "1", "state": 2, "passive": ["1/2/3"]},
{"write": "1", "state": 2, "passive": ["1/2/3"]},
),
(
{"write": False},
{"write": "1/2/3"},
{"state": None, "passive": []},
),
(
{"write": False},
{"state": "1/2/3"},
@ -54,11 +87,6 @@ INVALID = "invalid"
{"passive": ["1/2/3"]},
{"state": None, "passive": ["1/2/3"]},
),
(
{"passive": False},
{"passive": ["1/2/3"]},
{"write": None, "state": None},
),
(
{"passive": False},
{"write": "1/2/3"},
@ -68,12 +96,12 @@ INVALID = "invalid"
(
{"write_required": True},
{},
INVALID,
{INVALID: r"required key not provided*"},
),
(
{"state_required": True},
{},
INVALID,
{INVALID: r"required key not provided*"},
),
(
{"write_required": True},
@ -88,18 +116,18 @@ INVALID = "invalid"
(
{"write_required": True},
{"state": "1/2/3"},
INVALID,
{INVALID: r"required key not provided*"},
),
(
{"state_required": True},
{"write": "1/2/3"},
INVALID,
{INVALID: r"required key not provided*"},
),
# dpt key
(
{"dpt": ColorTempModes},
{"write": "1/2/3"},
INVALID,
{INVALID: r"required key not provided*"},
),
(
{"dpt": ColorTempModes},
@ -109,19 +137,19 @@ INVALID = "invalid"
(
{"dpt": ColorTempModes},
{"write": "1/2/3", "state": None, "passive": [], "dpt": "invalid"},
INVALID,
{INVALID: r"value must be one of ['5.001', '7.600', '9']*"},
),
],
)
def test_ga_selector(
selector_config: dict[str, Any],
data: dict[str, Any],
expected: str | dict[str, Any],
expected: dict[str, Any],
) -> None:
"""Test GASelector."""
selector = GASelector(**selector_config)
if expected == INVALID:
with pytest.raises(vol.Invalid):
if INVALID in expected:
with pytest.raises(vol.Invalid, match=expected[INVALID]):
selector(data)
else:
result = selector(data)