Use serialized schema from backend in UI entity configuration

This commit is contained in:
farmio 2025-07-26 20:06:43 +02:00
parent 40571dff3d
commit cb6c41bf95
8 changed files with 1699 additions and 236 deletions

View File

@ -27,7 +27,6 @@ from ..const import (
ColorTempModes, ColorTempModes,
CoverConf, CoverConf,
) )
from ..validation import sync_state_validator
from .const import ( from .const import (
CONF_COLOR, CONF_COLOR,
CONF_COLOR_TEMP_MAX, CONF_COLOR_TEMP_MAX,
@ -57,7 +56,14 @@ from .const import (
CONF_GA_WHITE_BRIGHTNESS, CONF_GA_WHITE_BRIGHTNESS,
CONF_GA_WHITE_SWITCH, CONF_GA_WHITE_SWITCH,
) )
from .knx_selector import GASelector, GroupSelect from .knx_selector import (
AllSerializeFirst,
GASelector,
GroupSelect,
GroupSelectOption,
KNXSectionFlat,
SyncStateSelector,
)
BASE_ENTITY_SCHEMA = vol.All( BASE_ENTITY_SCHEMA = vol.All(
{ {
@ -85,86 +91,86 @@ BASE_ENTITY_SCHEMA = vol.All(
) )
BINARY_SENSOR_SCHEMA = vol.Schema( BINARY_SENSOR_KNX_SCHEMA = vol.Schema(
{ {
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA, "section_binary_sensor": KNXSectionFlat(),
vol.Required(DOMAIN): { vol.Required(CONF_GA_SENSOR): GASelector(
vol.Required(CONF_GA_SENSOR): GASelector(write=False, state_required=True), write=False, state_required=True, valid_dpt="1"
vol.Required(CONF_RESPOND_TO_READ, default=False): bool, ),
vol.Required(CONF_SYNC_STATE, default=True): sync_state_validator, vol.Optional(CONF_INVERT): selector.BooleanSelector(),
vol.Optional(CONF_INVERT): selector.BooleanSelector(), "section_advanced_options": KNXSectionFlat(collapsible=True),
vol.Optional(CONF_IGNORE_INTERNAL_STATE): selector.BooleanSelector(), vol.Optional(CONF_IGNORE_INTERNAL_STATE): selector.BooleanSelector(),
vol.Optional(CONF_CONTEXT_TIMEOUT): selector.NumberSelector( vol.Optional(CONF_CONTEXT_TIMEOUT): selector.NumberSelector(
selector.NumberSelectorConfig( selector.NumberSelectorConfig(
min=0, max=10, step=0.1, unit_of_measurement="s" min=0, max=10, step=0.1, unit_of_measurement="s"
) )
), ),
vol.Optional(CONF_RESET_AFTER): selector.NumberSelector( vol.Optional(CONF_RESET_AFTER): selector.NumberSelector(
selector.NumberSelectorConfig( selector.NumberSelectorConfig(
min=0, max=600, step=0.1, unit_of_measurement="s" min=0, max=600, step=0.1, unit_of_measurement="s"
) )
), ),
}, vol.Required(CONF_SYNC_STATE, default=True): SyncStateSelector(),
} },
) )
COVER_SCHEMA = vol.Schema( COVER_KNX_SCHEMA = AllSerializeFirst(
{ vol.Schema(
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA, {
vol.Required(DOMAIN): vol.All( "section_binary_control": KNXSectionFlat(),
vol.Schema( vol.Optional(CONF_GA_UP_DOWN): GASelector(state=False),
{ vol.Optional(CoverConf.INVERT_UPDOWN): selector.BooleanSelector(),
vol.Optional(CONF_GA_UP_DOWN): GASelector(state=False), "section_stop_control": KNXSectionFlat(),
vol.Optional(CoverConf.INVERT_UPDOWN): selector.BooleanSelector(), vol.Optional(CONF_GA_STOP): GASelector(state=False),
vol.Optional(CONF_GA_STOP): GASelector(state=False), vol.Optional(CONF_GA_STEP): GASelector(state=False),
vol.Optional(CONF_GA_STEP): GASelector(state=False), "section_position_control": KNXSectionFlat(collapsible=True),
vol.Optional(CONF_GA_POSITION_SET): GASelector(state=False), vol.Optional(CONF_GA_POSITION_SET): GASelector(state=False),
vol.Optional(CONF_GA_POSITION_STATE): GASelector(write=False), vol.Optional(CONF_GA_POSITION_STATE): GASelector(write=False),
vol.Optional(CoverConf.INVERT_POSITION): selector.BooleanSelector(), vol.Optional(CoverConf.INVERT_POSITION): selector.BooleanSelector(),
vol.Optional(CONF_GA_ANGLE): GASelector(), "section_tilt_control": KNXSectionFlat(collapsible=True),
vol.Optional(CoverConf.INVERT_ANGLE): selector.BooleanSelector(), vol.Optional(CONF_GA_ANGLE): GASelector(),
vol.Optional( vol.Optional(CoverConf.INVERT_ANGLE): selector.BooleanSelector(),
CoverConf.TRAVELLING_TIME_DOWN, default=25 "section_travel_time": KNXSectionFlat(),
): selector.NumberSelector( vol.Optional(
selector.NumberSelectorConfig( CoverConf.TRAVELLING_TIME_UP, default=25
min=0, max=1000, step=0.1, unit_of_measurement="s" ): selector.NumberSelector(
) selector.NumberSelectorConfig(
), min=0, max=1000, step=0.1, unit_of_measurement="s"
vol.Optional( )
CoverConf.TRAVELLING_TIME_UP, default=25
): selector.NumberSelector(
selector.NumberSelectorConfig(
min=0, max=1000, step=0.1, unit_of_measurement="s"
)
),
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
},
extra=vol.REMOVE_EXTRA,
), ),
vol.Any( vol.Optional(
vol.Schema( CoverConf.TRAVELLING_TIME_DOWN, default=25
{ ): selector.NumberSelector(
vol.Required(CONF_GA_UP_DOWN): GASelector( selector.NumberSelectorConfig(
state=False, write_required=True min=0, max=1000, step=0.1, unit_of_measurement="s"
) )
},
extra=vol.ALLOW_EXTRA,
),
vol.Schema(
{
vol.Required(CONF_GA_POSITION_SET): GASelector(
state=False, write_required=True
)
},
extra=vol.ALLOW_EXTRA,
),
msg=(
"At least one of 'Up/Down control' or"
" 'Position - Set position' is required."
),
), ),
vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(),
},
extra=vol.REMOVE_EXTRA,
),
vol.Any(
vol.Schema(
{
vol.Required(CONF_GA_UP_DOWN): GASelector(
state=False, write_required=True
)
},
extra=vol.ALLOW_EXTRA,
), ),
} vol.Schema(
{
vol.Required(CONF_GA_POSITION_SET): GASelector(
state=False, write_required=True
)
},
extra=vol.ALLOW_EXTRA,
),
msg=(
"At least one of 'Open/Close control' or"
" 'Position - Set position' is required."
),
),
) )
@ -177,81 +183,91 @@ class LightColorMode(StrEnum):
XYY = "242.600" XYY = "242.600"
@unique
class LightColorModeSchema(StrEnum):
"""Enum for light color mode."""
DEFAULT = "default"
INDIVIDUAL = "individual"
HSV = "hsv"
_hs_color_inclusion_msg = ( _hs_color_inclusion_msg = (
"'Hue', 'Saturation' and 'Brightness' addresses are required for HSV configuration" "'Hue', 'Saturation' and 'Brightness' addresses are required for HSV configuration"
) )
LIGHT_KNX_SCHEMA = vol.All( LIGHT_KNX_SCHEMA = AllSerializeFirst(
vol.Schema( vol.Schema(
{ {
vol.Optional(CONF_GA_SWITCH): GASelector(write_required=True), "section_switch": KNXSectionFlat(),
vol.Optional(CONF_GA_BRIGHTNESS): GASelector(write_required=True), vol.Optional(CONF_GA_SWITCH): GASelector(
write_required=True, valid_dpt="1"
),
"section_brightness": KNXSectionFlat(),
vol.Optional(CONF_GA_BRIGHTNESS): GASelector(
write_required=True, valid_dpt="5.001"
),
"section_color_temp": KNXSectionFlat(collapsible=True),
vol.Optional(CONF_GA_COLOR_TEMP): GASelector( vol.Optional(CONF_GA_COLOR_TEMP): GASelector(
write_required=True, dpt=ColorTempModes write_required=True, dpt=ColorTempModes
), ),
vol.Required(CONF_COLOR_TEMP_MIN, default=2700): selector.NumberSelector(
selector.NumberSelectorConfig(
min=1, max=10000, step=1, unit_of_measurement="K"
)
),
vol.Required(CONF_COLOR_TEMP_MAX, default=6000): selector.NumberSelector(
selector.NumberSelectorConfig(
min=1, max=10000, step=1, unit_of_measurement="K"
)
),
vol.Optional(CONF_COLOR): GroupSelect( vol.Optional(CONF_COLOR): GroupSelect(
vol.Schema( GroupSelectOption(
{ translation_key="single_address",
schema={
vol.Optional(CONF_GA_COLOR): GASelector( vol.Optional(CONF_GA_COLOR): GASelector(
write_required=True, dpt=LightColorMode write_required=True, dpt=LightColorMode
) )
} },
), ),
vol.Schema( GroupSelectOption(
{ translation_key="individual_addresses",
vol.Required(CONF_GA_RED_BRIGHTNESS): GASelector( schema={
write_required=True "section_red": KNXSectionFlat(),
),
vol.Optional(CONF_GA_RED_SWITCH): GASelector( vol.Optional(CONF_GA_RED_SWITCH): GASelector(
write_required=False write_required=False, valid_dpt="1"
),
vol.Required(CONF_GA_RED_BRIGHTNESS): GASelector(
write_required=True, valid_dpt="5.001"
),
"section_green": KNXSectionFlat(),
vol.Optional(CONF_GA_GREEN_SWITCH): GASelector(
write_required=False, valid_dpt="1"
), ),
vol.Required(CONF_GA_GREEN_BRIGHTNESS): GASelector( vol.Required(CONF_GA_GREEN_BRIGHTNESS): GASelector(
write_required=True write_required=True, valid_dpt="5.001"
),
vol.Optional(CONF_GA_GREEN_SWITCH): GASelector(
write_required=False
), ),
"section_blue": KNXSectionFlat(),
vol.Required(CONF_GA_BLUE_BRIGHTNESS): GASelector( vol.Required(CONF_GA_BLUE_BRIGHTNESS): GASelector(
write_required=True write_required=True, valid_dpt="5.001"
), ),
vol.Optional(CONF_GA_BLUE_SWITCH): GASelector( vol.Optional(CONF_GA_BLUE_SWITCH): GASelector(
write_required=False write_required=False, valid_dpt="1"
), ),
"section_white": KNXSectionFlat(),
vol.Optional(CONF_GA_WHITE_BRIGHTNESS): GASelector( vol.Optional(CONF_GA_WHITE_BRIGHTNESS): GASelector(
write_required=True write_required=True, valid_dpt="5.001"
), ),
vol.Optional(CONF_GA_WHITE_SWITCH): GASelector( vol.Optional(CONF_GA_WHITE_SWITCH): GASelector(
write_required=False write_required=False, valid_dpt="1"
), ),
} },
), ),
vol.Schema( GroupSelectOption(
{ translation_key="hsv_addresses",
vol.Required(CONF_GA_HUE): GASelector(write_required=True), schema={
vol.Required(CONF_GA_HUE): GASelector(
write_required=True, valid_dpt="5.001"
),
vol.Required(CONF_GA_SATURATION): GASelector( vol.Required(CONF_GA_SATURATION): GASelector(
write_required=True write_required=True, valid_dpt="5.001"
), ),
} },
), ),
# msg="error in `color` config",
),
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
vol.Optional(CONF_COLOR_TEMP_MIN, default=2700): vol.All(
vol.Coerce(int), vol.Range(min=1)
),
vol.Optional(CONF_COLOR_TEMP_MAX, default=6000): vol.All(
vol.Coerce(int), vol.Range(min=1)
), ),
vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(),
} }
), ),
vol.Any( vol.Any(
@ -291,26 +307,22 @@ LIGHT_KNX_SCHEMA = vol.All(
), ),
) )
SWITCH_KNX_SCHEMA = vol.Schema(
LIGHT_SCHEMA = vol.Schema(
{ {
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA, "section_switch": KNXSectionFlat(),
vol.Required(DOMAIN): LIGHT_KNX_SCHEMA, vol.Required(CONF_GA_SWITCH): GASelector(write_required=True),
} vol.Optional(CONF_INVERT, default=False): selector.BooleanSelector(),
vol.Optional(CONF_RESPOND_TO_READ, default=False): selector.BooleanSelector(),
vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(),
},
) )
KNX_SCHEMA_FOR_PLATFORM = {
SWITCH_SCHEMA = vol.Schema( Platform.BINARY_SENSOR: BINARY_SENSOR_KNX_SCHEMA,
{ Platform.COVER: COVER_KNX_SCHEMA,
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA, Platform.LIGHT: LIGHT_KNX_SCHEMA,
vol.Required(DOMAIN): { Platform.SWITCH: SWITCH_KNX_SCHEMA,
vol.Optional(CONF_INVERT, default=False): bool, }
vol.Required(CONF_GA_SWITCH): GASelector(write_required=True),
vol.Optional(CONF_RESPOND_TO_READ, default=False): bool,
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
},
}
)
ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All( ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All(
vol.Schema( vol.Schema(
@ -326,18 +338,16 @@ ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All(
cv.key_value_schemas( cv.key_value_schemas(
CONF_PLATFORM, CONF_PLATFORM,
{ {
Platform.BINARY_SENSOR: vol.Schema( platform: vol.Schema(
{vol.Required(CONF_DATA): BINARY_SENSOR_SCHEMA}, extra=vol.ALLOW_EXTRA {
), vol.Required(CONF_DATA): {
Platform.COVER: vol.Schema( vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA,
{vol.Required(CONF_DATA): COVER_SCHEMA}, extra=vol.ALLOW_EXTRA vol.Required(DOMAIN): knx_schema,
), },
Platform.LIGHT: vol.Schema( },
{vol.Required(CONF_DATA): LIGHT_SCHEMA}, extra=vol.ALLOW_EXTRA extra=vol.ALLOW_EXTRA,
), )
Platform.SWITCH: vol.Schema( for platform, knx_schema in KNX_SCHEMA_FOR_PLATFORM.items()
{vol.Required(CONF_DATA): SWITCH_SCHEMA}, extra=vol.ALLOW_EXTRA
),
}, },
), ),
) )

View File

@ -6,11 +6,102 @@ from typing import Any
import voluptuous as vol import voluptuous as vol
from ..validation import ga_validator, maybe_ga_validator 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 .const import CONF_DPT, CONF_GA_PASSIVE, CONF_GA_STATE, CONF_GA_WRITE
class GroupSelect(vol.Any): class AllSerializeFirst(vol.All):
"""Use the first validated value for serialization.
This is a version of vol.All with custom error handling to
show proper invalid markers for sub-schema items in the UI.
"""
class KNXSelectorBase:
"""Base class for KNX selectors supporting optional nested schemas."""
schema: vol.Schema | vol.Any | vol.All
selector_type: str
# mark if self.schema should be serialized to `schema` key
serialize_subschema: bool = False
def __call__(self, data: Any) -> Any:
"""Validate the passed data."""
return self.schema(data)
def serialize(self) -> dict[str, Any]:
"""Serialize the selector to a dictionary."""
# don't use "name", "default", "optional" or "required" in base output
# as it will be overwritten by the parent keys attributes
# "schema" will be overwritten by knx serializer if `self.serialize_subschema` is True
raise NotImplementedError("Subclasses must implement this method.")
class KNXSectionFlat(KNXSelectorBase):
"""Generate a schema-neutral section with title and description for the following siblings."""
selector_type = "knx_section_flat"
schema = vol.Schema(None)
def __init__(
self,
collapsible: bool = False,
) -> None:
"""Initialize the section."""
self.collapsible = collapsible
def serialize(self) -> dict[str, Any]:
"""Serialize the selector to a dictionary."""
return {
"type": self.selector_type,
"collapsible": self.collapsible,
}
class KNXSection(KNXSelectorBase):
"""Configuration groups similar to DataEntryFlow sections but with more options."""
selector_type = "knx_section"
serialize_subschema = True
def __init__(
self,
schema: dict[str | vol.Marker, vol.Schemable],
collapsible: bool = True,
) -> None:
"""Initialize the section."""
self.collapsible = collapsible
self.schema = vol.Schema(schema)
def serialize(self) -> dict[str, Any]:
"""Serialize the section to a dictionary."""
return {
"type": self.selector_type,
"collapsible": self.collapsible,
}
class GroupSelectOption(KNXSelectorBase):
"""Schema for group select options."""
selector_type = "knx_group_select_option"
serialize_subschema: bool = True
def __init__(self, schema: vol.Schemable, translation_key: str) -> None:
"""Initialize the group select option schema."""
self.translation_key = translation_key
self.schema = vol.Schema(schema)
def serialize(self) -> dict[str, Any]:
"""Serialize the group select option to a dictionary."""
return {
"type": self.selector_type,
"translation_key": self.translation_key,
}
class GroupSelectSchema(vol.Any):
"""Use the first validated value. """Use the first validated value.
This is a version of vol.Any with custom error handling to This is a version of vol.Any with custom error handling to
@ -35,10 +126,33 @@ class GroupSelect(vol.Any):
raise vol.AnyInvalid(self.msg or "no valid value found", path=path) raise vol.AnyInvalid(self.msg or "no valid value found", path=path)
class GASelector: class GroupSelect(KNXSelectorBase):
"""Selector for group select options."""
selector_type = "knx_group_select"
serialize_subschema = True
def __init__(
self,
*options: GroupSelectOption,
collapsible: bool = True,
) -> None:
"""Initialize the group select selector."""
self.collapsible = collapsible
self.schema = GroupSelectSchema(*options)
def serialize(self) -> dict[str, Any]:
"""Serialize the group select to a dictionary."""
return {
"type": self.selector_type,
"collapsible": self.collapsible,
}
class GASelector(KNXSelectorBase):
"""Selector for a KNX group address structure.""" """Selector for a KNX group address structure."""
schema: vol.Schema selector_type = "knx_group_address"
def __init__( def __init__(
self, self,
@ -48,6 +162,7 @@ class GASelector:
write_required: bool = False, write_required: bool = False,
state_required: bool = False, state_required: bool = False,
dpt: type[Enum] | None = None, dpt: type[Enum] | None = None,
valid_dpt: str | Iterable[str] | None = None,
) -> None: ) -> None:
"""Initialize the group address selector.""" """Initialize the group address selector."""
self.write = write self.write = write
@ -56,12 +171,43 @@ class GASelector:
self.write_required = write_required self.write_required = write_required
self.state_required = state_required self.state_required = state_required
self.dpt = dpt self.dpt = dpt
# valid_dpt is used in frontend to filter dropdown menu - no validation is done
self.valid_dpt = (valid_dpt,) if isinstance(valid_dpt, str) else valid_dpt
self.schema = self.build_schema() self.schema = self.build_schema()
def __call__(self, data: Any) -> Any: def serialize(self) -> dict[str, Any]:
"""Validate the passed data.""" """Serialize the selector to a dictionary."""
return self.schema(data)
def dpt_to_dict(dpt: str) -> dict[str, int | None]:
"""Convert a DPT string to a dictionary."""
dpt_num = dpt.split(".")
return {
"main": int(dpt_num[0]),
"sub": int(dpt_num[1]) if len(dpt_num) > 1 else None,
}
options: dict[str, Any] = {
"write": {"required": self.write_required} if self.write else False,
"state": {"required": self.state_required} if self.state else False,
"passive": self.passive,
}
if self.dpt is not None:
options["dptSelect"] = [
{
"value": item.value,
"translation_key": item.value.replace(".", "_"),
"dpt": dpt_to_dict(item.value), # used to filter DPTs in dropdown
}
for item in self.dpt
]
if self.valid_dpt is not None:
options["validDPTs"] = [dpt_to_dict(dpt) for dpt in self.valid_dpt]
return {
"type": self.selector_type,
"options": options,
}
def build_schema(self) -> vol.Schema: def build_schema(self) -> vol.Schema:
"""Create the schema based on configuration.""" """Create the schema based on configuration."""
@ -118,3 +264,27 @@ class GASelector:
schema[vol.Required(CONF_DPT)] = vol.In({item.value for item in self.dpt}) schema[vol.Required(CONF_DPT)] = vol.In({item.value for item in self.dpt})
else: else:
schema[vol.Remove(CONF_DPT)] = object schema[vol.Remove(CONF_DPT)] = object
class SyncStateSelector(KNXSelectorBase):
"""Selector for knx sync state validation."""
schema = vol.Schema(sync_state_validator)
selector_type = "knx_sync_state"
def __init__(self, allow_false: bool = False) -> None:
"""Initialize the sync state validator."""
self.allow_false = allow_false
def serialize(self) -> dict[str, Any]:
"""Serialize the selector to a dictionary."""
return {
"type": self.selector_type,
"allow_false": self.allow_false,
}
def __call__(self, data: Any) -> Any:
"""Validate the passed data."""
if not self.allow_false and not data:
raise vol.Invalid(f"Sync state cannot be {data}")
return self.schema(data)

View File

@ -0,0 +1,42 @@
"""Custom serializer for KNX schemas."""
from typing import Any
import voluptuous as vol
from voluptuous_serialize import UNSUPPORTED, convert
from homeassistant.const import Platform
from homeassistant.helpers import selector
from .entity_store_schema import KNX_SCHEMA_FOR_PLATFORM
from .knx_selector import AllSerializeFirst, GroupSelectSchema, KNXSelectorBase
def knx_serializer(
schema: vol.Schema,
) -> dict[str, Any] | list[dict[str, Any]]:
"""Serialize KNX schema."""
if isinstance(schema, GroupSelectSchema):
return [
convert(option, custom_serializer=knx_serializer)
for option in schema.validators
]
if isinstance(schema, KNXSelectorBase):
result = schema.serialize()
if schema.serialize_subschema:
result["schema"] = convert(schema.schema, custom_serializer=knx_serializer)
return result
if isinstance(schema, AllSerializeFirst):
return convert(schema.validators[0], custom_serializer=knx_serializer) # type: ignore[no-any-return]
if isinstance(schema, selector.Selector):
return schema.serialize() | {"type": "ha_selector"}
return UNSUPPORTED # type: ignore[no-any-return]
def get_serialized_schema(platform: Platform) -> dict | list | None:
"""Get the schema for a specific platform."""
if knx_schema := KNX_SCHEMA_FOR_PLATFORM.get(platform):
return convert(knx_schema, custom_serializer=knx_serializer) # type: ignore[no-any-return]
return None

View File

@ -339,5 +339,275 @@
"name": "[%key:common::action::reload%]", "name": "[%key:common::action::reload%]",
"description": "Reloads the KNX integration." "description": "Reloads the KNX integration."
} }
},
"config_panel": {
"entities": {
"create": {
"type_selection": {
"title": "Select entity type",
"header": "Create KNX entity"
},
"header": "Create new entity",
"_": {
"entity": {
"title": "Entity configuration",
"description": "Home Assistant specific settings.",
"name_title": "Device and entity name",
"name_description": "Define how the entity should be named in Home Assistant.",
"device_description": "A device allows to group multiple entities. Select the device this entity belongs to or create a new one.",
"entity_label": "Entity name",
"entity_description": "Optional if a device is selected, otherwise required. If the entity is assigned to a device, the device name is used as prefix.",
"entity_category_title": "Entity category",
"entity_category_description": "Classification of a non-primary entity. Leave empty for standard behaviour."
},
"knx": {
"title": "KNX configuration",
"knx_group_address": {
"dpt": "Datapoint type",
"send_address": "Send address",
"state_address": "State address",
"passive_addresses": "Passive addresses",
"valid_dpts": "Valid DPTs"
},
"sync_state": {
"title": "State updater",
"description": "Actively request state updates from KNX bus for state addresses.",
"strategy": "Strategy",
"options": {
"true": "Use integration default",
"false": "Never",
"init": "Once when connection established",
"expire": "Expire after last value update",
"every": "Scheduled every"
}
}
}
},
"binary_sensor": {
"description": "Read-only entity for binary datapoints. Window or door states etc.",
"knx": {
"section_binary_sensor": {
"title": "Binary sensor",
"description": "DPT 1 group addresses representing binary states."
},
"invert": {
"label": "Invert",
"description": "Invert payload before processing."
},
"section_advanced_options": {
"title": "State properties",
"description": "Properties of the binary sensor state."
},
"ignore_internal_state": {
"label": "Force update",
"description": "Write each update to the state machine, even if the data is the same."
},
"context_timeout": {
"label": "Context timeout",
"description": "The time in seconds between multiple identical telegram payloads would count towards an internal counter. This can be used to automate on mulit-clicks of a button. `0` to disable this feature."
},
"reset_after": {
"label": "Reset after",
"description": "Reset back to “off” state after specified seconds."
}
}
},
"cover": {
"description": "The KNX cover platform is used as an interface to shutter actuators.",
"knx": {
"section_binary_control": {
"title": "Open/Close control",
"description": "DPT 1 group addresses triggering full movement."
},
"ga_up_down": {
"label": "Open/Close"
},
"invert_updown": {
"label": "Invert",
"description": "Default is UP (0) to open a cover and DOWN (1) to close a cover. Enable this to invert the open/close commands from/to your KNX actuator."
},
"section_stop_control": {
"title": "Stop",
"description": "DPT 1 group addresses for stopping movement."
},
"ga_stop": {
"label": "Stop"
},
"ga_step": {
"label": "Stepwise move"
},
"section_position_control": {
"title": "Position",
"description": "DPT 5 group addresses for cover position."
},
"ga_position_set": {
"label": "Set position"
},
"ga_position_state": {
"label": "Current Position"
},
"invert_position": {
"label": "Invert",
"description": "Invert payload before processing. Enable if KNX reports 0% as fully closed."
},
"section_tilt_control": {
"title": "Tilt",
"description": "DPT 5 group addresses for slat tilt angle."
},
"ga_angle": {
"label": "Tilt angle"
},
"invert_angle": {
"label": "[%key:component::knx::config_panel::entities::create::cover::knx::invert_position::label%]",
"description": "[%key:component::knx::config_panel::entities::create::cover::knx::invert_position::description%]"
},
"section_travel_time": {
"title": "Travel time",
"description": "Used to calculate intermediate positions of the cover while traveling."
},
"travelling_time_up": {
"label": "Travel time for opening",
"description": "Time the cover needs to fully open in seconds."
},
"travelling_time_down": {
"label": "Travel time for closing",
"description": "Time the cover needs to fully close in seconds."
}
}
},
"light": {
"description": "The KNX light platform is used as an interface to dimming actuators, LED controllers, DALI gateways and similar.",
"knx": {
"section_switch": {
"title": "Switch",
"description": "Turn the light on/off."
},
"ga_switch": {
"label": "Switch"
},
"section_brightness": {
"title": "Brightness",
"description": "Control the brightness of the light."
},
"ga_brightness": {
"label": "Brightness"
},
"section_color_temp": {
"title": "Color temperature",
"description": "Control the color temperature of the light."
},
"ga_color_temp": {
"label": "Color temperature",
"options": {
"5_001": "Percent",
"7_600": "Kelvin",
"9": "2-byte floating point"
}
},
"color_temp_min": {
"label": "Warmest possible color temperature"
},
"color_temp_max": {
"label": "Coldest possible color temperature"
},
"color": {
"title": "Color",
"description": "Control the color of the light.",
"options": {
"single_address": {
"label": "Single address",
"description": "RGB, RGBW or XYY color controlled by a single group address."
},
"individual_addresses": {
"label": "Individual addresses",
"description": "RGB(W) using individual state and brightness group addresses."
},
"hsv_addresses": {
"label": "HSV",
"description": "Hue, saturation and brightness using individual group addresses."
}
},
"ga_color": {
"label": "Color",
"options": {
"232_600": "RGB",
"242_600": "XYY",
"251_600": "RGBW"
}
},
"section_red": {
"title": "Red",
"description": "Control the lights red color. Brightness group address is required."
},
"ga_red_switch": {
"label": "Red switch"
},
"ga_red_brightness": {
"label": "Red brightness"
},
"section_green": {
"title": "Green",
"description": "Control the lights green color. Brightness group address is required."
},
"ga_green_switch": {
"label": "Green switch"
},
"ga_green_brightness": {
"label": "Green brightness"
},
"section_blue": {
"title": "Blue",
"description": "Control the lights blue color. Brightness group address is required."
},
"ga_blue_switch": {
"label": "Blue switch"
},
"ga_blue_brightness": {
"label": "Blue brightness"
},
"section_white": {
"title": "White",
"description": "Control the lights white color. Brightness group address is required."
},
"ga_white_switch": {
"label": "White switch"
},
"ga_white_brightness": {
"label": "White brightness"
},
"ga_hue": {
"label": "Hue",
"description": "Control the lights hue."
},
"ga_saturation": {
"label": "Saturation",
"description": "Control the lights saturation."
}
}
}
},
"switch": {
"description": "The KNX switch platform is used as an interface to switching actuators.",
"knx": {
"section_switch": {
"title": "Switching",
"description": "DPT 1 group addresses controlling the switch function."
},
"ga_switch": {
"label": "Switch",
"description": "Group address to switch the device on/off."
},
"invert": {
"label": "Invert",
"description": "Invert payloads before processing or sending."
},
"respond_to_read": {
"label": "Respond to read",
"description": "Respond to GroupValueRead telegrams received to the configured send address."
}
}
}
}
}
} }
} }

View File

@ -14,7 +14,7 @@ from xknxproject.exceptions import XknxProjectException
from homeassistant.components import panel_custom, websocket_api from homeassistant.components import panel_custom, websocket_api
from homeassistant.components.http import StaticPathConfig from homeassistant.components.http import StaticPathConfig
from homeassistant.const import CONF_ENTITY_ID, CONF_PLATFORM from homeassistant.const import CONF_ENTITY_ID, CONF_PLATFORM, Platform
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.dispatcher import async_dispatcher_connect
@ -33,6 +33,7 @@ from .storage.entity_store_validation import (
EntityStoreValidationSuccess, EntityStoreValidationSuccess,
validate_entity_data, validate_entity_data,
) )
from .storage.serialize import get_serialized_schema
from .telegrams import SIGNAL_KNX_TELEGRAM, TelegramDict from .telegrams import SIGNAL_KNX_TELEGRAM, TelegramDict
if TYPE_CHECKING: if TYPE_CHECKING:
@ -57,6 +58,7 @@ async def register_panel(hass: HomeAssistant) -> None:
websocket_api.async_register_command(hass, ws_get_entity_config) websocket_api.async_register_command(hass, ws_get_entity_config)
websocket_api.async_register_command(hass, ws_get_entity_entries) websocket_api.async_register_command(hass, ws_get_entity_entries)
websocket_api.async_register_command(hass, ws_create_device) websocket_api.async_register_command(hass, ws_create_device)
websocket_api.async_register_command(hass, ws_get_schema)
if DOMAIN not in hass.data.get("frontend_panels", {}): if DOMAIN not in hass.data.get("frontend_panels", {}):
await hass.http.async_register_static_paths( await hass.http.async_register_static_paths(
@ -363,6 +365,28 @@ def ws_validate_entity(
) )
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required("type"): "knx/get_schema",
vol.Required(CONF_PLATFORM): vol.Coerce(Platform),
}
)
@websocket_api.async_response
async def ws_get_schema(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict,
) -> None:
"""Provide serialized schema for platform."""
if schema := get_serialized_schema(msg[CONF_PLATFORM]):
connection.send_result(msg["id"], schema)
return
connection.send_error(
msg["id"], websocket_api.const.ERR_HOME_ASSISTANT_ERROR, "Unknown platform"
)
@websocket_api.require_admin @websocket_api.require_admin
@websocket_api.websocket_command( @websocket_api.websocket_command(
{ {

View File

@ -0,0 +1,756 @@
# serializer version: 1
# name: test_knx_get_schema[binary_sensor]
dict({
'id': 1,
'result': list([
dict({
'collapsible': False,
'name': 'section_binary_sensor',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_sensor',
'options': dict({
'passive': True,
'state': dict({
'required': True,
}),
'validDPTs': list([
dict({
'main': 1,
'sub': None,
}),
]),
'write': False,
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'name': 'invert',
'optional': True,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'collapsible': True,
'name': 'section_advanced_options',
'type': 'knx_section_flat',
}),
dict({
'name': 'ignore_internal_state',
'optional': True,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'name': 'context_timeout',
'optional': True,
'selector': dict({
'number': dict({
'max': 10.0,
'min': 0.0,
'mode': 'slider',
'step': 0.1,
'unit_of_measurement': 's',
}),
}),
'type': 'ha_selector',
}),
dict({
'name': 'reset_after',
'optional': True,
'selector': dict({
'number': dict({
'max': 600.0,
'min': 0.0,
'mode': 'slider',
'step': 0.1,
'unit_of_measurement': 's',
}),
}),
'type': 'ha_selector',
}),
dict({
'allow_false': False,
'default': True,
'name': 'sync_state',
'required': True,
'type': 'knx_sync_state',
}),
]),
'success': True,
'type': 'result',
})
# ---
# name: test_knx_get_schema[cover]
dict({
'id': 1,
'result': list([
dict({
'collapsible': False,
'name': 'section_binary_control',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_up_down',
'optional': True,
'options': dict({
'passive': True,
'state': False,
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'name': 'invert_updown',
'optional': True,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'collapsible': False,
'name': 'section_stop_control',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_stop',
'optional': True,
'options': dict({
'passive': True,
'state': False,
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'name': 'ga_step',
'optional': True,
'options': dict({
'passive': True,
'state': False,
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'collapsible': True,
'name': 'section_position_control',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_position_set',
'optional': True,
'options': dict({
'passive': True,
'state': False,
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'name': 'ga_position_state',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'write': False,
}),
'type': 'knx_group_address',
}),
dict({
'name': 'invert_position',
'optional': True,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'collapsible': True,
'name': 'section_tilt_control',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_angle',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'name': 'invert_angle',
'optional': True,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'collapsible': False,
'name': 'section_travel_time',
'type': 'knx_section_flat',
}),
dict({
'default': 25,
'name': 'travelling_time_up',
'optional': True,
'selector': dict({
'number': dict({
'max': 1000.0,
'min': 0.0,
'mode': 'slider',
'step': 0.1,
'unit_of_measurement': 's',
}),
}),
'type': 'ha_selector',
}),
dict({
'default': 25,
'name': 'travelling_time_down',
'optional': True,
'selector': dict({
'number': dict({
'max': 1000.0,
'min': 0.0,
'mode': 'slider',
'step': 0.1,
'unit_of_measurement': 's',
}),
}),
'type': 'ha_selector',
}),
dict({
'allow_false': False,
'default': True,
'name': 'sync_state',
'optional': True,
'type': 'knx_sync_state',
}),
]),
'success': True,
'type': 'result',
})
# ---
# name: test_knx_get_schema[light]
dict({
'id': 1,
'result': list([
dict({
'collapsible': False,
'name': 'section_switch',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_switch',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 1,
'sub': None,
}),
]),
'write': dict({
'required': True,
}),
}),
'type': 'knx_group_address',
}),
dict({
'collapsible': False,
'name': 'section_brightness',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_brightness',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 5,
'sub': 1,
}),
]),
'write': dict({
'required': True,
}),
}),
'type': 'knx_group_address',
}),
dict({
'collapsible': True,
'name': 'section_color_temp',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_color_temp',
'optional': True,
'options': dict({
'dptSelect': list([
dict({
'dpt': dict({
'main': 7,
'sub': 600,
}),
'translation_key': '7_600',
'value': '7.600',
}),
dict({
'dpt': dict({
'main': 9,
'sub': None,
}),
'translation_key': '9',
'value': '9',
}),
dict({
'dpt': dict({
'main': 5,
'sub': 1,
}),
'translation_key': '5_001',
'value': '5.001',
}),
]),
'passive': True,
'state': dict({
'required': False,
}),
'write': dict({
'required': True,
}),
}),
'type': 'knx_group_address',
}),
dict({
'default': 2700,
'name': 'color_temp_min',
'required': True,
'selector': dict({
'number': dict({
'max': 10000.0,
'min': 1.0,
'mode': 'slider',
'step': 1.0,
'unit_of_measurement': 'K',
}),
}),
'type': 'ha_selector',
}),
dict({
'default': 6000,
'name': 'color_temp_max',
'required': True,
'selector': dict({
'number': dict({
'max': 10000.0,
'min': 1.0,
'mode': 'slider',
'step': 1.0,
'unit_of_measurement': 'K',
}),
}),
'type': 'ha_selector',
}),
dict({
'collapsible': True,
'name': 'color',
'optional': True,
'schema': list([
dict({
'schema': list([
dict({
'name': 'ga_color',
'optional': True,
'options': dict({
'dptSelect': list([
dict({
'dpt': dict({
'main': 232,
'sub': 600,
}),
'translation_key': '232_600',
'value': '232.600',
}),
dict({
'dpt': dict({
'main': 251,
'sub': 600,
}),
'translation_key': '251_600',
'value': '251.600',
}),
dict({
'dpt': dict({
'main': 242,
'sub': 600,
}),
'translation_key': '242_600',
'value': '242.600',
}),
]),
'passive': True,
'state': dict({
'required': False,
}),
'write': dict({
'required': True,
}),
}),
'type': 'knx_group_address',
}),
]),
'translation_key': 'single_address',
'type': 'knx_group_select_option',
}),
dict({
'schema': list([
dict({
'collapsible': False,
'name': 'section_red',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_red_switch',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 1,
'sub': None,
}),
]),
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'name': 'ga_red_brightness',
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 5,
'sub': 1,
}),
]),
'write': dict({
'required': True,
}),
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'collapsible': False,
'name': 'section_green',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_green_switch',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 1,
'sub': None,
}),
]),
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'name': 'ga_green_brightness',
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 5,
'sub': 1,
}),
]),
'write': dict({
'required': True,
}),
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'collapsible': False,
'name': 'section_blue',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_blue_brightness',
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 5,
'sub': 1,
}),
]),
'write': dict({
'required': True,
}),
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'name': 'ga_blue_switch',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 1,
'sub': None,
}),
]),
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
dict({
'collapsible': False,
'name': 'section_white',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_white_brightness',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 5,
'sub': 1,
}),
]),
'write': dict({
'required': True,
}),
}),
'type': 'knx_group_address',
}),
dict({
'name': 'ga_white_switch',
'optional': True,
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 1,
'sub': None,
}),
]),
'write': dict({
'required': False,
}),
}),
'type': 'knx_group_address',
}),
]),
'translation_key': 'individual_addresses',
'type': 'knx_group_select_option',
}),
dict({
'schema': list([
dict({
'name': 'ga_hue',
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 5,
'sub': 1,
}),
]),
'write': dict({
'required': True,
}),
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'name': 'ga_saturation',
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'validDPTs': list([
dict({
'main': 5,
'sub': 1,
}),
]),
'write': dict({
'required': True,
}),
}),
'required': True,
'type': 'knx_group_address',
}),
]),
'translation_key': 'hsv_addresses',
'type': 'knx_group_select_option',
}),
]),
'type': 'knx_group_select',
}),
dict({
'allow_false': False,
'default': True,
'name': 'sync_state',
'optional': True,
'type': 'knx_sync_state',
}),
]),
'success': True,
'type': 'result',
})
# ---
# name: test_knx_get_schema[switch]
dict({
'id': 1,
'result': list([
dict({
'collapsible': False,
'name': 'section_switch',
'type': 'knx_section_flat',
}),
dict({
'name': 'ga_switch',
'options': dict({
'passive': True,
'state': dict({
'required': False,
}),
'write': dict({
'required': True,
}),
}),
'required': True,
'type': 'knx_group_address',
}),
dict({
'default': False,
'name': 'invert',
'optional': True,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'default': False,
'name': 'respond_to_read',
'optional': True,
'selector': dict({
'boolean': dict({
}),
}),
'type': 'ha_selector',
}),
dict({
'allow_false': False,
'default': True,
'name': 'sync_state',
'optional': True,
'type': 'knx_sync_state',
}),
]),
'success': True,
'type': 'result',
})
# ---
# name: test_knx_get_schema[tts]
dict({
'error': dict({
'code': 'home_assistant_error',
'message': 'Unknown platform',
}),
'id': 1,
'success': False,
'type': 'result',
})
# ---

