diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index b61a825e97f..0cdd92abe64 100644 --- a/homeassistant/components/knx/__init__.py +++ b/homeassistant/components/knx/__init__.py @@ -50,6 +50,7 @@ from .const import ( from .expose import KNXExposeSensor, KNXExposeTime, create_knx_exposure from .schema import ( BinarySensorSchema, + ButtonSchema, ClimateSchema, ConnectionSchema, CoverSchema, @@ -102,6 +103,7 @@ CONFIG_SCHEMA = vol.Schema( **EventSchema.SCHEMA, **ExposeSchema.platform_node(), **BinarySensorSchema.platform_node(), + **ButtonSchema.platform_node(), **ClimateSchema.platform_node(), **CoverSchema.platform_node(), **FanSchema.platform_node(), diff --git a/homeassistant/components/knx/button.py b/homeassistant/components/knx/button.py new file mode 100644 index 00000000000..72bb807abb0 --- /dev/null +++ b/homeassistant/components/knx/button.py @@ -0,0 +1,57 @@ +"""Support for KNX/IP buttons.""" +from __future__ import annotations + +from xknx import XKNX +from xknx.devices import RawValue as XknxRawValue + +from homeassistant.components.button import ButtonEntity +from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType + +from .const import CONF_PAYLOAD, CONF_PAYLOAD_LENGTH, DOMAIN, KNX_ADDRESS +from .knx_entity import KnxEntity + + +async def async_setup_platform( + hass: HomeAssistant, + config: ConfigType, + async_add_entities: AddEntitiesCallback, + discovery_info: DiscoveryInfoType | None = None, +) -> None: + """Set up buttons for KNX platform.""" + if not discovery_info or not discovery_info["platform_config"]: + return + platform_config = discovery_info["platform_config"] + xknx: XKNX = hass.data[DOMAIN].xknx + + async_add_entities( + KNXButton(xknx, entity_config) for entity_config in platform_config + ) + + +class KNXButton(KnxEntity, ButtonEntity): + """Representation of a KNX button.""" + + _device: XknxRawValue + + def __init__(self, xknx: XKNX, config: ConfigType) -> None: + """Initialize a KNX button.""" + super().__init__( + device=XknxRawValue( + xknx, + name=config[CONF_NAME], + payload_length=config[CONF_PAYLOAD_LENGTH], + group_address=config[KNX_ADDRESS], + ) + ) + self._payload = config[CONF_PAYLOAD] + self._attr_entity_category = config.get(CONF_ENTITY_CATEGORY) + self._attr_unique_id = ( + f"{self._device.remote_value.group_address}_{self._payload}" + ) + + async def async_press(self) -> None: + """Press the button.""" + await self._device.set(self._payload) diff --git a/homeassistant/components/knx/const.py b/homeassistant/components/knx/const.py index 7eae5f1c19f..1058f5abe07 100644 --- a/homeassistant/components/knx/const.py +++ b/homeassistant/components/knx/const.py @@ -31,6 +31,8 @@ CONF_KNX_EXPOSE: Final = "expose" CONF_KNX_INDIVIDUAL_ADDRESS: Final = "individual_address" CONF_KNX_ROUTING: Final = "routing" CONF_KNX_TUNNELING: Final = "tunneling" +CONF_PAYLOAD: Final = "payload" +CONF_PAYLOAD_LENGTH: Final = "payload_length" CONF_RESET_AFTER: Final = "reset_after" CONF_RESPOND_TO_READ: Final = "respond_to_read" CONF_STATE_ADDRESS: Final = "state_address" @@ -51,6 +53,7 @@ class SupportedPlatforms(Enum): """Supported platforms.""" BINARY_SENSOR = "binary_sensor" + BUTTON = "button" CLIMATE = "climate" COVER = "cover" FAN = "fan" diff --git a/homeassistant/components/knx/schema.py b/homeassistant/components/knx/schema.py index 6cd8fc6bc0d..d84531bbcaf 100644 --- a/homeassistant/components/knx/schema.py +++ b/homeassistant/components/knx/schema.py @@ -9,7 +9,7 @@ import voluptuous as vol from xknx import XKNX from xknx.devices.climate import SetpointShiftMode from xknx.dpt import DPTBase, DPTNumeric -from xknx.exceptions import CouldNotParseAddress +from xknx.exceptions import ConversionError, CouldNotParseAddress from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.telegram.address import IndividualAddress, parse_device_group_address @@ -40,6 +40,8 @@ from .const import ( CONF_KNX_INDIVIDUAL_ADDRESS, CONF_KNX_ROUTING, CONF_KNX_TUNNELING, + CONF_PAYLOAD, + CONF_PAYLOAD_LENGTH, CONF_RESET_AFTER, CONF_RESPOND_TO_READ, CONF_STATE_ADDRESS, @@ -123,20 +125,49 @@ def numeric_type_validator(value: Any) -> str | int: raise vol.Invalid(f"value '{value}' is not a valid numeric sensor type.") +def _max_payload_value(payload_length: int) -> int: + if payload_length == 0: + return 0x3F + return int(256 ** payload_length) - 1 + + +def button_payload_sub_validator(entity_config: OrderedDict) -> OrderedDict: + """Validate a button entity payload configuration.""" + if _type := entity_config.get(CONF_TYPE): + _payload = entity_config[ButtonSchema.CONF_VALUE] + if (transcoder := DPTBase.parse_transcoder(_type)) is None: + raise vol.Invalid(f"'type: {_type}' is not a valid sensor type.") + entity_config[CONF_PAYLOAD_LENGTH] = transcoder.payload_length + try: + entity_config[CONF_PAYLOAD] = int.from_bytes( + transcoder.to_knx(_payload), byteorder="big" + ) + except ConversionError as ex: + raise vol.Invalid( + f"'payload: {_payload}' not valid for 'type: {_type}'" + ) from ex + return entity_config + + _payload = entity_config[CONF_PAYLOAD] + _payload_length = entity_config[CONF_PAYLOAD_LENGTH] + if _payload > (max_payload := _max_payload_value(_payload_length)): + raise vol.Invalid( + f"'payload: {_payload}' exceeds possible maximum for " + f"payload_length {_payload_length}: {max_payload}" + ) + return entity_config + + def select_options_sub_validator(entity_config: OrderedDict) -> OrderedDict: """Validate a select entity options configuration.""" options_seen = set() payloads_seen = set() - payload_length = entity_config[SelectSchema.CONF_PAYLOAD_LENGTH] - if payload_length == 0: - max_payload = 0x3F - else: - max_payload = 256 ** payload_length - 1 + payload_length = entity_config[CONF_PAYLOAD_LENGTH] for opt in entity_config[SelectSchema.CONF_OPTIONS]: option = opt[SelectSchema.CONF_OPTION] - payload = opt[SelectSchema.CONF_PAYLOAD] - if payload > max_payload: + payload = opt[CONF_PAYLOAD] + if payload > (max_payload := _max_payload_value(payload_length)): raise vol.Invalid( f"'payload: {payload}' for 'option: {option}' exceeds possible" f" maximum of 'payload_length: {payload_length}': {max_payload}" @@ -284,6 +315,67 @@ class BinarySensorSchema(KNXPlatformSchema): ) +class ButtonSchema(KNXPlatformSchema): + """Voluptuous schema for KNX buttons.""" + + PLATFORM_NAME = SupportedPlatforms.BUTTON.value + + CONF_VALUE = "value" + DEFAULT_NAME = "KNX Button" + + payload_or_value_msg = f"Please use only one of `{CONF_PAYLOAD}` or `{CONF_VALUE}`" + length_or_type_msg = ( + f"Please use only one of `{CONF_PAYLOAD_LENGTH}` or `{CONF_TYPE}`" + ) + + ENTITY_SCHEMA = vol.All( + vol.Schema( + { + vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string, + vol.Required(KNX_ADDRESS): ga_validator, + vol.Exclusive( + CONF_PAYLOAD, "payload_or_value", msg=payload_or_value_msg + ): object, + vol.Exclusive( + CONF_VALUE, "payload_or_value", msg=payload_or_value_msg + ): object, + vol.Exclusive( + CONF_PAYLOAD_LENGTH, "length_or_type", msg=length_or_type_msg + ): object, + vol.Exclusive( + CONF_TYPE, "length_or_type", msg=length_or_type_msg + ): object, + vol.Optional(CONF_ENTITY_CATEGORY): ENTITY_CATEGORIES_SCHEMA, + } + ), + vol.Any( + vol.Schema( + # encoded value + { + vol.Required(CONF_VALUE): vol.Any(int, float, str), + vol.Required(CONF_TYPE): sensor_type_validator, + }, + extra=vol.ALLOW_EXTRA, + ), + vol.Schema( + # raw payload - default is DPT 1 style True + { + vol.Optional(CONF_PAYLOAD, default=1): cv.positive_int, + vol.Optional(CONF_PAYLOAD_LENGTH, default=0): vol.All( + vol.Coerce(int), vol.Range(min=0, max=14) + ), + vol.Optional(CONF_VALUE): None, + vol.Optional(CONF_TYPE): None, + }, + extra=vol.ALLOW_EXTRA, + ), + ), + # calculate raw CONF_PAYLOAD and CONF_PAYLOAD_LENGTH + # from CONF_VALUE and CONF_TYPE if given and check payload size + button_payload_sub_validator, + ) + + class ClimateSchema(KNXPlatformSchema): """Voluptuous schema for KNX climate devices.""" @@ -733,8 +825,6 @@ class SelectSchema(KNXPlatformSchema): CONF_OPTION = "option" CONF_OPTIONS = "options" - CONF_PAYLOAD = "payload" - CONF_PAYLOAD_LENGTH = "payload_length" DEFAULT_NAME = "KNX Select" ENTITY_SCHEMA = vol.All( diff --git a/homeassistant/components/knx/select.py b/homeassistant/components/knx/select.py index e548ad27c8a..f002bad37ce 100644 --- a/homeassistant/components/knx/select.py +++ b/homeassistant/components/knx/select.py @@ -17,6 +17,8 @@ from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from .const import ( + CONF_PAYLOAD, + CONF_PAYLOAD_LENGTH, CONF_RESPOND_TO_READ, CONF_STATE_ADDRESS, CONF_SYNC_STATE, @@ -49,7 +51,7 @@ def _create_raw_value(xknx: XKNX, config: ConfigType) -> RawValue: return RawValue( xknx, name=config[CONF_NAME], - payload_length=config[SelectSchema.CONF_PAYLOAD_LENGTH], + payload_length=config[CONF_PAYLOAD_LENGTH], group_address=config[KNX_ADDRESS], group_address_state=config.get(CONF_STATE_ADDRESS), respond_to_read=config[CONF_RESPOND_TO_READ], @@ -66,7 +68,7 @@ class KNXSelect(KnxEntity, SelectEntity, RestoreEntity): """Initialize a KNX select.""" super().__init__(_create_raw_value(xknx, config)) self._option_payloads: dict[str, int] = { - option[SelectSchema.CONF_OPTION]: option[SelectSchema.CONF_PAYLOAD] + option[SelectSchema.CONF_OPTION]: option[CONF_PAYLOAD] for option in config[SelectSchema.CONF_OPTIONS] } self._attr_options = list(self._option_payloads) diff --git a/tests/components/knx/test_button.py b/tests/components/knx/test_button.py new file mode 100644 index 00000000000..0e5f40670f7 --- /dev/null +++ b/tests/components/knx/test_button.py @@ -0,0 +1,91 @@ +"""Test KNX button.""" +from datetime import timedelta + +from homeassistant.components.knx.const import ( + CONF_PAYLOAD, + CONF_PAYLOAD_LENGTH, + KNX_ADDRESS, +) +from homeassistant.components.knx.schema import ButtonSchema +from homeassistant.const import CONF_NAME, CONF_TYPE +from homeassistant.core import HomeAssistant +from homeassistant.util import dt + +from .conftest import KNXTestKit + +from tests.common import async_capture_events, async_fire_time_changed + + +async def test_button_simple(hass: HomeAssistant, knx: KNXTestKit): + """Test KNX button with default payload.""" + events = async_capture_events(hass, "state_changed") + await knx.setup_integration( + { + ButtonSchema.PLATFORM_NAME: { + CONF_NAME: "test", + KNX_ADDRESS: "1/2/3", + } + } + ) + assert len(hass.states.async_all()) == 1 + assert len(events) == 1 + events.pop() + + # press button + await hass.services.async_call( + "button", "press", {"entity_id": "button.test"}, blocking=True + ) + await knx.assert_write("1/2/3", True) + assert len(events) == 1 + events.pop() + + # received telegrams on button GA are ignored by the entity + old_state = hass.states.get("button.test") + async_fire_time_changed(hass, dt.utcnow() + timedelta(seconds=3)) + await knx.receive_write("1/2/3", False) + await knx.receive_write("1/2/3", True) + new_state = hass.states.get("button.test") + assert old_state == new_state + assert len(events) == 0 + + # button does not respond to read + await knx.receive_read("1/2/3") + await knx.assert_telegram_count(0) + + +async def test_button_raw(hass: HomeAssistant, knx: KNXTestKit): + """Test KNX button with raw payload.""" + await knx.setup_integration( + { + ButtonSchema.PLATFORM_NAME: { + CONF_NAME: "test", + KNX_ADDRESS: "1/2/3", + CONF_PAYLOAD: False, + CONF_PAYLOAD_LENGTH: 0, + } + } + ) + # press button + await hass.services.async_call( + "button", "press", {"entity_id": "button.test"}, blocking=True + ) + await knx.assert_write("1/2/3", False) + + +async def test_button_type(hass: HomeAssistant, knx: KNXTestKit): + """Test KNX button with encoded payload.""" + await knx.setup_integration( + { + ButtonSchema.PLATFORM_NAME: { + CONF_NAME: "test", + KNX_ADDRESS: "1/2/3", + ButtonSchema.CONF_VALUE: 21.5, + CONF_TYPE: "2byte_float", + } + } + ) + # press button + await hass.services.async_call( + "button", "press", {"entity_id": "button.test"}, blocking=True + ) + await knx.assert_write("1/2/3", (0x0C, 0x33)) diff --git a/tests/components/knx/test_select.py b/tests/components/knx/test_select.py index d8089976aca..f7f80c2ebf3 100644 --- a/tests/components/knx/test_select.py +++ b/tests/components/knx/test_select.py @@ -4,6 +4,8 @@ from unittest.mock import patch import pytest from homeassistant.components.knx.const import ( + CONF_PAYLOAD, + CONF_PAYLOAD_LENGTH, CONF_RESPOND_TO_READ, CONF_STATE_ADDRESS, CONF_SYNC_STATE, @@ -19,9 +21,9 @@ from .conftest import KNXTestKit async def test_select_dpt_2_simple(hass: HomeAssistant, knx: KNXTestKit): """Test simple KNX select.""" _options = [ - {SelectSchema.CONF_PAYLOAD: 0b00, SelectSchema.CONF_OPTION: "No control"}, - {SelectSchema.CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"}, - {SelectSchema.CONF_PAYLOAD: 0b11, SelectSchema.CONF_OPTION: "Control - On"}, + {CONF_PAYLOAD: 0b00, SelectSchema.CONF_OPTION: "No control"}, + {CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"}, + {CONF_PAYLOAD: 0b11, SelectSchema.CONF_OPTION: "Control - On"}, ] test_address = "1/1/1" await knx.setup_integration( @@ -30,7 +32,7 @@ async def test_select_dpt_2_simple(hass: HomeAssistant, knx: KNXTestKit): CONF_NAME: "test", KNX_ADDRESS: test_address, CONF_SYNC_STATE: False, - SelectSchema.CONF_PAYLOAD_LENGTH: 0, + CONF_PAYLOAD_LENGTH: 0, SelectSchema.CONF_OPTIONS: _options, } } @@ -89,9 +91,9 @@ async def test_select_dpt_2_simple(hass: HomeAssistant, knx: KNXTestKit): async def test_select_dpt_2_restore(hass: HomeAssistant, knx: KNXTestKit): """Test KNX select with passive_address and respond_to_read restoring state.""" _options = [ - {SelectSchema.CONF_PAYLOAD: 0b00, SelectSchema.CONF_OPTION: "No control"}, - {SelectSchema.CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"}, - {SelectSchema.CONF_PAYLOAD: 0b11, SelectSchema.CONF_OPTION: "Control - On"}, + {CONF_PAYLOAD: 0b00, SelectSchema.CONF_OPTION: "No control"}, + {CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"}, + {CONF_PAYLOAD: 0b11, SelectSchema.CONF_OPTION: "Control - On"}, ] test_address = "1/1/1" test_passive_address = "3/3/3" @@ -107,7 +109,7 @@ async def test_select_dpt_2_restore(hass: HomeAssistant, knx: KNXTestKit): CONF_NAME: "test", KNX_ADDRESS: [test_address, test_passive_address], CONF_RESPOND_TO_READ: True, - SelectSchema.CONF_PAYLOAD_LENGTH: 0, + CONF_PAYLOAD_LENGTH: 0, SelectSchema.CONF_OPTIONS: _options, } } @@ -129,11 +131,11 @@ async def test_select_dpt_2_restore(hass: HomeAssistant, knx: KNXTestKit): async def test_select_dpt_20_103_all_options(hass: HomeAssistant, knx: KNXTestKit): """Test KNX select with state_address, passive_address and respond_to_read.""" _options = [ - {SelectSchema.CONF_PAYLOAD: 0, SelectSchema.CONF_OPTION: "Auto"}, - {SelectSchema.CONF_PAYLOAD: 1, SelectSchema.CONF_OPTION: "Legio protect"}, - {SelectSchema.CONF_PAYLOAD: 2, SelectSchema.CONF_OPTION: "Normal"}, - {SelectSchema.CONF_PAYLOAD: 3, SelectSchema.CONF_OPTION: "Reduced"}, - {SelectSchema.CONF_PAYLOAD: 4, SelectSchema.CONF_OPTION: "Off"}, + {CONF_PAYLOAD: 0, SelectSchema.CONF_OPTION: "Auto"}, + {CONF_PAYLOAD: 1, SelectSchema.CONF_OPTION: "Legio protect"}, + {CONF_PAYLOAD: 2, SelectSchema.CONF_OPTION: "Normal"}, + {CONF_PAYLOAD: 3, SelectSchema.CONF_OPTION: "Reduced"}, + {CONF_PAYLOAD: 4, SelectSchema.CONF_OPTION: "Off"}, ] test_address = "1/1/1" test_state_address = "2/2/2" @@ -146,7 +148,7 @@ async def test_select_dpt_20_103_all_options(hass: HomeAssistant, knx: KNXTestKi KNX_ADDRESS: [test_address, test_passive_address], CONF_STATE_ADDRESS: test_state_address, CONF_RESPOND_TO_READ: True, - SelectSchema.CONF_PAYLOAD_LENGTH: 1, + CONF_PAYLOAD_LENGTH: 1, SelectSchema.CONF_OPTIONS: _options, } }