Add text platform for KNX (#82811)

homeassistant/components/knx/__init__.py

@@ -83,6 +83,7 @@ from .schema import (
     SelectSchema,
     SensorSchema,
     SwitchSchema,
+    TextSchema,
     WeatherSchema,
     ga_validator,
     sensor_type_validator,
@@ -133,6 +134,7 @@ CONFIG_SCHEMA = vol.Schema(
                     **SelectSchema.platform_node(),
                     **SensorSchema.platform_node(),
                     **SwitchSchema.platform_node(),
+                    **TextSchema.platform_node(),
                     **WeatherSchema.platform_node(),
                 }
             ),

homeassistant/components/knx/const.py

@@ -119,6 +119,7 @@ SUPPORTED_PLATFORMS: Final = [
     Platform.SELECT,
     Platform.SENSOR,
     Platform.SWITCH,
+    Platform.TEXT,
     Platform.WEATHER,
 ]
 

homeassistant/components/knx/schema.py

@@ -29,6 +29,7 @@ from homeassistant.components.sensor import (
 from homeassistant.components.switch import (
     DEVICE_CLASSES_SCHEMA as SWITCH_DEVICE_CLASSES_SCHEMA,
 )
+from homeassistant.components.text import TextMode
 from homeassistant.const import (
     CONF_DEVICE_CLASS,
     CONF_ENTITY_CATEGORY,
@@ -887,6 +888,26 @@ class SwitchSchema(KNXPlatformSchema):
     )
 
 
+class TextSchema(KNXPlatformSchema):
+    """Voluptuous schema for KNX text."""
+
+    PLATFORM = Platform.TEXT
+
+    DEFAULT_NAME = "KNX Text"
+
+    ENTITY_SCHEMA = vol.Schema(
+        {
+            vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
+            vol.Optional(CONF_RESPOND_TO_READ, default=False): cv.boolean,
+            vol.Optional(CONF_TYPE, default="latin_1"): string_type_validator,
+            vol.Optional(CONF_MODE, default=TextMode.TEXT): vol.Coerce(TextMode),
+            vol.Required(KNX_ADDRESS): ga_list_validator,
+            vol.Optional(CONF_STATE_ADDRESS): ga_list_validator,
+            vol.Optional(CONF_ENTITY_CATEGORY): ENTITY_CATEGORIES_SCHEMA,
+        }
+    )
+
+
 class WeatherSchema(KNXPlatformSchema):
     """Voluptuous schema for KNX weather station."""
 

homeassistant/components/knx/text.py

@@ -0,0 +1,92 @@
+"""Support for KNX/IP text."""
+from __future__ import annotations
+
+from xknx import XKNX
+from xknx.devices import Notification as XknxNotification
+from xknx.dpt import DPTLatin1
+
+from homeassistant import config_entries
+from homeassistant.components.text import TextEntity
+from homeassistant.const import (
+    CONF_ENTITY_CATEGORY,
+    CONF_MODE,
+    CONF_NAME,
+    CONF_TYPE,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+    Platform,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.restore_state import RestoreEntity
+from homeassistant.helpers.typing import ConfigType
+
+from .const import (
+    CONF_RESPOND_TO_READ,
+    CONF_STATE_ADDRESS,
+    DATA_KNX_CONFIG,
+    DOMAIN,
+    KNX_ADDRESS,
+)
+from .knx_entity import KnxEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: config_entries.ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up sensor(s) for KNX platform."""
+    xknx: XKNX = hass.data[DOMAIN].xknx
+    config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][Platform.TEXT]
+
+    async_add_entities(KNXText(xknx, entity_config) for entity_config in config)
+
+
+def _create_notification(xknx: XKNX, config: ConfigType) -> XknxNotification:
+    """Return a KNX Notification to be used within XKNX."""
+    return XknxNotification(
+        xknx,
+        name=config[CONF_NAME],
+        group_address=config[KNX_ADDRESS],
+        group_address_state=config.get(CONF_STATE_ADDRESS),
+        respond_to_read=config[CONF_RESPOND_TO_READ],
+        value_type=config[CONF_TYPE],
+    )
+
+
+class KNXText(KnxEntity, TextEntity, RestoreEntity):
+    """Representation of a KNX text."""
+
+    _device: XknxNotification
+    _attr_native_max = 14
+
+    def __init__(self, xknx: XKNX, config: ConfigType) -> None:
+        """Initialize a KNX text."""
+        super().__init__(_create_notification(xknx, config))
+        self._attr_mode = config[CONF_MODE]
+        self._attr_pattern = (
+            r"[\u0000-\u00ff]*"  # Latin-1
+            if issubclass(self._device.remote_value.dpt_class, DPTLatin1)
+            else r"[\u0000-\u007f]*"  # ASCII
+        )
+        self._attr_entity_category = config.get(CONF_ENTITY_CATEGORY)
+        self._attr_unique_id = str(self._device.remote_value.group_address)
+
+    async def async_added_to_hass(self) -> None:
+        """Restore last state."""
+        await super().async_added_to_hass()
+        if not self._device.remote_value.readable and (
+            last_state := await self.async_get_last_state()
+        ):
+            if last_state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE):
+                self._device.remote_value.value = last_state.state
+
+    @property
+    def native_value(self) -> str | None:
+        """Return the value reported by the text."""
+        return self._device.message
+
+    async def async_set_value(self, value: str) -> None:
+        """Change the value."""
+        await self._device.set(value)

tests/components/knx/test_text.py

@@ -0,0 +1,100 @@
+"""Test KNX number."""
+from homeassistant.components.knx.const import CONF_RESPOND_TO_READ, KNX_ADDRESS
+from homeassistant.components.knx.schema import TextSchema
+from homeassistant.const import CONF_NAME
+from homeassistant.core import HomeAssistant, State
+
+from .conftest import KNXTestKit
+
+from tests.common import mock_restore_cache
+
+
+async def test_text(hass: HomeAssistant, knx: KNXTestKit):
+    """Test KNX text."""
+    test_address = "1/1/1"
+    await knx.setup_integration(
+        {
+            TextSchema.PLATFORM: {
+                CONF_NAME: "test",
+                KNX_ADDRESS: test_address,
+            }
+        }
+    )
+    # set value
+    await hass.services.async_call(
+        "text",
+        "set_value",
+        {"entity_id": "text.test", "value": "hello world"},
+        blocking=True,
+    )
+    await knx.assert_write(
+        test_address,
+        (
+            0x68,
+            0x65,
+            0x6C,
+            0x6C,
+            0x6F,
+            0x20,
+            0x77,
+            0x6F,
+            0x72,
+            0x6C,
+            0x64,
+            0x0,
+            0x0,
+            0x0,
+        ),
+    )
+    state = hass.states.get("text.test")
+    assert state.state == "hello world"
+
+    # update from KNX
+    await knx.receive_write(
+        test_address,
+        (0x68, 0x61, 0x6C, 0x6C, 0x6F, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0),
+    )
+    state = hass.states.get("text.test")
+    assert state.state == "hallo"
+
+
+async def test_text_restore_and_respond(hass: HomeAssistant, knx: KNXTestKit):
+    """Test KNX text with passive_address, restoring state and respond_to_read."""
+    test_address = "1/1/1"
+    test_passive_address = "3/3/3"
+
+    fake_state = State("text.test", "test test")
+    mock_restore_cache(hass, (fake_state,))
+
+    await knx.setup_integration(
+        {
+            TextSchema.PLATFORM: {
+                CONF_NAME: "test",
+                KNX_ADDRESS: [test_address, test_passive_address],
+                CONF_RESPOND_TO_READ: True,
+            }
+        }
+    )
+    # restored state - doesn't send telegram
+    state = hass.states.get("text.test")
+    assert state.state == "test test"
+    await knx.assert_telegram_count(0)
+
+    # respond with restored state
+    await knx.receive_read(test_address)
+    await knx.assert_response(
+        test_address,
+        (0x74, 0x65, 0x73, 0x74, 0x20, 0x74, 0x65, 0x73, 0x74, 0x0, 0x0, 0x0, 0x0, 0x0),
+    )
+
+    # don't respond to passive address
+    await knx.receive_read(test_passive_address)
+    await knx.assert_no_telegram()
+
+    # update from KNX passive address
+    await knx.receive_write(
+        test_passive_address,
+        (0x68, 0x61, 0x6C, 0x6C, 0x6F, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0),
+    )
+    state = hass.states.get("text.test")
+    assert state.state == "hallo"