diff --git a/homeassistant/components/knx/config_flow.py b/homeassistant/components/knx/config_flow.py index 6e3da8ad523..c5b5885bd9b 100644 --- a/homeassistant/components/knx/config_flow.py +++ b/homeassistant/components/knx/config_flow.py @@ -57,7 +57,7 @@ from .const import ( KNXConfigEntryData, ) from .helpers.keyring import DEFAULT_KNX_KEYRING_FILENAME, save_uploaded_knxkeys_file -from .schema import ia_validator, ip_v4_validator +from .validation import ia_validator, ip_v4_validator CONF_KNX_GATEWAY: Final = "gateway" CONF_MAX_RATE_LIMIT: Final = 60 diff --git a/homeassistant/components/knx/schema.py b/homeassistant/components/knx/schema.py index c7bcd90538f..885fe4a177f 100644 --- a/homeassistant/components/knx/schema.py +++ b/homeassistant/components/knx/schema.py @@ -3,15 +3,12 @@ from __future__ import annotations from abc import ABC from collections import OrderedDict -from collections.abc import Callable -import ipaddress -from typing import Any, ClassVar, Final +from typing import ClassVar, Final import voluptuous as vol from xknx.devices.climate import SetpointShiftMode -from xknx.dpt import DPTBase, DPTNumeric, DPTString -from xknx.exceptions import ConversionError, CouldNotParseAddress, CouldNotParseTelegram -from xknx.telegram.address import IndividualAddress, parse_device_group_address +from xknx.dpt import DPTBase, DPTNumeric +from xknx.exceptions import ConversionError, CouldNotParseTelegram from homeassistant.components.binary_sensor import ( DEVICE_CLASSES_SCHEMA as BINARY_SENSOR_DEVICE_CLASSES_SCHEMA, @@ -57,83 +54,19 @@ from .const import ( PRESET_MODES, ColorTempModes, ) - -################## -# KNX VALIDATORS -################## - - -def dpt_subclass_validator(dpt_base_class: type[DPTBase]) -> Callable[[Any], str | int]: - """Validate that value is parsable as given sensor type.""" - - def dpt_value_validator(value: Any) -> str | int: - """Validate that value is parsable as sensor type.""" - if ( - isinstance(value, (str, int)) - and dpt_base_class.parse_transcoder(value) is not None - ): - return value - raise vol.Invalid( - f"type '{value}' is not a valid DPT identifier for" - f" {dpt_base_class.__name__}." - ) - - return dpt_value_validator - - -numeric_type_validator = dpt_subclass_validator(DPTNumeric) # type: ignore[type-abstract] -sensor_type_validator = dpt_subclass_validator(DPTBase) # type: ignore[type-abstract] -string_type_validator = dpt_subclass_validator(DPTString) - - -def ga_validator(value: Any) -> str | int: - """Validate that value is parsable as GroupAddress or InternalGroupAddress.""" - if isinstance(value, (str, int)): - try: - parse_device_group_address(value) - return value - except CouldNotParseAddress: - pass - raise vol.Invalid( - f"value '{value}' is not a valid KNX group address '
//'," - " '
/' or '' (eg.'1/2/3', '9/234', '123'), nor xknx internal" - " address 'i-'." - ) - - -ga_list_validator = vol.All( - cv.ensure_list, - [ga_validator], - vol.IsTrue("value must be a group address or a list containing group addresses"), -) - -ia_validator = vol.Any( - vol.All(str, str.strip, cv.matches_regex(IndividualAddress.ADDRESS_RE.pattern)), - vol.All(vol.Coerce(int), vol.Range(min=1, max=65535)), - msg=( - "value does not match pattern for KNX individual address" - " '..' (eg.'1.1.100')" - ), +from .validation import ( + ga_list_validator, + ga_validator, + numeric_type_validator, + sensor_type_validator, + string_type_validator, + sync_state_validator, ) -def ip_v4_validator(value: Any, multicast: bool | None = None) -> str: - """Validate that value is parsable as IPv4 address. - - Optionally check if address is in a reserved multicast block or is explicitly not. - """ - try: - address = ipaddress.IPv4Address(value) - except ipaddress.AddressValueError as ex: - raise vol.Invalid(f"value '{value}' is not a valid IPv4 address: {ex}") from ex - if multicast is not None and address.is_multicast != multicast: - raise vol.Invalid( - f"value '{value}' is not a valid IPv4" - f" {'multicast' if multicast else 'unicast'} address" - ) - return str(address) - - +################## +# KNX SUB VALIDATORS +################## def number_limit_sub_validator(entity_config: OrderedDict) -> OrderedDict: """Validate a number entity configurations dependent on configured value type.""" value_type = entity_config[CONF_TYPE] @@ -227,12 +160,6 @@ def select_options_sub_validator(entity_config: OrderedDict) -> OrderedDict: return entity_config -sync_state_validator = vol.Any( - vol.All(vol.Coerce(int), vol.Range(min=2, max=1440)), - cv.boolean, - cv.matches_regex(r"^(init|expire|every)( \d*)?$"), -) - ######### # EVENT ######### diff --git a/homeassistant/components/knx/validation.py b/homeassistant/components/knx/validation.py new file mode 100644 index 00000000000..c0ac93d19eb --- /dev/null +++ b/homeassistant/components/knx/validation.py @@ -0,0 +1,89 @@ +"""Validation helpers for KNX config schemas.""" +from collections.abc import Callable +import ipaddress +from typing import Any + +import voluptuous as vol +from xknx.dpt import DPTBase, DPTNumeric, DPTString +from xknx.exceptions import CouldNotParseAddress +from xknx.telegram.address import IndividualAddress, parse_device_group_address + +import homeassistant.helpers.config_validation as cv + + +def dpt_subclass_validator(dpt_base_class: type[DPTBase]) -> Callable[[Any], str | int]: + """Validate that value is parsable as given sensor type.""" + + def dpt_value_validator(value: Any) -> str | int: + """Validate that value is parsable as sensor type.""" + if ( + isinstance(value, (str, int)) + and dpt_base_class.parse_transcoder(value) is not None + ): + return value + raise vol.Invalid( + f"type '{value}' is not a valid DPT identifier for" + f" {dpt_base_class.__name__}." + ) + + return dpt_value_validator + + +numeric_type_validator = dpt_subclass_validator(DPTNumeric) # type: ignore[type-abstract] +sensor_type_validator = dpt_subclass_validator(DPTBase) # type: ignore[type-abstract] +string_type_validator = dpt_subclass_validator(DPTString) + + +def ga_validator(value: Any) -> str | int: + """Validate that value is parsable as GroupAddress or InternalGroupAddress.""" + if isinstance(value, (str, int)): + try: + parse_device_group_address(value) + return value + except CouldNotParseAddress as exc: + raise vol.Invalid( + f"'{value}' is not a valid KNX group address: {exc.message}" + ) from exc + raise vol.Invalid( + f"'{value}' is not a valid KNX group address: Invalid type '{type(value).__name__}'" + ) + + +ga_list_validator = vol.All( + cv.ensure_list, + [ga_validator], + vol.IsTrue("value must be a group address or a list containing group addresses"), +) + +ia_validator = vol.Any( + vol.All(str, str.strip, cv.matches_regex(IndividualAddress.ADDRESS_RE.pattern)), + vol.All(vol.Coerce(int), vol.Range(min=1, max=65535)), + msg=( + "value does not match pattern for KNX individual address" + " '..' (eg.'1.1.100')" + ), +) + + +def ip_v4_validator(value: Any, multicast: bool | None = None) -> str: + """Validate that value is parsable as IPv4 address. + + Optionally check if address is in a reserved multicast block or is explicitly not. + """ + try: + address = ipaddress.IPv4Address(value) + except ipaddress.AddressValueError as ex: + raise vol.Invalid(f"value '{value}' is not a valid IPv4 address: {ex}") from ex + if multicast is not None and address.is_multicast != multicast: + raise vol.Invalid( + f"value '{value}' is not a valid IPv4" + f" {'multicast' if multicast else 'unicast'} address" + ) + return str(address) + + +sync_state_validator = vol.Any( + vol.All(vol.Coerce(int), vol.Range(min=2, max=1440)), + cv.boolean, + cv.matches_regex(r"^(init|expire|every)( \d*)?$"), +)