Type the common part of KNX UI entity configurations (#182728)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Matthias Alphart
2026-09-21 19:03:06 +02:00
committed by GitHub
co-authored by Claude Opus 5
parent 5c99cd45d3
commit 24fca373b9
22 changed files with 194 additions and 197 deletions
@@ -31,7 +31,6 @@ from .const import (
CONF_RESET_AFTER,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
DOMAIN,
KNX_MODULE_KEY,
)
from .entity import (
@@ -41,9 +40,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import BinarySensorKnxConfig
from .storage.entity_store_schema import BinarySensorKnxConfig, KnxEntityData
async def async_setup_entry(
@@ -158,12 +155,12 @@ class KnxUiBinarySensor(_KnxBinarySensor, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxBinarySensor(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address_state=knx_conf.ga_sensor.state_and_passive(),
sync_state=knx_conf.sync_state,
invert=knx_conf.invert,
+7 -7
View File
@@ -14,7 +14,7 @@ from homeassistant.helpers.entity_platform import (
)
from homeassistant.helpers.typing import ConfigType
from .const import CONF_PAYLOAD_LENGTH, CONF_VALUE, DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY
from .const import CONF_PAYLOAD_LENGTH, CONF_VALUE, KNX_ADDRESS, KNX_MODULE_KEY
from .entity import (
KnxUiEntity,
KnxUiEntityPlatformController,
@@ -22,8 +22,8 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_DATA, CONF_ENTITY, CONF_GA_SEND
from .storage.const import CONF_DATA, CONF_GA_SEND
from .storage.entity_store_schema import KnxEntityData
from .storage.util import ConfigExtractor
@@ -104,13 +104,13 @@ class KnxUiButton(_KnxButton, KnxUiEntity):
self, knx_module: KNXModule, unique_id: str, config: KnxEntityData[Any]
) -> None:
"""Initialize a KNX button."""
knx_conf = ConfigExtractor(config[DOMAIN])
knx_conf = ConfigExtractor(config.knx)
button_data = knx_conf.get(CONF_DATA)
if CONF_PAYLOAD in button_data and CONF_PAYLOAD_LENGTH in button_data:
self._payload = int(button_data[CONF_PAYLOAD], 16)
self._device = XknxRawValue(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
payload_length=button_data[CONF_PAYLOAD_LENGTH],
group_address=knx_conf.get_write(CONF_GA_SEND),
)
@@ -119,7 +119,7 @@ class KnxUiButton(_KnxButton, KnxUiEntity):
self._payload = button_data[CONF_VALUE]
self._device = XknxExposeSensor(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
value_type=dpt_string,
group_address=knx_conf.get_write(CONF_GA_SEND),
respond_to_read=False,
@@ -128,5 +128,5 @@ class KnxUiButton(_KnxButton, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
+8 -7
View File
@@ -37,7 +37,6 @@ from .const import (
CONF_SYNC_STATE,
CONTROLLER_MODES,
CURRENT_HVAC_ACTIONS,
DOMAIN,
KNX_MODULE_KEY,
ClimateConf,
)
@@ -50,9 +49,7 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import ClimateSchema
from .storage.config_store import KnxEntityData
from .storage.const import (
CONF_ENTITY,
CONF_GA_ACTIVE,
CONF_GA_CONTROLLER_MODE,
CONF_GA_CONTROLLER_STATUS,
@@ -74,7 +71,11 @@ from .storage.const import (
CONF_IGNORE_AUTO_MODE,
CONF_TARGET_TEMPERATURE,
)
from .storage.entity_store_schema import ConfClimateFanSpeedMode, ConfSetpointShiftMode
from .storage.entity_store_schema import (
ConfClimateFanSpeedMode,
ConfSetpointShiftMode,
KnxEntityData,
)
from .storage.util import ConfigExtractor
ATTR_COMMAND_VALUE = "command_value"
@@ -705,11 +706,11 @@ class KnxUiClimate(_KnxClimate, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = ConfigExtractor(config[DOMAIN])
knx_conf = ConfigExtractor(config.knx)
self._device = _create_climate_ui(
knx_module.xknx, knx_conf, config[CONF_ENTITY][CONF_NAME]
knx_module.xknx, knx_conf, config.entity.xknx_name
)
default_hvac_mode = HVACMode(knx_conf.get(ClimateConf.DEFAULT_CONTROLLER_MODE))
+4 -5
View File
@@ -30,7 +30,7 @@ from homeassistant.helpers.entity_platform import (
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType
from .const import CONF_SYNC_STATE, DOMAIN, KNX_MODULE_KEY, CoverConf
from .const import CONF_SYNC_STATE, KNX_MODULE_KEY, CoverConf
from .entity import (
KnxUiEntity,
KnxUiEntityPlatformController,
@@ -39,9 +39,7 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import CoverSchema
from .storage.config_store import KnxEntityData
from .storage.const import (
CONF_ENTITY,
CONF_GA_ANGLE,
CONF_GA_POSITION_SET,
CONF_GA_POSITION_STATE,
@@ -49,6 +47,7 @@ from .storage.const import (
CONF_GA_STOP,
CONF_GA_UP_DOWN,
)
from .storage.entity_store_schema import KnxEntityData
from .storage.util import ConfigExtractor
@@ -313,9 +312,9 @@ class KnxUiCover(_KnxCover, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
self._device = _create_ui_cover(
knx_module.xknx, config[DOMAIN], config[CONF_ENTITY][CONF_NAME]
knx_module.xknx, config.knx, config.entity.xknx_name
)
self.init_base()
+4 -7
View File
@@ -21,7 +21,6 @@ from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
DOMAIN,
KNX_ADDRESS,
KNX_MODULE_KEY,
)
@@ -32,9 +31,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import DateKnxConfig
from .storage.entity_store_schema import DateKnxConfig, KnxEntityData
async def async_setup_entry(
@@ -137,12 +134,12 @@ class KnxUiDate(_KNXDate, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxDateDevice(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
localtime=False,
group_address=knx_conf.ga_date.write,
group_address_state=knx_conf.ga_date.state_and_passive(),
+4 -7
View File
@@ -22,7 +22,6 @@ from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
DOMAIN,
KNX_ADDRESS,
KNX_MODULE_KEY,
)
@@ -33,9 +32,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import DatetimeKnxConfig
from .storage.entity_store_schema import DatetimeKnxConfig, KnxEntityData
async def async_setup_entry(
@@ -142,12 +139,12 @@ class KnxUiDateTime(_KNXDateTime, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxDateTimeDevice(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
localtime=False,
group_address=knx_conf.ga_datetime.write,
group_address_state=knx_conf.ga_datetime.state_and_passive(),
+9 -7
View File
@@ -27,8 +27,8 @@ from homeassistant.helpers.entity_platform import (
from homeassistant.helpers.entity_registry import RegistryEntry
from .const import CONF_DEFAULT_ENTITY_ID, DOMAIN
from .storage.config_store import KnxEntityData, PlatformControllerBase
from .storage.const import CONF_DEVICE_INFO
from .storage.config_store import PlatformControllerBase
from .storage.entity_store_schema import BaseEntityConfig, KnxEntityData
if TYPE_CHECKING:
from .knx_module import KNXModule
@@ -260,13 +260,15 @@ class KnxUiEntity(_KnxEntityBase):
"""Representation of a KNX UI entity."""
def __init__(
self, knx_module: KNXModule, unique_id: str, entity_config: dict[str, Any]
self, knx_module: KNXModule, unique_id: str, entity_config: BaseEntityConfig
) -> None:
"""Initialize the UI entity."""
self._knx_module = knx_module
self._attr_name = entity_config[CONF_NAME]
self._attr_name = entity_config.name
self._attr_unique_id = unique_id
self._attr_entity_category = entity_config[CONF_ENTITY_CATEGORY]
if device_info := entity_config[CONF_DEVICE_INFO]:
self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, device_info)})
self._attr_entity_category = entity_config.entity_category
if entity_config.device_info:
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, entity_config.device_info)}
)
+4 -5
View File
@@ -33,15 +33,14 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import FanSchema
from .storage.config_store import KnxEntityData
from .storage.const import (
CONF_ENTITY,
CONF_GA_OSCILLATION,
CONF_GA_SPEED,
CONF_GA_STEP,
CONF_GA_SWITCH,
CONF_SPEED,
)
from .storage.entity_store_schema import KnxEntityData
from .storage.util import ConfigExtractor
_LOGGER = logging.getLogger(__name__)
@@ -257,13 +256,13 @@ class KnxUiFan(_KnxFan, KnxUiEntity):
self, knx_module: KNXModule, unique_id: str, config: KnxEntityData[Any]
) -> None:
"""Initialize of KNX fan."""
knx_conf = ConfigExtractor(config[DOMAIN])
knx_conf = ConfigExtractor(config.knx)
# max_step is required for step mode, thus can be used to differentiate modes
max_step: int | None = knx_conf.get(CONF_SPEED, FanConf.MAX_STEP)
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
if max_step:
# step control
@@ -276,7 +275,7 @@ class KnxUiFan(_KnxFan, KnxUiEntity):
self._device = XknxFan(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address_speed=speed_write,
group_address_speed_state=speed_state,
group_address_oscillation=knx_conf.get_write(CONF_GA_OSCILLATION),
+6 -8
View File
@@ -27,7 +27,7 @@ from homeassistant.helpers.entity_platform import (
from homeassistant.helpers.typing import ConfigType
from homeassistant.util import color as color_util
from .const import CONF_SYNC_STATE, DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY, ColorTempModes
from .const import CONF_SYNC_STATE, KNX_ADDRESS, KNX_MODULE_KEY, ColorTempModes
from .entity import (
KnxUiEntity,
KnxUiEntityPlatformController,
@@ -36,12 +36,10 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import LightSchema
from .storage.config_store import KnxEntityData
from .storage.const import (
CONF_COLOR,
CONF_COLOR_TEMP_MAX,
CONF_COLOR_TEMP_MIN,
CONF_ENTITY,
CONF_GA_BLUE_BRIGHTNESS,
CONF_GA_BLUE_SWITCH,
CONF_GA_BRIGHTNESS,
@@ -57,7 +55,7 @@ from .storage.const import (
CONF_GA_WHITE_BRIGHTNESS,
CONF_GA_WHITE_SWITCH,
)
from .storage.entity_store_schema import LightColorMode
from .storage.entity_store_schema import KnxEntityData, LightColorMode
from .storage.util import ConfigExtractor
@@ -610,11 +608,11 @@ class KnxUiLight(_KnxLight, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
self._device = _create_ui_light(
knx_module.xknx, config[DOMAIN], config[CONF_ENTITY][CONF_NAME]
knx_module.xknx, config.knx, config.entity.xknx_name
)
self._attr_color_mode = next(iter(self.supported_color_modes))
self._attr_max_color_temp_kelvin: int = config[DOMAIN][CONF_COLOR_TEMP_MAX]
self._attr_min_color_temp_kelvin: int = config[DOMAIN][CONF_COLOR_TEMP_MIN]
self._attr_max_color_temp_kelvin: int = config.knx[CONF_COLOR_TEMP_MAX]
self._attr_min_color_temp_kelvin: int = config.knx[CONF_COLOR_TEMP_MIN]
+5 -7
View File
@@ -14,7 +14,7 @@ from homeassistant.helpers.entity_platform import (
)
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY
from .const import KNX_ADDRESS, KNX_MODULE_KEY
from .entity import (
KnxUiEntity,
KnxUiEntityPlatformController,
@@ -22,9 +22,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import NotifyKnxConfig
from .storage.entity_store_schema import KnxEntityData, NotifyKnxConfig
async def async_setup_entry(
@@ -107,12 +105,12 @@ class KnxUiNotify(_KnxNotify, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxNotification(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address=knx_conf.ga_send.write,
value_type=knx_conf.ga_send.dpt,
)
+4 -7
View File
@@ -28,7 +28,6 @@ from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
DOMAIN,
KNX_ADDRESS,
KNX_MODULE_KEY,
NumberConf,
@@ -41,9 +40,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import NumberKnxConfig
from .storage.entity_store_schema import KnxEntityData, NumberKnxConfig
async def async_setup_entry(
@@ -178,16 +175,16 @@ class KnxUiNumber(_KnxNumber, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
dpt_string = knx_conf.ga_sensor.dpt
assert dpt_string is not None # required for number
dpt_info = get_supported_dpts()[dpt_string]
self._device = NumericValue(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address=knx_conf.ga_sensor.write,
group_address_state=knx_conf.ga_sensor.state_and_passive(),
respond_to_read=knx_conf.respond_to_read,
+5 -7
View File
@@ -14,7 +14,7 @@ from homeassistant.helpers.entity_platform import (
)
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY
from .const import KNX_ADDRESS, KNX_MODULE_KEY
from .entity import (
KnxUiEntity,
KnxUiEntityPlatformController,
@@ -24,9 +24,7 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import SceneSchema
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import SceneKnxConfig
from .storage.entity_store_schema import KnxEntityData, SceneKnxConfig
async def async_setup_entry(
@@ -117,12 +115,12 @@ class KnxUiScene(_KnxScene, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxScene(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address=knx_conf.ga_scene.write,
scene_number=knx_conf.scene_number,
)
+4 -6
View File
@@ -29,7 +29,6 @@ from .const import (
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
CONF_VALUE,
DOMAIN,
KNX_ADDRESS,
KNX_MODULE_KEY,
SelectConf,
@@ -42,8 +41,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import KnxEntityData
from .storage.util import ConfigExtractor
_LOGGER = logging.getLogger(__name__)
@@ -206,7 +204,7 @@ class KnxUiSelect(_KNXSelect, KnxUiEntity):
self, knx_module: KNXModule, unique_id: str, config: KnxEntityData[Any]
) -> None:
"""Initialize a KNX select."""
knx_conf = ConfigExtractor(config[DOMAIN])
knx_conf = ConfigExtractor(config.knx)
source = knx_conf.get(SelectConf.OPTIONS_SOURCE)
# the group address key tells how options are defined
if SelectConf.GA_ENUM in source:
@@ -223,7 +221,7 @@ class KnxUiSelect(_KNXSelect, KnxUiEntity):
self._device = RawValue(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
payload_length=payload_length,
group_address=knx_conf.get_write(SelectConf.OPTIONS_SOURCE, ga_key),
group_address_state=knx_conf.get_state_and_passive(
@@ -235,6 +233,6 @@ class KnxUiSelect(_KNXSelect, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
self._attr_options = list(self._option_payloads)
+5 -7
View File
@@ -36,7 +36,7 @@ from homeassistant.helpers.entity_platform import (
from homeassistant.helpers.typing import ConfigType, StateType
from homeassistant.util.enum import try_parse_enum
from .const import ATTR_SOURCE, CONF_SYNC_STATE, DOMAIN, KNX_MODULE_KEY
from .const import ATTR_SOURCE, CONF_SYNC_STATE, KNX_MODULE_KEY
from .dpt import get_supported_dpts
from .entity import (
KnxUiEntity,
@@ -47,9 +47,7 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import SensorSchema
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import SensorKnxConfig
from .storage.entity_store_schema import KnxEntityData, SensorKnxConfig
SCAN_INTERVAL = timedelta(seconds=10)
@@ -256,16 +254,16 @@ class KnxUiSensor(_KnxSensor, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
dpt_string = knx_conf.ga_sensor.dpt
assert dpt_string is not None # required for sensor
dpt_info = get_supported_dpts()[dpt_string]
self._device = XknxSensor(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address_state=knx_conf.ga_sensor.state_and_passive(),
sync_state=knx_conf.sync_state,
always_callback=True,
@@ -9,7 +9,6 @@ from typing import (
Any,
Final,
TypedDict,
cast,
get_args,
get_origin,
get_type_hints,
@@ -30,6 +29,7 @@ from ..const import DOMAIN, KNX_MODULE_KEY
from ..repairs import async_create_entity_validation_issue
from . import migration
from .const import CONF_DATA, CONF_ENTITY
from .entity_store_schema import KnxEntityData
from .entity_store_validation import (
EntityStoreValidationException,
validate_entity_data,
@@ -58,24 +58,20 @@ class KNXConfigStoreModel(TypedDict):
time_server: KNXTimeServerStoreModel
class KnxEntityData[KnxT](TypedDict):
"""Validated entity data: the common `entity` and the platform `knx` part."""
entity: dict[str, Any]
knx: KnxT
def to_storage_dict(data: KnxEntityData[Any]) -> dict[str, Any]:
"""Render validated entity data to its JSON serializable storage form."""
knx_config = data[DOMAIN]
if isinstance(knx_config, dict):
return cast(dict[str, Any], data) # platform not yet migrated to a typed config
return {
CONF_ENTITY: data[CONF_ENTITY],
DOMAIN: {
name: encode(getattr(knx_config, name))
for name, encode in _storage_encoders(type(knx_config))
},
CONF_ENTITY: dataclasses.asdict(data.entity),
DOMAIN: _knx_to_storage(data.knx),
}
def _knx_to_storage(knx_config: Any) -> dict[str, Any]:
if isinstance(knx_config, dict):
return knx_config # platform not yet migrated to a typed config
return {
name: encode(getattr(knx_config, name))
for name, encode in _storage_encoders(type(knx_config))
}
@@ -227,9 +223,9 @@ class KNXConfigStore:
invalid.append(unique_id)
continue
data: KnxEntityData[Any] = result[CONF_DATA]
if config_type is not None and not isinstance(data[DOMAIN], config_type):
if config_type is not None and not isinstance(data.knx, config_type):
raise TypeError(
f"{platform} schema yields {type(data[DOMAIN]).__name__},"
f"{platform} schema yields {type(data.knx).__name__},"
f" not {config_type.__name__}"
)
validated[unique_id] = data
@@ -3,7 +3,7 @@
from collections.abc import Hashable
from dataclasses import dataclass
from enum import StrEnum, unique
from typing import Annotated
from typing import Annotated, Any
import probatio
from probatio import Key
@@ -23,9 +23,9 @@ from homeassistant.components.text import TextMode
from homeassistant.const import (
CONF_ENTITY_CATEGORY,
CONF_ENTITY_ID,
CONF_NAME,
CONF_PAYLOAD,
CONF_PLATFORM,
EntityCategory,
Platform,
)
from homeassistant.helpers import selector
@@ -47,7 +47,8 @@ from ..const import (
)
from ..dpt import get_supported_dpts, raw_payload_length
from ..validation import (
entity_category_validator,
entity_category_supported,
parse_entity_category,
validate_number_attributes,
validate_sensor_attributes,
)
@@ -56,7 +57,6 @@ from .const import (
CONF_COLOR_TEMP_MAX,
CONF_COLOR_TEMP_MIN,
CONF_DATA,
CONF_DEVICE_INFO,
CONF_DPT,
CONF_ENTITY,
CONF_GA_ACTIVE,
@@ -135,34 +135,52 @@ SyncState = Annotated[bool | str | int, SyncStateSelector()]
SyncStateAllowFalse = Annotated[bool | str | int, SyncStateSelector(allow_false=True)]
@dataclass(kw_only=True, slots=True)
class BaseEntityConfig:
"""Common UI configuration of a KNX entity."""
name: str | None = None
device_info: str | None = None
entity_category: Annotated[
EntityCategory | None, probatio.Coerce(parse_entity_category)
] = None
@property
def xknx_name(self) -> str:
"""Name of the xknx device, empty when HA names the entity after its device."""
return self.name or ""
def _name_or_device_required(config: BaseEntityConfig) -> BaseEntityConfig:
"""Require a name, unless the entity is named after its device."""
if not config.name and config.device_info is None:
raise probatio.AnyInvalid("One of `Device` or `Name` is required")
return config
def base_entity_schema(platform: Platform) -> probatio.All:
"""Return the base entity schema for a platform."""
return probatio.All(
{
probatio.Optional(CONF_NAME, default=None): probatio.Maybe(str),
probatio.Optional(CONF_DEVICE_INFO, default=None): probatio.Maybe(str),
probatio.Optional(
CONF_ENTITY_CATEGORY, default=None
): entity_category_validator(platform),
},
probatio.Any(
probatio.Schema(
{
probatio.Required(CONF_NAME): probatio.All(str, probatio.IsTrue()),
},
extra=probatio.ALLOW_EXTRA,
),
probatio.Schema(
{
probatio.Required(CONF_DEVICE_INFO): str,
},
extra=probatio.ALLOW_EXTRA,
),
msg="One of `Device` or `Name` is required",
probatio.DataclassSchema(
BaseEntityConfig,
{CONF_ENTITY_CATEGORY: entity_category_supported(platform)},
),
_name_or_device_required,
)
@dataclass(kw_only=True, slots=True)
class KnxEntityData[KnxT]:
"""Validated UI entity data: the common `entity` and the platform `knx` part."""
entity: BaseEntityConfig
knx: KnxT
def _to_entity_data(data: dict[str, Any]) -> KnxEntityData[Any]:
return KnxEntityData(entity=data[CONF_ENTITY], knx=data[DOMAIN])
@dataclass(kw_only=True, slots=True)
class BinarySensorKnxConfig:
"""UI configuration of a KNX binary sensor."""
@@ -1118,14 +1136,17 @@ ENTITY_STORE_DATA_SCHEMA = probatio.All(
{
platform: probatio.Schema(
{
probatio.Required(CONF_DATA): probatio.Schema(
{
probatio.Required(CONF_ENTITY): base_entity_schema(
platform
),
probatio.Required(DOMAIN): knx_schema,
},
extra=probatio.PREVENT_EXTRA, # restrict in data key for yaml edit
probatio.Required(CONF_DATA): probatio.All(
probatio.Schema(
{
probatio.Required(CONF_ENTITY): base_entity_schema(
platform
),
probatio.Required(DOMAIN): knx_schema,
},
extra=probatio.PREVENT_EXTRA, # restrict in data key for yaml edit
),
_to_entity_data,
),
},
extra=probatio.ALLOW_EXTRA, # eg. "type" from WS-endpoint when validating directly
+5 -13
View File
@@ -22,13 +22,7 @@ from homeassistant.helpers.entity_platform import (
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_RESPOND_TO_READ,
CONF_SYNC_STATE,
DOMAIN,
KNX_ADDRESS,
KNX_MODULE_KEY,
)
from .const import CONF_RESPOND_TO_READ, CONF_SYNC_STATE, KNX_ADDRESS, KNX_MODULE_KEY
from .entity import (
KnxUiEntity,
KnxUiEntityPlatformController,
@@ -37,9 +31,7 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import SwitchSchema
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import SwitchKnxConfig
from .storage.entity_store_schema import KnxEntityData, SwitchKnxConfig
async def async_setup_entry(
@@ -145,12 +137,12 @@ class KnxUiSwitch(_KnxSwitch, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxSwitch(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address=knx_conf.ga_switch.write,
group_address_state=knx_conf.ga_switch.state_and_passive(),
respond_to_read=knx_conf.respond_to_read,
+4 -7
View File
@@ -28,7 +28,6 @@ from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
DOMAIN,
KNX_ADDRESS,
KNX_MODULE_KEY,
)
@@ -39,9 +38,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import TextKnxConfig
from .storage.entity_store_schema import KnxEntityData, TextKnxConfig
async def async_setup_entry(
@@ -155,12 +152,12 @@ class KnxUiText(_KnxText, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxNotification(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
group_address=knx_conf.ga_text.write,
group_address_state=knx_conf.ga_text.state_and_passive(),
respond_to_read=knx_conf.respond_to_read,
+4 -7
View File
@@ -21,7 +21,6 @@ from .const import (
CONF_RESPOND_TO_READ,
CONF_STATE_ADDRESS,
CONF_SYNC_STATE,
DOMAIN,
KNX_ADDRESS,
KNX_MODULE_KEY,
)
@@ -32,9 +31,7 @@ from .entity import (
build_yaml_unique_id,
)
from .knx_module import KNXModule
from .storage.config_store import KnxEntityData
from .storage.const import CONF_ENTITY
from .storage.entity_store_schema import TimeKnxConfig
from .storage.entity_store_schema import KnxEntityData, TimeKnxConfig
async def async_setup_entry(
@@ -137,12 +134,12 @@ class KnxUiTime(_KNXTime, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = config[DOMAIN]
knx_conf = config.knx
self._device = XknxTimeDevice(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
localtime=False,
group_address=knx_conf.ga_time.write,
group_address_state=knx_conf.ga_time.state_and_passive(),
+23 -16
View File
@@ -65,28 +65,27 @@ string_type_validator = dpt_subclass_validator(DPTString)
sensor_type_validator = probatio.Any(numeric_type_validator, string_type_validator)
def entity_category_validator(
platform: Platform,
) -> Callable[[Any], EntityCategory | None]:
"""Validate the entity category is supported by the platform.
def parse_entity_category(value: Any) -> EntityCategory | None:
"""Parse an entity category; `None` and "" (the UI clears with it) mean none."""
if value is None or value == "":
return None
try:
return EntityCategory(value)
except ValueError:
raise probatio.Invalid(f"'{value}' is not a valid entity category") from None
Works for both, UI and YAML configuration schema.
"""
def entity_category_supported(
platform: Platform,
) -> Callable[[EntityCategory | None], EntityCategory | None]:
"""Validate a parsed entity category is supported by the platform."""
valid_categories = set(EntityCategory)
if platform in PLATFORMS_WITHOUT_CONFIG_CATEGORY:
valid_categories -= {EntityCategory.CONFIG}
def validate(value: Any) -> EntityCategory | None:
def validate(entity_category: EntityCategory | None) -> EntityCategory | None:
"""Validate the entity category."""
if value is None or value == "": # UI sends an empty value to clear it
return None
try:
entity_category = EntityCategory(value)
except ValueError:
raise probatio.Invalid(
f"'{value}' is not a valid entity category"
) from None
if entity_category not in valid_categories:
if entity_category is not None and entity_category not in valid_categories:
_options = ", ".join(sorted(valid_categories))
raise probatio.Invalid(
f"Entity category '{entity_category}' is not supported by the"
@@ -97,6 +96,14 @@ def entity_category_validator(
return validate
def entity_category_validator(platform: Platform) -> probatio.All:
"""Validate the entity category is supported by the platform.
Works for both, UI and YAML configuration schema.
"""
return probatio.All(parse_entity_category, entity_category_supported(platform))
def ga_validator(value: Any) -> str | int:
"""Validate that value is parsable as GroupAddress or InternalGroupAddress."""
if not isinstance(value, (str, int)):
+5 -6
View File
@@ -20,7 +20,7 @@ from homeassistant.helpers.entity_platform import (
)
from homeassistant.helpers.typing import ConfigType
from .const import CONF_SYNC_STATE, DOMAIN, KNX_MODULE_KEY
from .const import CONF_SYNC_STATE, KNX_MODULE_KEY
from .entity import (
KnxUiEntity,
KnxUiEntityPlatformController,
@@ -29,9 +29,7 @@ from .entity import (
)
from .knx_module import KNXModule
from .schema import WeatherSchema
from .storage.config_store import KnxEntityData
from .storage.const import (
CONF_ENTITY,
CONF_GA_AIR_PRESSURE,
CONF_GA_BRIGHTNESS_EAST,
CONF_GA_BRIGHTNESS_NORTH,
@@ -47,6 +45,7 @@ from .storage.const import (
CONF_GA_WIND_SPEED,
CONF_INVERT_DAY_NIGHT,
)
from .storage.entity_store_schema import KnxEntityData
from .storage.util import ConfigExtractor
@@ -197,12 +196,12 @@ class KnxUiWeather(_KnxWeather, KnxUiEntity):
super().__init__(
knx_module=knx_module,
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
entity_config=config.entity,
)
knx_conf = ConfigExtractor(config[DOMAIN])
knx_conf = ConfigExtractor(config.knx)
self._device = XknxWeather(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
name=config.entity.xknx_name,
sync_state=knx_conf.get(CONF_SYNC_STATE),
group_address_temperature=knx_conf.get_state_and_passive(
CONF_GA_TEMPERATURE
+12 -3
View File
@@ -17,9 +17,11 @@ from homeassistant.components.knx.storage.config_store import (
)
from homeassistant.components.knx.storage.const import CONF_DATA
from homeassistant.components.knx.storage.entity_store_schema import (
BaseEntityConfig,
BinarySensorKnxConfig,
DateKnxConfig,
DatetimeKnxConfig,
KnxEntityData,
NotifyKnxConfig,
NumberKnxConfig,
SceneKnxConfig,
@@ -751,7 +753,7 @@ async def test_load_applies_schema_defaults_and_coercion(
)
assert hass.states.get("light.missing_defaults") is not None
config_store = hass.data[KNX_MODULE_KEY].config_store
light_config = config_store.get_entity_configs(Platform.LIGHT)[LIGHT_UID][DOMAIN]
light_config = config_store.get_entity_configs(Platform.LIGHT)[LIGHT_UID].knx
assert light_config["color_temp_min"] == 2700
assert light_config["color_temp_max"] == 6000
@@ -980,16 +982,23 @@ def test_typed_config_storage_roundtrip(
validated = validate_entity_data(
{CONF_PLATFORM: platform, CONF_DATA: {"entity": entity_input, "knx": knx_input}}
)[CONF_DATA]
assert isinstance(validated[DOMAIN], config_type)
assert isinstance(validated, KnxEntityData)
assert validated.entity == BaseEntityConfig(name="test")
assert isinstance(validated.knx, config_type)
stored = to_storage_dict(validated)
assert stored["entity"] == {
"name": "test",
"device_info": None,
"entity_category": None,
}
assert stored["knx"] == knx_stored
assert json.loads(json.dumps(stored)) == stored # storage is JSON
reloaded = validate_entity_data({CONF_PLATFORM: platform, CONF_DATA: stored})[
CONF_DATA
]
assert reloaded[DOMAIN] == validated[DOMAIN]
assert reloaded == validated
assert to_storage_dict(reloaded) == stored