Use GroupSelect class for conditional selectors

This commit is contained in:
farmio
2025-06-02 11:10:46 +02:00
parent c01f521199
commit 4ba0d54f63
11 changed files with 369 additions and 211 deletions

View File

@@ -39,7 +39,8 @@ from .const import (
KNX_MODULE_KEY,
)
from .entity import KnxUiEntity, KnxUiEntityPlatformController, KnxYamlEntity
from .storage.const import CONF_ENTITY, CONF_GA_PASSIVE, CONF_GA_SENSOR, CONF_GA_STATE
from .storage.const import CONF_ENTITY, CONF_GA_SENSOR
from .storage.util import ConfigExtractor
async def async_setup_entry(
@@ -146,17 +147,17 @@ class KnxUiBinarySensor(_KnxBinarySensor, KnxUiEntity):
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
)
knx_conf = ConfigExtractor(config[DOMAIN])
self._device = XknxBinarySensor(
xknx=knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
group_address_state=[
config[DOMAIN][CONF_GA_SENSOR][CONF_GA_STATE],
*config[DOMAIN][CONF_GA_SENSOR][CONF_GA_PASSIVE],
],
sync_state=config[DOMAIN][CONF_SYNC_STATE],
invert=config[DOMAIN].get(CONF_INVERT, False),
ignore_internal_state=config[DOMAIN].get(CONF_IGNORE_INTERNAL_STATE, False),
context_timeout=config[DOMAIN].get(CONF_CONTEXT_TIMEOUT),
reset_after=config[DOMAIN].get(CONF_RESET_AFTER),
group_address_state=knx_conf.get_state_and_passive(CONF_GA_SENSOR),
sync_state=knx_conf.get(CONF_SYNC_STATE),
invert=knx_conf.get(CONF_INVERT, default=False),
ignore_internal_state=knx_conf.get(
CONF_IGNORE_INTERNAL_STATE, default=False
),
context_timeout=knx_conf.get(CONF_CONTEXT_TIMEOUT),
reset_after=knx_conf.get(CONF_RESET_AFTER),
)
self._attr_force_update = self._device.ignore_internal_state

View File

