Compare commits

...

6 Commits

Author SHA1 Message Date
farmio
1b7c9afb2c validate device_class, state_class and unit 2025-12-10 23:08:09 +01:00
farmio
6974a70607 sort in BE for test consistency 2025-12-10 21:11:57 +01:00
farmio
9334d4b108 unit_o_m, device_class translations 2025-12-10 21:02:24 +01:00
Matthias Alphart
57ae4c8656 Update strings.json
Co-authored-by: Norbert Rittel <norbert@rittel.de>
2025-12-10 15:55:38 +01:00
farmio
76b9a99bdd add tests 2025-12-10 11:19:09 +01:00
farmio
47e9c5785f Support KNX sensor entity configuration from UI 2025-12-09 22:57:20 +01:00
11 changed files with 1005 additions and 52 deletions

View File

@@ -165,6 +165,7 @@ SUPPORTED_PLATFORMS_UI: Final = {
Platform.DATE,
Platform.DATETIME,
Platform.LIGHT,
Platform.SENSOR,
Platform.SWITCH,
Platform.TIME,
}

View File

@@ -0,0 +1,142 @@
"""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,
"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,
"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,
)

View File

@@ -6,8 +6,8 @@ from collections.abc import Callable
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 @@ from homeassistant.const import (
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)
@@ -116,58 +128,41 @@ 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(KnxYamlEntity, RestoreSensor):
class _KnxSensor(RestoreSensor, _KnxEntityBase):
"""Representation of a KNX sensor."""
_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),
)
if device_class := config.get(CONF_DEVICE_CLASS):
self._attr_device_class = device_class
else:
self._attr_device_class = try_parse_enum(
SensorDeviceClass, self._device.ha_device_class()
)
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 = {}
async def async_added_to_hass(self) -> None:
"""Restore last state."""
if (
@@ -192,6 +187,89 @@ class KNXSensor(KnxYamlEntity, RestoreSensor):
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=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
else:
self._attr_device_class = try_parse_enum(
SensorDeviceClass, self._device.ha_device_class()
)
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 = {}
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
)
else:
self._attr_device_class = dpt_info["sensor_device_class"]
if state_class_override := knx_conf.get(CONF_STATE_CLASS):
self._attr_state_class = try_parse_enum(
SensorStateClass, state_class_override
)
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):
"""Representation of a KNX system sensor."""

View File

@@ -65,3 +65,6 @@ CONF_GA_WHITE_BRIGHTNESS: Final = "ga_white_brightness"
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"

View File

@@ -5,11 +5,21 @@ from enum import StrEnum, unique
import voluptuous as vol
from homeassistant.components.climate import HVACMode
from homeassistant.components.sensor import (
CONF_STATE_CLASS as CONF_SENSOR_STATE_CLASS,
DEVICE_CLASS_STATE_CLASSES,
DEVICE_CLASS_UNITS,
STATE_CLASS_UNITS,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_ENTITY_CATEGORY,
CONF_ENTITY_ID,
CONF_NAME,
CONF_PLATFORM,
CONF_UNIT_OF_MEASUREMENT,
Platform,
)
from homeassistant.helpers import config_validation as cv, selector
@@ -30,12 +40,15 @@ from ..const import (
CoverConf,
FanZeroMode,
)
from ..dpt import get_supported_dpts
from .const import (
CONF_ALWAYS_CALLBACK,
CONF_COLOR,
CONF_COLOR_TEMP_MAX,
CONF_COLOR_TEMP_MIN,
CONF_DATA,
CONF_DEVICE_INFO,
CONF_DPT,
CONF_ENTITY,
CONF_GA_ACTIVE,
CONF_GA_ANGLE,
@@ -507,6 +520,114 @@ CLIMATE_KNX_SCHEMA = vol.Schema(
},
)
def _validate_sensor_attributes(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(
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 := 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(
vol.Schema(
{
vol.Required(CONF_GA_SENSOR): GASelector(
write=False, state_required=True, dpt=["numeric", "string"]
),
"section_advanced_options": KNXSectionFlat(collapsible=True),
vol.Optional(CONF_UNIT_OF_MEASUREMENT): selector.SelectSelector(
selector.SelectSelectorConfig(
options=sorted(
{
str(unit)
for units in DEVICE_CLASS_UNITS.values()
for unit in units
if unit is not None
}
),
mode=selector.SelectSelectorMode.DROPDOWN,
translation_key="component.knx.selector.sensor_unit_of_measurement",
custom_value=True,
),
),
vol.Optional(CONF_DEVICE_CLASS): selector.SelectSelector(
selector.SelectSelectorConfig(
options=[
cls.value
for cls in SensorDeviceClass
if cls != SensorDeviceClass.ENUM
],
translation_key="component.knx.selector.sensor_device_class",
sort=True,
)
),
vol.Optional(CONF_SENSOR_STATE_CLASS): selector.SelectSelector(
selector.SelectSelectorConfig(
options=list(SensorStateClass),
translation_key="component.knx.selector.sensor_state_class",
mode=selector.SelectSelectorMode.DROPDOWN,
)
),
vol.Optional(CONF_ALWAYS_CALLBACK): selector.BooleanSelector(),
vol.Required(CONF_SYNC_STATE, default=True): SyncStateSelector(
allow_false=True
),
},
),
_validate_sensor_attributes,
)
KNX_SCHEMA_FOR_PLATFORM = {
Platform.BINARY_SENSOR: BINARY_SENSOR_KNX_SCHEMA,
Platform.CLIMATE: CLIMATE_KNX_SCHEMA,
@@ -514,6 +635,7 @@ KNX_SCHEMA_FOR_PLATFORM = {
Platform.DATE: DATE_KNX_SCHEMA,
Platform.DATETIME: DATETIME_KNX_SCHEMA,
Platform.LIGHT: LIGHT_KNX_SCHEMA,
Platform.SENSOR: SENSOR_KNX_SCHEMA,
Platform.SWITCH: SWITCH_KNX_SCHEMA,
Platform.TIME: TIME_KNX_SCHEMA,
}

View File

@@ -6,6 +6,7 @@ from typing import Any
import voluptuous as vol
from ..dpt import HaDptClass, get_supported_dpts
from ..validation import ga_validator, maybe_ga_validator, sync_state_validator
from .const import CONF_DPT, CONF_GA_PASSIVE, CONF_GA_STATE, CONF_GA_WRITE
from .util import dpt_string_to_dict
@@ -162,7 +163,7 @@ class GASelector(KNXSelectorBase):
passive: bool = True,
write_required: bool = False,
state_required: bool = False,
dpt: type[Enum] | None = None,
dpt: type[Enum] | list[HaDptClass] | None = None,
valid_dpt: str | Iterable[str] | None = None,
) -> None:
"""Initialize the group address selector."""
@@ -186,14 +187,17 @@ class GASelector(KNXSelectorBase):
"passive": self.passive,
}
if self.dpt is not None:
options["dptSelect"] = [
{
"value": item.value,
"translation_key": item.value.replace(".", "_"),
"dpt": dpt_string_to_dict(item.value), # used for filtering GAs
}
for item in self.dpt
]
if isinstance(self.dpt, list):
options["dptClasses"] = self.dpt
else:
options["dptSelect"] = [
{
"value": item.value,
"translation_key": item.value.replace(".", "_"),
"dpt": dpt_string_to_dict(item.value), # used for filtering GAs
}
for item in self.dpt
]
if self.valid_dpt is not None:
options["validDPTs"] = [dpt_string_to_dict(dpt) for dpt in self.valid_dpt]
@@ -254,7 +258,12 @@ class GASelector(KNXSelectorBase):
def _add_dpt(self, schema: dict[vol.Marker, Any]) -> None:
"""Add DPT validator to the schema."""
if self.dpt is not None:
schema[vol.Required(CONF_DPT)] = vol.In({item.value for item in self.dpt})
if isinstance(self.dpt, list):
schema[vol.Required(CONF_DPT)] = vol.In(get_supported_dpts())
else:
schema[vol.Required(CONF_DPT)] = vol.In(
{item.value for item in self.dpt}
)
else:
schema[vol.Remove(CONF_DPT)] = object

View File

@@ -558,6 +558,35 @@
}
}
},
"sensor": {
"description": "Read-only entity for numeric or string datapoints. Temperature, percent etc.",
"knx": {
"always_callback": {
"description": "Write each update to the state machine, even if the data is the same.",
"label": "Force update"
},
"device_class": {
"description": "Override the DPTs default device class.",
"label": "Device class"
},
"ga_sensor": {
"description": "Group address representing state.",
"label": "State"
},
"section_advanced_options": {
"description": "Override default DPT-based sensor attributes.",
"title": "Overrides"
},
"state_class": {
"description": "Override the DPTs default state class.",
"label": "[%key:component::sensor::entity_component::_::state_attributes::state_class::name%]"
},
"unit_of_measurement": {
"description": "Override the DPTs default unit of measurement.",
"label": "Unit of measurement"
}
}
},
"switch": {
"description": "The KNX switch platform is used as an interface to switching actuators.",
"knx": {
@@ -688,6 +717,79 @@
}
}
},
"selector": {
"sensor_device_class": {
"options": {
"absolute_humidity": "[%key:component::sensor::entity_component::absolute_humidity::name%]",
"apparent_power": "[%key:component::sensor::entity_component::apparent_power::name%]",
"aqi": "[%key:component::sensor::entity_component::aqi::name%]",
"area": "[%key:component::sensor::entity_component::area::name%]",
"atmospheric_pressure": "[%key:component::sensor::entity_component::atmospheric_pressure::name%]",
"battery": "[%key:component::sensor::entity_component::battery::name%]",
"blood_glucose_concentration": "[%key:component::sensor::entity_component::blood_glucose_concentration::name%]",
"carbon_dioxide": "[%key:component::sensor::entity_component::carbon_dioxide::name%]",
"carbon_monoxide": "[%key:component::sensor::entity_component::carbon_monoxide::name%]",
"conductivity": "[%key:component::sensor::entity_component::conductivity::name%]",
"current": "[%key:component::sensor::entity_component::current::name%]",
"data_rate": "[%key:component::sensor::entity_component::data_rate::name%]",
"data_size": "[%key:component::sensor::entity_component::data_size::name%]",
"date": "[%key:component::sensor::entity_component::date::name%]",
"distance": "[%key:component::sensor::entity_component::distance::name%]",
"duration": "[%key:component::sensor::entity_component::duration::name%]",
"energy": "[%key:component::sensor::entity_component::energy::name%]",
"energy_distance": "[%key:component::sensor::entity_component::energy_distance::name%]",
"energy_storage": "[%key:component::sensor::entity_component::energy_storage::name%]",
"frequency": "[%key:component::sensor::entity_component::frequency::name%]",
"gas": "[%key:component::sensor::entity_component::gas::name%]",
"humidity": "[%key:component::sensor::entity_component::humidity::name%]",
"illuminance": "[%key:component::sensor::entity_component::illuminance::name%]",
"irradiance": "[%key:component::sensor::entity_component::irradiance::name%]",
"moisture": "[%key:component::sensor::entity_component::moisture::name%]",
"monetary": "[%key:component::sensor::entity_component::monetary::name%]",
"nitrogen_dioxide": "[%key:component::sensor::entity_component::nitrogen_dioxide::name%]",
"nitrogen_monoxide": "[%key:component::sensor::entity_component::nitrogen_monoxide::name%]",
"nitrous_oxide": "[%key:component::sensor::entity_component::nitrous_oxide::name%]",
"ozone": "[%key:component::sensor::entity_component::ozone::name%]",
"ph": "[%key:component::sensor::entity_component::ph::name%]",
"pm1": "[%key:component::sensor::entity_component::pm1::name%]",
"pm10": "[%key:component::sensor::entity_component::pm10::name%]",
"pm25": "[%key:component::sensor::entity_component::pm25::name%]",
"pm4": "[%key:component::sensor::entity_component::pm4::name%]",
"power": "[%key:component::sensor::entity_component::power::name%]",
"power_factor": "[%key:component::sensor::entity_component::power_factor::name%]",
"precipitation": "[%key:component::sensor::entity_component::precipitation::name%]",
"precipitation_intensity": "[%key:component::sensor::entity_component::precipitation_intensity::name%]",
"pressure": "[%key:component::sensor::entity_component::pressure::name%]",
"reactive_energy": "[%key:component::sensor::entity_component::reactive_energy::name%]",
"reactive_power": "[%key:component::sensor::entity_component::reactive_power::name%]",
"signal_strength": "[%key:component::sensor::entity_component::signal_strength::name%]",
"sound_pressure": "[%key:component::sensor::entity_component::sound_pressure::name%]",
"speed": "[%key:component::sensor::entity_component::speed::name%]",
"sulphur_dioxide": "[%key:component::sensor::entity_component::sulphur_dioxide::name%]",
"temperature": "[%key:component::sensor::entity_component::temperature::name%]",
"temperature_delta": "[%key:component::sensor::entity_component::temperature_delta::name%]",
"timestamp": "[%key:component::sensor::entity_component::timestamp::name%]",
"volatile_organic_compounds": "[%key:component::sensor::entity_component::volatile_organic_compounds::name%]",
"volatile_organic_compounds_parts": "[%key:component::sensor::entity_component::volatile_organic_compounds_parts::name%]",
"voltage": "[%key:component::sensor::entity_component::voltage::name%]",
"volume": "[%key:component::sensor::entity_component::volume::name%]",
"volume_flow_rate": "[%key:component::sensor::entity_component::volume_flow_rate::name%]",
"volume_storage": "[%key:component::sensor::entity_component::volume_storage::name%]",
"water": "[%key:component::sensor::entity_component::water::name%]",
"weight": "[%key:component::sensor::entity_component::weight::name%]",
"wind_direction": "[%key:component::sensor::entity_component::wind_direction::name%]",
"wind_speed": "[%key:component::sensor::entity_component::wind_speed::name%]"
}
},
"sensor_state_class": {
"options": {
"measurement": "[%key:component::sensor::entity_component::_::state_attributes::state_class::state::measurement%]",
"measurement_angle": "[%key:component::sensor::entity_component::_::state_attributes::state_class::state::measurement_angle%]",
"total": "[%key:component::sensor::entity_component::_::state_attributes::state_class::state::total%]",
"total_increasing": "[%key:component::sensor::entity_component::_::state_attributes::state_class::state::total_increasing%]"
}
}
},
"services": {
"event_register": {
"description": "Adds or removes group addresses to knx_event filter for triggering `knx_event`s. Only addresses added with this action can be removed.",

View File

@@ -22,6 +22,7 @@ from homeassistant.helpers.typing import UNDEFINED
from homeassistant.util.ulid import ulid_now
from .const import DOMAIN, KNX_MODULE_KEY, SUPPORTED_PLATFORMS_UI
from .dpt import get_supported_dpts
from .storage.config_store import ConfigStoreException
from .storage.const import CONF_DATA
from .storage.entity_store_schema import (
@@ -186,6 +187,7 @@ def ws_get_base_data(
msg["id"],
{
"connection_info": connection_info,
"dpt_metadata": get_supported_dpts(),
"project_info": _project_info,
"supported_platforms": sorted(SUPPORTED_PLATFORMS_UI),
},

View File

@@ -0,0 +1,26 @@
{
"version": 2,
"minor_version": 2,
"key": "knx/config_store.json",
"data": {
"entities": {
"sensor": {
"knx_es_01KC2F5CP5S4QCE3FZ49EF7CSJ": {
"entity": {
"name": "Test",
"entity_category": null,
"device_info": null
},
"knx": {
"ga_sensor": {
"state": "1/1/1",
"dpt": "7.600",
"passive": []
},
"sync_state": true
}
}
}
}
}
}

View File

@@ -1442,6 +1442,338 @@
'type': 'result',
})
# ---
# name: test_knx_get_schema[sensor]
dict({
'id': 1,
'result': list([
dict({
'name': 'ga_sensor',
'options': dict({
'dptClasses': list([
'numeric',
'string',
]),
'passive': True,
'state': dict({
'required': True,
}),
'write': False,
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'collapsible': True,
'name': 'section_advanced_options',
'required': False,
'type': 'knx_section_flat',
}),
dict({
'name': 'unit_of_measurement',
'optional': True,
'required': False,
'selector': dict({
'select': dict({
'custom_value': True,
'mode': 'dropdown',
'multiple': False,
'options': list([
'%',
'A',
'B',
'B/s',
'BTU/(h⋅ft²)',
'Beaufort',
'CCF',
'EB',
'EiB',
'GB',
'GB/s',
'GHz',
'GJ',
'GW',
'GWh',
'Gbit',
'Gbit/s',
'Gcal',
'GiB',
'GiB/s',
'Hz',
'J',
'K',
'KiB',
'KiB/s',
'L',
'L/h',
'L/min',
'L/s',
'MB',
'MB/s',
'MCF',
'MHz',
'MJ',
'MV',
'MW',
'MWh',
'Mbit',
'Mbit/s',
'Mcal',
'MiB',
'MiB/s',
'PB',
'Pa',
'PiB',
'S/cm',
'TB',
'TW',
'TWh',
'TiB',
'V',
'VA',
'W',
'W/m²',
'Wh',
'Wh/km',
'YB',
'YiB',
'ZB',
'ZiB',
'ac',
'bar',
'bit',
'bit/s',
'cal',
'cbar',
'cm',
'cm²',
'd',
'dB',
'dBA',
'dBm',
'fl. oz.',
'ft',
'ft/s',
'ft²',
'ft³',
'ft³/min',
'g',
'g/m³',
'gal',
'gal/d',
'gal/h',
'gal/min',
'h',
'hPa',
'ha',
'in',
'in/d',
'in/h',
'in/s',
'inHg',
'inH₂O',
'in²',
'kB',
'kB/s',
'kHz',
'kJ',
'kPa',
'kV',
'kVA',
'kW',
'kWh',
'kWh/100km',
'kbit',
'kbit/s',
'kcal',
'kg',
'km',
'km/h',
'km/kWh',
'km²',
'kn',
'kvar',
'kvarh',
'lb',
'lx',
'm',
'm/min',
'm/s',
'mA',
'mL',
'mL/s',
'mPa',
'mS/cm',
'mV',
'mVA',
'mW',
'mWh',
'mbar',
'mg',
'mg/dL',
'mg/m³',
'mi',
'mi/kWh',
'min',
'mi²',
'mm',
'mm/d',
'mm/h',
'mm/s',
'mmHg',
'mmol/L',
'mm²',
'mph',
'ms',
'mvar',
'm²',
'm³',
'm³/h',
'm³/min',
'm³/s',
'nmi',
'oz',
'ppb',
'ppm',
'psi',
's',
'st',
'var',
'varh',
'yd',
'yd²',
'°',
'°C',
'°F',
'μS/cm',
'μV',
'μg',
'μg/m³',
'μs',
]),
'sort': False,
'translation_key': 'component.knx.selector.sensor_unit_of_measurement',
}),
}),
'type': 'ha_selector',
}),
dict({
'name': 'device_class',
'optional': True,
'required': False,
'selector': dict({
'select': dict({
'custom_value': False,
'multiple': False,
'options': list([
'date',
'timestamp',
'absolute_humidity',
'apparent_power',
'aqi',
'area',
'atmospheric_pressure',
'battery',
'blood_glucose_concentration',
'carbon_monoxide',
'carbon_dioxide',
'conductivity',
'current',
'data_rate',
'data_size',
'distance',
'duration',
'energy',
'energy_distance',
'energy_storage',
'frequency',
'gas',
'humidity',
'illuminance',
'irradiance',
'moisture',
'monetary',
'nitrogen_dioxide',
'nitrogen_monoxide',
'nitrous_oxide',
'ozone',
'ph',
'pm1',
'pm10',
'pm25',
'pm4',
'power_factor',
'power',
'precipitation',
'precipitation_intensity',
'pressure',
'reactive_energy',
'reactive_power',
'signal_strength',
'sound_pressure',
'speed',
'sulphur_dioxide',
'temperature',
'temperature_delta',
'volatile_organic_compounds',
'volatile_organic_compounds_parts',
'voltage',
'volume',
'volume_storage',
'volume_flow_rate',
'water',
'weight',
'wind_direction',
'wind_speed',
]),
'sort': True,
'translation_key': 'component.knx.selector.sensor_device_class',
}),
}),
'type': 'ha_selector',
}),
dict({
'name': 'state_class',
'optional': True,
'required': False,
'selector': dict({
'select': dict({
'custom_value': False,
'mode': 'dropdown',
'multiple': False,
'options': list([
'measurement',
'measurement_angle',
'total',
'total_increasing',
]),
'sort': False,
'translation_key': 'component.knx.selector.sensor_state_class',
}),
}),
'type': 'ha_selector',
}),
dict({
'name': 'always_callback',
'optional': True,
'required': False,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'allow_false': True,
'default': True,
'name': 'sync_state',
'required': True,
'type': 'knx_sync_state',
}),
]),
'success': True,
'type': 'result',
})
# ---
# name: test_knx_get_schema[switch]
dict({
'id': 1,

View File

@@ -1,6 +1,9 @@
"""Test KNX sensor."""
from typing import Any
from freezegun.api import FrozenDateTimeFactory
import pytest
from homeassistant.components.knx.const import (
ATTR_SOURCE,
@@ -8,9 +11,10 @@ from homeassistant.components.knx.const import (
CONF_SYNC_STATE,
)
from homeassistant.components.knx.schema import SensorSchema
from homeassistant.const import CONF_NAME, CONF_TYPE, STATE_UNKNOWN
from homeassistant.const import CONF_NAME, CONF_TYPE, STATE_UNKNOWN, Platform
from homeassistant.core import HomeAssistant, State
from . import KnxEntityGenerator
from .conftest import KNXTestKit
from tests.common import (
@@ -166,3 +170,135 @@ async def test_always_callback(hass: HomeAssistant, knx: KNXTestKit) -> None:
await knx.receive_write("1/1/1", (0xFA,))
await knx.receive_write("2/2/2", (0xFA,))
assert len(events) == 6
@pytest.mark.parametrize(
("knx_config", "response_payload", "expected_state"),
[
(
{
"ga_sensor": {
"state": "1/1/1",
"passive": [],
"dpt": "9.001", # temperature 2 byte float
},
},
(0, 0),
{
"state": "0.0",
"device_class": "temperature",
"state_class": "measurement",
"unit_of_measurement": "°C",
},
),
(
{
"ga_sensor": {
"state": "1/1/1",
"passive": [],
"dpt": "12", # generic 4byte uint
},
"state_class": "total_increasing",
"device_class": "energy",
"unit_of_measurement": "Mcal",
"sync_state": True,
},
(1, 2, 3, 4),
{
"state": "16909060",
"device_class": "energy",
"state_class": "total_increasing",
},
),
],
)
async def test_sensor_ui_create(
hass: HomeAssistant,
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
knx_config: dict[str, Any],
response_payload: tuple[int, ...],
expected_state: dict[str, Any],
) -> None:
"""Test creating a sensor."""
await knx.setup_integration()
await create_ui_entity(
platform=Platform.SENSOR,
entity_data={"name": "test"},
knx_data=knx_config,
)
# created entity sends read-request to KNX bus
await knx.assert_read("1/1/1")
await knx.receive_response("1/1/1", response_payload)
knx.assert_state("sensor.test", **expected_state)
async def test_sensor_ui_load(knx: KNXTestKit) -> None:
"""Test loading a sensor from storage."""
await knx.setup_integration(config_store_fixture="config_store_sensor.json")
await knx.assert_read("1/1/1", response=(0, 0), ignore_order=True)
knx.assert_state(
"sensor.test",
"0",
device_class=None, # 7.600 color temperature has no sensor device class
state_class="measurement",
unit_of_measurement="K",
)
@pytest.mark.parametrize(
"knx_config",
[
(
{
"ga_sensor": {
"state": "1/1/1",
"passive": [],
"dpt": "9.001", # temperature 2 byte float
},
"state_class": "totoal_increasing", # invalid for temperature
}
),
(
{
"ga_sensor": {
"state": "1/1/1",
"passive": [],
"dpt": "12", # generic 4byte uint
},
"state_class": "total_increasing",
"device_class": "energy", # requires unit_of_measurement
"sync_state": True,
}
),
(
{
"ga_sensor": {
"state": "1/1/1",
"passive": [],
"dpt": "9.001", # temperature 2 byte float
},
"state_class": "measurement_angle", # requires degree unit
"sync_state": True,
}
),
],
)
async def test_sensor_ui_create_attribute_validation(
hass: HomeAssistant,
knx: KNXTestKit,
create_ui_entity: KnxEntityGenerator,
knx_config: dict[str, Any],
) -> None:
"""Test creating a sensor with invalid unit, state_class or device_class."""
await knx.setup_integration()
with pytest.raises(AssertionError) as err:
await create_ui_entity(
platform=Platform.SENSOR,
entity_data={"name": "test"},
knx_data=knx_config,
)
assert "success" in err.value.args[0]
assert "error_base" in err.value.args[0]
assert "path" in err.value.args[0]