Add a custom panel for KNX with a group monitor

homeassistant/components/knx/__init__.py

@@ -71,6 +71,7 @@
 )
 from .device import KNXInterfaceDevice
 from .expose import KNXExposeSensor, KNXExposeTime, create_knx_exposure
+from .project import KNXProject
 from .schema import (
     BinarySensorSchema,
     ButtonSchema,
@@ -91,6 +92,7 @@
     ga_validator,
     sensor_type_validator,
 )
+from .websocket import register_panel
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -222,6 +224,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
 
     conf = dict(conf)
     hass.data[DATA_KNX_CONFIG] = conf
+
     return True
 
 
@@ -304,6 +307,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         schema=SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA,
     )
 
+    await register_panel(hass)
+
     return True
 
 
@@ -368,6 +373,8 @@ def __init__(
         self.service_exposures: dict[str, KNXExposeSensor | KNXExposeTime] = {}
         self.entry = entry
 
+        self.project = KNXProject(hass=hass, entry=entry)
+
         self.xknx = XKNX(
             connection_config=self.connection_config(),
             rate_limit=self.entry.data[CONF_KNX_RATE_LIMIT],
@@ -393,6 +400,7 @@ def __init__(
 
     async def start(self) -> None:
         """Start XKNX object. Connect to tunneling or Routing device."""
+        await self.project.load_project()
         await self.xknx.start()
 
     async def stop(self, event: Event | None = None) -> None:

homeassistant/components/knx/const.py

@@ -1,9 +1,12 @@
 """Constants for the KNX integration."""
 from __future__ import annotations
 
+from collections.abc import Awaitable, Callable
 from enum import Enum
 from typing import Final, TypedDict
 
+from xknx.telegram import Telegram
+
 from homeassistant.components.climate import (
     PRESET_AWAY,
     PRESET_COMFORT,
@@ -76,6 +79,9 @@
 ATTR_COUNTER: Final = "counter"
 ATTR_SOURCE: Final = "source"
 
+AsyncMessageCallbackType = Callable[[Telegram], Awaitable[None]]
+MessageCallbackType = Callable[[Telegram], None]
+
 
 class KNXConfigEntryData(TypedDict, total=False):
     """Config entry for the KNX integration."""
@@ -101,6 +107,20 @@ class KNXConfigEntryData(TypedDict, total=False):
     sync_latency_tolerance: int | None
 
 
+class KNXBusMonitorMessage(TypedDict):
+    """KNX bus monitor message."""
+
+    destination_address: str
+    destination_text: str | None
+    payload: str
+    type: str
+    value: str | None
+    source_address: str
+    source_text: str | None
+    direction: str
+    timestamp: str
+
+
 class ColorTempModes(Enum):
     """Color temperature modes for config validation."""
 

homeassistant/components/knx/diagnostics.py

@@ -40,6 +40,11 @@ async def async_get_config_entry_diagnostics(
 
     diag["config_entry_data"] = async_redact_data(dict(config_entry.data), TO_REDACT)
 
+    if proj_info := knx_module.project.info:
+        diag["project_info"] = async_redact_data(proj_info, "name")
+    else:
+        diag["project_info"] = None
+
     raw_config = await conf_util.async_hass_config_yaml(hass)
     diag["configuration_yaml"] = raw_config.get(DOMAIN)
     try:

homeassistant/components/knx/manifest.json

@@ -1,13 +1,18 @@
 {
   "domain": "knx",
   "name": "KNX",
+  "after_dependencies": ["panel_custom"],
   "codeowners": ["@Julius2342", "@farmio", "@marvin-w"],
   "config_flow": true,
-  "dependencies": ["file_upload"],
+  "dependencies": ["file_upload", "websocket_api"],
   "documentation": "https://www.home-assistant.io/integrations/knx",
   "integration_type": "hub",
   "iot_class": "local_push",
-  "loggers": ["xknx"],
+  "loggers": ["xknx", "xknxproject"],
   "quality_scale": "platinum",
-  "requirements": ["xknx==2.9.0"]
+  "requirements": [
+    "xknx==2.9.0",
+    "xknxproject==3.0.0",
+    "knx_frontend==2023.5.2.143855"
+  ]
 }

homeassistant/components/knx/project.py

@@ -0,0 +1,117 @@
+"""Handle KNX project data."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+from typing import Final
+
+from xknx.dpt import DPTBase
+from xknxproject import XKNXProj
+from xknxproject.models import (
+    Device,
+    GroupAddress as GroupAddressModel,
+    KNXProject as KNXProjectModel,
+    ProjectInfo,
+)
+
+from homeassistant.components.file_upload import process_uploaded_file
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.storage import Store
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STORAGE_VERSION: Final = 1
+STORAGE_KEY: Final = f"{DOMAIN}/knx_project.json"
+
+
+@dataclass
+class GroupAddressInfo:
+    """Group address info for runtime usage."""
+
+    address: str
+    name: str
+    description: str
+    dpt_main: int | None
+    dpt_sub: int | None
+    transcoder: type[DPTBase] | None
+
+
+def _create_group_address_info(ga_model: GroupAddressModel) -> GroupAddressInfo:
+    """Convert GroupAddress dict value into GroupAddressInfo instance."""
+    dpt = ga_model["dpt"]
+    transcoder = DPTBase.transcoder_by_dpt(dpt["main"], dpt.get("sub")) if dpt else None
+    return GroupAddressInfo(
+        address=ga_model["address"],
+        name=ga_model["name"],
+        description=ga_model["description"],
+        transcoder=transcoder,
+        dpt_main=dpt["main"] if dpt else None,
+        dpt_sub=dpt["sub"] if dpt else None,
+    )
+
+
+class KNXProject:
+    """Manage KNX project data."""
+
+    loaded: bool
+    devices: dict[str, Device]
+    group_addresses: dict[str, GroupAddressInfo]
+    info: ProjectInfo | None
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry: ConfigEntry,
+    ) -> None:
+        """Initialize project data."""
+        self.hass = hass
+        self._store = Store[KNXProjectModel](hass, STORAGE_VERSION, STORAGE_KEY)
+
+        self.initial_state()
+
+    def initial_state(self) -> None:
+        """Set initial state for project data."""
+        self.loaded = False
+        self.devices = {}
+        self.group_addresses = {}
+        self.info = None
+
+    async def load_project(self, data: KNXProjectModel | None = None) -> None:
+        """Load project data from storage."""
+        if project := data or await self._store.async_load():
+            self.devices = project["devices"]
+            self.info = project["info"]
+
+            for ga_model in project["group_addresses"].values():
+                ga_info = _create_group_address_info(ga_model)
+                self.group_addresses[ga_info.address] = ga_info
+
+            _LOGGER.debug(
+                "Loaded KNX project data with %s group addresses from storage",
+                len(self.group_addresses),
+            )
+            self.loaded = True
+
+    async def process_project_file(self, file_id: str, password: str) -> None:
+        """Process an uploaded project file."""
+
+        def _parse_project() -> KNXProjectModel:
+            with process_uploaded_file(self.hass, file_id) as file_path:
+                xknxproj = XKNXProj(
+                    file_path,
+                    password=password,
+                    language=self.hass.config.language,
+                )
+                return xknxproj.parse()
+
+        project = await self.hass.async_add_executor_job(_parse_project)
+        await self._store.async_save(project)
+        await self.load_project(data=project)
+
+    async def remove_project_file(self) -> None:
+        """Remove project file from storage."""
+        await self._store.async_remove()
+        self.initial_state()