Support KNX sensor entity configuration from UI

homeassistant/components/knx/const.py

@@ -166,6 +166,7 @@ class FanZeroMode(StrEnum):
     Platform.FAN,
     Platform.DATETIME,
     Platform.LIGHT,
+    Platform.SENSOR,
     Platform.SWITCH,
     Platform.TIME,
 }

homeassistant/components/knx/dpt.py

@@ -0,0 +1,146 @@
+"""KNX DPT serializer."""
+
+from collections.abc import Mapping
+from functools import cache
+from typing import Literal, TypedDict
+
+from xknx.dpt import DPTBase, DPTComplex, DPTEnum, DPTNumeric
+from xknx.dpt.dpt_16 import DPTString
+
+from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
+
+HaDptClass = Literal["numeric", "enum", "complex", "string"]
+
+
+class DPTInfo(TypedDict):
+    """DPT information."""
+
+    dpt_class: HaDptClass
+    main: int
+    sub: int | None
+    name: str | None
+    unit: str | None
+    sensor_device_class: SensorDeviceClass | None
+    sensor_state_class: SensorStateClass | None
+
+
+@cache
+def get_supported_dpts() -> Mapping[str, DPTInfo]:
+    """Return a mapping of supported DPTs with HA specific attributes."""
+    dpts = {}
+    for dpt_class in DPTBase.dpt_class_tree():
+        dpt_number_str = dpt_class.dpt_number_str()
+        ha_dpt_class = _ha_dpt_class(dpt_class)
+        dpts[dpt_number_str] = DPTInfo(
+            dpt_class=ha_dpt_class,
+            main=dpt_class.dpt_main_number,  # type: ignore[typeddict-item] # checked in xknx unit tests
+            sub=dpt_class.dpt_sub_number,
+            name=dpt_class.value_type,
+            unit=dpt_class.unit,
+            sensor_device_class=_sensor_device_classes.get(dpt_number_str),
+            sensor_state_class=_get_sensor_state_class(ha_dpt_class, dpt_number_str),
+        )
+    return dpts
+
+
+def _ha_dpt_class(dpt_cls: type[DPTBase]) -> HaDptClass:
+    """Return the DPT class identifier string."""
+    if issubclass(dpt_cls, DPTNumeric):
+        return "numeric"
+    if issubclass(dpt_cls, DPTEnum):
+        return "enum"
+    if issubclass(dpt_cls, DPTComplex):
+        return "complex"
+    if issubclass(dpt_cls, DPTString):
+        return "string"
+    raise ValueError("Unsupported DPT class")
+
+
+_sensor_device_classes: Mapping[str, SensorDeviceClass] = {
+    "7.011": SensorDeviceClass.DISTANCE,
+    "7.012": SensorDeviceClass.CURRENT,
+    "7.013": SensorDeviceClass.ILLUMINANCE,
+    "8.012": SensorDeviceClass.DISTANCE,
+    "9.001": SensorDeviceClass.TEMPERATURE,
+    "9.002": SensorDeviceClass.TEMPERATURE_DELTA,
+    "9.004": SensorDeviceClass.ILLUMINANCE,
+    "9.005": SensorDeviceClass.WIND_SPEED,
+    "9.006": SensorDeviceClass.PRESSURE,
+    "9.007": SensorDeviceClass.HUMIDITY,
+    "9.020": SensorDeviceClass.VOLTAGE,
+    "9.021": SensorDeviceClass.CURRENT,
+    "9.024": SensorDeviceClass.POWER,
+    "9.025": SensorDeviceClass.VOLUME_FLOW_RATE,
+    "9.027": SensorDeviceClass.TEMPERATURE,
+    "9.028": SensorDeviceClass.WIND_SPEED,
+    "9.029": SensorDeviceClass.ABSOLUTE_HUMIDITY,
+    "12.1200": SensorDeviceClass.VOLUME,
+    "12.1201": SensorDeviceClass.VOLUME,
+    "13.002": SensorDeviceClass.VOLUME_FLOW_RATE,
+    "13.010": SensorDeviceClass.ENERGY,
+    "13.012": SensorDeviceClass.REACTIVE_ENERGY,
+    "13.013": SensorDeviceClass.ENERGY,
+    "13.015": SensorDeviceClass.REACTIVE_ENERGY,
+    "13.016": SensorDeviceClass.ENERGY,
+    "13.1200": SensorDeviceClass.VOLUME,
+    "13.1201": SensorDeviceClass.VOLUME,
+    "14.010": SensorDeviceClass.AREA,
+    "14.019": SensorDeviceClass.CURRENT,
+    "14.027": SensorDeviceClass.VOLTAGE,
+    "14.028": SensorDeviceClass.VOLTAGE,
+    "14.030": SensorDeviceClass.VOLTAGE,
+    "14.031": SensorDeviceClass.ENERGY,
+    "14.033": SensorDeviceClass.FREQUENCY,
+    "14.037": SensorDeviceClass.ENERGY_STORAGE,
+    "14.039": SensorDeviceClass.DISTANCE,
+    "14.051": SensorDeviceClass.WEIGHT,
+    "14.056": SensorDeviceClass.POWER,
+    "14.057": SensorDeviceClass.POWER_FACTOR,
+    "14.058": SensorDeviceClass.PRESSURE,
+    "14.065": SensorDeviceClass.SPEED,
+    "14.068": SensorDeviceClass.TEMPERATURE,
+    "14.069": SensorDeviceClass.TEMPERATURE,
+    "14.070": SensorDeviceClass.TEMPERATURE_DELTA,
+    "14.076": SensorDeviceClass.VOLUME,
+    "14.077": SensorDeviceClass.VOLUME_FLOW_RATE,
+    "14.080": SensorDeviceClass.APPARENT_POWER,
+    "14.1200": SensorDeviceClass.VOLUME_FLOW_RATE,
+    "14.1201": SensorDeviceClass.VOLUME_FLOW_RATE,
+    "29.010": SensorDeviceClass.ENERGY,
+    "29.012": SensorDeviceClass.REACTIVE_ENERGY,
+}
+
+_sensor_state_class_overrides: Mapping[str, SensorStateClass | None] = {
+    "5.003": SensorStateClass.MEASUREMENT_ANGLE,  # DPTAngle
+    "5.006": None,  # DPTTariff
+    "7.010": None,  # DPTPropDataType
+    "8.011": SensorStateClass.MEASUREMENT_ANGLE,  # DPTRotationAngle
+    "9.026": SensorStateClass.TOTAL_INCREASING,  # DPTRainAmount
+    "12.1200": SensorStateClass.TOTAL,  # DPTVolumeLiquidLitre
+    "12.1201": SensorStateClass.TOTAL,  # DPTVolumeM3
+    "13.010": SensorStateClass.TOTAL,  # DPTActiveEnergy
+    "13.011": SensorStateClass.TOTAL,  # DPTApparantEnergy
+    "13.012": SensorStateClass.TOTAL,  # DPTReactiveEnergy
+    "14.007": SensorStateClass.MEASUREMENT_ANGLE,  # DPTAngleDeg
+    "14.037": SensorStateClass.TOTAL,  # DPTHeatQuantity
+    "14.051": SensorStateClass.TOTAL,  # DPTMass
+    "14.055": SensorStateClass.MEASUREMENT_ANGLE,  # DPTPhaseAngleDeg
+    "14.031": SensorStateClass.TOTAL_INCREASING,  # DPTEnergy
+    "17.001": None,  # DPTSceneNumber
+    "29.010": SensorStateClass.TOTAL,  # DPTActiveEnergy8Byte
+    "29.011": SensorStateClass.TOTAL,  # DPTApparantEnergy8Byte
+    "29.012": SensorStateClass.TOTAL,  # DPTReactiveEnergy8Byte
+}
+
+
+def _get_sensor_state_class(
+    ha_dpt_class: HaDptClass, dpt_number_str: str
+) -> SensorStateClass | None:
+    """Return the SensorStateClass for a given DPT."""
+    if ha_dpt_class != "numeric":
+        return None
+
+    return _sensor_state_class_overrides.get(
+        dpt_number_str,
+        SensorStateClass.MEASUREMENT,
+    )

