Compare commits

...

4 Commits

Author SHA1 Message Date
farmio
f8ed52fe31 typo 2026-01-29 12:20:33 +01:00
farmio
34589fcda7 update snapshots 2026-01-29 12:18:37 +01:00
farmio
da027c063d tests 2026-01-29 12:03:09 +01:00
farmio
66ba2819e1 Add KNX time server configuration from UI 2026-01-29 09:51:10 +01:00
24 changed files with 462 additions and 50 deletions

View File

@@ -120,6 +120,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data[KNX_MODULE_KEY] = knx_module
knx_module.ui_time_server_controller.start(
knx_module.xknx, knx_module.config_store.get_time_server_config()
)
if CONF_KNX_EXPOSE in config:
knx_module.yaml_exposures.extend(
create_combined_knx_exposure(hass, knx_module.xknx, config[CONF_KNX_EXPOSE])
@@ -153,6 +156,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
exposure.async_remove()
for exposure in knx_module.service_exposures.values():
exposure.async_remove()
knx_module.ui_time_server_controller.stop()
configured_platforms_yaml = {
platform

View File

@@ -6,7 +6,7 @@ from asyncio import TaskGroup
from collections.abc import Callable, Iterable
from dataclasses import dataclass
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
from xknx import XKNX
from xknx.devices import DateDevice, DateTimeDevice, ExposeSensor, TimeDevice
@@ -43,6 +43,9 @@ from homeassistant.util import dt as dt_util
from .const import CONF_RESPOND_TO_READ, KNX_ADDRESS
from .schema import ExposeSchema
if TYPE_CHECKING:
from .storage.time_server import KNXTimeServerStoreModel
_LOGGER = logging.getLogger(__name__)
@@ -59,7 +62,7 @@ def create_knx_exposure(
):
exposure = KnxExposeTime(
xknx=xknx,
config=config,
options=_yaml_config_to_expose_time_options(config),
)
else:
exposure = KnxExposeEntity(
@@ -85,7 +88,7 @@ def create_combined_knx_exposure(
if value_type.lower() in ExposeSchema.EXPOSE_TIME_TYPES:
time_exposure = KnxExposeTime(
xknx=xknx,
config=config,
options=_yaml_config_to_expose_time_options(config),
)
time_exposure.async_register()
exposures.append(time_exposure)
@@ -291,26 +294,82 @@ class KnxExposeEntity:
)
@dataclass
class KnxExposeTimeOptions:
"""Options for KNX Expose time."""
device_cls: type[DateDevice | DateTimeDevice | TimeDevice]
group_address: GroupAddress | InternalGroupAddress
name: str
def _yaml_config_to_expose_time_options(config: ConfigType) -> KnxExposeTimeOptions:
"""Convert single yaml expose time config to KnxExposeTimeOptions."""
ga = parse_device_group_address(config[KNX_ADDRESS])
expose_type: str = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
xknx_device_cls: type[DateDevice | DateTimeDevice | TimeDevice]
match expose_type.lower():
case ExposeSchema.CONF_DATE:
xknx_device_cls = DateDevice
case ExposeSchema.CONF_DATETIME:
xknx_device_cls = DateTimeDevice
case ExposeSchema.CONF_TIME:
xknx_device_cls = TimeDevice
return KnxExposeTimeOptions(
name=expose_type.capitalize(),
group_address=ga,
device_cls=xknx_device_cls,
)
@callback
def create_time_server_exposures(
xknx: XKNX,
config: KNXTimeServerStoreModel,
) -> list[KnxExposeTime]:
"""Create exposures from UI config store time server config."""
exposures: list[KnxExposeTime] = []
device_cls: type[DateDevice | DateTimeDevice | TimeDevice]
for expose_type, data in config.items():
if not data or (ga := data.get("write")) is None: # type: ignore[attr-defined]
continue
match expose_type:
case "time":
device_cls = TimeDevice
case "date":
device_cls = DateDevice
case "datetime":
device_cls = DateTimeDevice
case _:
continue
exposures.append(
KnxExposeTime(
xknx=xknx,
options=KnxExposeTimeOptions(
name=f"timeserver_{expose_type}",
group_address=parse_device_group_address(ga),
device_cls=device_cls,
),
)
)
for exposure in exposures:
exposure.async_register()
return exposures
class KnxExposeTime:
"""Object to Expose Time/Date object to KNX bus."""
def __init__(self, xknx: XKNX, config: ConfigType) -> None:
__slots__ = ("device", "xknx")
def __init__(self, xknx: XKNX, options: KnxExposeTimeOptions) -> None:
"""Initialize of Expose class."""
self.xknx = xknx
expose_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
xknx_device_cls: type[DateDevice | DateTimeDevice | TimeDevice]
match expose_type:
case ExposeSchema.CONF_DATE:
xknx_device_cls = DateDevice
case ExposeSchema.CONF_DATETIME:
xknx_device_cls = DateTimeDevice
case ExposeSchema.CONF_TIME:
xknx_device_cls = TimeDevice
self.device = xknx_device_cls(
self.device = options.device_cls(
self.xknx,
name=expose_type.capitalize(),
name=options.name,
localtime=dt_util.get_default_time_zone(),
group_address=config[KNX_ADDRESS],
group_address=options.group_address,
)
@property

View File

@@ -58,6 +58,7 @@ from .expose import KnxExposeEntity, KnxExposeTime
from .project import KNXProject
from .repairs import data_secure_group_key_issue_dispatcher
from .storage.config_store import KNXConfigStore
from .storage.time_server import TimeServerController
from .telegrams import Telegrams
_LOGGER = logging.getLogger(__name__)
@@ -75,6 +76,7 @@ class KNXModule:
self.connected = False
self.yaml_exposures: list[KnxExposeEntity | KnxExposeTime] = []
self.service_exposures: dict[str, KnxExposeEntity | KnxExposeTime] = {}
self.ui_time_server_controller = TimeServerController()
self.entry = entry
self.project = KNXProject(hass=hass, entry=entry)

View File

@@ -12,13 +12,14 @@ from homeassistant.helpers.storage import Store
from homeassistant.util.ulid import ulid_now
from ..const import DOMAIN
from . import migration
from .const import CONF_DATA
from .migration import migrate_1_to_2, migrate_2_1_to_2_2
from .time_server import KNXTimeServerStoreModel
_LOGGER = logging.getLogger(__name__)
STORAGE_VERSION: Final = 2
STORAGE_VERSION_MINOR: Final = 2
STORAGE_VERSION_MINOR: Final = 3
STORAGE_KEY: Final = f"{DOMAIN}/config_store.json"
type KNXPlatformStoreModel = dict[str, dict[str, Any]] # unique_id: configuration
@@ -31,6 +32,7 @@ class KNXConfigStoreModel(TypedDict):
"""Represent KNX configuration store data."""
entities: KNXEntityStoreModel
time_server: KNXTimeServerStoreModel
class PlatformControllerBase(ABC):
@@ -56,11 +58,15 @@ class _KNXConfigStoreStorage(Store[KNXConfigStoreModel]):
"""Migrate to the new version."""
if old_major_version == 1:
# version 2.1 introduced in 2025.8
migrate_1_to_2(old_data)
migration.migrate_1_to_2(old_data)
if old_major_version <= 2 and old_minor_version < 2:
# version 2.2 introduced in 2025.9.2
migrate_2_1_to_2_2(old_data)
migration.migrate_2_1_to_2_2(old_data)
if old_major_version <= 2 and old_minor_version < 3:
# version 2.3 introduced in 2026.3
migration.migrate_2_2_to_2_3(old_data)
return old_data
@@ -79,7 +85,10 @@ class KNXConfigStore:
self._store = _KNXConfigStoreStorage(
hass, STORAGE_VERSION, STORAGE_KEY, minor_version=STORAGE_VERSION_MINOR
)
self.data = KNXConfigStoreModel(entities={})
self.data = KNXConfigStoreModel( # initialize with default structure
entities={},
time_server={},
)
self._platform_controllers: dict[Platform, PlatformControllerBase] = {}
async def load_data(self) -> None:
@@ -174,6 +183,19 @@ class KNXConfigStore:
if registry_entry.unique_id in unique_ids
]
@callback
def get_time_server_config(self) -> KNXTimeServerStoreModel:
"""Return KNX time server configuration."""
return self.data["time_server"]
async def update_time_server_config(self, config: KNXTimeServerStoreModel) -> None:
"""Update time server configuration."""
self.data["time_server"] = config
knx_module = self.hass.data.get(DOMAIN)
if knx_module:
knx_module.ui_time_server_controller.start(knx_module.xknx, config)
await self._store.async_save(self.data)
class ConfigStoreException(Exception):
"""KNX config store exception."""

View File

@@ -4,6 +4,8 @@ from typing import Literal, TypedDict
import voluptuous as vol
from homeassistant.helpers.typing import VolSchemaType
from .entity_store_schema import ENTITY_STORE_DATA_SCHEMA
@@ -37,14 +39,14 @@ def parse_invalid(exc: vol.Invalid) -> _ErrorDescription:
)
def validate_entity_data(entity_data: dict) -> dict:
"""Validate entity data.
def validate_config_store_data(schema: VolSchemaType, entity_data: dict) -> dict:
"""Validate data for config store.
Return validated data or raise EntityStoreValidationException.
"""
try:
# return so defaults are applied
return ENTITY_STORE_DATA_SCHEMA(entity_data) # type: ignore[no-any-return]
return schema(entity_data) # type: ignore[no-any-return]
except vol.MultipleInvalid as exc:
raise EntityStoreValidationException(
validation_error={
@@ -63,6 +65,14 @@ def validate_entity_data(entity_data: dict) -> dict:
) from exc
def validate_entity_data(entity_data: dict) -> dict:
"""Validate entity data.
Return validated data or raise EntityStoreValidationException.
"""
return validate_config_store_data(ENTITY_STORE_DATA_SCHEMA, entity_data)
class EntityStoreValidationException(Exception):
"""Entity store validation exception."""

View File

@@ -50,3 +50,8 @@ def migrate_2_1_to_2_2(data: dict[str, Any]) -> None:
# "respond_to_read" was never used for binary_sensor and is not valid
# in the new schema. It was set as default in Store schema v1 and v2.1
b_sensor["knx"].pop(CONF_RESPOND_TO_READ, None)
def migrate_2_2_to_2_3(data: dict[str, Any]) -> None:
"""Migrate from schema 2.2 to schema 2.3."""
data.setdefault("time_server", {})

View File

@@ -0,0 +1,64 @@
"""Time server controller for KNX integration."""
from __future__ import annotations
from typing import Any, TypedDict
import voluptuous as vol
from xknx import XKNX
from ..expose import KnxExposeTime, create_time_server_exposures
from .entity_store_validation import validate_config_store_data
from .knx_selector import GASelector
class KNXTimeServerStoreModel(TypedDict, total=False):
"""Represent KNX time server configuration store data."""
time: dict[str, Any] | None
date: dict[str, Any] | None
datetime: dict[str, Any] | None
TIME_SERVER_CONFIG_SCHEMA = vol.Schema(
{
vol.Optional("time"): GASelector(
state=False, passive=False, valid_dpt="10.001"
),
vol.Optional("date"): GASelector(
state=False, passive=False, valid_dpt="11.001"
),
vol.Optional("datetime"): GASelector(
state=False, passive=False, valid_dpt="19.001"
),
}
)
def validate_time_server_data(time_server_data: dict) -> KNXTimeServerStoreModel:
"""Validate time server data.
Return validated data or raise EntityStoreValidationException.
"""
return validate_config_store_data(TIME_SERVER_CONFIG_SCHEMA, time_server_data) # type: ignore[return-value]
class TimeServerController:
"""Controller class for UI time exposures."""
def __init__(self) -> None:
"""Initialize time server controller."""
self.time_exposes: list[KnxExposeTime] = []
def stop(self) -> None:
"""Shutdown time server controller."""
for expose in self.time_exposes:
expose.async_remove()
self.time_exposes.clear()
def start(self, xknx: XKNX, config: KNXTimeServerStoreModel) -> None:
"""Update time server configuration."""
if self.time_exposes:
self.stop()
self.time_exposes = create_time_server_exposures(xknx, config)

View File

@@ -36,6 +36,7 @@ from .storage.entity_store_validation import (
validate_entity_data,
)
from .storage.serialize import get_serialized_schema
from .storage.time_server import validate_time_server_data
from .telegrams import (
SIGNAL_KNX_DATA_SECURE_ISSUE_TELEGRAM,
SIGNAL_KNX_TELEGRAM,
@@ -65,6 +66,8 @@ async def register_panel(hass: HomeAssistant) -> None:
websocket_api.async_register_command(hass, ws_get_entity_entries)
websocket_api.async_register_command(hass, ws_create_device)
websocket_api.async_register_command(hass, ws_get_schema)
websocket_api.async_register_command(hass, ws_get_time_server_config)
websocket_api.async_register_command(hass, ws_update_time_server_config)
if DOMAIN not in hass.data.get("frontend_panels", {}):
await hass.http.async_register_static_paths(
@@ -583,3 +586,55 @@ def ws_create_device(
configuration_url=f"homeassistant://knx/entities/view?device_id={_device.id}",
)
connection.send_result(msg["id"], _device.dict_repr)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/get_time_server_config",
}
)
@provide_knx
@callback
def ws_get_time_server_config(
hass: HomeAssistant,
knx: KNXModule,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Get time server configuration from entity store."""
config_info = knx.config_store.get_time_server_config()
connection.send_result(msg["id"], config_info)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/update_time_server_config",
vol.Required("config"): dict, # validation done in handler
}
)
@websocket_api.async_response
@provide_knx
async def ws_update_time_server_config(
hass: HomeAssistant,
knx: KNXModule,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Update entity in entity store and reload it."""
try:
validated_config = validate_time_server_data(msg["config"])
except EntityStoreValidationException as exc:
connection.send_result(msg["id"], exc.validation_error)
return
try:
await knx.config_store.update_time_server_config(validated_config)
except ConfigStoreException as err:
connection.send_error(
msg["id"], websocket_api.const.ERR_HOME_ASSISTANT_ERROR, str(err)
)
return
connection.send_result(
msg["id"], EntityStoreValidationSuccess(success=True, entity_id=None)
)

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -21,6 +21,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -131,6 +131,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 1,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -77,6 +77,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -38,6 +38,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -38,6 +38,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -46,6 +46,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -137,6 +137,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 1,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -42,6 +42,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -19,6 +19,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -21,6 +21,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -24,6 +24,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -1,6 +1,6 @@
{
"version": 2,
"minor_version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {
@@ -38,6 +38,7 @@
}
}
}
}
},
"time_server": {}
}
}

View File

@@ -0,0 +1,13 @@
{
"version": 2,
"minor_version": 3,
"key": "knx/config_store.json",
"data": {
"entities": {},
"time_server": {
"time": { "write": "1/1/1" },
"date": { "write": "2/2/2" },
"datetime": { "write": "3/3/3" }
}
}
}

View File

@@ -12,6 +12,8 @@
'config_store': dict({
'entities': dict({
}),
'time_server': dict({
}),
}),
'configuration_yaml': dict({
'wrong_key': dict({
@@ -42,6 +44,8 @@
'config_store': dict({
'entities': dict({
}),
'time_server': dict({
}),
}),
'configuration_yaml': None,
'project_info': None,
@@ -65,6 +69,8 @@
'config_store': dict({
'entities': dict({
}),
'time_server': dict({
}),
}),
'configuration_yaml': None,
'project_info': None,
@@ -128,6 +134,8 @@
}),
}),
}),
'time_server': dict({
}),
}),
'configuration_yaml': None,
'project_info': dict({

View File

@@ -460,12 +460,12 @@ async def test_migration_1_to_2(
assert hass_storage[KNX_CONFIG_STORAGE_KEY] == new_data
async def test_migration_2_1_to_2_2(
async def test_migration_2_1_to_2_3(
hass: HomeAssistant,
knx: KNXTestKit,
hass_storage: dict[str, Any],
) -> None:
"""Test migration from schema 2.1 to schema 2.2."""
"""Test migration from schema 2.1 to schema 2.3."""
await knx.setup_integration(
config_store_fixture="config_store_binarysensor_v2_1.json",
state_updater=False,

View File

@@ -0,0 +1,158 @@
"""Test KNX time server."""
import pytest
from homeassistant.core import HomeAssistant
from .conftest import KNXTestKit
from tests.typing import WebSocketGenerator
# Freeze time: 2026-1-29 11:02:03 UTC -> Europe/Vienna (UTC+1) = 12:02:03 Thursday
FREEZE_TIME = "2026-1-29 11:02:03"
# KNX Time DPT 10.001: Day of week + time
# 0x8C = 0b10001100 = Thursday (100) + 12 hours (01100)
# 0x02 = 2 minutes
# 0x03 = 3 seconds
RAW_TIME = (0x8C, 0x02, 0x03)
# KNX Date DPT 11.001
# 0x1D = 29th day
# 0x01 = January (month 1)
# 0x1A = 26 (2026 - 2000)
RAW_DATE = (0x1D, 0x01, 0x1A)
# KNX DateTime DPT 19.001: Year, Month, Day, Hour+DoW, Minutes, Seconds, Flags, Quality
# 0x7E = 126 (offset from 1900)
# 0x01 = January
# 0x1D = 29th day
# 0x8C = Thursday + 12 hours
# 0x02 = 2 minutes
# 0x03 = 3 seconds
# 0x20 = ignore working day flag, no DST
# 0xC0 = external sync, reliable source
RAW_DATETIME = (0x7E, 0x01, 0x1D, 0x8C, 0x02, 0x03, 0x20, 0xC0)
@pytest.mark.freeze_time(FREEZE_TIME)
@pytest.mark.parametrize(
("config", "expected_telegrams"),
[
(
{"time": {"write": "1/1/1"}},
{"1/1/1": RAW_TIME},
),
(
{"date": {"write": "2/2/2"}},
{"2/2/2": RAW_DATE},
),
(
{"datetime": {"write": "3/3/3"}},
{"3/3/3": RAW_DATETIME},
),
(
{"time": {"write": "1/1/1"}, "date": {"write": "2/2/2"}},
{
"1/1/1": RAW_TIME,
"2/2/2": RAW_DATE,
},
),
(
{"date": {"write": "2/2/2"}, "datetime": {"write": "3/3/3"}},
{
"2/2/2": RAW_DATE,
"3/3/3": RAW_DATETIME,
},
),
],
)
async def test_time_server_write_format(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
config: dict,
expected_telegrams: dict[str, tuple],
) -> None:
"""Test time server writes each format when configured."""
await hass.config.async_set_time_zone("Europe/Vienna")
await knx.setup_integration({})
client = await hass_ws_client(hass)
# Get initial empty configuration
await client.send_json_auto_id({"type": "knx/get_time_server_config"})
res = await client.receive_json()
assert res["success"], res
assert res["result"] == {}
# Update time server config to enable format
await client.send_json_auto_id(
{"type": "knx/update_time_server_config", "config": config}
)
res = await client.receive_json()
assert res["success"], res
# Verify telegrams are written
for address, expected_value in expected_telegrams.items():
await knx.assert_write(address, expected_value)
# Verify read responses work
for address, expected_value in expected_telegrams.items():
await knx.receive_read(address)
await knx.assert_response(address, expected_value)
@pytest.mark.freeze_time(FREEZE_TIME)
async def test_time_server_load_from_config_store(
hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator
) -> None:
"""Test time server is loaded correctly from config store."""
await hass.config.async_set_time_zone("Europe/Vienna")
await knx.setup_integration(
{}, config_store_fixture="config_store_time_server.json"
)
# Verify all three formats are written on startup
await knx.assert_write("1/1/1", RAW_TIME)
await knx.assert_write("2/2/2", RAW_DATE)
await knx.assert_write("3/3/3", RAW_DATETIME)
client = await hass_ws_client(hass)
# Verify configuration was loaded
await client.send_json_auto_id({"type": "knx/get_time_server_config"})
res = await client.receive_json()
assert res["success"], res
assert res["result"] == {
"time": {"write": "1/1/1"},
"date": {"write": "2/2/2"},
"datetime": {"write": "3/3/3"},
}
@pytest.mark.parametrize(
"invalid_config",
[
{"invalid": 1},
{"time": {"state": "1/2/3"}},
{"time": {"write": "not_an_address"}},
{"date": {"passive": ["1/2/3"]}},
{"datetime": {}},
{"time": {"write": "1/2/3"}, "invalid_key": "value"},
],
)
async def test_time_server_invalid_config(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
invalid_config: dict,
) -> None:
"""Test invalid time server configuration is rejected."""
await knx.setup_integration({})
client = await hass_ws_client(hass)
# Try to update with invalid configuration
await client.send_json_auto_id(
{"type": "knx/update_time_server_config", "config": invalid_config}
)
res = await client.receive_json()
assert res["success"] # uses custom error handling
assert not res["result"]["success"]
assert "errors" in res["result"]