Add time platform to KNX (#95302)

homeassistant/components/knx/__init__.py

@@ -90,6 +90,7 @@ from .schema import (
     SensorSchema,
     SwitchSchema,
     TextSchema,
+    TimeSchema,
     WeatherSchema,
     ga_validator,
     sensor_type_validator,
@@ -143,6 +144,7 @@ CONFIG_SCHEMA = vol.Schema(
                     **SensorSchema.platform_node(),
                     **SwitchSchema.platform_node(),
                     **TextSchema.platform_node(),
+                    **TimeSchema.platform_node(),
                     **WeatherSchema.platform_node(),
                 }
             ),

homeassistant/components/knx/const.py

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

homeassistant/components/knx/manifest.json

@@ -11,7 +11,7 @@
   "loggers": ["xknx", "xknxproject"],
   "quality_scale": "platinum",
   "requirements": [
-    "xknx==2.11.0",
+    "xknx==2.11.1",
     "xknxproject==3.2.0",
     "knx-frontend==2023.6.23.191712"
   ]

homeassistant/components/knx/schema.py

@@ -936,6 +936,25 @@ class TextSchema(KNXPlatformSchema):
     )
 
 
+class TimeSchema(KNXPlatformSchema):
+    """Voluptuous schema for KNX time."""
+
+    PLATFORM = Platform.TIME
+
+    DEFAULT_NAME = "KNX Time"
+
+    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 WeatherSchema(KNXPlatformSchema):
     """Voluptuous schema for KNX weather station."""
 

homeassistant/components/knx/time.py

@@ -0,0 +1,104 @@
+"""Support for KNX/IP time."""
+from __future__ import annotations
+
+from datetime import time as dt_time
+import time as time_time
+from typing import Final
+
+from xknx import XKNX
+from xknx.devices import DateTime as XknxDateTime
+
+from homeassistant import config_entries
+from homeassistant.components.time import TimeEntity
+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
+
+from .const import (
+    CONF_RESPOND_TO_READ,
+    CONF_STATE_ADDRESS,
+    CONF_SYNC_STATE,
+    DATA_KNX_CONFIG,
+    DOMAIN,
+    KNX_ADDRESS,
+)
+from .knx_entity import KnxEntity
+
+_TIME_TRANSLATION_FORMAT: Final = "%H:%M:%S"
+
+
+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.TIME]
+
+    async_add_entities(KNXTime(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="TIME",
+        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 KNXTime(KnxEntity, TimeEntity, RestoreEntity):
+    """Representation of a KNX time."""
+
+    _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 = time_time.strptime(
+                last_state.state, _TIME_TRANSLATION_FORMAT
+            )
+
+    @property
+    def native_value(self) -> dt_time | None:
+        """Return the latest value."""
+        if (time_struct := self._device.remote_value.value) is None:
+            return None
+        return dt_time(
+            hour=time_struct.tm_hour,
+            minute=time_struct.tm_min,
+            second=min(time_struct.tm_sec, 59),  # account for leap seconds
+        )
+
+    async def async_set_value(self, value: dt_time) -> None:
+        """Change the value."""
+        time_struct = time_time.strptime(
+            value.strftime(_TIME_TRANSLATION_FORMAT),
+            _TIME_TRANSLATION_FORMAT,
+        )
+        await self._device.set(time_struct)

requirements_all.txt

@@ -2687,7 +2687,7 @@ xbox-webapi==2.0.11
 xiaomi-ble==0.17.2
 
 # homeassistant.components.knx
-xknx==2.11.0
+xknx==2.11.1
 
 # homeassistant.components.knx
 xknxproject==3.2.0

requirements_test_all.txt

@@ -1969,7 +1969,7 @@ xbox-webapi==2.0.11
 xiaomi-ble==0.17.2
 
 # homeassistant.components.knx
-xknx==2.11.0
+xknx==2.11.1
 
 # homeassistant.components.knx
 xknxproject==3.2.0

tests/components/knx/test_time.py

@@ -0,0 +1,86 @@
+"""Test KNX time."""
+from homeassistant.components.knx.const import CONF_RESPOND_TO_READ, KNX_ADDRESS
+from homeassistant.components.knx.schema import TimeSchema
+from homeassistant.components.time import ATTR_TIME, DOMAIN, SERVICE_SET_VALUE
+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_time(hass: HomeAssistant, knx: KNXTestKit) -> None:
+    """Test KNX time."""
+    test_address = "1/1/1"
+    await knx.setup_integration(
+        {
+            TimeSchema.PLATFORM: {
+                CONF_NAME: "test",
+                KNX_ADDRESS: test_address,
+            }
+        }
+    )
+    # set value
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_SET_VALUE,
+        {"entity_id": "time.test", ATTR_TIME: "01:02:03"},
+        blocking=True,
+    )
+    await knx.assert_write(
+        test_address,
+        (0x01, 0x02, 0x03),
+    )
+    state = hass.states.get("time.test")
+    assert state.state == "01:02:03"
+
+    # update from KNX
+    await knx.receive_write(
+        test_address,
+        (0x0C, 0x10, 0x3B),
+    )
+    state = hass.states.get("time.test")
+    assert state.state == "12:16:59"
+
+
+async def test_time_restore_and_respond(hass: HomeAssistant, knx: KNXTestKit) -> None:
+    """Test KNX time with passive_address, restoring state and respond_to_read."""
+    test_address = "1/1/1"
+    test_passive_address = "3/3/3"
+
+    fake_state = State("time.test", "01:02:03")
+    mock_restore_cache(hass, (fake_state,))
+
+    await knx.setup_integration(
+        {
+            TimeSchema.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("time.test")
+    assert state.state == "01:02:03"
+    await knx.assert_telegram_count(0)
+
+    # respond with restored state
+    await knx.receive_read(test_address)
+    await knx.assert_response(
+        test_address,
+        (0x01, 0x02, 0x03),
+    )
+
+    # 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,
+        (0x0C, 0x00, 0x00),
+    )
+    state = hass.states.get("time.test")
+    assert state.state == "12:00:00"