homeassistant/components/knx/sensor.py

@@ -6,8 +6,8 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from functools import partial
+from typing import Any
 
-from xknx import XKNX
 from xknx.core.connection_state import XknxConnectionState, XknxConnectionType
 from xknx.devices import Device as XknxDevice, Sensor as XknxSensor
 
@@ -25,20 +25,32 @@
     CONF_ENTITY_CATEGORY,
     CONF_NAME,
     CONF_TYPE,
+    CONF_UNIT_OF_MEASUREMENT,
     STATE_UNAVAILABLE,
     STATE_UNKNOWN,
     EntityCategory,
     Platform,
 )
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.entity_platform import (
+    AddConfigEntryEntitiesCallback,
+    async_get_current_platform,
+)
 from homeassistant.helpers.typing import ConfigType, StateType
 from homeassistant.util.enum import try_parse_enum
 
-from .const import ATTR_SOURCE, KNX_MODULE_KEY
-from .entity import KnxYamlEntity
+from .const import ATTR_SOURCE, CONF_SYNC_STATE, DOMAIN, KNX_MODULE_KEY
+from .dpt import get_supported_dpts
+from .entity import (
+    KnxUiEntity,
+    KnxUiEntityPlatformController,
+    KnxYamlEntity,
+    _KnxEntityBase,
+)
 from .knx_module import KNXModule
 from .schema import SensorSchema