@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Any, Literal
from typing import Any
from xknx import XKNX
from xknx.devices import Cover as XknxCover
@@ -35,15 +35,13 @@ from .schema import CoverSchema
from .storage.const import (
CONF_ENTITY,
CONF_GA_ANGLE,
CONF_GA_PASSIVE,
CONF_GA_POSITION_SET,
CONF_GA_POSITION_STATE,
CONF_GA_STATE,
CONF_GA_STEP,
CONF_GA_STOP,
CONF_GA_UP_DOWN,
CONF_GA_WRITE,
)
from .storage.util import ConfigExtractor
async def async_setup_entry(
@@ -230,38 +228,24 @@ class KnxYamlCover(_KnxCover, KnxYamlEntity):
def _create_ui_cover(xknx: XKNX, knx_config: ConfigType, name: str) -> XknxCover:
"""Return a KNX Light device to be used within XKNX."""
def get_address(
key: str, address_type: Literal["write", "state"] = CONF_GA_WRITE
) -> str | None:
"""Get a single group address for given key."""
return knx_config[key][address_type] if key in knx_config else None
def get_addresses(
key: str, address_type: Literal["write", "state"] = CONF_GA_STATE
) -> list[Any] | None:
"""Get group address including passive addresses as list."""
return (
[knx_config[key][address_type], *knx_config[key][CONF_GA_PASSIVE]]
if key in knx_config
else None
)
conf = ConfigExtractor(knx_config)
return XknxCover(
xknx=xknx,
name=name,
group_address_long=get_addresses(CONF_GA_UP_DOWN, CONF_GA_WRITE),
group_address_short=get_addresses(CONF_GA_STEP, CONF_GA_WRITE),
group_address_stop=get_addresses(CONF_GA_STOP, CONF_GA_WRITE),
group_address_position=get_addresses(CONF_GA_POSITION_SET, CONF_GA_WRITE),
group_address_position_state=get_addresses(CONF_GA_POSITION_STATE),
group_address_angle=get_address(CONF_GA_ANGLE),
group_address_angle_state=get_addresses(CONF_GA_ANGLE),
travel_time_down=knx_config[CoverConf.TRAVELLING_TIME_DOWN],
travel_time_up=knx_config[CoverConf.TRAVELLING_TIME_UP],
invert_updown=knx_config.get(CoverConf.INVERT_UPDOWN, False),
invert_position=knx_config.get(CoverConf.INVERT_POSITION, False),
invert_angle=knx_config.get(CoverConf.INVERT_ANGLE, False),
sync_state=knx_config[CONF_SYNC_STATE],
group_address_long=conf.get_write_and_passive(CONF_GA_UP_DOWN),
group_address_short=conf.get_write_and_passive(CONF_GA_STEP),
group_address_stop=conf.get_write_and_passive(CONF_GA_STOP),
group_address_position=conf.get_write_and_passive(CONF_GA_POSITION_SET),
group_address_position_state=conf.get_state_and_passive(CONF_GA_POSITION_STATE),
group_address_angle=conf.get_write(CONF_GA_ANGLE),
group_address_angle_state=conf.get_state_and_passive(CONF_GA_ANGLE),
travel_time_down=conf.get(CoverConf.TRAVELLING_TIME_DOWN),
travel_time_up=conf.get(CoverConf.TRAVELLING_TIME_UP),
invert_updown=conf.get(CoverConf.INVERT_UPDOWN, default=False),
invert_position=conf.get(CoverConf.INVERT_POSITION, default=False),
invert_angle=conf.get(CoverConf.INVERT_ANGLE, default=False),
sync_state=conf.get(CONF_SYNC_STATE),
)

View File

@@ -33,9 +33,9 @@ from .const import CONF_SYNC_STATE, DOMAIN, KNX_ADDRESS, KNX_MODULE_KEY, ColorTe
from .entity import KnxUiEntity, KnxUiEntityPlatformController, KnxYamlEntity
from .schema import LightSchema
from .storage.const import (
CONF_COLOR,
CONF_COLOR_TEMP_MAX,
CONF_COLOR_TEMP_MIN,
CONF_DPT,
CONF_ENTITY,
CONF_GA_BLUE_BRIGHTNESS,
CONF_GA_BLUE_SWITCH,
@@ -45,17 +45,15 @@ from .storage.const import (
CONF_GA_GREEN_BRIGHTNESS,
CONF_GA_GREEN_SWITCH,
CONF_GA_HUE,
CONF_GA_PASSIVE,
CONF_GA_RED_BRIGHTNESS,
CONF_GA_RED_SWITCH,
CONF_GA_SATURATION,
CONF_GA_STATE,
CONF_GA_SWITCH,
CONF_GA_WHITE_BRIGHTNESS,
CONF_GA_WHITE_SWITCH,
CONF_GA_WRITE,
)
from .storage.entity_store_schema import LightColorMode
from .storage.util import ConfigExtractor
async def async_setup_entry(
@@ -203,94 +201,98 @@ def _create_yaml_light(xknx: XKNX, config: ConfigType) -> XknxLight:
def _create_ui_light(xknx: XKNX, knx_config: ConfigType, name: str) -> XknxLight:
"""Return a KNX Light device to be used within XKNX."""
def get_write(key: str) -> str | None:
"""Get the write group address."""
return knx_config[key][CONF_GA_WRITE] if key in knx_config else None
def get_state(key: str) -> list[Any] | None:
"""Get the state group address."""
return (
[knx_config[key][CONF_GA_STATE], *knx_config[key][CONF_GA_PASSIVE]]
if key in knx_config
else None
)
def get_dpt(key: str) -> str | None:
"""Get the DPT."""
return knx_config[key].get(CONF_DPT) if key in knx_config else None
conf = ConfigExtractor(knx_config)
group_address_tunable_white = None
group_address_tunable_white_state = None
group_address_color_temp = None
group_address_color_temp_state = None
color_temperature_type = ColorTemperatureType.UINT_2_BYTE
if ga_color_temp := knx_config.get(CONF_GA_COLOR_TEMP):
if ga_color_temp[CONF_DPT] == ColorTempModes.RELATIVE.value:
group_address_tunable_white = ga_color_temp[CONF_GA_WRITE]
group_address_tunable_white_state = [
ga_color_temp[CONF_GA_STATE],
*ga_color_temp[CONF_GA_PASSIVE],
]
if _color_temp_dpt := conf.get_dpt(CONF_GA_COLOR_TEMP):
if _color_temp_dpt == ColorTempModes.RELATIVE.value:
group_address_tunable_white = conf.get_write(CONF_GA_COLOR_TEMP)
group_address_tunable_white_state = conf.get_state_and_passive(
CONF_GA_COLOR_TEMP
)
else:
# absolute uint or float
group_address_color_temp = ga_color_temp[CONF_GA_WRITE]
group_address_color_temp_state = [
ga_color_temp[CONF_GA_STATE],
*ga_color_temp[CONF_GA_PASSIVE],
]
if ga_color_temp[CONF_DPT] == ColorTempModes.ABSOLUTE_FLOAT.value:
group_address_color_temp = conf.get_write(CONF_GA_COLOR_TEMP)
group_address_color_temp_state = conf.get_state_and_passive(
CONF_GA_COLOR_TEMP
)
if _color_temp_dpt == ColorTempModes.ABSOLUTE_FLOAT.value:
color_temperature_type = ColorTemperatureType.FLOAT_2_BYTE
_color_dpt = get_dpt(CONF_GA_COLOR)
color_dpt = conf.get_dpt(CONF_COLOR, CONF_GA_COLOR)
return XknxLight(
xknx,
name=name,
group_address_switch=get_write(CONF_GA_SWITCH),
group_address_switch_state=get_state(CONF_GA_SWITCH),
group_address_brightness=get_write(CONF_GA_BRIGHTNESS),
group_address_brightness_state=get_state(CONF_GA_BRIGHTNESS),
group_address_color=get_write(CONF_GA_COLOR)
if _color_dpt == LightColorMode.RGB
group_address_switch=conf.get_write(CONF_GA_SWITCH),
group_address_switch_state=conf.get_state_and_passive(CONF_GA_SWITCH),
group_address_brightness=conf.get_write(CONF_GA_BRIGHTNESS),
group_address_brightness_state=conf.get_state_and_passive(CONF_GA_BRIGHTNESS),
group_address_color=conf.get_write(CONF_COLOR, CONF_GA_COLOR)
if color_dpt == LightColorMode.RGB
else None,
group_address_color_state=get_state(CONF_GA_COLOR)
if _color_dpt == LightColorMode.RGB
group_address_color_state=conf.get_state_and_passive(CONF_COLOR, CONF_GA_COLOR)
if color_dpt == LightColorMode.RGB
else None,
group_address_rgbw=get_write(CONF_GA_COLOR)
if _color_dpt == LightColorMode.RGBW
group_address_rgbw=conf.get_write(CONF_COLOR, CONF_GA_COLOR)
if color_dpt == LightColorMode.RGBW
else None,
group_address_rgbw_state=get_state(CONF_GA_COLOR)
if _color_dpt == LightColorMode.RGBW
group_address_rgbw_state=conf.get_state_and_passive(CONF_COLOR, CONF_GA_COLOR)
if color_dpt == LightColorMode.RGBW
else None,
group_address_hue=get_write(CONF_GA_HUE),
group_address_hue_state=get_state(CONF_GA_HUE),
group_address_saturation=get_write(CONF_GA_SATURATION),
group_address_saturation_state=get_state(CONF_GA_SATURATION),
group_address_xyy_color=get_write(CONF_GA_COLOR)
if _color_dpt == LightColorMode.XYY
group_address_hue=conf.get_write(CONF_COLOR, CONF_GA_HUE),
group_address_hue_state=conf.get_state_and_passive(CONF_COLOR, CONF_GA_HUE),
group_address_saturation=conf.get_write(CONF_COLOR, CONF_GA_SATURATION),
group_address_saturation_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_SATURATION
),
group_address_xyy_color=conf.get_write(CONF_COLOR, CONF_GA_COLOR)
if color_dpt == LightColorMode.XYY
else None,
group_address_xyy_color_state=get_write(CONF_GA_COLOR)
if _color_dpt == LightColorMode.XYY
group_address_xyy_color_state=conf.get_write(CONF_COLOR, CONF_GA_COLOR)
if color_dpt == LightColorMode.XYY
else None,
group_address_tunable_white=group_address_tunable_white,
group_address_tunable_white_state=group_address_tunable_white_state,
group_address_color_temperature=group_address_color_temp,
group_address_color_temperature_state=group_address_color_temp_state,
group_address_switch_red=get_write(CONF_GA_RED_SWITCH),
group_address_switch_red_state=get_state(CONF_GA_RED_SWITCH),
group_address_brightness_red=get_write(CONF_GA_RED_BRIGHTNESS),
group_address_brightness_red_state=get_state(CONF_GA_RED_BRIGHTNESS),
group_address_switch_green=get_write(CONF_GA_GREEN_SWITCH),
group_address_switch_green_state=get_state(CONF_GA_GREEN_SWITCH),
group_address_brightness_green=get_write(CONF_GA_GREEN_BRIGHTNESS),
group_address_brightness_green_state=get_state(CONF_GA_GREEN_BRIGHTNESS),
group_address_switch_blue=get_write(CONF_GA_BLUE_SWITCH),
group_address_switch_blue_state=get_state(CONF_GA_BLUE_SWITCH),
group_address_brightness_blue=get_write(CONF_GA_BLUE_BRIGHTNESS),
group_address_brightness_blue_state=get_state(CONF_GA_BLUE_BRIGHTNESS),
group_address_switch_white=get_write(CONF_GA_WHITE_SWITCH),
group_address_switch_white_state=get_state(CONF_GA_WHITE_SWITCH),
group_address_brightness_white=get_write(CONF_GA_WHITE_BRIGHTNESS),
group_address_brightness_white_state=get_state(CONF_GA_WHITE_BRIGHTNESS),
group_address_switch_red=conf.get_write(CONF_COLOR, CONF_GA_RED_SWITCH),
group_address_switch_red_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_RED_SWITCH
),
group_address_brightness_red=conf.get_write(CONF_COLOR, CONF_GA_RED_BRIGHTNESS),
group_address_brightness_red_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_RED_BRIGHTNESS
),
group_address_switch_green=conf.get_write(CONF_COLOR, CONF_GA_GREEN_SWITCH),
group_address_switch_green_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_GREEN_SWITCH
),
group_address_brightness_green=conf.get_write(CONF_GA_GREEN_BRIGHTNESS),
group_address_brightness_green_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_GREEN_BRIGHTNESS
),
group_address_switch_blue=conf.get_write(CONF_GA_BLUE_SWITCH),
group_address_switch_blue_state=conf.get_state_and_passive(CONF_GA_BLUE_SWITCH),
group_address_brightness_blue=conf.get_write(CONF_GA_BLUE_BRIGHTNESS),
group_address_brightness_blue_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_BLUE_BRIGHTNESS
),
group_address_switch_white=conf.get_write(CONF_COLOR, CONF_GA_WHITE_SWITCH),
group_address_switch_white_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_WHITE_SWITCH
),
group_address_brightness_white=conf.get_write(
CONF_COLOR, CONF_GA_WHITE_BRIGHTNESS
),
group_address_brightness_white_state=conf.get_state_and_passive(
CONF_COLOR, CONF_GA_WHITE_BRIGHTNESS
),
color_temperature_type=color_temperature_type,
min_kelvin=knx_config[CONF_COLOR_TEMP_MIN],
max_kelvin=knx_config[CONF_COLOR_TEMP_MAX],

