Create, update and delete KNX entities from UI / WS-commands

homeassistant/components/knx/__init__.py

@@ -31,6 +31,7 @@
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import discovery
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.device_registry import DeviceEntry
 from homeassistant.helpers.reload import async_integration_yaml_config
 from homeassistant.helpers.storage import STORAGE_DIR
 from homeassistant.helpers.typing import ConfigType
@@ -62,7 +63,8 @@
     DATA_KNX_CONFIG,
     DOMAIN,
     KNX_ADDRESS,
-    SUPPORTED_PLATFORMS,
+    SUPPORTED_PLATFORMS_UI,
+    SUPPORTED_PLATFORMS_YAML,
     TELEGRAM_LOG_DEFAULT,
 )
 from .device import KNXInterfaceDevice
@@ -90,6 +92,7 @@
     WeatherSchema,
 )
 from .services import register_knx_services
+from .storage.config_store import KNXConfigStore
 from .telegrams import STORAGE_KEY as TELEGRAMS_STORAGE_KEY, Telegrams
 from .websocket import register_panel
 
@@ -190,10 +193,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             knx_module.exposures.append(
                 create_knx_exposure(hass, knx_module.xknx, expose_config)
             )
-    # always forward sensor for system entities (telegram counter, etc.)
-    platforms = {platform for platform in SUPPORTED_PLATFORMS if platform in config}
-    platforms.add(Platform.SENSOR)
-    await hass.config_entries.async_forward_entry_setups(entry, platforms)
+    await hass.config_entries.async_forward_entry_setups(
+        entry,
+        {
+            Platform.SENSOR,  # always forward sensor for system entities (telegram counter, etc.)
+            *SUPPORTED_PLATFORMS_UI,  # forward all platforms that support UI entity management
+            *{  # forward yaml-only managed platforms on demand
+                platform for platform in SUPPORTED_PLATFORMS_YAML if platform in config
+            },
+        },
+    )
 
     # set up notify service for backwards compatibility - remove 2024.11
     if NotifySchema.PLATFORM in config:
@@ -220,15 +229,15 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     unload_ok = await hass.config_entries.async_unload_platforms(
         entry,
-        [
+        {
             Platform.SENSOR,  # always unload system entities (telegram counter, etc.)
-            *[
+            *SUPPORTED_PLATFORMS_UI,  # unload all platforms that support UI entity management
+            *{  # unload yaml-only managed platforms if configured
                 platform
-                for platform in SUPPORTED_PLATFORMS
+                for platform in SUPPORTED_PLATFORMS_YAML
                 if platform in hass.data[DATA_KNX_CONFIG]
-                and platform is not Platform.SENSOR
-            ],
-        ],
+            },
+        },
     )
     if unload_ok:
         await knx_module.stop()
@@ -263,6 +272,22 @@ def remove_files(storage_dir: Path, knxkeys_filename: str | None) -> None:
     await hass.async_add_executor_job(remove_files, storage_dir, knxkeys_filename)
 
 
+async def async_remove_config_entry_device(
+    hass: HomeAssistant, config_entry: ConfigEntry, device_entry: DeviceEntry
+) -> bool:
+    """Remove a config entry from a device."""
+    knx_module: KNXModule = hass.data[DOMAIN]
+    if not device_entry.identifiers.isdisjoint(
+        knx_module.interface_device.device_info["identifiers"]
+    ):
+        # can not remove interface device
+        return False
+    for entity in knx_module.config_store.get_entity_entries():
+        if entity.device_id == device_entry.id:
+            await knx_module.config_store.delete_entity(entity.entity_id)
+    return True
+
+
 class KNXModule:
     """Representation of KNX Object."""
 