+from .storage.const import CONF_ALWAYS_CALLBACK, CONF_ENTITY, CONF_GA_SENSOR
+from .storage.util import ConfigExtractor
 
 SCAN_INTERVAL = timedelta(seconds=10)
 
@@ -122,43 +134,82 @@ async def async_setup_entry(
     config_entry: config_entries.ConfigEntry,
     async_add_entities: AddConfigEntryEntitiesCallback,
 ) -> None:
-    """Set up sensor(s) for KNX platform."""
+    """Set up entities for KNX platform."""
     knx_module = hass.data[KNX_MODULE_KEY]
+    platform = async_get_current_platform()
+    knx_module.config_store.add_platform(
+        platform=Platform.SENSOR,
+        controller=KnxUiEntityPlatformController(
+            knx_module=knx_module,
+            entity_platform=platform,
+            entity_class=KnxUiSensor,
+        ),
+    )
+
     entities: list[SensorEntity] = []
     entities.extend(
         KNXSystemSensor(knx_module, description)
         for description in SYSTEM_ENTITY_DESCRIPTIONS
     )
-    config: list[ConfigType] | None = knx_module.config_yaml.get(Platform.SENSOR)
-    if config:
+    if yaml_platform_config := knx_module.config_yaml.get(Platform.SENSOR):
         entities.extend(
-            KNXSensor(knx_module, entity_config) for entity_config in config
+            KnxYamlSensor(knx_module, entity_config)
+            for entity_config in yaml_platform_config
+        )
+    if ui_config := knx_module.config_store.data["entities"].get(Platform.SENSOR):
+        entities.extend(
+            KnxUiSensor(knx_module, unique_id, config)
+            for unique_id, config in ui_config.items()
         )
     async_add_entities(entities)
 
 
-def _create_sensor(xknx: XKNX, config: ConfigType) -> XknxSensor:
-    """Return a KNX sensor to be used within XKNX."""
-    return XknxSensor(
-        xknx,
-        name=config[CONF_NAME],
-        group_address_state=config[SensorSchema.CONF_STATE_ADDRESS],
-        sync_state=config[SensorSchema.CONF_SYNC_STATE],
-        always_callback=True,
-        value_type=config[CONF_TYPE],
-    )
+class _KnxSensor(RestoreSensor, _KnxEntityBase):
+    """Representation of a KNX sensor."""
 
+    _device: XknxSensor
 