View File

@@ -13,10 +13,11 @@ from homeassistant.util.ulid import ulid_now
from ..const import DOMAIN
from .const import CONF_DATA
from .migration import migrate_1_to_2
_LOGGER = logging.getLogger(__name__)
STORAGE_VERSION: Final = 1
STORAGE_VERSION: Final = 2
STORAGE_KEY: Final = f"{DOMAIN}/config_store.json"
type KNXPlatformStoreModel = dict[str, dict[str, Any]] # unique_id: configuration
@@ -45,6 +46,20 @@ class PlatformControllerBase(ABC):
"""Update an existing entities configuration."""
class _KNXConfigStoreStorage(Store[KNXConfigStoreModel]):
"""Storage handler for KNXConfigStore."""
async def _async_migrate_func(
self, old_major_version: int, old_minor_version: int, old_data: dict[str, Any]
) -> dict[str, Any]:
"""Migrate to the new version."""
if old_major_version == 1:
# version 2 introduced in 2025.7
migrate_1_to_2(old_data)
return old_data
class KNXConfigStore:
"""Manage KNX config store data."""
@@ -56,7 +71,7 @@ class KNXConfigStore:
"""Initialize config store."""
self.hass = hass
self.config_entry = config_entry
self._store = Store[KNXConfigStoreModel](hass, STORAGE_VERSION, STORAGE_KEY)
self._store = _KNXConfigStoreStorage(hass, STORAGE_VERSION, STORAGE_KEY)
self.data = KNXConfigStoreModel(entities={})
self._platform_controllers: dict[Platform, PlatformControllerBase] = {}

