KNX Sensor: set device and state class for YAML entities based on DPT (#159465)

homeassistant/components/knx/schema.py

@@ -22,7 +22,7 @@ from homeassistant.components.cover import (
 )
 from homeassistant.components.number import NumberMode
 from homeassistant.components.sensor import (
-    CONF_STATE_CLASS,
+    CONF_STATE_CLASS as CONF_SENSOR_STATE_CLASS,
     DEVICE_CLASSES_SCHEMA as SENSOR_DEVICE_CLASSES_SCHEMA,
     STATE_CLASSES_SCHEMA,
 )
@@ -64,6 +64,7 @@ from .const import (
     NumberConf,
     SceneConf,
 )
+from .dpt import get_supported_dpts
 from .validation import (
     backwards_compatible_xknx_climate_enum_member,
     dpt_base_type_validator,
@@ -74,6 +75,7 @@ from .validation import (
     string_type_validator,
     sync_state_validator,
     validate_number_attributes,
+    validate_sensor_attributes,
 )
 
 
@@ -143,6 +145,13 @@ def select_options_sub_validator(entity_config: OrderedDict) -> OrderedDict:
     return entity_config
 
 
+def _sensor_attribute_sub_validator(config: dict) -> dict:
+    """Validate that state_class is compatible with device_class and unit_of_measurement."""
+    transcoder: type[DPTBase] = DPTBase.parse_transcoder(config[CONF_TYPE])  # type: ignore[assignment]  # already checked in sensor_type_validator
+    dpt_metadata = get_supported_dpts()[transcoder.dpt_number_str()]
+    return validate_sensor_attributes(dpt_metadata, config)
+
+
 #########
 # EVENT
 #########
@@ -848,17 +857,20 @@ class SensorSchema(KNXPlatformSchema):
     CONF_SYNC_STATE = CONF_SYNC_STATE
     DEFAULT_NAME = "KNX Sensor"
 
-    ENTITY_SCHEMA = vol.Schema(
+    ENTITY_SCHEMA = vol.All(
-        {
+        vol.Schema(
-            vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
+            {
-            vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
+                vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
-            vol.Optional(CONF_ALWAYS_CALLBACK, default=False): cv.boolean,
+                vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
-            vol.Optional(CONF_STATE_CLASS): STATE_CLASSES_SCHEMA,
+                vol.Optional(CONF_ALWAYS_CALLBACK, default=False): cv.boolean,
-            vol.Required(CONF_TYPE): sensor_type_validator,
+                vol.Optional(CONF_SENSOR_STATE_CLASS): STATE_CLASSES_SCHEMA,
-            vol.Required(CONF_STATE_ADDRESS): ga_list_validator,
+                vol.Required(CONF_TYPE): sensor_type_validator,
-            vol.Optional(CONF_DEVICE_CLASS): SENSOR_DEVICE_CLASSES_SCHEMA,
+                vol.Required(CONF_STATE_ADDRESS): ga_list_validator,
-            vol.Optional(CONF_ENTITY_CATEGORY): ENTITY_CATEGORIES_SCHEMA,
+                vol.Optional(CONF_DEVICE_CLASS): SENSOR_DEVICE_CLASSES_SCHEMA,
-        }
+                vol.Optional(CONF_ENTITY_CATEGORY): ENTITY_CATEGORIES_SCHEMA,
+            }
+        ),
+        _sensor_attribute_sub_validator,
     )
 
 

homeassistant/components/knx/sensor.py

@@ -213,18 +213,22 @@ class KnxYamlSensor(_KnxSensor, KnxYamlEntity):
                 value_type=config[CONF_TYPE],
             ),
         )
+        dpt_string = self._device.sensor_value.dpt_class.dpt_number_str()
+        dpt_info = get_supported_dpts()[dpt_string]
+
         if device_class := config.get(CONF_DEVICE_CLASS):
             self._attr_device_class = device_class
         else:
-            self._attr_device_class = try_parse_enum(
+            self._attr_device_class = dpt_info["sensor_device_class"]
-                SensorDeviceClass, self._device.ha_device_class()
-            )
 
+        self._attr_state_class = (
+            config.get(CONF_STATE_CLASS) or dpt_info["sensor_state_class"]
+        )
+
+        self._attr_native_unit_of_measurement = dpt_info["unit"]
         self._attr_force_update = config[SensorSchema.CONF_ALWAYS_CALLBACK]
         self._attr_entity_category = config.get(CONF_ENTITY_CATEGORY)
         self._attr_unique_id = str(self._device.sensor_value.group_address_state)
-        self._attr_native_unit_of_measurement = self._device.unit_of_measurement()
-        self._attr_state_class = config.get(CONF_STATE_CLASS)
         self._attr_extra_state_attributes = {}
 
 

homeassistant/components/knx/storage/entity_store_schema.py

@@ -13,9 +13,7 @@ from homeassistant.components.number import (
 )
 from homeassistant.components.sensor import (
     CONF_STATE_CLASS as CONF_SENSOR_STATE_CLASS,
-    DEVICE_CLASS_STATE_CLASSES,
     DEVICE_CLASS_UNITS as SENSOR_DEVICE_CLASS_UNITS,
-    STATE_CLASS_UNITS,
     SensorDeviceClass,
     SensorStateClass,
 )
@@ -52,7 +50,7 @@ from ..const import (
     SceneConf,
 )
 from ..dpt import get_supported_dpts
-from ..validation import validate_number_attributes
+from ..validation import validate_number_attributes, validate_sensor_attributes
 from .const import (
     CONF_ALWAYS_CALLBACK,
     CONF_COLOR,
@@ -684,62 +682,11 @@ CLIMATE_KNX_SCHEMA = vol.Schema(
 )
 
 
-def _validate_sensor_attributes(config: dict) -> dict:
+def _sensor_attribute_sub_validator(config: dict) -> dict:
     """Validate that state_class is compatible with device_class and unit_of_measurement."""
     dpt = config[CONF_GA_SENSOR][CONF_DPT]
     dpt_metadata = get_supported_dpts()[dpt]
-    state_class = config.get(
+    return validate_sensor_attributes(dpt_metadata, config)
-        CONF_SENSOR_STATE_CLASS,
-        dpt_metadata["sensor_state_class"],
-    )
-    device_class = config.get(
-        CONF_DEVICE_CLASS,
-        dpt_metadata["sensor_device_class"],
-    )
-    unit_of_measurement = config.get(
-        CONF_UNIT_OF_MEASUREMENT,
-        dpt_metadata["unit"],
-    )
-    if (
-        state_class
-        and device_class
-        and (state_classes := DEVICE_CLASS_STATE_CLASSES.get(device_class)) is not None
-        and state_class not in state_classes
-    ):
-        raise vol.Invalid(
-            f"State class '{state_class}' is not valid for device class '{device_class}'. "
-            f"Valid options are: {', '.join(sorted(map(str, state_classes), key=str.casefold))}",
-            path=[CONF_SENSOR_STATE_CLASS],
-        )
-    if (
-        device_class
-        and (d_c_units := SENSOR_DEVICE_CLASS_UNITS.get(device_class)) is not None
-        and unit_of_measurement not in d_c_units
-    ):
-        raise vol.Invalid(
-            f"Unit of measurement '{unit_of_measurement}' is not valid for device class '{device_class}'. "
-            f"Valid options are: {', '.join(sorted(map(str, d_c_units), key=str.casefold))}",
-            path=(
-                [CONF_DEVICE_CLASS]
-                if CONF_DEVICE_CLASS in config
-                else [CONF_UNIT_OF_MEASUREMENT]
-            ),
-        )
-    if (
-        state_class
-        and (s_c_units := STATE_CLASS_UNITS.get(state_class)) is not None
-        and unit_of_measurement not in s_c_units
-    ):
-        raise vol.Invalid(
-            f"Unit of measurement '{unit_of_measurement}' is not valid for state class '{state_class}'. "
-            f"Valid options are: {', '.join(sorted(map(str, s_c_units), key=str.casefold))}",
-            path=(
-                [CONF_SENSOR_STATE_CLASS]
-                if CONF_SENSOR_STATE_CLASS in config
-                else [CONF_UNIT_OF_MEASUREMENT]
-            ),
-        )
-    return config
 
 
 SENSOR_KNX_SCHEMA = AllSerializeFirst(
@@ -788,7 +735,7 @@ SENSOR_KNX_SCHEMA = AllSerializeFirst(
             ),
         },
     ),
-    _validate_sensor_attributes,
+    _sensor_attribute_sub_validator,
 )
 
 KNX_SCHEMA_FOR_PLATFORM = {

homeassistant/components/knx/validation.py

@@ -14,11 +14,17 @@ from xknx.telegram.address import IndividualAddress, parse_device_group_address
 from homeassistant.components.number import (
     DEVICE_CLASS_UNITS as NUMBER_DEVICE_CLASS_UNITS,
 )
+from homeassistant.components.sensor import (
+    CONF_STATE_CLASS as CONF_SENSOR_STATE_CLASS,
+    DEVICE_CLASS_STATE_CLASSES,
+    DEVICE_CLASS_UNITS,
+    STATE_CLASS_UNITS,
+)
 from homeassistant.const import CONF_DEVICE_CLASS, CONF_UNIT_OF_MEASUREMENT
 from homeassistant.helpers import config_validation as cv
 
 from .const import NumberConf
-from .dpt import get_supported_dpts
+from .dpt import DPTInfo, get_supported_dpts
 
 
 def dpt_subclass_validator(dpt_base_class: type[DPTBase]) -> Callable[[Any], str | int]:
@@ -219,3 +225,65 @@ def validate_number_attributes(
         )
 
     return config
+
+
+def validate_sensor_attributes(
+    dpt_info: DPTInfo, config: dict[str, Any]
+) -> dict[str, Any]:
+    """Validate that state_class is compatible with device_class and unit_of_measurement.
+
+    Works for both, UI and YAML configuration schema since they
+    share same names for all tested attributes.
+    """
+    state_class = config.get(
+        CONF_SENSOR_STATE_CLASS,
+        dpt_info["sensor_state_class"],
+    )
+    device_class = config.get(
+        CONF_DEVICE_CLASS,
+        dpt_info["sensor_device_class"],
+    )
+    unit_of_measurement = config.get(
+        CONF_UNIT_OF_MEASUREMENT,
+        dpt_info["unit"],
+    )
+    if (
+        state_class
+        and device_class
+        and (state_classes := DEVICE_CLASS_STATE_CLASSES.get(device_class)) is not None
+        and state_class not in state_classes
+    ):
+        raise vol.Invalid(
+            f"State class '{state_class}' is not valid for device class '{device_class}'. "
+            f"Valid options are: {', '.join(sorted(map(str, state_classes), key=str.casefold))}",
+            path=[CONF_SENSOR_STATE_CLASS],
+        )
+    if (
+        device_class
+        and (d_c_units := DEVICE_CLASS_UNITS.get(device_class)) is not None
+        and unit_of_measurement not in d_c_units
+    ):
+        raise vol.Invalid(
+            f"Unit of measurement '{unit_of_measurement}' is not valid for device class '{device_class}'. "
+            f"Valid options are: {', '.join(sorted(map(str, d_c_units), key=str.casefold))}",
+            path=(
+                [CONF_DEVICE_CLASS]
+                if CONF_DEVICE_CLASS in config
+                else [CONF_UNIT_OF_MEASUREMENT]
+            ),
+        )
+    if (
+        state_class
+        and (s_c_units := STATE_CLASS_UNITS.get(state_class)) is not None
+        and unit_of_measurement not in s_c_units
+    ):
+        raise vol.Invalid(
+            f"Unit of measurement '{unit_of_measurement}' is not valid for state class '{state_class}'. "
+            f"Valid options are: {', '.join(sorted(map(str, s_c_units), key=str.casefold))}",
+            path=(
+                [CONF_SENSOR_STATE_CLASS]
+                if CONF_SENSOR_STATE_CLASS in config
+                else [CONF_UNIT_OF_MEASUREMENT]
+            ),
+        )
+    return config

tests/components/knx/test_sensor.py

@@ -1,5 +1,6 @@
 """Test KNX sensor."""
 
+import logging
 from typing import Any
 
 from freezegun.api import FrozenDateTimeFactory
@@ -11,6 +12,11 @@ from homeassistant.components.knx.const import (
     CONF_SYNC_STATE,
 )
 from homeassistant.components.knx.schema import SensorSchema
+from homeassistant.components.sensor import (
+    CONF_STATE_CLASS as CONF_SENSOR_STATE_CLASS,
+    SensorDeviceClass,
+    SensorStateClass,
+)
 from homeassistant.const import CONF_NAME, CONF_TYPE, STATE_UNKNOWN, Platform
 from homeassistant.core import HomeAssistant, State
 
@@ -42,13 +48,18 @@ async def test_sensor(hass: HomeAssistant, knx: KNXTestKit) -> None:
     # StateUpdater initialize state
     await knx.assert_read("1/1/1")
     await knx.receive_response("1/1/1", (0, 40))
-    state = hass.states.get("sensor.test")
+    knx.assert_state(
-    assert state.state == "40"
+        "sensor.test",
+        "40",
+        # default values for DPT type "current"
+        device_class=SensorDeviceClass.CURRENT,
+        state_class=SensorStateClass.MEASUREMENT,
+        unit_of_measurement="mA",
+    )
 
     # update from KNX
     await knx.receive_write("1/1/1", (0x03, 0xE8))
-    state = hass.states.get("sensor.test")
+    knx.assert_state("sensor.test", "1000")
-    assert state.state == "1000"
 
     # don't answer to GroupValueRead requests
     await knx.receive_read("1/1/1")
@@ -172,6 +183,38 @@ async def test_always_callback(hass: HomeAssistant, knx: KNXTestKit) -> None:
     assert len(events) == 6
 
 
+async def test_sensor_yaml_attribute_validation(
+    hass: HomeAssistant,
+    caplog: pytest.LogCaptureFixture,
+    knx: KNXTestKit,
+) -> None:
+    """Test creating a sensor with invalid unit, state_class or device_class."""
+    with caplog.at_level(logging.ERROR):
+        await knx.setup_integration(
+            {
+                SensorSchema.PLATFORM: {
+                    CONF_NAME: "test",
+                    CONF_STATE_ADDRESS: "1/1/1",
+                    CONF_TYPE: "9.001",  # temperature 2 byte float
+                    CONF_SENSOR_STATE_CLASS: "total_increasing",  # invalid for temperature
+                }
+            }
+        )
+    assert len(caplog.messages) == 2
+    record = caplog.records[0]
+    assert record.levelname == "ERROR"
+    assert (
+        "Invalid config for 'knx': State class 'total_increasing' is not valid for device class"
+        in record.message
+    )
+
+    record = caplog.records[1]
+    assert record.levelname == "ERROR"
+    assert "Setup failed for 'knx': Invalid config." in record.message
+
+    assert hass.states.get("sensor.test") is None
+
+
 @pytest.mark.parametrize(
     ("knx_config", "response_payload", "expected_state"),
     [
@@ -186,8 +229,8 @@ async def test_always_callback(hass: HomeAssistant, knx: KNXTestKit) -> None:
             (0, 0),
             {
                 "state": "0.0",
-                "device_class": "temperature",
+                "device_class": SensorDeviceClass.TEMPERATURE,
-                "state_class": "measurement",
+                "state_class": SensorStateClass.MEASUREMENT,
                 "unit_of_measurement": "°C",
             },
         ),
@@ -206,8 +249,8 @@ async def test_always_callback(hass: HomeAssistant, knx: KNXTestKit) -> None:
             (1, 2, 3, 4),
             {
                 "state": "16909060",
-                "device_class": "energy",
+                "device_class": SensorDeviceClass.ENERGY,
-                "state_class": "total_increasing",
+                "state_class": SensorStateClass.TOTAL_INCREASING,
             },
         ),
     ],