-class KNXSensor(KnxYamlEntity, RestoreSensor):
-    """Representation of a KNX sensor."""
+    async def async_added_to_hass(self) -> None:
+        """Restore last state."""
+        if (
+            (last_state := await self.async_get_last_state())
+            and last_state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
+            and (
+                (last_sensor_data := await self.async_get_last_sensor_data())
+                is not None
+            )
+        ):
+            self._attr_native_value = last_sensor_data.native_value
+            self._attr_extra_state_attributes.update(last_state.attributes)
+        await super().async_added_to_hass()
+
+    def after_update_callback(self, device: XknxDevice) -> None:
+        """Call after device was updated."""
+        self._attr_native_value = self._device.resolve_state()
+        if telegram := self._device.last_telegram:
+            self._attr_extra_state_attributes[ATTR_SOURCE] = str(
+                telegram.source_address
+            )
+        super().after_update_callback(device)
+
+
+class KnxYamlSensor(_KnxSensor, KnxYamlEntity):
+    """Representation of a KNX sensor configured from YAML."""
 
     _device: XknxSensor
 
     def __init__(self, knx_module: KNXModule, config: ConfigType) -> None:
         """Initialize of a KNX sensor."""
         super().__init__(
             knx_module=knx_module,
-            device=_create_sensor(knx_module.xknx, config),
+            device=XknxSensor(
+                knx_module.xknx,
+                name=config[CONF_NAME],
+                group_address_state=config[SensorSchema.CONF_STATE_ADDRESS],
+                sync_state=config[CONF_SYNC_STATE],
+                always_callback=True,
+                value_type=config[CONF_TYPE],
+            ),
         )
         if device_class := config.get(CONF_DEVICE_CLASS):
             self._attr_device_class = device_class
@@ -174,28 +225,55 @@ def __init__(self, knx_module: KNXModule, config: ConfigType) -> None:
         self._attr_state_class = config.get(CONF_STATE_CLASS)
         self._attr_extra_state_attributes = {}
 
-    async def async_added_to_hass(self) -> None:
-        """Restore last state."""
-        if (
-            (last_state := await self.async_get_last_state())
-            and last_state.state not in (STATE_UNKNOWN, STATE_UNAVAILABLE)
-            and (
-                (last_sensor_data := await self.async_get_last_sensor_data())
-                is not None
+
+class KnxUiSensor(_KnxSensor, KnxUiEntity):
+    """Representation of a KNX sensor configured from the UI."""
+
+    _device: XknxSensor
+
+    def __init__(
+        self, knx_module: KNXModule, unique_id: str, config: dict[str, Any]
+    ) -> None:
+        """Initialize KNX sensor."""
+        super().__init__(
+            knx_module=knx_module,
+            unique_id=unique_id,
+            entity_config=config[CONF_ENTITY],
+        )
+        knx_conf = ConfigExtractor(config[DOMAIN])
+        dpt_string = knx_conf.get_dpt(CONF_GA_SENSOR)
+        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],
+            group_address_state=knx_conf.get_state_and_passive(CONF_GA_SENSOR),
+            sync_state=knx_conf.get(CONF_SYNC_STATE),
+            always_callback=True,
+            value_type=dpt_string,
+        )
+
+        if device_class_override := knx_conf.get(CONF_DEVICE_CLASS):
+            self._attr_device_class = try_parse_enum(
+                SensorDeviceClass, device_class_override
             )
-        ):
-            self._attr_native_value = last_sensor_data.native_value
-            self._attr_extra_state_attributes.update(last_state.attributes)
-        await super().async_added_to_hass()
+        else:
+            self._attr_device_class = dpt_info["sensor_device_class"]
 
-    def after_update_callback(self, device: XknxDevice) -> None:
-        """Call after device was updated."""
-        self._attr_native_value = self._device.resolve_state()
-        if telegram := self._device.last_telegram:
-            self._attr_extra_state_attributes[ATTR_SOURCE] = str(
-                telegram.source_address
+        if state_class_override := knx_conf.get(CONF_STATE_CLASS):
+            self._attr_state_class = try_parse_enum(
+                SensorStateClass, state_class_override
             )
-        super().after_update_callback(device)
+        else:
+            self._attr_state_class = dpt_info["sensor_state_class"]
+
+        self._attr_native_unit_of_measurement = (
+            knx_conf.get(CONF_UNIT_OF_MEASUREMENT) or dpt_info["unit"]
+        )
+
+        self._attr_force_update = knx_conf.get(CONF_ALWAYS_CALLBACK, default=False)
+        self._attr_extra_state_attributes = {}
 
 
 class KNXSystemSensor(SensorEntity):

homeassistant/components/knx/storage/const.py

@@ -71,3 +71,6 @@
 CONF_GA_WHITE_SWITCH: Final = "ga_white_switch"
 CONF_GA_HUE: Final = "ga_hue"
 CONF_GA_SATURATION: Final = "ga_saturation"
+
+# Sensor
+CONF_ALWAYS_CALLBACK: Final = "always_callback"