View File

@@ -2,6 +2,7 @@
from typing import Final
# Common
CONF_DATA: Final = "data"
CONF_ENTITY: Final = "entity"
CONF_DEVICE_INFO: Final = "device_info"
@@ -12,10 +13,22 @@ CONF_DPT: Final = "dpt"
CONF_GA_SENSOR: Final = "ga_sensor"
CONF_GA_SWITCH: Final = "ga_switch"
CONF_GA_COLOR_TEMP: Final = "ga_color_temp"
# Cover
CONF_GA_UP_DOWN: Final = "ga_up_down"
CONF_GA_STOP: Final = "ga_stop"
CONF_GA_STEP: Final = "ga_step"
CONF_GA_POSITION_SET: Final = "ga_position_set"
CONF_GA_POSITION_STATE: Final = "ga_position_state"
CONF_GA_ANGLE: Final = "ga_angle"
# Light
CONF_COLOR_TEMP_MIN: Final = "color_temp_min"
CONF_COLOR_TEMP_MAX: Final = "color_temp_max"
CONF_GA_BRIGHTNESS: Final = "ga_brightness"
CONF_GA_COLOR_TEMP: Final = "ga_color_temp"
# Light/color
CONF_COLOR: Final = "color"
CONF_GA_COLOR: Final = "ga_color"
CONF_GA_RED_BRIGHTNESS: Final = "ga_red_brightness"
CONF_GA_RED_SWITCH: Final = "ga_red_switch"
@@ -27,9 +40,3 @@ CONF_GA_WHITE_BRIGHTNESS: Final = "ga_white_brightness"
CONF_GA_WHITE_SWITCH: Final = "ga_white_switch"
CONF_GA_HUE: Final = "ga_hue"
CONF_GA_SATURATION: Final = "ga_saturation"
CONF_GA_UP_DOWN: Final = "ga_up_down"
CONF_GA_STOP: Final = "ga_stop"
CONF_GA_STEP: Final = "ga_step"
CONF_GA_POSITION_SET: Final = "ga_position_set"
CONF_GA_POSITION_STATE: Final = "ga_position_state"
CONF_GA_ANGLE: Final = "ga_angle"