@@ -278,6 +303,7 @@ def __init__(
         self.entry = entry
 
         self.project = KNXProject(hass=hass, entry=entry)
+        self.config_store = KNXConfigStore(hass=hass, entry=entry)
 
         self.xknx = XKNX(
             connection_config=self.connection_config(),
@@ -309,6 +335,7 @@ def __init__(
     async def start(self) -> None:
         """Start XKNX object. Connect to tunneling or Routing device."""
         await self.project.load_project()
+        await self.config_store.load_data()
         await self.telegrams.load_history()
         await self.xknx.start()
 

homeassistant/components/knx/config_flow.py

@@ -62,7 +62,7 @@
     TELEGRAM_LOG_MAX,
     KNXConfigEntryData,
 )
-from .helpers.keyring import DEFAULT_KNX_KEYRING_FILENAME, save_uploaded_knxkeys_file
+from .storage.keyring import DEFAULT_KNX_KEYRING_FILENAME, save_uploaded_knxkeys_file
 from .validation import ia_validator, ip_v4_validator
 
 CONF_KNX_GATEWAY: Final = "gateway"

homeassistant/components/knx/const.py

@@ -127,12 +127,13 @@ class KNXConfigEntryData(TypedDict, total=False):
 class ColorTempModes(Enum):
     """Color temperature modes for config validation."""
 
-    ABSOLUTE = "DPT-7.600"
-    ABSOLUTE_FLOAT = "DPT-9"
-    RELATIVE = "DPT-5.001"
+    # YAML uses Enum.name (with vol.Upper), UI uses Enum.value for lookup
+    ABSOLUTE = "7.600"
+    ABSOLUTE_FLOAT = "9"
+    RELATIVE = "5.001"
 
 
-SUPPORTED_PLATFORMS: Final = [
+SUPPORTED_PLATFORMS_YAML: Final = {
     Platform.BINARY_SENSOR,
     Platform.BUTTON,
     Platform.CLIMATE,
@@ -150,7 +151,9 @@ class ColorTempModes(Enum):
     Platform.TEXT,
     Platform.TIME,
     Platform.WEATHER,
-]
+}
+
+SUPPORTED_PLATFORMS_UI: Final = {Platform.SWITCH}
 
 # Map KNX controller modes to HA modes. This list might not be complete.
 CONTROLLER_MODES: Final = {

homeassistant/components/knx/storage/config_store.py

@@ -0,0 +1,145 @@
+"""KNX entity configuration store."""
+
+from collections.abc import Callable
+import logging
+from typing import TYPE_CHECKING, Any, Final, TypedDict
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_PLATFORM, Platform
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.storage import Store
+from homeassistant.util.ulid import ulid_now
+
+from ..const import DOMAIN
+from .const import CONF_DATA
+
+if TYPE_CHECKING:
+    from ..knx_entity import KnxEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+STORAGE_VERSION: Final = 1
+STORAGE_KEY: Final = f"{DOMAIN}/config_store.json"
+
+KNXPlatformStoreModel = dict[str, dict[str, Any]]  # unique_id: configuration
+KNXEntityStoreModel = dict[
+    str, KNXPlatformStoreModel
+]  # platform: KNXPlatformStoreModel
+
+
+class KNXConfigStoreModel(TypedDict):
+    """Represent KNX configuration store data."""
+
+    entities: KNXEntityStoreModel
+
+
+class KNXConfigStore:
+    """Manage KNX config store data."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry: ConfigEntry,
+    ) -> None:
+        """Initialize config store."""
+        self.hass = hass
+        self._store = Store[KNXConfigStoreModel](hass, STORAGE_VERSION, STORAGE_KEY)
+        self.data = KNXConfigStoreModel(entities={})
+
+        # entities and async_add_entity are filled by platform setups
+        self.entities: dict[str, KnxEntity] = {}  # unique_id as key
+        self.async_add_entity: dict[
+            Platform, Callable[[str, dict[str, Any]], None]
+        ] = {}
+
+    async def load_data(self) -> None:
+        """Load config store data from storage."""
+        if data := await self._store.async_load():
+            self.data = KNXConfigStoreModel(**data)
+            _LOGGER.debug(
+                "Loaded KNX config data from storage. %s entity platforms",
+                len(self.data["entities"]),
+            )
+
+    async def create_entity(
+        self, platform: Platform, data: dict[str, Any]
+    ) -> str | None:
+        """Create a new entity."""
+        if platform not in self.async_add_entity:
+            raise ConfigStoreException(f"Entity platform not ready: {platform}")
+        unique_id = f"knx_es_{ulid_now()}"
+        self.async_add_entity[platform](unique_id, data)
+        # store data after entity was added to be sure config didn't raise exceptions
+        self.data["entities"].setdefault(platform, {})[unique_id] = data
+        await self._store.async_save(self.data)
+
+        entity_registry = er.async_get(self.hass)
+        return entity_registry.async_get_entity_id(platform, DOMAIN, unique_id)
+
+    @callback
+    def get_entity_config(self, entity_id: str) -> dict[str, Any]:
+        """Return KNX entity configuration."""
+        entity_registry = er.async_get(self.hass)
+        if (entry := entity_registry.async_get(entity_id)) is None:
+            raise ConfigStoreException(f"Entity not found: {entity_id}")
+        try:
+            return {
+                CONF_PLATFORM: entry.domain,
+                CONF_DATA: self.data["entities"][entry.domain][entry.unique_id],
+            }
+        except KeyError as err:
+            raise ConfigStoreException(f"Entity data not found: {entity_id}") from err
+
+    async def update_entity(
+        self, platform: Platform, entity_id: str, data: dict[str, Any]
+    ) -> None:
+        """Update an existing entity."""
+        if platform not in self.async_add_entity:
+            raise ConfigStoreException(f"Entity platform not ready: {platform}")
+        entity_registry = er.async_get(self.hass)
+        if (entry := entity_registry.async_get(entity_id)) is None:
+            raise ConfigStoreException(f"Entity not found: {entity_id}")
+        unique_id = entry.unique_id
+        if (
+            platform not in self.data["entities"]
+            or unique_id not in self.data["entities"][platform]
+        ):
+            raise ConfigStoreException(
+                f"Entity not found in storage: {entity_id} - {unique_id}"
+            )
+        await self.entities.pop(unique_id).async_remove()
+        self.async_add_entity[platform](unique_id, data)
+        # store data after entity is added to make sure config doesn't raise exceptions
+        self.data["entities"][platform][unique_id] = data
+        await self._store.async_save(self.data)
+
+    async def delete_entity(self, entity_id: str) -> None:
+        """Delete an existing entity."""
+        entity_registry = er.async_get(self.hass)
+        if (entry := entity_registry.async_get(entity_id)) is None:
+            raise ConfigStoreException(f"Entity not found: {entity_id}")
+        try:
+            del self.data["entities"][entry.domain][entry.unique_id]
+        except KeyError as err:
+            raise ConfigStoreException(
+                f"Entity not found in {entry.domain}: {entry.unique_id}"
+            ) from err
+        try:
+            del self.entities[entry.unique_id]
+        except KeyError:
+            _LOGGER.warning("Entity not initialized when deleted: %s", entity_id)
+        entity_registry.async_remove(entity_id)
+        await self._store.async_save(self.data)
+
+    def get_entity_entries(self) -> list[er.RegistryEntry]:
+        """Get entity_ids of all configured entities by platform."""
+        return [
+            entity.registry_entry
+            for entity in self.entities.values()
+            if entity.registry_entry is not None
+        ]
+
+
+class ConfigStoreException(Exception):
+    """KNX config store exception."""

homeassistant/components/knx/storage/const.py

@@ -0,0 +1,14 @@
+"""Constants used in KNX config store."""
+
+from typing import Final
+
+CONF_DATA: Final = "data"
+CONF_ENTITY: Final = "entity"
+CONF_DEVICE_INFO: Final = "device_info"
+CONF_GA_WRITE: Final = "write"
+CONF_GA_STATE: Final = "state"
+CONF_GA_PASSIVE: Final = "passive"
+CONF_DPT: Final = "dpt"
+
+
+CONF_GA_SWITCH: Final = "ga_switch"