"""KNX DPT serializer.""" from collections.abc import Mapping from functools import cache from typing import Literal, NotRequired, TypedDict, cast from xknx.dpt import ( DPTBase, DPTBinary, DPTComplex, DPTComplexFieldSchema, DPTEnum, DPTNumeric, ) from xknx.dpt.dpt_16 import DPTString from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass from homeassistant.const import UnitOfReactiveEnergy type 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 payload_length: int # numeric specific min: NotRequired[float] max: NotRequired[float] step: NotRequired[float] # enum specific options: NotRequired[list[str]] # complex specific schema: NotRequired[list[DPTComplexFieldSchema]] @cache def get_supported_dpts() -> Mapping[str, DPTInfo]: """Return a mapping of supported DPTs with HA specific attributes.""" dpts: dict[str, DPTInfo] = {} for dpt_class in DPTBase.dpt_class_tree(): dpt_number_str = dpt_class.dpt_number_str() ha_dpt_class = _ha_dpt_class(dpt_class) info = 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=_sensor_unit_overrides.get(dpt_number_str, 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), payload_length=dpt_class.payload_length, ) if ha_dpt_class == "numeric": _add_numeric_details(info, cast(type[DPTNumeric], dpt_class)) elif ha_dpt_class == "enum": _add_enum_details(info, cast(type[DPTEnum], dpt_class)) elif ha_dpt_class == "complex": _add_complex_details(info, cast(type[DPTComplex], dpt_class)) dpts[dpt_number_str] = info return dpts def raw_payload_length(dpt_cls: type[DPTBase]) -> int: """Return the payload length to use for a `RawValue` of a DPT. DPT 1, 2 and 3 are integrated in the APDU header. Their transcoders report a bit count while `RawValue` uses `0` to mark such binary payloads. """ if dpt_cls.payload_type is DPTBinary: return 0 return dpt_cls.payload_length 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") def _add_numeric_details(dpt_info: DPTInfo, dpt_cls: type[DPTNumeric]) -> None: """Add numeric specific details to the DPTInfo.""" dpt_info["min"] = dpt_cls.value_min dpt_info["max"] = dpt_cls.value_max dpt_info["step"] = dpt_cls.resolution def _add_enum_details(dpt_info: DPTInfo, dpt_cls: type[DPTEnum]) -> None: """Add enum specific details to the DPTInfo.""" dpt_info["options"] = [o.name.lower() for o in dpt_cls.get_valid_values()] def _add_complex_details(dpt_info: DPTInfo, dpt_cls: type[DPTComplex]) -> None: """Add complex specific details to the DPTInfo.""" dpt_info["schema"] = dpt_cls.get_dict_schema() _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, # DPTActiveEnergy "13.012": SensorDeviceClass.REACTIVE_ENERGY, # DPTReactiveEnergy "13.013": SensorDeviceClass.ENERGY, # DPTActiveEnergykWh "13.015": SensorDeviceClass.REACTIVE_ENERGY, # DPTReactiveEnergykVARh "13.016": SensorDeviceClass.ENERGY, # DPTActiveEnergyMWh "13.1200": SensorDeviceClass.VOLUME, # DPTDeltaVolumeLiquidLitre "13.1201": SensorDeviceClass.VOLUME, # DPTDeltaVolumeM3 "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, # DPTHeatQuantity "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, # DPTVolume "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 "13.013": SensorStateClass.TOTAL, # DPTActiveEnergykWh "13.015": SensorStateClass.TOTAL, # DPTReactiveEnergykVARh "13.016": SensorStateClass.TOTAL, # DPTActiveEnergyMWh "13.1200": SensorStateClass.TOTAL, # DPTDeltaVolumeLiquidLitre "13.1201": SensorStateClass.TOTAL, # DPTDeltaVolumeM3 "14.007": SensorStateClass.MEASUREMENT_ANGLE, # DPTAngleDeg "14.051": SensorStateClass.TOTAL, # DPTMass "14.055": SensorStateClass.MEASUREMENT_ANGLE, # DPTPhaseAngleDeg "14.031": SensorStateClass.TOTAL_INCREASING, # DPTEnergy "14.076": SensorStateClass.TOTAL, # DPTVolume "17.001": None, # DPTSceneNumber "29.010": SensorStateClass.TOTAL, # DPTActiveEnergy8Byte "29.011": SensorStateClass.TOTAL, # DPTApparantEnergy8Byte "29.012": SensorStateClass.TOTAL, # DPTReactiveEnergy8Byte } _sensor_unit_overrides: Mapping[str, str] = { # DPTReactiveEnergy (VARh in KNX) "13.012": UnitOfReactiveEnergy.VOLT_AMPERE_REACTIVE_HOUR, # DPTReactiveEnergykVARh (kVARh in KNX) "13.015": UnitOfReactiveEnergy.KILO_VOLT_AMPERE_REACTIVE_HOUR, # DPTReactiveEnergy8Byte (VARh in KNX) "29.012": UnitOfReactiveEnergy.VOLT_AMPERE_REACTIVE_HOUR, } 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, )