expose.py

Go to the documentation of this file.

"""Exposures to KNX bus."""
 
from __future__ import annotations
 
from collections.abc import Callable
import logging
 
from xknx import XKNX
from xknx.devices import DateDevice, DateTimeDevice, ExposeSensor, TimeDevice
from xknx.dpt import DPTNumeric, DPTString
from xknx.exceptions import ConversionError
from xknx.remote_value import RemoteValueSensor
 
from homeassistant.const import (
    CONF_ENTITY_ID,
    CONF_VALUE_TEMPLATE,
    STATE_OFF,
    STATE_ON,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
)
from homeassistant.core import (
    Event,
    EventStateChangedData,
    HomeAssistant,
    State,
    callback,
)
from homeassistant.exceptions import TemplateError
from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.helpers.template import Template
from homeassistant.helpers.typing import ConfigType, StateType
 
from .const import CONF_RESPOND_TO_READ, KNX_ADDRESS
from .schema import ExposeSchema
 
_LOGGER = logging.getLogger(__name__)
 
 
@callback
def create_knx_exposure(
    hass: HomeAssistant, xknx: XKNX, config: ConfigType
) -> KNXExposeSensor | KNXExposeTime:
    """Create exposures from config."""
 
    expose_type = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
 
    exposure: KNXExposeSensor | KNXExposeTime
    if (
        isinstance(expose_type, str)
        and expose_type.lower() in ExposeSchema.EXPOSE_TIME_TYPES
    ):
        exposure = KNXExposeTime(
            xknx=xknx,
            config=config,
        )
    else:
        exposure = KNXExposeSensor(
            hass,
            xknx=xknx,
            config=config,
        )
    exposure.async_register()
    return exposure
 
 
class KNXExposeSensor:
    """Object to Expose Home Assistant entity to KNX bus."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        xknx: XKNX,
        config: ConfigType,
    ) -> None:
        """Initialize of Expose class."""
        self.hasshass = hass
        self.xknxxknx = xknx
 
        self.entity_id: str = config[CONF_ENTITY_ID]
        self.expose_attribute: str | None = config.get(
            ExposeSchema.CONF_KNX_EXPOSE_ATTRIBUTE
        )
        self.expose_defaultexpose_default = config.get(ExposeSchema.CONF_KNX_EXPOSE_DEFAULT)
        self.expose_typeexpose_type: int | str = config[ExposeSchema.CONF_KNX_EXPOSE_TYPE]
        self.value_template: Template | None = config.get(CONF_VALUE_TEMPLATE)
 
        self._remove_listener_remove_listener: Callable[[], None] | None = None
        self.device: ExposeSensor = ExposeSensor(
            xknx=self.xknxxknx,
            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_typeexpose_type,
            cooldown=config[ExposeSchema.CONF_KNX_EXPOSE_COOLDOWN],
        )
 
    @callback
    def async_register(self) -> None:
        """Register listener."""
        self._remove_listener_remove_listener = async_track_state_change_event(
            self.hasshass, [self.entity_id], self._async_entity_changed_async_entity_changed
        )
        self.xknxxknx.devices.async_add(self.device)
        self._init_expose_state_init_expose_state()
 
    @callback
    def _init_expose_state(self) -> None:
        """Initialize state of the exposure."""
        init_state = self.hasshass.states.get(self.entity_id)
        state_value = self._get_expose_value_get_expose_value(init_state)
        try:
            self.device.sensor_value.value = state_value
        except ConversionError:
            _LOGGER.exception("Error during sending of expose sensor value")
 
    @callback
    def async_remove(self) -> None:
        """Prepare for deletion."""
        if self._remove_listener_remove_listener is not None:
            self._remove_listener_remove_listener()
            self._remove_listener_remove_listener = None
        self.xknxxknx.devices.async_remove(self.device)
 
    def _get_expose_value(self, state: State | None) -> bool | int | float | str | None:
        """Extract value from state."""
        if state is None or state.state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
            if self.expose_defaultexpose_default is None:
                return None
            value = self.expose_defaultexpose_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_defaultexpose_default
        else:
            value = state.state
 
        if self.value_template is not None:
            try:
                value = self.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,
                    err,
                )
                return None
 
        if self.expose_typeexpose_type == "binary":
            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)
        ):
            try:
                if issubclass(self.device.sensor_value.dpt_class, DPTNumeric):
                    return float(value)
                if issubclass(self.device.sensor_value.dpt_class, 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",
                    value,
                    err,
                )
                return None
        return value  # type: ignore[no-any-return]
 
    async def _async_entity_changed(self, event: Event[EventStateChangedData]) -> None:
        """Handle entity change."""
        new_state = event.data["new_state"]
        if (new_value := self._get_expose_value_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_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_async_set_knx_value(new_value)
 
    async def _async_set_knx_value(self, value: StateType) -> None:
        """Set new value on xknx ExposeSensor."""
        try:
            await self.device.set(value)
        except ConversionError as err:
            _LOGGER.warning(
                'Could not expose %s %s value "%s" to KNX: %s',
                self.entity_id,
                self.expose_attribute or "state",
                value,
                err,
            )
 
 
class KNXExposeTime:
    """Object to Expose Time/Date object to KNX bus."""
 
    def __init__(self, xknx: XKNX, config: ConfigType) -> None:
        """Initialize of Expose class."""
        self.xknxxknx = 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.devicedevice = xknx_device_cls(
            self.xknxxknx,
            name=expose_type.capitalize(),
            localtime=True,
            group_address=config[KNX_ADDRESS],
        )
 
    @callback
    def async_register(self) -> None:
        """Register listener."""
        self.xknxxknx.devices.async_add(self.devicedevice)
 
    @callback
    def async_remove(self) -> None:
        """Prepare for deletion."""
        self.xknxxknx.devices.async_remove(self.devicedevice)