diff --git a/homeassistant/components/knx/__init__.py b/homeassistant/components/knx/__init__.py index 91d1d9fa1c59e3..8fab84c0aea6a4 100644 --- a/homeassistant/components/knx/__init__.py +++ b/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: diff --git a/homeassistant/components/knx/const.py b/homeassistant/components/knx/const.py index d006637abd179d..858f1cefea07b0 100644 --- a/homeassistant/components/knx/const.py +++ b/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.""" diff --git a/homeassistant/components/knx/diagnostics.py b/homeassistant/components/knx/diagnostics.py index 60a41c9a408e04..2fada718d31337 100644 --- a/homeassistant/components/knx/diagnostics.py +++ b/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: diff --git a/homeassistant/components/knx/manifest.json b/homeassistant/components/knx/manifest.json index d3aeced46c9988..ad61b812386ef6 100644 --- a/homeassistant/components/knx/manifest.json +++ b/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.1.0", + "knx_frontend==2023.5.2.143855" + ] } diff --git a/homeassistant/components/knx/project.py b/homeassistant/components/knx/project.py new file mode 100644 index 00000000000000..274ef5cb9a3b6b --- /dev/null +++ b/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() diff --git a/homeassistant/components/knx/websocket.py b/homeassistant/components/knx/websocket.py new file mode 100644 index 00000000000000..d5a41dce1461e3 --- /dev/null +++ b/homeassistant/components/knx/websocket.py @@ -0,0 +1,251 @@ +"""KNX Websocket API.""" +from __future__ import annotations + +from collections.abc import Callable +from typing import Final + +from knx_frontend import get_build_id, locate_dir +import voluptuous as vol +from xknx.dpt import DPTArray +from xknx.exceptions import XKNXException +from xknx.telegram import Telegram, TelegramDirection +from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite +from xknxproject.exceptions import XknxProjectException + +from homeassistant.components import panel_custom, websocket_api +from homeassistant.core import HomeAssistant, callback +import homeassistant.util.dt as dt_util + +from .const import ( + DOMAIN, + AsyncMessageCallbackType, + KNXBusMonitorMessage, + MessageCallbackType, +) +from .project import KNXProject + +URL_BASE: Final = "/knx_static" + + +async def register_panel(hass: HomeAssistant) -> None: + """Register the KNX Panel and Websocket API.""" + websocket_api.async_register_command(hass, ws_info) + websocket_api.async_register_command(hass, ws_project_file_process) + websocket_api.async_register_command(hass, ws_project_file_remove) + websocket_api.async_register_command(hass, ws_group_monitor_info) + websocket_api.async_register_command(hass, ws_subscribe_telegram) + + if DOMAIN not in hass.data.get("frontend_panels", {}): + path = locate_dir() + build_id = get_build_id() + hass.http.register_static_path( + URL_BASE, path, cache_headers=(build_id != "dev") + ) + await panel_custom.async_register_panel( + hass=hass, + frontend_url_path=DOMAIN, + webcomponent_name="knx-frontend", + sidebar_title=DOMAIN.upper(), + sidebar_icon="mdi:bus-electric", + module_url=f"{URL_BASE}/entrypoint-{build_id}.js", + embed_iframe=True, + require_admin=True, + ) + + +@websocket_api.websocket_command( + { + vol.Required("type"): "knx/info", + } +) +@callback +def ws_info( + hass: HomeAssistant, + connection: websocket_api.ActiveConnection, + msg: dict, +) -> None: + """Handle get info command.""" + xknx = hass.data[DOMAIN].xknx + + _project_info = None + if project_info := hass.data[DOMAIN].project.info: + _project_info = { + "name": project_info["name"], + "last_modified": project_info["last_modified"], + "tool_version": project_info["tool_version"], + } + + connection.send_result( + msg["id"], + { + "version": xknx.version, + "connected": xknx.connection_manager.connected.is_set(), + "current_address": str(xknx.current_address), + "project": _project_info, + }, + ) + + +@websocket_api.require_admin +@websocket_api.websocket_command( + { + vol.Required("type"): "knx/project_file_process", + vol.Required("file_id"): str, + vol.Required("password"): str, + } +) +@websocket_api.async_response +async def ws_project_file_process( + hass: HomeAssistant, + connection: websocket_api.ActiveConnection, + msg: dict, +) -> None: + """Handle get info command.""" + knx_project = hass.data[DOMAIN].project + try: + await knx_project.process_project_file( + file_id=msg["file_id"], + password=msg["password"], + ) + except (ValueError, XknxProjectException) as err: + # ValueError could raise from file_upload integration + connection.send_error( + msg["id"], websocket_api.const.ERR_HOME_ASSISTANT_ERROR, str(err) + ) + return + + connection.send_result(msg["id"]) + + +@websocket_api.require_admin +@websocket_api.websocket_command( + { + vol.Required("type"): "knx/project_file_remove", + } +) +@websocket_api.async_response +async def ws_project_file_remove( + hass: HomeAssistant, + connection: websocket_api.ActiveConnection, + msg: dict, +) -> None: + """Handle get info command.""" + knx_project = hass.data[DOMAIN].project + await knx_project.remove_project_file() + connection.send_result(msg["id"]) + + +@websocket_api.websocket_command( + { + vol.Required("type"): "knx/group_monitor_info", + } +) +@callback +def ws_group_monitor_info( + hass: HomeAssistant, + connection: websocket_api.ActiveConnection, + msg: dict, +) -> None: + """Handle get info command of group monitor.""" + project_loaded = hass.data[DOMAIN].project.loaded + connection.send_result( + msg["id"], + {"project_loaded": bool(project_loaded)}, + ) + + +@websocket_api.websocket_command( + { + vol.Required("type"): "knx/subscribe_telegrams", + } +) +@callback +def ws_subscribe_telegram( + hass: HomeAssistant, + connection: websocket_api.ActiveConnection, + msg: dict, +) -> None: + """Subscribe to incoming and outgoing KNX telegrams.""" + project: KNXProject = hass.data[DOMAIN].project + + async def forward_telegrams(telegram: Telegram) -> None: + """Forward events to websocket.""" + payload: str + dpt_payload = None + if isinstance(telegram.payload, (GroupValueWrite, GroupValueResponse)): + dpt_payload = telegram.payload.value + if isinstance(dpt_payload, DPTArray): + payload = f"0x{bytes(dpt_payload.value).hex()}" + else: + payload = f"{dpt_payload.value:d}" + elif isinstance(telegram.payload, GroupValueRead): + payload = "" + else: + return + + direction = ( + "group_monitor_incoming" + if telegram.direction is TelegramDirection.INCOMING + else "group_monitor_outgoing" + ) + dst = str(telegram.destination_address) + src = str(telegram.source_address) + bus_message: KNXBusMonitorMessage = KNXBusMonitorMessage( + destination_address=dst, + destination_text=None, + payload=payload, + type=str(telegram.payload.__class__.__name__), + value=None, + source_address=src, + source_text=None, + direction=direction, + timestamp=dt_util.as_local(dt_util.utcnow()).strftime("%H:%M:%S.%f")[:-3], + ) + if project.loaded: + if ga_infos := project.group_addresses.get(dst): + bus_message["destination_text"] = ga_infos.name + if dpt_payload is not None and ga_infos.transcoder is not None: + try: + value = ga_infos.transcoder.from_knx(dpt_payload) + except XKNXException: + bus_message["value"] = "Error decoding value" + else: + unit = ( + f" {ga_infos.transcoder.unit}" + if ga_infos.transcoder.unit is not None + else "" + ) + bus_message["value"] = f"{value}{unit}" + if ia_infos := project.devices.get(src): + bus_message[ + "source_text" + ] = f"{ia_infos['manufacturer_name']} {ia_infos['name']}" + + connection.send_event( + msg["id"], + bus_message, + ) + + connection.subscriptions[msg["id"]] = async_subscribe_telegrams( + hass, forward_telegrams + ) + + connection.send_result(msg["id"]) + + +def async_subscribe_telegrams( + hass: HomeAssistant, + telegram_callback: AsyncMessageCallbackType | MessageCallbackType, +) -> Callable[[], None]: + """Subscribe to telegram received callback.""" + xknx = hass.data[DOMAIN].xknx + + unregister = xknx.telegram_queue.register_telegram_received_cb( + telegram_callback, match_for_outgoing=True + ) + + def async_remove() -> None: + """Remove callback.""" + xknx.telegram_queue.unregister_telegram_received_cb(unregister) + + return async_remove diff --git a/requirements_all.txt b/requirements_all.txt index 95504b93ea5722..195b438db10615 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1021,6 +1021,9 @@ kegtron-ble==0.4.0 # homeassistant.components.kiwi kiwiki-client==0.1.1 +# homeassistant.components.knx +knx_frontend==2023.5.2.143855 + # homeassistant.components.konnected konnected==1.2.0 @@ -2663,6 +2666,9 @@ xiaomi-ble==0.17.0 # homeassistant.components.knx xknx==2.9.0 +# homeassistant.components.knx +xknxproject==3.1.0 + # homeassistant.components.bluesound # homeassistant.components.fritz # homeassistant.components.rest diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 9364c3e66fd70f..ec93d3319f061b 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -777,6 +777,9 @@ justnimbus==0.6.0 # homeassistant.components.kegtron kegtron-ble==0.4.0 +# homeassistant.components.knx +knx_frontend==2023.5.2.143855 + # homeassistant.components.konnected konnected==1.2.0 @@ -1930,6 +1933,9 @@ xiaomi-ble==0.17.0 # homeassistant.components.knx xknx==2.9.0 +# homeassistant.components.knx +xknxproject==3.1.0 + # homeassistant.components.bluesound # homeassistant.components.fritz # homeassistant.components.rest diff --git a/tests/components/knx/conftest.py b/tests/components/knx/conftest.py index 9cf325086a2f71..fcad9a7be89bd3 100644 --- a/tests/components/knx/conftest.py +++ b/tests/components/knx/conftest.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio +import json from unittest.mock import DEFAULT, AsyncMock, Mock, patch import pytest @@ -11,7 +12,13 @@ from xknx.io import DEFAULT_MCAST_GRP, DEFAULT_MCAST_PORT from xknx.telegram import Telegram, TelegramDirection from xknx.telegram.address import GroupAddress, IndividualAddress -from xknx.telegram.apci import APCI, GroupValueRead, GroupValueResponse, GroupValueWrite +from xknx.telegram.apci import ( + APCI, + GroupValueRead, + GroupValueResponse, + GroupValueWrite, + IndividualAddressRead, +) from homeassistant.components.knx.const import ( CONF_KNX_AUTOMATIC, @@ -26,10 +33,13 @@ DEFAULT_ROUTING_IA, DOMAIN as KNX_DOMAIN, ) +from homeassistant.components.knx.project import STORAGE_KEY as KNX_PROJECT_STORAGE_KEY from homeassistant.core import HomeAssistant from homeassistant.setup import async_setup_component -from tests.common import MockConfigEntry +from tests.common import MockConfigEntry, load_fixture + +FIXTURE_PROJECT_DATA = json.loads(load_fixture("project.json", KNX_DOMAIN)) class KNXTestKit: @@ -181,39 +191,72 @@ def _payload_value(payload: int | tuple[int, ...]) -> DPTArray | DPTBinary: return DPTBinary(payload) return DPTArray(payload) - async def _receive_telegram(self, group_address: str, payload: APCI) -> None: + async def _receive_telegram( + self, + group_address: str, + payload: APCI, + source: str | None = None, + ) -> None: """Inject incoming KNX telegram.""" self.xknx.telegrams.put_nowait( Telegram( destination_address=GroupAddress(group_address), direction=TelegramDirection.INCOMING, payload=payload, - source_address=IndividualAddress(self.INDIVIDUAL_ADDRESS), + source_address=IndividualAddress(source or self.INDIVIDUAL_ADDRESS), ) ) await self.xknx.telegrams.join() await self.hass.async_block_till_done() - async def receive_read( - self, - group_address: str, - ) -> None: + async def receive_individual_address_read(self, source: str | None = None): + """Inject incoming IndividualAddressRead telegram.""" + self.xknx.telegrams.put_nowait( + Telegram( + destination_address=IndividualAddress(self.INDIVIDUAL_ADDRESS), + direction=TelegramDirection.INCOMING, + payload=IndividualAddressRead(), + source_address=IndividualAddress(source or "1.3.5"), + ) + ) + await self.xknx.telegrams.join() + await self.hass.async_block_till_done() + + async def receive_read(self, group_address: str, source: str | None = None) -> None: """Inject incoming GroupValueRead telegram.""" - await self._receive_telegram(group_address, GroupValueRead()) + await self._receive_telegram( + group_address, + GroupValueRead(), + source=source, + ) async def receive_response( - self, group_address: str, payload: int | tuple[int, ...] + self, + group_address: str, + payload: int | tuple[int, ...], + source: str | None = None, ) -> None: """Inject incoming GroupValueResponse telegram.""" payload_value = self._payload_value(payload) - await self._receive_telegram(group_address, GroupValueResponse(payload_value)) + await self._receive_telegram( + group_address, + GroupValueResponse(payload_value), + source=source, + ) async def receive_write( - self, group_address: str, payload: int | tuple[int, ...] + self, + group_address: str, + payload: int | tuple[int, ...], + source: str | None = None, ) -> None: """Inject incoming GroupValueWrite telegram.""" payload_value = self._payload_value(payload) - await self._receive_telegram(group_address, GroupValueWrite(payload_value)) + await self._receive_telegram( + group_address, + GroupValueWrite(payload_value), + source=source, + ) @pytest.fixture @@ -239,3 +282,13 @@ async def knx(request, hass, mock_config_entry: MockConfigEntry): knx_test_kit = KNXTestKit(hass, mock_config_entry) yield knx_test_kit await knx_test_kit.assert_no_telegram() + + +@pytest.fixture +def load_knxproj(hass_storage): + """Mock KNX project data.""" + hass_storage[KNX_PROJECT_STORAGE_KEY] = { + "version": 1, + "data": FIXTURE_PROJECT_DATA, + } + return diff --git a/tests/components/knx/fixtures/project.json b/tests/components/knx/fixtures/project.json new file mode 100644 index 00000000000000..60798d1b245f0e --- /dev/null +++ b/tests/components/knx/fixtures/project.json @@ -0,0 +1,502 @@ +{ + "info": { + "project_id": "P-04FF", + "name": "Fixture", + "last_modified": "2023-04-30T09:04:04.4043671Z", + "group_address_style": "ThreeLevel", + "guid": "6a019e80-5945-489e-95a3-378735c642d1", + "created_by": "ETS5", + "schema_version": "20", + "tool_version": "5.7.1428.39779", + "xknxproject_version": "3.1.0", + "language_code": "de-DE" + }, + "communication_objects": { + "1.0.9/O-57_R-21": { + "name": "Ch A Current Setpoint", + "number": 57, + "text": "Kanal A - Regler", + "function_text": "aktueller Sollwert", + "description": "", + "device_address": "1.0.9", + "dpts": [ + { + "main": 9, + "sub": 1 + } + ], + "object_size": "2 Bytes", + "flags": { + "read": true, + "write": false, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/0/2"] + }, + "1.0.9/O-73_R-29": { + "name": "Ch A On/Off Request Master", + "number": 73, + "text": "Kanal A - Regler", + "function_text": "Regelung aktivieren/deaktivieren", + "description": "", + "device_address": "1.0.9", + "dpts": [ + { + "main": 1, + "sub": 1 + } + ], + "object_size": "1 Bit", + "flags": { + "read": false, + "write": true, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": false + }, + "group_address_links": ["0/0/1"] + }, + "1.1.6/O-4_R-4": { + "name": "DayNight_General_1_GO", + "number": 4, + "text": "Zeit", + "function_text": "Tag (0) / Nacht (1)", + "description": "", + "device_address": "1.1.6", + "dpts": [ + { + "main": 1, + "sub": 24 + } + ], + "object_size": "1 Bit", + "flags": { + "read": false, + "write": true, + "communication": true, + "update": true, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/0/1"] + }, + "1.1.6/O-1_R-1": { + "name": "Time_General_1_GO", + "number": 1, + "text": "Zeit", + "function_text": "Uhrzeit", + "description": "", + "device_address": "1.1.6", + "dpts": [ + { + "main": 10, + "sub": 1 + } + ], + "object_size": "3 Bytes", + "flags": { + "read": false, + "write": true, + "communication": true, + "update": true, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/1/2"] + }, + "1.1.6/O-241_R-124": { + "name": "StatusOnOff_RGB_1_GO", + "number": 241, + "text": "RGB:", + "function_text": "Status An/Aus", + "description": "", + "device_address": "1.1.6", + "dpts": [ + { + "main": 1, + "sub": 1 + } + ], + "object_size": "1 Bit", + "flags": { + "read": true, + "write": false, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/1/0"] + }, + "2.0.5/O-107_R-61": { + "name": "UHRZEIT", + "number": 107, + "text": "Uhrzeit", + "function_text": "Eingang / Ausgang", + "description": "", + "device_address": "2.0.5", + "dpts": [ + { + "main": 10, + "sub": 1 + } + ], + "object_size": "3 Bytes", + "flags": { + "read": true, + "write": true, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/0/3"] + }, + "2.0.5/O-123_R-3923": { + "name": "T_MW_INTERN", + "number": 123, + "text": "Temp.Sensor: Messwert", + "function_text": "Ausgang", + "description": "", + "device_address": "2.0.5", + "dpts": [ + { + "main": 9, + "sub": 1 + } + ], + "object_size": "2 Bytes", + "flags": { + "read": true, + "write": false, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/0/2"] + }, + "2.0.5/O-331_R-254": { + "name": "NACHT_SA", + "number": 331, + "text": "Nacht: Schaltausgang", + "function_text": "Ausgang", + "description": "", + "device_address": "2.0.5", + "dpts": [ + { + "main": 1, + "sub": 1 + } + ], + "object_size": "1 Bit", + "flags": { + "read": true, + "write": false, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/0/1"] + }, + "2.0.15/O-1_R-0": { + "name": "Time", + "number": 1, + "text": "Uhrzeit", + "function_text": "Senden", + "description": "", + "device_address": "2.0.15", + "dpts": [ + { + "main": 10, + "sub": 1 + } + ], + "object_size": "3 Bytes", + "flags": { + "read": false, + "write": false, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": true + }, + "group_address_links": ["0/1/2"] + }, + "2.0.15/O-3_R-2": { + "name": "Trigger send date/time", + "number": 3, + "text": "Trigger sende Datum/Uhrzeit", + "function_text": "Empfangen", + "description": "", + "device_address": "2.0.15", + "dpts": [ + { + "main": 1, + "sub": 17 + } + ], + "object_size": "1 Bit", + "flags": { + "read": false, + "write": true, + "communication": true, + "update": false, + "read_on_init": false, + "transmit": false + }, + "group_address_links": ["0/1/0"] + } + }, + "topology": { + "0": { + "name": "Backbone Bereich", + "description": null, + "lines": { + "0": { + "name": "Bereichslinie", + "description": null, + "devices": [], + "medium_type": "KNXnet/IP (IP)" + } + } + }, + "1": { + "name": "Eins", + "description": null, + "lines": { + "0": { + "name": "Hauptlinie", + "description": null, + "devices": ["1.0.0", "1.0.9"], + "medium_type": "Twisted Pair (TP)" + }, + "1": { + "name": "L1", + "description": null, + "devices": ["1.1.0", "1.1.1", "1.1.6"], + "medium_type": "Twisted Pair (TP)" + } + } + }, + "2": { + "name": "Zwei", + "description": null, + "lines": { + "0": { + "name": "Hauptlinie", + "description": null, + "devices": ["2.0.0", "2.0.5", "2.0.6", "2.0.15"], + "medium_type": "Twisted Pair (TP)" + } + } + } + }, + "devices": { + "1.0.0": { + "name": "KNX IP Router 752 secure", + "hardware_name": "KNX IP Router 752 secure", + "description": "", + "manufacturer_name": "Weinzierl Engineering GmbH", + "individual_address": "1.0.0", + "project_uid": 6, + "communication_object_ids": [] + }, + "1.0.9": { + "name": "HCC/S2.2.1.1 Heiz-/Kühlkreis Controller,3-Punkt,2-fach,REG", + "hardware_name": "HCC/S2.2.1.1 Heiz-/Kühlkreis Controller,3-Punkt,2-fach,REG", + "description": "", + "manufacturer_name": "ABB", + "individual_address": "1.0.9", + "project_uid": 30, + "communication_object_ids": ["1.0.9/O-57_R-21", "1.0.9/O-73_R-29"] + }, + "1.1.0": { + "name": "Bereichs-/Linienkoppler REG", + "hardware_name": "Bereichs-/Linienkoppler REG", + "description": "", + "manufacturer_name": "Albrecht Jung", + "individual_address": "1.1.0", + "project_uid": 23, + "communication_object_ids": [] + }, + "1.1.1": { + "name": "SCN-IP000.03 IP Interface mit Secure", + "hardware_name": "IP Interface Secure", + "description": "", + "manufacturer_name": "MDT technologies", + "individual_address": "1.1.1", + "project_uid": 24, + "communication_object_ids": [] + }, + "1.1.6": { + "name": "Enertex KNX LED Dimmsequenzer 20A/5x REG", + "hardware_name": "LED Dimmsequenzer 20A/5x REG/DK", + "description": "", + "manufacturer_name": "Enertex Bayern GmbH", + "individual_address": "1.1.6", + "project_uid": 29, + "communication_object_ids": [ + "1.1.6/O-4_R-4", + "1.1.6/O-1_R-1", + "1.1.6/O-241_R-124" + ] + }, + "2.0.0": { + "name": "KNX/IP-Router", + "hardware_name": "IP Router", + "description": "", + "manufacturer_name": "GIRA Giersiepen", + "individual_address": "2.0.0", + "project_uid": 17, + "communication_object_ids": [] + }, + "2.0.5": { + "name": "Suntracer KNX pro", + "hardware_name": "KNX Suntracer Pro", + "description": "", + "manufacturer_name": "Elsner Elektronik GmbH", + "individual_address": "2.0.5", + "project_uid": 31, + "communication_object_ids": [ + "2.0.5/O-107_R-61", + "2.0.5/O-123_R-3923", + "2.0.5/O-331_R-254" + ] + }, + "2.0.6": { + "name": "KNX Modbus TCP Gateway 716", + "hardware_name": "KNX Modbus TCP Gateway 716", + "description": "", + "manufacturer_name": "Weinzierl Engineering GmbH", + "individual_address": "2.0.6", + "project_uid": 32, + "communication_object_ids": [] + }, + "2.0.15": { + "name": "KNX/IP-Router", + "hardware_name": "Router Applications", + "description": "", + "manufacturer_name": "GIRA Giersiepen", + "individual_address": "2.0.15", + "project_uid": 50, + "communication_object_ids": ["2.0.15/O-1_R-0", "2.0.15/O-3_R-2"] + } + }, + "group_addresses": { + "0/0/1": { + "name": "Binary", + "identifier": "GA-1", + "raw_address": 1, + "address": "0/0/1", + "project_uid": 43, + "dpt": { + "main": 1, + "sub": 1 + }, + "communication_object_ids": [ + "1.0.9/O-73_R-29", + "1.1.6/O-4_R-4", + "2.0.5/O-331_R-254" + ], + "description": "" + }, + "0/0/2": { + "name": "2-byte float", + "identifier": "GA-2", + "raw_address": 2, + "address": "0/0/2", + "project_uid": 44, + "dpt": { + "main": 9, + "sub": 1 + }, + "communication_object_ids": ["1.0.9/O-57_R-21", "2.0.5/O-123_R-3923"], + "description": "" + }, + "0/0/3": { + "name": "daytime", + "identifier": "GA-3", + "raw_address": 3, + "address": "0/0/3", + "project_uid": 45, + "dpt": { + "main": 10, + "sub": 1 + }, + "communication_object_ids": ["2.0.5/O-107_R-61"], + "description": "" + }, + "0/0/4": { + "name": "RGB color", + "identifier": "GA-7", + "raw_address": 4, + "address": "0/0/4", + "project_uid": 69, + "dpt": { + "main": 232, + "sub": 600 + }, + "communication_object_ids": [], + "description": "" + }, + "0/1/0": { + "name": "binary (1.017)", + "identifier": "GA-4", + "raw_address": 256, + "address": "0/1/0", + "project_uid": 47, + "dpt": { + "main": 1, + "sub": 17 + }, + "communication_object_ids": ["1.1.6/O-241_R-124", "2.0.15/O-3_R-2"], + "description": "" + }, + "0/1/1": { + "name": "percent", + "identifier": "GA-5", + "raw_address": 257, + "address": "0/1/1", + "project_uid": 48, + "dpt": { + "main": 5, + "sub": 1 + }, + "communication_object_ids": [], + "description": "" + }, + "0/1/2": { + "name": "daytime", + "identifier": "GA-6", + "raw_address": 258, + "address": "0/1/2", + "project_uid": 49, + "dpt": { + "main": 10, + "sub": 1 + }, + "communication_object_ids": ["1.1.6/O-1_R-1", "2.0.15/O-1_R-0"], + "description": "" + } + }, + "locations": { + "Neues Projekt": { + "type": "Building", + "identifier": "P-04FF-0_BP-1", + "name": "Neues Projekt", + "usage_id": null, + "usage_text": "", + "number": "", + "description": "", + "project_uid": 3, + "devices": [], + "spaces": {} + } + } +} diff --git a/tests/components/knx/test_diagnostic.py b/tests/components/knx/test_diagnostic.py index 99ce43998916fe..df8cb71d4af958 100644 --- a/tests/components/knx/test_diagnostic.py +++ b/tests/components/knx/test_diagnostic.py @@ -55,6 +55,7 @@ async def test_diagnostics( }, "configuration_error": None, "configuration_yaml": None, + "project_info": None, "xknx": {"current_address": "0.0.0", "version": "1.0.0"}, } @@ -85,6 +86,7 @@ async def test_diagnostic_config_error( }, "configuration_error": "extra keys not allowed @ data['knx']['wrong_key']", "configuration_yaml": {"wrong_key": {}}, + "project_info": None, "xknx": {"current_address": "0.0.0", "version": "1.0.0"}, } @@ -134,5 +136,34 @@ async def test_diagnostic_redact( }, "configuration_error": None, "configuration_yaml": None, + "project_info": None, "xknx": {"current_address": "0.0.0", "version": "1.0.0"}, } + + +@pytest.mark.parametrize("hass_config", [{}]) +async def test_diagnostics_project( + hass: HomeAssistant, + hass_client: ClientSessionGenerator, + mock_config_entry: MockConfigEntry, + knx: KNXTestKit, + mock_hass_config: None, + load_knxproj: None, +) -> None: + """Test diagnostics.""" + await knx.setup_integration({}) + diag = await get_diagnostics_for_config_entry(hass, hass_client, mock_config_entry) + + assert "config_entry_data" in diag + assert "configuration_error" in diag + assert "configuration_yaml" in diag + assert "project_info" in diag + assert "xknx" in diag + # project specific fields + assert "created_by" in diag["project_info"] + assert "group_address_style" in diag["project_info"] + assert "last_modified" in diag["project_info"] + assert "schema_version" in diag["project_info"] + assert "tool_version" in diag["project_info"] + assert "language_code" in diag["project_info"] + assert diag["project_info"]["name"] == "**REDACTED**" diff --git a/tests/components/knx/test_websocket.py b/tests/components/knx/test_websocket.py new file mode 100644 index 00000000000000..c053e4fa9cbd2f --- /dev/null +++ b/tests/components/knx/test_websocket.py @@ -0,0 +1,282 @@ +"""KNX Websocket Tests.""" +from typing import Any +from unittest.mock import patch + +from homeassistant.components.knx import DOMAIN, KNX_ADDRESS, SwitchSchema +from homeassistant.const import CONF_NAME +from homeassistant.core import HomeAssistant + +from .conftest import FIXTURE_PROJECT_DATA, KNXTestKit + +from tests.typing import WebSocketGenerator + + +async def test_knx_info_command( + hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator +): + """Test knx/info command.""" + await knx.setup_integration({}) + client = await hass_ws_client(hass) + await client.send_json({"id": 6, "type": "knx/info"}) + + res = await client.receive_json() + assert res["success"], res + assert res["result"]["version"] is not None + assert res["result"]["connected"] + assert res["result"]["current_address"] == "0.0.0" + assert res["result"]["project"] is None + + +async def test_knx_info_command_with_project( + hass: HomeAssistant, + knx: KNXTestKit, + hass_ws_client: WebSocketGenerator, + load_knxproj: None, +): + """Test knx/info command with loaded project.""" + await knx.setup_integration({}) + client = await hass_ws_client(hass) + await client.send_json({"id": 6, "type": "knx/info"}) + + res = await client.receive_json() + assert res["success"], res + assert res["result"]["version"] is not None + assert res["result"]["connected"] + assert res["result"]["current_address"] == "0.0.0" + assert res["result"]["project"] is not None + assert res["result"]["project"]["name"] == "Fixture" + assert res["result"]["project"]["last_modified"] == "2023-04-30T09:04:04.4043671Z" + assert res["result"]["project"]["tool_version"] == "5.7.1428.39779" + + +async def test_knx_project_file_process( + hass: HomeAssistant, + knx: KNXTestKit, + hass_ws_client: WebSocketGenerator, + hass_storage: dict[str, Any], +): + """Test knx/project_file_process command for storing and loading new data.""" + _file_id = "1234" + _password = "pw-test" + _parse_result = FIXTURE_PROJECT_DATA + + await knx.setup_integration({}) + client = await hass_ws_client(hass) + assert not hass.data[DOMAIN].project.loaded + + await client.send_json( + { + "id": 6, + "type": "knx/project_file_process", + "file_id": _file_id, + "password": _password, + } + ) + with patch( + "homeassistant.components.knx.project.process_uploaded_file", + ) as file_upload_mock, patch( + "xknxproject.XKNXProj.parse", return_value=_parse_result + ) as parse_mock: + file_upload_mock.return_value.__enter__.return_value = "" + res = await client.receive_json() + + file_upload_mock.assert_called_once_with(hass, _file_id) + parse_mock.assert_called_once_with() + + assert res["success"], res + assert hass.data[DOMAIN].project.loaded + + +async def test_knx_project_file_process_error( + hass: HomeAssistant, + knx: KNXTestKit, + hass_ws_client: WebSocketGenerator, +): + """Test knx/project_file_process exception handling.""" + await knx.setup_integration({}) + client = await hass_ws_client(hass) + assert not hass.data[DOMAIN].project.loaded + + await client.send_json( + { + "id": 6, + "type": "knx/project_file_process", + "file_id": "1234", + "password": "", + } + ) + with patch( + "homeassistant.components.knx.project.process_uploaded_file", + ) as file_upload_mock, patch( + "xknxproject.XKNXProj.parse", side_effect=ValueError + ) as parse_mock: + file_upload_mock.return_value.__enter__.return_value = "" + res = await client.receive_json() + parse_mock.assert_called_once_with() + + assert res["error"], res + assert not hass.data[DOMAIN].project.loaded + + +async def test_knx_project_file_remove( + hass: HomeAssistant, + knx: KNXTestKit, + hass_ws_client: WebSocketGenerator, + load_knxproj: None, +): + """Test knx/project_file_remove command.""" + await knx.setup_integration({}) + client = await hass_ws_client(hass) + assert hass.data[DOMAIN].project.loaded + + await client.send_json({"id": 6, "type": "knx/project_file_remove"}) + with patch("homeassistant.helpers.storage.Store.async_remove") as remove_mock: + res = await client.receive_json() + remove_mock.assert_called_once_with() + + assert res["success"], res + assert not hass.data[DOMAIN].project.loaded + + +async def test_knx_group_monitor_info_command( + hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator +): + """Test knx/group_monitor_info command.""" + await knx.setup_integration({}) + client = await hass_ws_client(hass) + + await client.send_json({"id": 6, "type": "knx/group_monitor_info"}) + + res = await client.receive_json() + assert res["success"], res + assert res["result"]["project_loaded"] is False + + +async def test_knx_subscribe_telegrams_command_no_project( + hass: HomeAssistant, knx: KNXTestKit, hass_ws_client: WebSocketGenerator +): + """Test knx/subscribe_telegrams command without project data.""" + await knx.setup_integration( + { + SwitchSchema.PLATFORM: { + CONF_NAME: "test", + KNX_ADDRESS: "1/2/4", + } + } + ) + client = await hass_ws_client(hass) + await client.send_json({"id": 6, "type": "knx/subscribe_telegrams"}) + res = await client.receive_json() + assert res["success"], res + + # send incoming events + await knx.receive_read("1/2/3") + await knx.receive_write("1/3/4", True) + await knx.receive_write("1/3/4", False) + await knx.receive_individual_address_read() + await knx.receive_write("1/3/8", (0x34, 0x45)) + # send outgoing events + await hass.services.async_call( + "switch", "turn_on", {"entity_id": "switch.test"}, blocking=True + ) + await knx.assert_write("1/2/4", 1) + + # receive events + res = await client.receive_json() + assert res["event"]["destination_address"] == "1/2/3" + assert res["event"]["payload"] == "" + assert res["event"]["type"] == "GroupValueRead" + assert res["event"]["source_address"] == "1.2.3" + assert res["event"]["direction"] == "group_monitor_incoming" + assert res["event"]["timestamp"] is not None + + res = await client.receive_json() + assert res["event"]["destination_address"] == "1/3/4" + assert res["event"]["payload"] == "1" + assert res["event"]["type"] == "GroupValueWrite" + assert res["event"]["source_address"] == "1.2.3" + assert res["event"]["direction"] == "group_monitor_incoming" + assert res["event"]["timestamp"] is not None + + res = await client.receive_json() + assert res["event"]["destination_address"] == "1/3/4" + assert res["event"]["payload"] == "0" + assert res["event"]["type"] == "GroupValueWrite" + assert res["event"]["source_address"] == "1.2.3" + assert res["event"]["direction"] == "group_monitor_incoming" + assert res["event"]["timestamp"] is not None + + res = await client.receive_json() + assert res["event"]["destination_address"] == "1/3/8" + assert res["event"]["payload"] == "0x3445" + assert res["event"]["type"] == "GroupValueWrite" + assert res["event"]["source_address"] == "1.2.3" + assert res["event"]["direction"] == "group_monitor_incoming" + assert res["event"]["timestamp"] is not None + + res = await client.receive_json() + assert res["event"]["destination_address"] == "1/2/4" + assert res["event"]["payload"] == "1" + assert res["event"]["type"] == "GroupValueWrite" + assert ( + res["event"]["source_address"] == "0.0.0" + ) # needs to be the IA currently connected to + assert res["event"]["direction"] == "group_monitor_outgoing" + assert res["event"]["timestamp"] is not None + + +async def test_knx_subscribe_telegrams_command_project( + hass: HomeAssistant, + knx: KNXTestKit, + hass_ws_client: WebSocketGenerator, + load_knxproj: None, +): + """Test knx/subscribe_telegrams command with project data.""" + await knx.setup_integration({}) + client = await hass_ws_client(hass) + await client.send_json({"id": 6, "type": "knx/subscribe_telegrams"}) + res = await client.receive_json() + assert res["success"], res + + # incoming DPT 1 telegram + await knx.receive_write("0/0/1", True) + res = await client.receive_json() + assert res["event"]["destination_address"] == "0/0/1" + assert res["event"]["destination_text"] == "Binary" + assert res["event"]["payload"] == "1" + assert res["event"]["type"] == "GroupValueWrite" + assert res["event"]["source_address"] == "1.2.3" + assert res["event"]["direction"] == "group_monitor_incoming" + assert res["event"]["timestamp"] is not None + + # incoming DPT 5 telegram + await knx.receive_write("0/1/1", (0x50,), source="1.1.6") + res = await client.receive_json() + assert res["event"]["destination_address"] == "0/1/1" + assert res["event"]["destination_text"] == "percent" + assert res["event"]["payload"] == "0x50" + assert res["event"]["value"] == "31 %" + assert res["event"]["type"] == "GroupValueWrite" + assert res["event"]["source_address"] == "1.1.6" + assert ( + res["event"]["source_text"] + == "Enertex Bayern GmbH Enertex KNX LED Dimmsequenzer 20A/5x REG" + ) + assert res["event"]["direction"] == "group_monitor_incoming" + assert res["event"]["timestamp"] is not None + + # incoming undecodable telegram (wrong payload type) + await knx.receive_write("0/1/1", True, source="1.1.6") + res = await client.receive_json() + assert res["event"]["destination_address"] == "0/1/1" + assert res["event"]["destination_text"] == "percent" + assert res["event"]["payload"] == "1" + assert res["event"]["value"] == "Error decoding value" + assert res["event"]["type"] == "GroupValueWrite" + assert res["event"]["source_address"] == "1.1.6" + assert ( + res["event"]["source_text"] + == "Enertex Bayern GmbH Enertex KNX LED Dimmsequenzer 20A/5x REG" + ) + assert res["event"]["direction"] == "group_monitor_incoming" + assert res["event"]["timestamp"] is not None