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

Author: farmio

homeassistant/components/knx/binary_sensor.py

@@ -39,7 +39,8 @@
     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 @@ def __init__(
             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

homeassistant/components/knx/cover.py

@@ -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 .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 @@ def __init__(self, knx_module: KNXModule, config: ConfigType) -> None:
 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),
     )
 
 

homeassistant/components/knx/light.py

@@ -35,7 +35,6 @@
 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 @@
     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 individual_color_addresses(color: str, feature: str) -> Any | None:
 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],

homeassistant/components/knx/storage/util.py

@@ -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]

homeassistant/components/knx/switch.py

@@ -36,13 +36,8 @@
 )
 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 @@ def __init__(
             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),
         )