mirror of
https://github.com/home-assistant/core.git
synced 2026-02-03 22:05:35 +01:00
Compare commits
4 Commits
pr-162044
...
knx-time-s
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f8ed52fe31 | ||
|
|
34589fcda7 | ||
|
|
da027c063d | ||
|
|
66ba2819e1 |
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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."""
|
||||
|
||||
|
||||
@@ -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", {})
|
||||
|
||||
64
homeassistant/components/knx/storage/time_server.py
Normal file
64
homeassistant/components/knx/storage/time_server.py
Normal 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)
|
||||
@@ -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)
|
||||
)
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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": {}
|
||||
}
|
||||
}
|
||||
|
||||
13
tests/components/knx/fixtures/config_store_time_server.json
Normal file
13
tests/components/knx/fixtures/config_store_time_server.json
Normal 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" }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
158
tests/components/knx/test_time_server.py
Normal file
158
tests/components/knx/test_time_server.py
Normal 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"]
|
||||
Reference in New Issue
Block a user