From d173d25072e5ca6542192dbe028f40996e4ffa02 Mon Sep 17 00:00:00 2001 From: Matthias Alphart Date: Tue, 13 Jan 2026 17:25:46 +0100 Subject: [PATCH] Refactor KNX expose entity class (#160705) --- homeassistant/components/knx/__init__.py | 14 +- homeassistant/components/knx/expose.py | 244 ++++++++++++++------- homeassistant/components/knx/knx_module.py | 6 +- homeassistant/components/knx/services.py | 4 +- 4 files changed, 183 insertions(+), 85 deletions(-) diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index ead846735c9..cf91107852a 100644 --- a/homeassistant/components/knx/__init__.py +++ b/homeassistant/components/knx/__init__.py @@ -27,7 +27,7 @@ from .const import ( SUPPORTED_PLATFORMS_UI, SUPPORTED_PLATFORMS_YAML, ) -from .expose import create_knx_exposure +from .expose import create_combined_knx_exposure from .knx_module import KNXModule from .project import STORAGE_KEY as PROJECT_STORAGE_KEY from .schema import ( @@ -121,10 +121,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: hass.data[KNX_MODULE_KEY] = knx_module if CONF_KNX_EXPOSE in config: - for expose_config in config[CONF_KNX_EXPOSE]: - knx_module.exposures.append( - create_knx_exposure(hass, knx_module.xknx, expose_config) - ) + knx_module.yaml_exposures.extend( + create_combined_knx_exposure(hass, knx_module.xknx, config[CONF_KNX_EXPOSE]) + ) + configured_platforms_yaml = { platform for platform in SUPPORTED_PLATFORMS_YAML if platform in config } @@ -149,7 +149,9 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # if not loaded directly return return True - for exposure in knx_module.exposures: + for exposure in knx_module.yaml_exposures: + exposure.async_remove() + for exposure in knx_module.service_exposures.values(): exposure.async_remove() configured_platforms_yaml = { diff --git a/homeassistant/components/knx/expose.py b/homeassistant/components/knx/expose.py index 0a42b6018ba..e8d44385e42 100644 --- a/homeassistant/components/knx/expose.py +++ b/homeassistant/components/knx/expose.py @@ -2,14 +2,22 @@ from __future__ import annotations -from collections.abc import Callable +from asyncio import TaskGroup +from collections.abc import Callable, Iterable +from dataclasses import dataclass import logging +from typing import Any from xknx import XKNX from xknx.devices import DateDevice, DateTimeDevice, ExposeSensor, TimeDevice -from xknx.dpt import DPTNumeric, DPTString +from xknx.dpt import DPTBase, DPTNumeric, DPTString +from xknx.dpt.dpt_1 import DPT1BitEnum, DPTSwitch from xknx.exceptions import ConversionError -from xknx.remote_value import RemoteValueSensor +from xknx.telegram.address import ( + GroupAddress, + InternalGroupAddress, + parse_device_group_address, +) from homeassistant.const import ( CONF_ENTITY_ID, @@ -41,79 +49,159 @@ _LOGGER = logging.getLogger(__name__) @callback def create_knx_exposure( hass: HomeAssistant, xknx: XKNX, config: ConfigType -) -> KNXExposeSensor | KNXExposeTime: - """Create exposures from config.""" - +) -> KnxExposeEntity | KnxExposeTime: + """Create single exposure.""" expose_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE] - - exposure: KNXExposeSensor | KNXExposeTime + exposure: KnxExposeEntity | KnxExposeTime if ( isinstance(expose_type, str) and expose_type.lower() in ExposeSchema.EXPOSE_TIME_TYPES ): - exposure = KNXExposeTime( + exposure = KnxExposeTime( xknx=xknx, config=config, ) else: - exposure = KNXExposeSensor( - hass, + exposure = KnxExposeEntity( + hass=hass, xknx=xknx, - config=config, + entity_id=config[CONF_ENTITY_ID], + options=(_yaml_config_to_expose_options(config),), ) exposure.async_register() return exposure -class KNXExposeSensor: - """Object to Expose Home Assistant entity to KNX bus.""" +@callback +def create_combined_knx_exposure( + hass: HomeAssistant, xknx: XKNX, configs: list[ConfigType] +) -> list[KnxExposeEntity | KnxExposeTime]: + """Create exposures from YAML config combined by entity_id.""" + exposures: list[KnxExposeEntity | KnxExposeTime] = [] + entity_exposure_map: dict[str, list[KnxExposeOptions]] = {} + + for config in configs: + value_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE] + if value_type.lower() in ExposeSchema.EXPOSE_TIME_TYPES: + time_exposure = KnxExposeTime( + xknx=xknx, + config=config, + ) + time_exposure.async_register() + exposures.append(time_exposure) + continue + + entity_id = config[CONF_ENTITY_ID] + option = _yaml_config_to_expose_options(config) + entity_exposure_map.setdefault(entity_id, []).append(option) + + for entity_id, options in entity_exposure_map.items(): + entity_exposure = KnxExposeEntity( + hass=hass, + xknx=xknx, + entity_id=entity_id, + options=options, + ) + entity_exposure.async_register() + exposures.append(entity_exposure) + return exposures + + +@dataclass(slots=True) +class KnxExposeOptions: + """Options for KNX Expose.""" + + attribute: str | None + group_address: GroupAddress | InternalGroupAddress + dpt: type[DPTBase] + respond_to_read: bool + cooldown: float + default: Any | None + value_template: Template | None + + +def _yaml_config_to_expose_options(config: ConfigType) -> KnxExposeOptions: + """Convert single yaml expose config to KnxExposeOptions.""" + value_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE] + dpt: type[DPTBase] + if value_type == "binary": + # HA yaml expose flag for DPT-1 (no explicit DPT 1 definitions in xknx back then) + dpt = DPTSwitch + else: + dpt = DPTBase.parse_transcoder(config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]) # type: ignore[assignment] # checked by schema validation + ga = parse_device_group_address(config[KNX_ADDRESS]) + return KnxExposeOptions( + attribute=config.get(ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE), + group_address=ga, + dpt=dpt, + respond_to_read=config[CONF_RESPOND_TO_READ], + cooldown=config[ExposeSchema.CONF_KNX_EXPOSE_COOLDOWN], + default=config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT), + value_template=config.get(CONF_VALUE_TEMPLATE), + ) + + +class KnxExposeEntity: + """Expose Home Assistant entity values to KNX bus.""" def __init__( self, hass: HomeAssistant, xknx: XKNX, - config: ConfigType, + entity_id: str, + options: Iterable[KnxExposeOptions], ) -> None: - """Initialize of Expose class.""" + """Initialize KnxExposeEntity class.""" self.hass = hass self.xknx = xknx - - self.entity_id: str = config[CONF_ENTITY_ID] - self.expose_attribute: str | None = config.get( - ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE - ) - self.expose_default = config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT) - self.expose_type: int | str = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE] - self.value_template: Template | None = config.get(CONF_VALUE_TEMPLATE) + self.entity_id = entity_id self._remove_listener: Callable[[], None] | None = None - self.device: ExposeSensor = ExposeSensor( - xknx=self.xknx, - name=f"{self.entity_id}__{self.expose_attribute or 'state'}", - group_address=config[KNX_ADDRESS], - respond_to_read=config[CONF_RESPOND_TO_READ], - value_type=self.expose_type, - cooldown=config[ExposeSchema.CONF_KNX_EXPOSE_COOLDOWN], + self._exposures = tuple( + ( + option, + ExposeSensor( + xknx=self.xknx, + name=f"{self.entity_id} {option.attribute or 'state'}", + group_address=option.group_address, + respond_to_read=option.respond_to_read, + value_type=option.dpt, + cooldown=option.cooldown, + ), + ) + for option in options ) + @property + def name(self) -> str: + """Return name of the expose entity.""" + expose_names = [opt.attribute or "state" for opt, _ in self._exposures] + return f"{self.entity_id}__{'__'.join(expose_names)}" + @callback def async_register(self) -> None: - """Register listener.""" + """Register listener and XKNX devices.""" self._remove_listener = async_track_state_change_event( self.hass, [self.entity_id], self._async_entity_changed ) - self.xknx.devices.async_add(self.device) + for _option, xknx_expose in self._exposures: + self.xknx.devices.async_add(xknx_expose) self._init_expose_state() @callback def _init_expose_state(self) -> None: - """Initialize state of the exposure.""" + """Initialize state of all exposures.""" init_state = self.hass.states.get(self.entity_id) - state_value = self._get_expose_value(init_state) - try: - self.device.sensor_value.value = state_value - except ConversionError: - _LOGGER.exception("Error during sending of expose sensor value") + for option, xknx_expose in self._exposures: + state_value = self._get_expose_value(init_state, option) + try: + xknx_expose.sensor_value.value = state_value + except ConversionError: + _LOGGER.exception( + "Error setting value %s for expose sensor %s", + state_value, + xknx_expose.name, + ) @callback def async_remove(self) -> None: @@ -121,53 +209,57 @@ class KNXExposeSensor: if self._remove_listener is not None: self._remove_listener() self._remove_listener = None - self.xknx.devices.async_remove(self.device) + for _option, xknx_expose in self._exposures: + self.xknx.devices.async_remove(xknx_expose) - def _get_expose_value(self, state: State | None) -> bool | int | float | str | None: - """Extract value from state.""" + def _get_expose_value( + self, state: State | None, option: KnxExposeOptions + ) -> bool | int | float | str | None: + """Extract value from state for a specific option.""" if state is None or state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE): - if self.expose_default is None: + if option.default is None: return None - value = self.expose_default - elif self.expose_attribute is not None: - _attr = state.attributes.get(self.expose_attribute) - value = _attr if _attr is not None else self.expose_default + value = option.default + elif option.attribute is not None: + _attr = state.attributes.get(option.attribute) + value = _attr if _attr is not None else option.default else: value = state.state - if self.value_template is not None: + if option.value_template is not None: try: - value = self.value_template.async_render_with_possible_json_value( + value = option.value_template.async_render_with_possible_json_value( value, error_value=None ) except (TemplateError, TypeError, ValueError) as err: _LOGGER.warning( - "Error rendering value template for KNX expose %s %s: %s", - self.device.name, - self.value_template.template, + "Error rendering value template for KNX expose %s %s %s: %s", + self.entity_id, + option.attribute or "state", + option.value_template.template, err, ) return None - if self.expose_type == "binary": + if issubclass(option.dpt, DPT1BitEnum): if value in (1, STATE_ON, "True"): return True if value in (0, STATE_OFF, "False"): return False - if value is not None and ( - isinstance(self.device.sensor_value, RemoteValueSensor) - ): + + # Handle numeric and string DPT conversions + if value is not None: try: - if issubclass(self.device.sensor_value.dpt_class, DPTNumeric): + if issubclass(option.dpt, DPTNumeric): return float(value) - if issubclass(self.device.sensor_value.dpt_class, DPTString): + if issubclass(option.dpt, DPTString): # DPT 16.000 only allows up to 14 Bytes return str(value)[:14] except (ValueError, TypeError) as err: _LOGGER.warning( 'Could not expose %s %s value "%s" to KNX: Conversion failed: %s', self.entity_id, - self.expose_attribute or "state", + option.attribute or "state", value, err, ) @@ -175,32 +267,31 @@ class KNXExposeSensor: return value # type: ignore[no-any-return] async def _async_entity_changed(self, event: Event[EventStateChangedData]) -> None: - """Handle entity change.""" + """Handle entity change for all options.""" new_state = event.data["new_state"] - if (new_value := self._get_expose_value(new_state)) is None: - return - old_state = event.data["old_state"] - # don't use default value for comparison on first state change (old_state is None) - old_value = self._get_expose_value(old_state) if old_state is not None else None - # don't send same value sequentially - if new_value != old_value: - await self._async_set_knx_value(new_value) + async with TaskGroup() as tg: + for option, xknx_expose in self._exposures: + expose_value = self._get_expose_value(new_state, option) + if expose_value is None: + continue + tg.create_task(self._async_set_knx_value(xknx_expose, expose_value)) - async def _async_set_knx_value(self, value: StateType) -> None: + async def _async_set_knx_value( + self, xknx_expose: ExposeSensor, value: StateType + ) -> None: """Set new value on xknx ExposeSensor.""" try: - await self.device.set(value) + await xknx_expose.set(value, skip_unchanged=True) except ConversionError as err: _LOGGER.warning( - 'Could not expose %s %s value "%s" to KNX: %s', - self.entity_id, - self.expose_attribute or "state", + 'Could not expose %s value "%s" to KNX: %s', + xknx_expose.name, value, err, ) -class KNXExposeTime: +class KnxExposeTime: """Object to Expose Time/Date object to KNX bus.""" def __init__(self, xknx: XKNX, config: ConfigType) -> None: @@ -222,6 +313,11 @@ class KNXExposeTime: group_address=config[KNX_ADDRESS], ) + @property + def name(self) -> str: + """Return name of the time expose object.""" + return f"expose_{self.device.name}" + @callback def async_register(self) -> None: """Register listener.""" diff --git a/homeassistant/components/knx/knx_module.py b/homeassistant/components/knx/knx_module.py index 42c14eae2a8..33e08badf51 100644 --- a/homeassistant/components/knx/knx_module.py +++ b/homeassistant/components/knx/knx_module.py @@ -54,7 +54,7 @@ from .const import ( TELEGRAM_LOG_DEFAULT, ) from .device import KNXInterfaceDevice -from .expose import KNXExposeSensor, KNXExposeTime +from .expose import KnxExposeEntity, KnxExposeTime from .project import KNXProject from .repairs import data_secure_group_key_issue_dispatcher from .storage.config_store import KNXConfigStore @@ -73,8 +73,8 @@ class KNXModule: self.hass = hass self.config_yaml = config self.connected = False - self.exposures: list[KNXExposeSensor | KNXExposeTime] = [] - self.service_exposures: dict[str, KNXExposeSensor | KNXExposeTime] = {} + self.yaml_exposures: list[KnxExposeEntity | KnxExposeTime] = [] + self.service_exposures: dict[str, KnxExposeEntity | KnxExposeTime] = {} self.entry = entry self.project = KNXProject(hass=hass, entry=entry) diff --git a/homeassistant/components/knx/services.py b/homeassistant/components/knx/services.py index ebb01e0ef28..0e6798e1584 100644 --- a/homeassistant/components/knx/services.py +++ b/homeassistant/components/knx/services.py @@ -193,7 +193,7 @@ async def service_exposure_register_modify(call: ServiceCall) -> None: " for '%s' - %s" ), group_address, - replaced_exposure.device.name, + replaced_exposure.name, ) replaced_exposure.async_remove() exposure = create_knx_exposure(knx_module.hass, knx_module.xknx, call.data) @@ -201,7 +201,7 @@ async def service_exposure_register_modify(call: ServiceCall) -> None: _LOGGER.debug( "Service exposure_register registered exposure for '%s' - %s", group_address, - exposure.device.name, + exposure.name, )