jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 19:27:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

KNX: Use a ConfigExtractor helper class for value retrieval (#147983)

Browse Source
This commit is contained in:
Matthias Alphart 2025-07-02 21:49:24 +02:00 committed by GitHub
parent 78c39f8a06
commit 53d2f6b0c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 150 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
homeassistant/components/knx/binary_sensor.py
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

48
homeassistant/components/knx/cover.py
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),
)

133
homeassistant/components/knx/light.py
View File

@ -35,7 +35,6 @@ from .schema import LightSchema
from .storage.const import (
CONF_COLOR_TEMP_MAX,
CONF_COLOR_TEMP_MIN,
CONF_DPT,
CONF_ENTITY,
CONF_GA_BLUE_BRIGHTNESS,
CONF_GA_BLUE_SWITCH,
@ -45,17 +44,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 +200,92 @@ 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_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_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_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_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_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_GA_HUE),
group_address_hue_state=conf.get_state_and_passive(CONF_GA_HUE),
group_address_saturation=conf.get_write(CONF_GA_SATURATION),
group_address_saturation_state=conf.get_state_and_passive(CONF_GA_SATURATION),
group_address_xyy_color=conf.get_write(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_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_GA_RED_SWITCH),
group_address_switch_red_state=conf.get_state_and_passive(CONF_GA_RED_SWITCH),
group_address_brightness_red=conf.get_write(CONF_GA_RED_BRIGHTNESS),
group_address_brightness_red_state=conf.get_state_and_passive(
CONF_GA_RED_BRIGHTNESS
),
group_address_switch_green=conf.get_write(CONF_GA_GREEN_SWITCH),
group_address_switch_green_state=conf.get_state_and_passive(
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_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_GA_BLUE_BRIGHTNESS
),
group_address_switch_white=conf.get_write(CONF_GA_WHITE_SWITCH),
group_address_switch_white_state=conf.get_state_and_passive(
CONF_GA_WHITE_SWITCH
),
group_address_brightness_white=conf.get_write(CONF_GA_WHITE_BRIGHTNESS),
group_address_brightness_white_state=conf.get_state_and_passive(
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],

51
homeassistant/components/knx/storage/util.py Normal file
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]

23
homeassistant/components/knx/switch.py
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),
)