Add datetime platform to KNX (#97190)

homeassistant/components/knx/__init__.py

@@ -81,6 +81,7 @@ from .schema import (
     ClimateSchema,
     CoverSchema,
     DateSchema,
+    DateTimeSchema,
     EventSchema,
     ExposeSchema,
     FanSchema,
@@ -138,6 +139,7 @@ CONFIG_SCHEMA = vol.Schema(
                     **ClimateSchema.platform_node(),
                     **CoverSchema.platform_node(),
                     **DateSchema.platform_node(),
+                    **DateTimeSchema.platform_node(),
                     **FanSchema.platform_node(),
                     **LightSchema.platform_node(),
                     **NotifySchema.platform_node(),

homeassistant/components/knx/const.py

@@ -128,6 +128,7 @@ SUPPORTED_PLATFORMS: Final = [
     Platform.CLIMATE,
     Platform.COVER,
     Platform.DATE,
+    Platform.DATETIME,
     Platform.FAN,
     Platform.LIGHT,
     Platform.NOTIFY,

homeassistant/components/knx/datetime.py

@@ -0,0 +1,103 @@
+"""Support for KNX/IP datetime."""
+from __future__ import annotations
+
+from datetime import datetime
+
+from xknx import XKNX
+from xknx.devices import DateTime as XknxDateTime
+
+from homeassistant import config_entries
+from homeassistant.components.datetime import DateTimeEntity
+from homeassistant.const import (
+    CONF_ENTITY_CATEGORY,
+    CONF_NAME,
+    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
+import homeassistant.util.dt as dt_util
+
+from .const import (
+    CONF_RESPOND_TO_READ,
+    CONF_STATE_ADDRESS,
+    CONF_SYNC_STATE,
+    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 entities for KNX platform."""
+    xknx: XKNX = hass.data[DOMAIN].xknx
+    config: list[ConfigType] = hass.data[DATA_KNX_CONFIG][Platform.DATETIME]
+
+    async_add_entities(KNXDateTime(xknx, entity_config) for entity_config in config)
+
+
+def _create_xknx_device(xknx: XKNX, config: ConfigType) -> XknxDateTime:
+    """Return a XKNX DateTime object to be used within XKNX."""
+    return XknxDateTime(
+        xknx,
+        name=config[CONF_NAME],
+        broadcast_type="DATETIME",
+        localtime=False,
+        group_address=config[KNX_ADDRESS],
+        group_address_state=config.get(CONF_STATE_ADDRESS),
+        respond_to_read=config[CONF_RESPOND_TO_READ],
+        sync_state=config[CONF_SYNC_STATE],
+    )
+
+
+class KNXDateTime(KnxEntity, DateTimeEntity, RestoreEntity):
+    """Representation of a KNX datetime."""
+
+    _device: XknxDateTime
+
+    def __init__(self, xknx: XKNX, config: ConfigType) -> None:
+        """Initialize a KNX time."""
+        super().__init__(_create_xknx_device(xknx, config))
+        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()) is not None
+            and last_state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
+        ):
+            self._device.remote_value.value = (
+                datetime.fromisoformat(last_state.state)
+                .astimezone(dt_util.DEFAULT_TIME_ZONE)
+                .timetuple()
+            )
+
+    @property
+    def native_value(self) -> datetime | None:
+        """Return the latest value."""
+        if (time_struct := self._device.remote_value.value) is None:
+            return None
+        return datetime(
+            year=time_struct.tm_year,
+            month=time_struct.tm_mon,
+            day=time_struct.tm_mday,
+            hour=time_struct.tm_hour,
+            minute=time_struct.tm_min,
+            second=min(time_struct.tm_sec, 59),  # account for leap seconds
+            tzinfo=dt_util.DEFAULT_TIME_ZONE,
+        )
+
+    async def async_set_value(self, value: datetime) -> None:
+        """Change the value."""
+        await self._device.set(value.astimezone(dt_util.DEFAULT_TIME_ZONE).timetuple())

homeassistant/components/knx/schema.py

@@ -575,6 +575,25 @@ class DateSchema(KNXPlatformSchema):
     )
 
 
+class DateTimeSchema(KNXPlatformSchema):
+    """Voluptuous schema for KNX date."""
+
+    PLATFORM = Platform.DATETIME
+
+    DEFAULT_NAME = "KNX DateTime"
+
+    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_SYNC_STATE, default=True): sync_state_validator,
+            vol.Required(KNX_ADDRESS): ga_list_validator,
+            vol.Optional(CONF_STATE_ADDRESS): ga_list_validator,
+            vol.Optional(CONF_ENTITY_CATEGORY): ENTITY_CATEGORIES_SCHEMA,
+        }
+    )
+
+
 class ExposeSchema(KNXPlatformSchema):
     """Voluptuous schema for KNX exposures."""
 

tests/components/knx/test_datetime.py

@@ -0,0 +1,89 @@
+"""Test KNX date."""
+from homeassistant.components.datetime import ATTR_DATETIME, DOMAIN, SERVICE_SET_VALUE
+from homeassistant.components.knx.const import CONF_RESPOND_TO_READ, KNX_ADDRESS
+from homeassistant.components.knx.schema import DateTimeSchema
+from homeassistant.const import CONF_NAME
+from homeassistant.core import HomeAssistant, State
+
+from .conftest import KNXTestKit
+
+from tests.common import mock_restore_cache
+
+# KNX DPT 19.001 doesn't provide timezone information so we send local time
+
+
+async def test_datetime(hass: HomeAssistant, knx: KNXTestKit) -> None:
+    """Test KNX datetime."""
+    # default timezone in tests is US/Pacific
+    test_address = "1/1/1"
+    await knx.setup_integration(
+        {
+            DateTimeSchema.PLATFORM: {
+                CONF_NAME: "test",
+                KNX_ADDRESS: test_address,
+            }
+        }
+    )
+    # set value
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_SET_VALUE,
+        {"entity_id": "datetime.test", ATTR_DATETIME: "2020-01-02T03:04:05+00:00"},
+        blocking=True,
+    )
+    await knx.assert_write(
+        test_address,
+        (0x78, 0x01, 0x01, 0x73, 0x04, 0x05, 0x20, 0x80),
+    )
+    state = hass.states.get("datetime.test")
+    assert state.state == "2020-01-02T03:04:05+00:00"
+
+    # update from KNX
+    await knx.receive_write(
+        test_address,
+        (0x7B, 0x07, 0x19, 0x49, 0x28, 0x08, 0x00, 0x00),
+    )
+    state = hass.states.get("datetime.test")
+    assert state.state == "2023-07-25T16:40:08+00:00"
+
+
+async def test_date_restore_and_respond(hass: HomeAssistant, knx: KNXTestKit) -> None:
+    """Test KNX datetime with passive_address, restoring state and respond_to_read."""
+    hass.config.set_time_zone("Europe/Vienna")
+    test_address = "1/1/1"
+    test_passive_address = "3/3/3"
+    fake_state = State("datetime.test", "2022-03-03T03:04:05+00:00")
+    mock_restore_cache(hass, (fake_state,))
+
+    await knx.setup_integration(
+        {
+            DateTimeSchema.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("datetime.test")
+    assert state.state == "2022-03-03T03:04:05+00:00"
+    await knx.assert_telegram_count(0)
+
+    # respond with restored state
+    await knx.receive_read(test_address)
+    await knx.assert_response(
+        test_address,
+        (0x7A, 0x03, 0x03, 0x84, 0x04, 0x05, 0x20, 0x80),
+    )
+
+    # 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,
+        (0x78, 0x01, 0x01, 0x73, 0x04, 0x05, 0x20, 0x80),
+    )
+    state = hass.states.get("datetime.test")
+    assert state.state == "2020-01-01T18:04:05+00:00"