diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index c8b9ca40bd7..5d2ab031323 100644 --- a/homeassistant/components/knx/__init__.py +++ b/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(), } ), diff --git a/homeassistant/components/knx/const.py b/homeassistant/components/knx/const.py index f98d42fbd9e..058223bfaa1 100644 --- a/homeassistant/components/knx/const.py +++ b/homeassistant/components/knx/const.py @@ -119,6 +119,7 @@ SUPPORTED_PLATFORMS: Final = [ Platform.SELECT, Platform.SENSOR, Platform.SWITCH, + Platform.TEXT, Platform.WEATHER, ] diff --git a/homeassistant/components/knx/schema.py b/homeassistant/components/knx/schema.py index 370eaa7303e..87a7b6fdab5 100644 --- a/homeassistant/components/knx/schema.py +++ b/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.""" diff --git a/homeassistant/components/knx/text.py b/homeassistant/components/knx/text.py new file mode 100644 index 00000000000..abd3f44ae6b --- /dev/null +++ b/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) diff --git a/tests/components/knx/test_text.py b/tests/components/knx/test_text.py new file mode 100644 index 00000000000..0f5169054b3 --- /dev/null +++ b/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"