Add Button platform to KNX integration (#59082)

* add button platform

* default values for payload and payload_length

* allow `type` configuration for encoded payloads

* add test for type configuration

* move common constants to const.py

- CONF_PAYLOAD
- CONF_PAYLOAD_LENGTH

* validate payload for payload_length or type

* c&p errors

* fix unique_id and pylint

* fix validator

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(),

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)

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"

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]
+    payload_length = entity_config[CONF_PAYLOAD_LENGTH]
-    if payload_length == 0:
-        max_payload = 0x3F
-    else:
-        max_payload = 256 ** payload_length - 1
 
     for opt in entity_config[SelectSchema.CONF_OPTIONS]:
         option = opt[SelectSchema.CONF_OPTION]
-        payload = opt[SelectSchema.CONF_PAYLOAD]
+        payload = opt[CONF_PAYLOAD]
-        if payload > max_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(

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)

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))

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"},
+        {CONF_PAYLOAD: 0b00, SelectSchema.CONF_OPTION: "No control"},
-        {SelectSchema.CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"},
+        {CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"},
-        {SelectSchema.CONF_PAYLOAD: 0b11, SelectSchema.CONF_OPTION: "Control - On"},
+        {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"},
+        {CONF_PAYLOAD: 0b00, SelectSchema.CONF_OPTION: "No control"},
-        {SelectSchema.CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"},
+        {CONF_PAYLOAD: 0b10, SelectSchema.CONF_OPTION: "Control - Off"},
-        {SelectSchema.CONF_PAYLOAD: 0b11, SelectSchema.CONF_OPTION: "Control - On"},
+        {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"},
+        {CONF_PAYLOAD: 0, SelectSchema.CONF_OPTION: "Auto"},
-        {SelectSchema.CONF_PAYLOAD: 1, SelectSchema.CONF_OPTION: "Legio protect"},
+        {CONF_PAYLOAD: 1, SelectSchema.CONF_OPTION: "Legio protect"},
-        {SelectSchema.CONF_PAYLOAD: 2, SelectSchema.CONF_OPTION: "Normal"},
+        {CONF_PAYLOAD: 2, SelectSchema.CONF_OPTION: "Normal"},
-        {SelectSchema.CONF_PAYLOAD: 3, SelectSchema.CONF_OPTION: "Reduced"},
+        {CONF_PAYLOAD: 3, SelectSchema.CONF_OPTION: "Reduced"},
-        {SelectSchema.CONF_PAYLOAD: 4, SelectSchema.CONF_OPTION: "Off"},
+        {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,
             }
         }