mirror of
https://github.com/home-assistant/core.git
synced 2026-06-25 16:15:28 +02:00
Compare commits
11 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 73967ec3b8 | |||
| 7e7c846c78 | |||
| 7ff7c45b18 | |||
| 4481e90ee6 | |||
| 7670dcb227 | |||
| db0f9d1f54 | |||
| f1535bf8a4 | |||
| c460100fa1 | |||
| 53e2a9341f | |||
| 0e713d549c | |||
| a068574fe2 |
@@ -1,19 +1,24 @@
|
||||
"""Support for KNX button entities."""
|
||||
|
||||
from typing import override
|
||||
from typing import Any, override
|
||||
|
||||
from xknx.devices import RawValue as XknxRawValue
|
||||
from xknx.devices import ExposeSensor as XknxExposeSensor, RawValue as XknxRawValue
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.button import ButtonEntity
|
||||
from homeassistant.const import CONF_ENTITY_CATEGORY, CONF_NAME, CONF_PAYLOAD, 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
|
||||
|
||||
from .const import CONF_PAYLOAD_LENGTH, KNX_ADDRESS, KNX_MODULE_KEY
|
||||
from .entity import KnxYamlEntity
|
||||
from .const import CONF_PAYLOAD_LENGTH, CONF_VALUE, DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY
|
||||
from .entity import KnxUiEntity, KnxUiEntityPlatformController, KnxYamlEntity
|
||||
from .knx_module import KNXModule
|
||||
from .storage.const import CONF_DATA, CONF_ENTITY, CONF_GA_SEND
|
||||
from .storage.util import ConfigExtractor
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
@@ -21,27 +26,60 @@ async def async_setup_entry(
|
||||
config_entry: config_entries.ConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up the KNX binary sensor platform."""
|
||||
"""Set up button(s) for KNX platform."""
|
||||
knx_module = hass.data[KNX_MODULE_KEY]
|
||||
config: list[ConfigType] = knx_module.config_yaml[Platform.BUTTON]
|
||||
platform = async_get_current_platform()
|
||||
knx_module.config_store.add_platform(
|
||||
platform=Platform.BUTTON,
|
||||
controller=KnxUiEntityPlatformController(
|
||||
knx_module=knx_module,
|
||||
entity_platform=platform,
|
||||
entity_class=KnxUiButton,
|
||||
),
|
||||
)
|
||||
|
||||
async_add_entities(KNXButton(knx_module, entity_config) for entity_config in config)
|
||||
entities: list[KnxYamlEntity | KnxUiEntity] = []
|
||||
if yaml_platform_config := knx_module.config_yaml.get(Platform.BUTTON):
|
||||
entities.extend(
|
||||
KnxYamlButton(knx_module, entity_config)
|
||||
for entity_config in yaml_platform_config
|
||||
)
|
||||
if ui_config := knx_module.config_store.data["entities"].get(Platform.BUTTON):
|
||||
entities.extend(
|
||||
KnxUiButton(knx_module, unique_id, config)
|
||||
for unique_id, config in ui_config.items()
|
||||
)
|
||||
if entities:
|
||||
async_add_entities(entities)
|
||||
|
||||
|
||||
class KNXButton(KnxYamlEntity, ButtonEntity):
|
||||
class _KnxButton(ButtonEntity):
|
||||
"""Representation of a KNX button."""
|
||||
|
||||
_device: XknxRawValue | XknxExposeSensor
|
||||
_payload: Any
|
||||
|
||||
@override
|
||||
async def async_press(self) -> None:
|
||||
"""Press the button."""
|
||||
await self._device.set(self._payload)
|
||||
|
||||
|
||||
class KnxYamlButton(_KnxButton, KnxYamlEntity):
|
||||
"""Representation of a KNX button configured via YAML."""
|
||||
|
||||
_device: XknxRawValue
|
||||
|
||||
def __init__(self, knx_module: KNXModule, config: ConfigType) -> None:
|
||||
"""Initialize a KNX button."""
|
||||
# dpt-value to payload conversion is done in schema validation for yaml config
|
||||
self._payload = config[CONF_PAYLOAD]
|
||||
self._device = XknxRawValue(
|
||||
xknx=knx_module.xknx,
|
||||
name=config[CONF_NAME],
|
||||
payload_length=config[CONF_PAYLOAD_LENGTH],
|
||||
group_address=config[KNX_ADDRESS],
|
||||
)
|
||||
self._payload = config[CONF_PAYLOAD]
|
||||
super().__init__(
|
||||
knx_module=knx_module,
|
||||
unique_id=f"{self._device.remote_value.group_address}_{self._payload}",
|
||||
@@ -49,7 +87,39 @@ class KNXButton(KnxYamlEntity, ButtonEntity):
|
||||
entity_category=config.get(CONF_ENTITY_CATEGORY),
|
||||
)
|
||||
|
||||
@override
|
||||
async def async_press(self) -> None:
|
||||
"""Press the button."""
|
||||
await self._device.set(self._payload)
|
||||
|
||||
class KnxUiButton(_KnxButton, KnxUiEntity):
|
||||
"""Representation of a KNX button configured via the UI."""
|
||||
|
||||
_device: XknxRawValue | XknxExposeSensor
|
||||
|
||||
def __init__(
|
||||
self, knx_module: KNXModule, unique_id: str, config: dict[str, Any]
|
||||
) -> None:
|
||||
"""Initialize a KNX button."""
|
||||
knx_conf = ConfigExtractor(config[DOMAIN])
|
||||
button_data = knx_conf.get(CONF_DATA)
|
||||
if CONF_PAYLOAD in button_data and CONF_PAYLOAD_LENGTH in button_data:
|
||||
self._payload = int(button_data[CONF_PAYLOAD], 16)
|
||||
self._device = XknxRawValue(
|
||||
xknx=knx_module.xknx,
|
||||
name=config[CONF_ENTITY][CONF_NAME],
|
||||
payload_length=button_data[CONF_PAYLOAD_LENGTH],
|
||||
group_address=knx_conf.get_write(CONF_GA_SEND),
|
||||
)
|
||||
else:
|
||||
dpt_string = knx_conf.get_dpt(CONF_GA_SEND)
|
||||
self._payload = button_data[CONF_VALUE]
|
||||
self._device = XknxExposeSensor(
|
||||
xknx=knx_module.xknx,
|
||||
name=config[CONF_ENTITY][CONF_NAME],
|
||||
value_type=dpt_string,
|
||||
group_address=knx_conf.get_write(CONF_GA_SEND),
|
||||
respond_to_read=False,
|
||||
)
|
||||
|
||||
super().__init__(
|
||||
knx_module=knx_module,
|
||||
unique_id=unique_id,
|
||||
entity_config=config[CONF_ENTITY],
|
||||
)
|
||||
|
||||
@@ -26,6 +26,7 @@ KNX_ADDRESS: Final = "address"
|
||||
CONF_INVERT: Final = "invert"
|
||||
CONF_KNX_EXPOSE: Final = "expose"
|
||||
CONF_KNX_INDIVIDUAL_ADDRESS: Final = "individual_address"
|
||||
CONF_VALUE: Final = "value"
|
||||
|
||||
##
|
||||
# Connection constants
|
||||
@@ -178,6 +179,7 @@ SUPPORTED_PLATFORMS_YAML: Final = {
|
||||
|
||||
SUPPORTED_PLATFORMS_UI: Final = {
|
||||
Platform.BINARY_SENSOR,
|
||||
Platform.BUTTON,
|
||||
Platform.CLIMATE,
|
||||
Platform.COVER,
|
||||
Platform.DATE,
|
||||
|
||||
@@ -2,9 +2,9 @@
|
||||
|
||||
from collections.abc import Mapping
|
||||
from functools import cache
|
||||
from typing import Literal, TypedDict
|
||||
from typing import Literal, NotRequired, TypedDict, cast
|
||||
|
||||
from xknx.dpt import DPTBase, DPTComplex, DPTEnum, DPTNumeric
|
||||
from xknx.dpt import DPTBase, DPTComplex, DPTComplexFieldSchema, DPTEnum, DPTNumeric
|
||||
from xknx.dpt.dpt_16 import DPTString
|
||||
|
||||
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass
|
||||
@@ -24,15 +24,28 @@ class DPTInfo(TypedDict):
|
||||
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 = {}
|
||||
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)
|
||||
dpts[dpt_number_str] = DPTInfo(
|
||||
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,
|
||||
@@ -40,7 +53,15 @@ def get_supported_dpts() -> Mapping[str, DPTInfo]:
|
||||
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
|
||||
|
||||
|
||||
@@ -57,6 +78,23 @@ def _ha_dpt_class(dpt_cls: type[DPTBase]) -> HaDptClass:
|
||||
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,
|
||||
|
||||
@@ -57,6 +57,7 @@ from .const import (
|
||||
CONF_RESPOND_TO_READ,
|
||||
CONF_STATE_ADDRESS,
|
||||
CONF_SYNC_STATE,
|
||||
CONF_VALUE,
|
||||
KNX_ADDRESS,
|
||||
ClimateConf,
|
||||
ColorTempModes,
|
||||
@@ -98,9 +99,12 @@ def _max_payload_value(payload_length: int) -> int:
|
||||
|
||||
|
||||
def button_payload_sub_validator(entity_config: OrderedDict) -> OrderedDict:
|
||||
"""Validate a button entity payload configuration."""
|
||||
"""Validate a button entity payload configuration.
|
||||
|
||||
Returns raw payload and length from value and type (DPT), if given.
|
||||
"""
|
||||
if _type := entity_config.get(CONF_TYPE):
|
||||
_payload = entity_config[ButtonSchema.CONF_VALUE]
|
||||
_payload = entity_config[CONF_VALUE]
|
||||
if (transcoder := DPTBase.parse_transcoder(_type)) is None:
|
||||
raise vol.Invalid(f"'type: {_type}' is not a valid sensor type.")
|
||||
entity_config[CONF_PAYLOAD_LENGTH] = transcoder.payload_length
|
||||
@@ -234,8 +238,6 @@ class ButtonSchema(KNXPlatformSchema):
|
||||
|
||||
PLATFORM = Platform.BUTTON
|
||||
|
||||
CONF_VALUE = "value"
|
||||
|
||||
payload_or_value_msg = f"Please use only one of `{CONF_PAYLOAD}` or `{CONF_VALUE}`"
|
||||
length_or_type_msg = (
|
||||
f"Please use only one of `{CONF_PAYLOAD_LENGTH}` or `{CONF_TYPE}`"
|
||||
|
||||
@@ -19,6 +19,9 @@ CONF_GA_TIME: Final = "ga_time"
|
||||
|
||||
CONF_GA_STEP: Final = "ga_step"
|
||||
|
||||
# Button
|
||||
CONF_GA_SEND: Final = "ga_send"
|
||||
|
||||
# Climate
|
||||
CONF_GA_TEMPERATURE_CURRENT: Final = "ga_temperature_current"
|
||||
CONF_GA_HUMIDITY_CURRENT: Final = "ga_humidity_current"
|
||||
|
||||
@@ -3,7 +3,8 @@
|
||||
from enum import StrEnum, unique
|
||||
|
||||
import voluptuous as vol
|
||||
from xknx.dpt import DPTNumeric
|
||||
from xknx.dpt import DPTBase, DPTBinary, DPTNumeric
|
||||
from xknx.exceptions import ConversionError
|
||||
|
||||
from homeassistant.components.climate import HVACMode
|
||||
from homeassistant.components.number import (
|
||||
@@ -36,9 +37,11 @@ from ..const import (
|
||||
CONF_CONTEXT_TIMEOUT,
|
||||
CONF_IGNORE_INTERNAL_STATE,
|
||||
CONF_INVERT,
|
||||
CONF_PAYLOAD_LENGTH,
|
||||
CONF_RESET_AFTER,
|
||||
CONF_RESPOND_TO_READ,
|
||||
CONF_SYNC_STATE,
|
||||
CONF_VALUE,
|
||||
DOMAIN,
|
||||
SUPPORTED_PLATFORMS_UI,
|
||||
ClimateConf,
|
||||
@@ -92,6 +95,7 @@ from .const import (
|
||||
CONF_GA_RED_SWITCH,
|
||||
CONF_GA_SATURATION,
|
||||
CONF_GA_SCENE,
|
||||
CONF_GA_SEND,
|
||||
CONF_GA_SENSOR,
|
||||
CONF_GA_SETPOINT_SHIFT,
|
||||
CONF_GA_SPEED,
|
||||
@@ -115,6 +119,7 @@ from .knx_selector import (
|
||||
GASelector,
|
||||
GroupSelect,
|
||||
GroupSelectOption,
|
||||
KnxPayloadSelector,
|
||||
KNXSectionFlat,
|
||||
SyncStateSelector,
|
||||
)
|
||||
@@ -169,6 +174,55 @@ BINARY_SENSOR_KNX_SCHEMA = vol.Schema(
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _button_data_sub_validator(config: dict) -> dict:
|
||||
"""Validate data matching configured DPT."""
|
||||
dpt = config[CONF_GA_SEND].get(CONF_DPT)
|
||||
transcoder = None
|
||||
if dpt:
|
||||
transcoder = DPTBase.parse_transcoder(dpt)
|
||||
assert transcoder is not None # already checked by GASelector
|
||||
|
||||
if CONF_VALUE in config[CONF_DATA]:
|
||||
try:
|
||||
transcoder.to_knx(config[CONF_DATA][CONF_VALUE])
|
||||
except ConversionError as ex:
|
||||
raise vol.Invalid(
|
||||
f"Value invalid for DPT {transcoder.dpt_number_str()}",
|
||||
path=([CONF_DATA]),
|
||||
) from ex
|
||||
elif CONF_PAYLOAD_LENGTH in config[CONF_DATA]:
|
||||
length = config[CONF_DATA][CONF_PAYLOAD_LENGTH]
|
||||
if length != transcoder.payload_length or (
|
||||
length != 0 and transcoder.payload_type is DPTBinary
|
||||
):
|
||||
raise vol.Invalid(
|
||||
f"Payload length invalid for DPT {transcoder.dpt_number_str()}",
|
||||
path=([CONF_DATA]),
|
||||
)
|
||||
return config
|
||||
# without DPT only raw allowed -> payload + payload_length (checked by KnxPayloadSelector)
|
||||
if CONF_PAYLOAD_LENGTH in config[CONF_DATA]:
|
||||
return config
|
||||
raise vol.Invalid("Invalid configuration for button entity")
|
||||
|
||||
|
||||
BUTTON_KNX_SCHEMA = AllSerializeFirst(
|
||||
vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_GA_SEND): GASelector(
|
||||
state=False,
|
||||
write_required=True,
|
||||
passive=False,
|
||||
dpt=["numeric", "enum", "complex", "string"],
|
||||
dpt_required=False, # for raw payload support
|
||||
),
|
||||
vol.Required(CONF_DATA): KnxPayloadSelector(ga_path=CONF_GA_SEND),
|
||||
},
|
||||
),
|
||||
_button_data_sub_validator,
|
||||
)
|
||||
|
||||
COVER_KNX_SCHEMA = AllSerializeFirst(
|
||||
vol.Schema(
|
||||
{
|
||||
@@ -741,6 +795,7 @@ SENSOR_KNX_SCHEMA = AllSerializeFirst(
|
||||
|
||||
KNX_SCHEMA_FOR_PLATFORM = {
|
||||
Platform.BINARY_SENSOR: BINARY_SENSOR_KNX_SCHEMA,
|
||||
Platform.BUTTON: BUTTON_KNX_SCHEMA,
|
||||
Platform.CLIMATE: CLIMATE_KNX_SCHEMA,
|
||||
Platform.COVER: COVER_KNX_SCHEMA,
|
||||
Platform.DATE: DATE_KNX_SCHEMA,
|
||||
|
||||
@@ -6,6 +6,9 @@ from typing import Any, override
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.const import CONF_PAYLOAD
|
||||
|
||||
from ..const import CONF_PAYLOAD_LENGTH, CONF_VALUE
|
||||
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
|
||||
@@ -159,7 +162,11 @@ class GroupSelect(KNXSelectorBase):
|
||||
|
||||
|
||||
class GASelector(KNXSelectorBase):
|
||||
"""Selector for a KNX group address structure."""
|
||||
"""Selector for a KNX group address structure.
|
||||
|
||||
`dpt_required` optional dpt only apply to dpt-class lists, enums are always required.
|
||||
`valid_dpt` is used in frontend to filter dropdown menu - no validation is done.
|
||||
"""
|
||||
|
||||
selector_type = "knx_group_address"
|
||||
|
||||
@@ -171,6 +178,7 @@ class GASelector(KNXSelectorBase):
|
||||
write_required: bool = False,
|
||||
state_required: bool = False,
|
||||
dpt: type[Enum] | list[HaDptClass] | None = None,
|
||||
dpt_required: bool = True,
|
||||
valid_dpt: str | Iterable[str] | None = None,
|
||||
) -> None:
|
||||
"""Initialize the group address selector."""
|
||||
@@ -180,7 +188,7 @@ class GASelector(KNXSelectorBase):
|
||||
self.write_required = write_required
|
||||
self.state_required = state_required
|
||||
self.dpt = dpt
|
||||
# valid_dpt is used in frontend to filter dropdown menu - no validation is done
|
||||
self.dpt_required = dpt_required
|
||||
self.valid_dpt = (valid_dpt,) if isinstance(valid_dpt, str) else valid_dpt
|
||||
|
||||
self.schema = self.build_schema()
|
||||
@@ -196,6 +204,7 @@ class GASelector(KNXSelectorBase):
|
||||
}
|
||||
if self.dpt is not None:
|
||||
if isinstance(self.dpt, list):
|
||||
# optional / required is not passed to FE - only validated in BE
|
||||
options["dptClasses"] = self.dpt
|
||||
else:
|
||||
options["dptSelect"] = [
|
||||
@@ -267,7 +276,8 @@ class GASelector(KNXSelectorBase):
|
||||
"""Add DPT validator to the schema."""
|
||||
if self.dpt is not None:
|
||||
if isinstance(self.dpt, list):
|
||||
schema[vol.Required(CONF_DPT)] = vol.In(get_supported_dpts())
|
||||
marker = vol.Required if self.dpt_required else vol.Optional
|
||||
schema[marker(CONF_DPT)] = vol.In(get_supported_dpts())
|
||||
else:
|
||||
schema[vol.Required(CONF_DPT)] = vol.In(
|
||||
{item.value for item in self.dpt}
|
||||
@@ -300,3 +310,64 @@ class SyncStateSelector(KNXSelectorBase):
|
||||
if not self.allow_false and not data:
|
||||
raise vol.Invalid(f"Sync state cannot be {data}")
|
||||
return self.schema(data)
|
||||
|
||||
|
||||
class KnxPayloadSelector(KNXSelectorBase):
|
||||
"""Selector for KNX payload configuration.
|
||||
|
||||
Raw payloads are stored as hex strings.
|
||||
"""
|
||||
|
||||
schema = vol.Any(
|
||||
{
|
||||
vol.Required(CONF_VALUE): object,
|
||||
},
|
||||
{
|
||||
vol.Required(CONF_PAYLOAD): str,
|
||||
vol.Required(CONF_PAYLOAD_LENGTH): vol.All(int, vol.Range(min=0, max=14)),
|
||||
},
|
||||
)
|
||||
selector_type = "knx_payload"
|
||||
|
||||
def __init__(self, ga_path: str) -> None:
|
||||
"""Initialize the KNX payload selector."""
|
||||
self.ga_path = ga_path
|
||||
|
||||
@override
|
||||
def serialize(self) -> dict[str, Any]:
|
||||
"""Serialize the selector to a dictionary."""
|
||||
return {
|
||||
"type": self.selector_type,
|
||||
"ga_path": self.ga_path,
|
||||
}
|
||||
|
||||
@override
|
||||
def __call__(self, data: Any) -> Any:
|
||||
"""Validate the passed data."""
|
||||
validated = self.schema(data)
|
||||
if CONF_PAYLOAD in validated and CONF_PAYLOAD_LENGTH in validated:
|
||||
payload = validated[CONF_PAYLOAD]
|
||||
payload_length = validated[CONF_PAYLOAD_LENGTH]
|
||||
try:
|
||||
int_payload = int(payload, 16)
|
||||
except ValueError as ex:
|
||||
raise vol.Invalid(f"Invalid payload format: {payload}") from ex
|
||||
validated[CONF_PAYLOAD] = hex(int_payload) # prepends "0x" if not present
|
||||
|
||||
if int_payload < 0:
|
||||
raise vol.Invalid(f"Payload cannot be negative: {payload}")
|
||||
if payload_length == 0:
|
||||
# DPT 1,2,3 is marked length 0, has 6 bit size
|
||||
if int_payload > 63:
|
||||
raise vol.Invalid(
|
||||
f"Payload exceeds DPT 1,2,3 limit of 0x3f (63): {payload}"
|
||||
)
|
||||
else:
|
||||
max_payload = (1 << (payload_length * 8)) - 1
|
||||
if int_payload > max_payload:
|
||||
raise vol.Invalid(
|
||||
f"Payload {payload} exceeds possible maximum for "
|
||||
f"length {payload_length}: {hex(max_payload)}"
|
||||
)
|
||||
# CONF_VALUE branch needs subvalidator as we don't have the DPT available here
|
||||
return validated
|
||||
|
||||
@@ -453,6 +453,19 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"button": {
|
||||
"description": "Entity for sending predefined values.",
|
||||
"knx": {
|
||||
"data": {
|
||||
"description": "The value sent when the button is pressed. The format of the value depends on the DPT of the configured address.",
|
||||
"label": "Data"
|
||||
},
|
||||
"ga_send": {
|
||||
"description": "Group address the value is sent to.",
|
||||
"label": "Address"
|
||||
}
|
||||
}
|
||||
},
|
||||
"climate": {
|
||||
"description": "The KNX climate platform is used as an interface to heating actuators, HVAC gateways, etc.",
|
||||
"knx": {
|
||||
@@ -1014,6 +1027,19 @@
|
||||
"project": {
|
||||
"description": "Inspect imported group addresses",
|
||||
"title": "Project"
|
||||
},
|
||||
"selectors": {
|
||||
"knx-payload-selector": {
|
||||
"dpt_missing": "No DPT selected – Typed mode not available",
|
||||
"mode": {
|
||||
"label": "Payload format",
|
||||
"raw": "Raw payload",
|
||||
"typed": "Typed value"
|
||||
},
|
||||
"raw_length": "Payload length",
|
||||
"raw_length_description": "Length of the raw payload in bytes. For DPT 1, 2 and 3 use `0`.",
|
||||
"raw_payload": "Raw payload"
|
||||
}
|
||||
}
|
||||
},
|
||||
"device_automation": {
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
{
|
||||
"version": 2,
|
||||
"minor_version": 3,
|
||||
"key": "knx/config_store.json",
|
||||
"data": {
|
||||
"entities": {
|
||||
"button": {
|
||||
"knx_es_01KVFEGP54VJW94TR9GQW2XA4R": {
|
||||
"entity": {
|
||||
"name": "test raw",
|
||||
"device_info": null,
|
||||
"entity_category": null
|
||||
},
|
||||
"knx": {
|
||||
"data": {
|
||||
"payload": "0x1",
|
||||
"payload_length": 1
|
||||
},
|
||||
"ga_send": {
|
||||
"write": "1/1/1"
|
||||
}
|
||||
}
|
||||
},
|
||||
"knx_es_01KVFEHE937CQGWP81RZNQ6D8E": {
|
||||
"entity": {
|
||||
"name": "test typed",
|
||||
"device_info": null,
|
||||
"entity_category": null
|
||||
},
|
||||
"knx": {
|
||||
"ga_send": {
|
||||
"write": "1/1/2",
|
||||
"dpt": "1.001"
|
||||
},
|
||||
"data": {
|
||||
"value": "on"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"time_server": {}
|
||||
}
|
||||
}
|
||||
@@ -129,6 +129,39 @@
|
||||
'type': 'result',
|
||||
})
|
||||
# ---
|
||||
# name: test_knx_get_schema[button]
|
||||
dict({
|
||||
'id': 1,
|
||||
'result': list([
|
||||
dict({
|
||||
'name': 'ga_send',
|
||||
'options': dict({
|
||||
'dptClasses': list([
|
||||
'numeric',
|
||||
'enum',
|
||||
'complex',
|
||||
'string',
|
||||
]),
|
||||
'passive': False,
|
||||
'state': False,
|
||||
'write': dict({
|
||||
'required': True,
|
||||
}),
|
||||
}),
|
||||
'required': True,
|
||||
'type': 'knx_group_address',
|
||||
}),
|
||||
dict({
|
||||
'ga_path': 'ga_send',
|
||||
'name': 'data',
|
||||
'required': True,
|
||||
'type': 'knx_payload',
|
||||
}),
|
||||
]),
|
||||
'success': True,
|
||||
'type': 'result',
|
||||
})
|
||||
# ---
|
||||
# name: test_knx_get_schema[climate]
|
||||
dict({
|
||||
'id': 1,
|
||||
|
||||
@@ -2,22 +2,32 @@
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.knx.const import (
|
||||
CONF_PAYLOAD_LENGTH,
|
||||
CONF_VALUE,
|
||||
KNX_ADDRESS,
|
||||
KNX_MODULE_KEY,
|
||||
)
|
||||
from homeassistant.components.knx.schema import ButtonSchema
|
||||
from homeassistant.const import CONF_NAME, CONF_PAYLOAD, CONF_TYPE
|
||||
from homeassistant.const import (
|
||||
CONF_NAME,
|
||||
CONF_PAYLOAD,
|
||||
CONF_TYPE,
|
||||
STATE_UNKNOWN,
|
||||
Platform,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import KnxEntityGenerator
|
||||
from .conftest import KNXTestKit
|
||||
|
||||
from tests.common import async_capture_events, async_fire_time_changed
|
||||
from tests.typing import WebSocketGenerator
|
||||
|
||||
|
||||
async def test_button_simple(
|
||||
@@ -83,7 +93,7 @@ async def test_button_type(hass: HomeAssistant, knx: KNXTestKit) -> None:
|
||||
ButtonSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
KNX_ADDRESS: "1/2/3",
|
||||
ButtonSchema.CONF_VALUE: 21.5,
|
||||
CONF_VALUE: 21.5,
|
||||
CONF_TYPE: "2byte_float",
|
||||
}
|
||||
}
|
||||
@@ -125,7 +135,7 @@ async def test_button_invalid(
|
||||
ButtonSchema.PLATFORM: {
|
||||
CONF_NAME: "test",
|
||||
KNX_ADDRESS: "1/2/3",
|
||||
ButtonSchema.CONF_VALUE: conf_value,
|
||||
CONF_VALUE: conf_value,
|
||||
CONF_TYPE: conf_type,
|
||||
}
|
||||
}
|
||||
@@ -139,3 +149,128 @@ async def test_button_invalid(
|
||||
assert "Setup failed for 'knx': Invalid config." in record.message
|
||||
assert hass.states.get("button.test") is None
|
||||
assert hass.data.get(KNX_MODULE_KEY) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"knx_config",
|
||||
[
|
||||
(
|
||||
{
|
||||
"ga_send": {"write": "1/1/1"},
|
||||
"data": {"payload": "1", "payload_length": 1}, # raw payload
|
||||
}
|
||||
),
|
||||
(
|
||||
{
|
||||
"ga_send": {"write": "1/1/1", "dpt": "5"}, # generic 1byte uint
|
||||
"data": {"payload": "0x01", "payload_length": 1}, # raw payload
|
||||
}
|
||||
),
|
||||
(
|
||||
{
|
||||
"ga_send": {"write": "1/1/1", "dpt": "5"}, # generic 1byte uint
|
||||
"data": {"value": 1}, # typed value
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_button_ui_create(
|
||||
hass: HomeAssistant,
|
||||
knx: KNXTestKit,
|
||||
create_ui_entity: KnxEntityGenerator,
|
||||
knx_config: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test creating a button."""
|
||||
await knx.setup_integration()
|
||||
await create_ui_entity(
|
||||
platform=Platform.BUTTON,
|
||||
entity_data={"name": "test"},
|
||||
knx_data=knx_config,
|
||||
)
|
||||
await hass.services.async_call(
|
||||
"button", "press", {"entity_id": "button.test"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/1/1", (1,))
|
||||
|
||||
|
||||
async def test_button_ui_load(hass: HomeAssistant, knx: KNXTestKit) -> None:
|
||||
"""Test loading a button from storage."""
|
||||
await knx.setup_integration(config_store_fixture="config_store_button.json")
|
||||
|
||||
# Raw button configuration
|
||||
knx.assert_state(
|
||||
"button.test_raw",
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
await hass.services.async_call(
|
||||
"button", "press", {"entity_id": "button.test_raw"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/1/1", (1,))
|
||||
|
||||
# Typed button configuration
|
||||
knx.assert_state(
|
||||
"button.test_typed",
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
await hass.services.async_call(
|
||||
"button", "press", {"entity_id": "button.test_typed"}, blocking=True
|
||||
)
|
||||
await knx.assert_write("1/1/2", True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"knx_config",
|
||||
[
|
||||
{ # missing data
|
||||
"ga_send": {"write": "1/1/1", "dpt": "9.001"},
|
||||
},
|
||||
{ # missing DPT
|
||||
"ga_send": {"write": "1/1/1"},
|
||||
"data": {"value": 1},
|
||||
},
|
||||
{ # invalid value for DPT
|
||||
"ga_send": {"write": "1/1/1", "dpt": "9.001"},
|
||||
"data": {"value": "not_valid"},
|
||||
},
|
||||
{ # invalid length for DPT
|
||||
"ga_send": {"write": "1/1/1", "dpt": "9.001"},
|
||||
"data": {"payload": "0x1", "payload_length": 1},
|
||||
},
|
||||
{ # out of bound value for DPT
|
||||
"ga_send": {"write": "1/1/1", "dpt": "5.001"},
|
||||
"data": {"value": 101},
|
||||
},
|
||||
{ # out of bound value for length
|
||||
"ga_send": {"write": "1/1/1"},
|
||||
"data": {"payload": "0x100", "payload_length": 1},
|
||||
},
|
||||
{ # out of bound value for zero-length
|
||||
"ga_send": {"write": "1/1/1"},
|
||||
"data": {"payload": "0x40", "payload_length": 0},
|
||||
},
|
||||
],
|
||||
)
|
||||
async def test_button_ui_create_data_validation(
|
||||
hass: HomeAssistant,
|
||||
knx: KNXTestKit,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
knx_config: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test creating a button with invalid data."""
|
||||
await knx.setup_integration()
|
||||
client = await hass_ws_client(hass)
|
||||
await client.send_json_auto_id(
|
||||
{
|
||||
"type": "knx/create_entity",
|
||||
"platform": Platform.BUTTON,
|
||||
"data": {
|
||||
"entity": {"name": "test"},
|
||||
"knx": knx_config,
|
||||
},
|
||||
}
|
||||
)
|
||||
res = await client.receive_json()
|
||||
assert res["success"], res
|
||||
assert res["result"]["success"] is False
|
||||
assert res["result"]["error_base"]
|
||||
assert res["result"]["errors"][0]["path"]
|
||||
|
||||
Reference in New Issue
Block a user