View File

@@ -29,6 +29,7 @@ from ..const import (
)
from ..validation import sync_state_validator
from .const import (
CONF_COLOR,
CONF_COLOR_TEMP_MAX,
CONF_COLOR_TEMP_MIN,
CONF_DATA,
@@ -59,7 +60,7 @@ from .const import (
CONF_GA_WHITE_SWITCH,
CONF_GA_WRITE,
)
from .knx_selector import GASelector
from .knx_selector import GASelector, GroupSelect
BASE_ENTITY_SCHEMA = vol.All(
{
@@ -134,16 +135,14 @@ COVER_SCHEMA = vol.Schema(
vol.Required(DOMAIN): vol.All(
vol.Schema(
{
**optional_ga_schema(CONF_GA_UP_DOWN, GASelector(state=False)),
vol.Optional(CONF_GA_UP_DOWN): GASelector(state=False),
vol.Optional(CoverConf.INVERT_UPDOWN): selector.BooleanSelector(),
**optional_ga_schema(CONF_GA_STOP, GASelector(state=False)),
**optional_ga_schema(CONF_GA_STEP, GASelector(state=False)),
**optional_ga_schema(CONF_GA_POSITION_SET, GASelector(state=False)),
**optional_ga_schema(
CONF_GA_POSITION_STATE, GASelector(write=False)
),
vol.Optional(CONF_GA_STOP): GASelector(state=False),
vol.Optional(CONF_GA_STEP): GASelector(state=False),
vol.Optional(CONF_GA_POSITION_SET): GASelector(state=False),
vol.Optional(CONF_GA_POSITION_STATE): GASelector(write=False),
vol.Optional(CoverConf.INVERT_POSITION): selector.BooleanSelector(),
**optional_ga_schema(CONF_GA_ANGLE, GASelector()),
vol.Optional(CONF_GA_ANGLE): GASelector(),
vol.Optional(CoverConf.INVERT_ANGLE): selector.BooleanSelector(),
vol.Optional(
CoverConf.TRAVELLING_TIME_DOWN, default=25
@@ -208,73 +207,112 @@ class LightColorModeSchema(StrEnum):
HSV = "hsv"
_LIGHT_COLOR_MODE_SCHEMA = "_light_color_mode_schema"
_hs_color_inclusion_msg = (
"'Hue', 'Saturation' and 'Brightness' addresses are required for HSV configuration"
)
_COMMON_LIGHT_SCHEMA = vol.Schema(
LIGHT_KNX_SCHEMA = vol.All(
vol.Schema(
{
vol.Optional(CONF_SYNC_STATE, default=True): sync_state_validator,
**optional_ga_schema(
CONF_GA_COLOR_TEMP, GASelector(write_required=True, dpt=ColorTempModes)
vol.Optional(CONF_GA_SWITCH): GASelector(write_required=True),
vol.Optional(CONF_GA_BRIGHTNESS): GASelector(write_required=True),
vol.Optional(CONF_GA_COLOR_TEMP): GASelector(
write_required=True, dpt=ColorTempModes
),
vol.Optional(CONF_COLOR): GroupSelect(
vol.Schema(
{
vol.Optional(CONF_GA_COLOR): GASelector(
write_required=True, dpt=LightColorMode
)
}
),
vol.Schema(
{
vol.Required(CONF_GA_RED_BRIGHTNESS): GASelector(
write_required=True
),
vol.Optional(CONF_GA_RED_SWITCH): GASelector(
write_required=False
),
vol.Required(CONF_GA_GREEN_BRIGHTNESS): GASelector(
write_required=True
),
vol.Optional(CONF_GA_GREEN_SWITCH): GASelector(
write_required=False
),
vol.Required(CONF_GA_BLUE_BRIGHTNESS): GASelector(
write_required=True
),
vol.Optional(CONF_GA_BLUE_SWITCH): GASelector(
write_required=False
),
vol.Optional(CONF_GA_WHITE_BRIGHTNESS): GASelector(
write_required=True
),
vol.Optional(CONF_GA_WHITE_SWITCH): GASelector(
write_required=False
),
}
),
vol.Schema(
{
vol.Required(CONF_GA_HUE): GASelector(write_required=True),
vol.Required(CONF_GA_SATURATION): GASelector(
write_required=True
),
}
),
# 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)
),
},
extra=vol.REMOVE_EXTRA,
)
_DEFAULT_LIGHT_SCHEMA = _COMMON_LIGHT_SCHEMA.extend(
{
vol.Required(_LIGHT_COLOR_MODE_SCHEMA): LightColorModeSchema.DEFAULT.value,
vol.Required(CONF_GA_SWITCH): GASelector(write_required=True),
**optional_ga_schema(CONF_GA_BRIGHTNESS, GASelector(write_required=True)),
**optional_ga_schema(
CONF_GA_COLOR,
GASelector(write_required=True, dpt=LightColorMode),
}
),
}
)
_INDIVIDUAL_LIGHT_SCHEMA = _COMMON_LIGHT_SCHEMA.extend(
vol.Any(
vol.Schema(
{vol.Required(CONF_GA_SWITCH): object},
extra=vol.ALLOW_EXTRA,
),
vol.Schema( # brightness addresses are required in INDIVIDUAL_COLOR_SCHEMA
{vol.Required(CONF_COLOR): {vol.Required(CONF_GA_RED_BRIGHTNESS): object}},
extra=vol.ALLOW_EXTRA,
),
msg="either 'address' or 'individual_colors' is required",
),
vol.Any(
vol.Schema( # 'brightness' is non-optional for hs-color
{
vol.Required(_LIGHT_COLOR_MODE_SCHEMA): LightColorModeSchema.INDIVIDUAL.value,
**optional_ga_schema(CONF_GA_SWITCH, GASelector(write_required=True)),
**optional_ga_schema(CONF_GA_BRIGHTNESS, GASelector(write_required=True)),
vol.Required(CONF_GA_RED_BRIGHTNESS): GASelector(write_required=True),
**optional_ga_schema(CONF_GA_RED_SWITCH, GASelector(write_required=False)),
vol.Required(CONF_GA_GREEN_BRIGHTNESS): GASelector(write_required=True),
**optional_ga_schema(CONF_GA_GREEN_SWITCH, GASelector(write_required=False)),
vol.Required(CONF_GA_BLUE_BRIGHTNESS): GASelector(write_required=True),
**optional_ga_schema(CONF_GA_BLUE_SWITCH, GASelector(write_required=False)),
**optional_ga_schema(CONF_GA_WHITE_BRIGHTNESS, GASelector(write_required=True)),
**optional_ga_schema(CONF_GA_WHITE_SWITCH, GASelector(write_required=False)),
}
)
_HSV_LIGHT_SCHEMA = _COMMON_LIGHT_SCHEMA.extend(
{
vol.Required(_LIGHT_COLOR_MODE_SCHEMA): LightColorModeSchema.HSV.value,
vol.Required(CONF_GA_SWITCH): GASelector(write_required=True),
vol.Required(CONF_GA_BRIGHTNESS): GASelector(write_required=True),
vol.Required(CONF_GA_HUE): GASelector(write_required=True),
vol.Required(CONF_GA_SATURATION): GASelector(write_required=True),
}
)
LIGHT_KNX_SCHEMA = cv.key_value_schemas(
_LIGHT_COLOR_MODE_SCHEMA,
default_schema=_DEFAULT_LIGHT_SCHEMA,
value_schemas={
LightColorModeSchema.DEFAULT: _DEFAULT_LIGHT_SCHEMA,
LightColorModeSchema.INDIVIDUAL: _INDIVIDUAL_LIGHT_SCHEMA,
LightColorModeSchema.HSV: _HSV_LIGHT_SCHEMA,
vol.Required(CONF_GA_BRIGHTNESS, msg=_hs_color_inclusion_msg): object,
vol.Required(CONF_COLOR): {
vol.Required(CONF_GA_HUE, msg=_hs_color_inclusion_msg): object,
vol.Required(
CONF_GA_SATURATION, msg=_hs_color_inclusion_msg
): object,
},
},
extra=vol.ALLOW_EXTRA,
),
vol.Schema( # hs-colors not used
{
vol.Optional(CONF_COLOR): {
vol.Optional(CONF_GA_HUE): None,
vol.Optional(CONF_GA_SATURATION): None,
},
},
extra=vol.ALLOW_EXTRA,
),
msg=_hs_color_inclusion_msg,
),
)
LIGHT_SCHEMA = vol.Schema(
{
vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA,

View File

@@ -1,5 +1,6 @@
"""Selectors for KNX."""
from collections.abc import Hashable, Iterable
from enum import Enum
from typing import Any
@@ -9,6 +10,31 @@ from ..validation import ga_validator, maybe_ga_validator
from .const import CONF_DPT, CONF_GA_PASSIVE, CONF_GA_STATE, CONF_GA_WRITE
class GroupSelect(vol.Any):
"""Use the first validated value.
This is a version of vol.Any with custom error handling to
show proper invalid markers for sub-schema items in the UI.
"""
def _exec(self, funcs: Iterable, v: Any, path: list[Hashable] | None = None) -> Any:
"""Execute the validation functions."""
errors = []
for func in funcs:
try:
if path is None:
return func(v)
return func(path, v)
except vol.Invalid as e:
errors.append(e)
if errors:
raise next(
(err for err in errors if "extra keys not allowed" not in err.msg),
errors[0],
)
raise vol.AnyInvalid(self.msg or "no valid value found", path=path)
class GASelector:
"""Selector for a KNX group address structure."""

View File

@@ -0,0 +1,42 @@
"""Migration functions for KNX config store schema."""
from typing import Any
from homeassistant.const import Platform
from . import const as store_const
def migrate_1_to_2(data: dict[str, Any]) -> None:
"""Migrate from schema 1 to schema 2."""
if lights := data.get("entities", {}).get(Platform.LIGHT):
for light in lights.values():
_migrate_light_schema_1_to_2(light["knx"])
def _migrate_light_schema_1_to_2(light_knx_data: dict[str, Any]) -> None:
"""Migrate light color mode schema."""
# Remove no more needed helper data from schema
light_knx_data.pop("_light_color_mode_schema", None)
# Move color related group addresses to new "color" key
color = {}
for color_key in (
# optional / required and exclusive keys are the same in old and new schema
store_const.CONF_GA_COLOR,
store_const.CONF_GA_HUE,
store_const.CONF_GA_SATURATION,
store_const.CONF_GA_RED_BRIGHTNESS,
store_const.CONF_GA_RED_SWITCH,
store_const.CONF_GA_GREEN_BRIGHTNESS,
store_const.CONF_GA_GREEN_SWITCH,
store_const.CONF_GA_BLUE_BRIGHTNESS,
store_const.CONF_GA_BLUE_SWITCH,
store_const.CONF_GA_WHITE_BRIGHTNESS,
store_const.CONF_GA_WHITE_SWITCH,
):
if color_key in light_knx_data:
color[color_key] = light_knx_data.pop(color_key)
if color:
light_knx_data[store_const.CONF_COLOR] = color

View File

@@ -0,0 +1,51 @@
"""Utility functions for the KNX integration."""
from functools import partial
from typing import Any
from homeassistant.helpers.typing import ConfigType
from .const import CONF_DPT, CONF_GA_PASSIVE, CONF_GA_STATE, CONF_GA_WRITE
def nested_get(dic: ConfigType, *keys: str, default: Any | None = None) -> Any:
"""Get the value from a nested dictionary."""
for key in keys:
if key not in dic:
return default
dic = dic[key]
return dic
class ConfigExtractor:
"""Helper class for extracting values from a knx config store dictionary."""
__slots__ = ("get",)
def __init__(self, config: ConfigType) -> None:
"""Initialize the extractor."""
self.get = partial(nested_get, config)
def get_write(self, *path: str) -> str | None:
"""Get the write group address."""
return self.get(*path, CONF_GA_WRITE) # type: ignore[no-any-return]
def get_state(self, *path: str) -> str | None:
"""Get the state group address."""
return self.get(*path, CONF_GA_STATE) # type: ignore[no-any-return]
def get_write_and_passive(self, *path: str) -> list[Any | None]:
"""Get the group addresses of write and passive."""
write = self.get(*path, CONF_GA_WRITE)
passive = self.get(*path, CONF_GA_PASSIVE)
return [write, *passive] if passive else [write]
def get_state_and_passive(self, *path: str) -> list[Any | None]:
"""Get the group addresses of state and passive."""
state = self.get(*path, CONF_GA_STATE)
passive = self.get(*path, CONF_GA_PASSIVE)
return [state, *passive] if passive else [state]
def get_dpt(self, *path: str) -> str | None:
"""Get the data point type of a group address config key."""
return self.get(*path, CONF_DPT) # type: ignore[no-any-return]

View File

@@ -36,13 +36,8 @@ from .const import (
)
from .entity import KnxUiEntity, KnxUiEntityPlatformController, KnxYamlEntity
from .schema import SwitchSchema
from .storage.const import (
CONF_ENTITY,
CONF_GA_PASSIVE,
CONF_GA_STATE,
CONF_GA_SWITCH,
CONF_GA_WRITE,
)
from .storage.const import CONF_ENTITY, CONF_GA_SWITCH
from .storage.util import ConfigExtractor
async def async_setup_entry(
@@ -142,15 +137,13 @@ class KnxUiSwitch(_KnxSwitch, KnxUiEntity):
unique_id=unique_id,
entity_config=config[CONF_ENTITY],
)
knx_conf = ConfigExtractor(config[DOMAIN])
self._device = XknxSwitch(
knx_module.xknx,
name=config[CONF_ENTITY][CONF_NAME],
group_address=config[DOMAIN][CONF_GA_SWITCH][CONF_GA_WRITE],
group_address_state=[
config[DOMAIN][CONF_GA_SWITCH][CONF_GA_STATE],
*config[DOMAIN][CONF_GA_SWITCH][CONF_GA_PASSIVE],
],
respond_to_read=config[DOMAIN][CONF_RESPOND_TO_READ],
sync_state=config[DOMAIN][CONF_SYNC_STATE],
invert=config[DOMAIN][CONF_INVERT],
group_address=knx_conf.get_write(CONF_GA_SWITCH),
group_address_state=knx_conf.get_state_and_passive(CONF_GA_SWITCH),
respond_to_read=knx_conf.get(CONF_RESPOND_TO_READ),
sync_state=knx_conf.get(CONF_SYNC_STATE),
invert=knx_conf.get(CONF_INVERT),
)

View File

@@ -1182,7 +1182,6 @@ async def test_light_ui_create(
entity_data={"name": "test"},
knx_data={
"ga_switch": {"write": "1/1/1", "state": "2/2/2"},
"_light_color_mode_schema": "default",
"sync_state": True,
},
)
@@ -1223,7 +1222,6 @@ async def test_light_ui_color_temp(
"write": "3/3/3",
"dpt": color_temp_mode,
},
"_light_color_mode_schema": "default",
"sync_state": True,
},
)
@@ -1257,7 +1255,6 @@ async def test_light_ui_multi_mode(
knx_data={
"color_temp_min": 2700,
"color_temp_max": 6000,
"_light_color_mode_schema": "default",
"ga_switch": {
"write": "1/1/1",
"passive": [],
@@ -1275,6 +1272,7 @@ async def test_light_ui_multi_mode(
"state": "0/6/3",
"passive": [],
},
"color": {
"ga_color": {
"write": "0/6/4",
"dpt": "251.600",
@@ -1282,6 +1280,7 @@ async def test_light_ui_multi_mode(
"passive": [],
},
},
},
)
await knx.assert_read("2/2/2", True)
await knx.assert_read("0/6/1", (0xFF,))