View File

@ -4,9 +4,20 @@ from typing import Any
import pytest import pytest
import voluptuous as vol import voluptuous as vol
from voluptuous_serialize import convert
from homeassistant.components.knx.const import ColorTempModes from homeassistant.components.knx.const import ColorTempModes
from homeassistant.components.knx.storage.knx_selector import GASelector from homeassistant.components.knx.storage.knx_selector import (
AllSerializeFirst,
GASelector,
GroupSelect,
GroupSelectOption,
KNXSection,
KNXSectionFlat,
SyncStateSelector,
)
from homeassistant.components.knx.storage.serialize import knx_serializer
from homeassistant.helpers import selector
INVALID = "invalid" INVALID = "invalid"
@ -14,48 +25,6 @@ INVALID = "invalid"
@pytest.mark.parametrize( @pytest.mark.parametrize(
("selector_config", "data", "expected"), ("selector_config", "data", "expected"),
[ [
# empty data is invalid
(
{},
{},
{INVALID: "At least one group address must be set"},
),
(
{"write": False},
{},
{INVALID: "At least one group address must be set"},
),
(
{"passive": False},
{},
{INVALID: "At least one group address must be set"},
),
(
{"write": False, "state": False, "passive": False},
{},
{INVALID: "At least one group address must be set"},
),
# stale data is invalid
(
{"write": False},
{"write": "1/2/3"},
{INVALID: "At least one group address must be set"},
),
(
{"write": False},
{"passive": []},
{INVALID: "At least one group address must be set"},
),
(
{"state": False},
{"write": None},
{INVALID: "At least one group address must be set"},
),
(
{"passive": False},
{"passive": ["1/2/3"]},
{INVALID: "At least one group address must be set"},
),
# valid data # valid data
( (
{}, {},
@ -93,16 +62,6 @@ INVALID = "invalid"
{"write": "1/2/3", "state": None}, {"write": "1/2/3", "state": None},
), ),
# required keys # required keys
(
{"write_required": True},
{},
{INVALID: r"required key not provided*"},
),
(
{"state_required": True},
{},
{INVALID: r"required key not provided*"},
),
( (
{"write_required": True}, {"write_required": True},
{"write": "1/2/3"}, {"write": "1/2/3"},
@ -113,32 +72,12 @@ INVALID = "invalid"
{"state": "1/2/3"}, {"state": "1/2/3"},
{"write": None, "state": "1/2/3", "passive": []}, {"write": None, "state": "1/2/3", "passive": []},
), ),
(
{"write_required": True},
{"state": "1/2/3"},
{INVALID: r"required key not provided*"},
),
(
{"state_required": True},
{"write": "1/2/3"},
{INVALID: r"required key not provided*"},
),
# dpt key # dpt key
(
{"dpt": ColorTempModes},
{"write": "1/2/3"},
{INVALID: r"required key not provided*"},
),
( (
{"dpt": ColorTempModes}, {"dpt": ColorTempModes},
{"write": "1/2/3", "dpt": "7.600"}, {"write": "1/2/3", "dpt": "7.600"},
{"write": "1/2/3", "state": None, "passive": [], "dpt": "7.600"}, {"write": "1/2/3", "state": None, "passive": [], "dpt": "7.600"},
), ),
(
{"dpt": ColorTempModes},
{"write": "1/2/3", "state": None, "passive": [], "dpt": "invalid"},
{INVALID: r"value must be one of ['5.001', '7.600', '9']*"},
),
], ],
) )
def test_ga_selector( def test_ga_selector(
@ -148,9 +87,240 @@ def test_ga_selector(
) -> None: ) -> None:
"""Test GASelector.""" """Test GASelector."""
selector = GASelector(**selector_config) selector = GASelector(**selector_config)
if INVALID in expected: result = selector(data)
with pytest.raises(vol.Invalid, match=expected[INVALID]): assert result == expected
selector(data)
else:
result = selector(data) @pytest.mark.parametrize(
assert result == expected ("selector_config", "data", "error_str"),
[
# empty data is invalid
(
{},
{},
"At least one group address must be set",
),
(
{"write": False},
{},
"At least one group address must be set",
),
(
{"passive": False},
{},
"At least one group address must be set",
),
(
{"write": False, "state": False, "passive": False},
{},
"At least one group address must be set",
),
# stale data is invalid
(
{"write": False},
{"write": "1/2/3"},
"At least one group address must be set",
),
(
{"write": False},
{"passive": []},
"At least one group address must be set",
),
(
{"state": False},
{"write": None},
"At least one group address must be set",
),
(
{"passive": False},
{"passive": ["1/2/3"]},
"At least one group address must be set",
),
# required keys
(
{"write_required": True},
{},
r"required key not provided*",
),
(
{"state_required": True},
{},
r"required key not provided*",
),
(
{"write_required": True},
{"state": "1/2/3"},
r"required key not provided*",
),
(
{"state_required": True},
{"write": "1/2/3"},
r"required key not provided*",
),
# dpt key
(
{"dpt": ColorTempModes},
{"write": "1/2/3"},
r"required key not provided*",
),
(
{"dpt": ColorTempModes},
{"write": "1/2/3", "state": None, "passive": [], "dpt": "invalid"},
r"value must be one of ['5.001', '7.600', '9']*",
),
],
)
def test_ga_selector_invalid(
selector_config: dict[str, Any],
data: dict[str, Any],
error_str: str,
) -> None:
"""Test GASelector."""
selector = GASelector(**selector_config)
with pytest.raises(vol.Invalid, match=error_str):
selector(data)
def test_sync_state_selector() -> None:
"""Test SyncStateSelector."""
selector = SyncStateSelector()
assert selector("expire 50") == "expire 50"
with pytest.raises(vol.Invalid):
selector("invalid")
with pytest.raises(vol.Invalid, match="Sync state cannot be False"):
selector(False)
false_allowed = SyncStateSelector(allow_false=True)
assert false_allowed(False) is False
@pytest.mark.parametrize(
("selector", "serialized"),
[
(
GASelector(),
{
"type": "knx_group_address",
"options": {
"write": {"required": False},
"state": {"required": False},
"passive": True,
},
},
),
(
GASelector(
state=False, write_required=True, passive=False, valid_dpt="5.001"
),
{
"type": "knx_group_address",
"options": {
"write": {"required": True},
"state": False,
"passive": False,
"validDPTs": [{"main": 5, "sub": 1}],
},
},
),
(
GASelector(dpt=ColorTempModes),
{
"type": "knx_group_address",
"options": {
"write": {"required": False},
"state": {"required": False},
"passive": True,
"dptSelect": [
{
"value": "7.600",
"translation_key": "7_600",
"dpt": {"main": 7, "sub": 600},
},
{
"value": "9",
"translation_key": "9",
"dpt": {"main": 9, "sub": None},
},
{
"value": "5.001",
"translation_key": "5_001",
"dpt": {"main": 5, "sub": 1},
},
],
},
},
),
],
)
def test_ga_selector_serialization(
selector: GASelector, serialized: dict[str, Any]
) -> None:
"""Test GASelector serialization."""
assert selector.serialize() == serialized
@pytest.mark.parametrize(
("schema", "serialized"),
[
(
AllSerializeFirst(vol.Schema({"key": int}), vol.Schema({"ignored": str})),
[{"name": "key", "type": "integer"}],
),
(
KNXSectionFlat(collapsible=True),
{"type": "knx_section_flat", "collapsible": True},
),
(
KNXSection(
collapsible=True,
schema={"key": int},
),
{
"type": "knx_section",
"collapsible": True,
"schema": [{"name": "key", "type": "integer"}],
},
),
(
GroupSelect(
GroupSelectOption(translation_key="option_1", schema={"key_1": str}),
GroupSelectOption(translation_key="option_2", schema={"key_2": int}),
),
{
"type": "knx_group_select",
"collapsible": True,
"schema": [
{
"type": "knx_group_select_option",
"translation_key": "option_1",
"schema": [{"name": "key_1", "type": "string"}],
},
{
"type": "knx_group_select_option",
"translation_key": "option_2",
"schema": [{"name": "key_2", "type": "integer"}],
},
],
},
),
(
SyncStateSelector(),
{
"type": "knx_sync_state",
"allow_false": False,
},
),
(
selector.BooleanSelector(),
{
"type": "ha_selector",
"selector": {"boolean": {}},
},
),
],
)
def test_serialization(schema: Any, serialized: dict[str, Any]) -> None:
"""Test serialization of the selector."""
assert convert(schema, custom_serializer=knx_serializer) == serialized

View File

@ -4,8 +4,13 @@ from typing import Any
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.knx.const import KNX_ADDRESS, KNX_MODULE_KEY from homeassistant.components.knx.const import (
KNX_ADDRESS,
KNX_MODULE_KEY,
SUPPORTED_PLATFORMS_UI,
)
from homeassistant.components.knx.project import STORAGE_KEY as KNX_PROJECT_STORAGE_KEY from homeassistant.components.knx.project import STORAGE_KEY as KNX_PROJECT_STORAGE_KEY
from homeassistant.components.knx.schema import SwitchSchema from homeassistant.components.knx.schema import SwitchSchema
from homeassistant.const import CONF_NAME from homeassistant.const import CONF_NAME
@ -390,6 +395,22 @@ async def test_knx_subscribe_telegrams_command_project(
assert res["event"]["timestamp"] is not None assert res["event"]["timestamp"] is not None
@pytest.mark.parametrize("platform", {*SUPPORTED_PLATFORMS_UI, "tts"})
async def test_knx_get_schema(
hass: HomeAssistant,
knx: KNXTestKit,
hass_ws_client: WebSocketGenerator,
snapshot: SnapshotAssertion,
platform: str,
) -> None:
"""Test knx/get_schema command returning proper schema data."""
await knx.setup_integration()
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "knx/get_schema", "platform": platform})
res = await client.receive_json()
assert res == snapshot
@pytest.mark.parametrize( @pytest.mark.parametrize(
"endpoint", "endpoint",
[ [