From 297d318e154f66be7f83129acb1579d91969b986 Mon Sep 17 00:00:00 2001 From: bahmcloud Date: Fri, 13 Feb 2026 09:26:56 +0100 Subject: [PATCH] Initial HA KNX Bridge scaffold --- .gitignore | 10 + README.md | 46 +++ custom_components/ha_knx_bridge/__init__.py | 31 ++ custom_components/ha_knx_bridge/bridge.py | 307 ++++++++++++++++++ .../ha_knx_bridge/config_flow.py | 134 ++++++++ custom_components/ha_knx_bridge/const.py | 13 + custom_components/ha_knx_bridge/manifest.json | 13 + custom_components/ha_knx_bridge/strings.json | 46 +++ hacs.json | 5 + 9 files changed, 605 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 custom_components/ha_knx_bridge/__init__.py create mode 100644 custom_components/ha_knx_bridge/bridge.py create mode 100644 custom_components/ha_knx_bridge/config_flow.py create mode 100644 custom_components/ha_knx_bridge/const.py create mode 100644 custom_components/ha_knx_bridge/manifest.json create mode 100644 custom_components/ha_knx_bridge/strings.json create mode 100644 hacs.json diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..01d0600 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.idea/ +.DS_Store +__pycache__/ +*.pyc +.pytest_cache/ +.ruff_cache/ +.mypy_cache/ +.vscode/ +*.log +custom_components/**/__pycache__/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..cd2c3ad --- /dev/null +++ b/README.md @@ -0,0 +1,46 @@ +# HA KNX Bridge + +Bridge existing Home Assistant entities to KNX group addresses. The integration lets you +add "Ports" that bind an HA entity to KNX addresses for state updates and commands. + +Current minimal scope: +- Binary Sensor port (state -> KNX) +- Cover port (KNX -> HA commands, HA state -> KNX) + +## Requirements +- Home Assistant 2025.12 or newer (tested for compatibility with 2026.2). +- The built-in KNX integration must be set up and running. + +## Install (HACS) +1. Add this repository as a custom integration in HACS. +2. Install the integration. +3. Restart Home Assistant. +4. Add the "HA KNX Bridge" integration from the UI. + +## Configure +1. During setup, select the existing Home Assistant KNX integration entry. +2. Add Ports from the integration's configuration page. + +### Binary Sensor Port +- `entity_id`: the HA binary_sensor to mirror. +- `state_address` (DPT 1): KNX group address that receives the HA state. + +### Cover Port +- `entity_id`: the HA cover to control/monitor. +- `move_long_address` (DPT 1.008 Up/Down): KNX group address for open/close commands. +- `stop_address` (DPT 1): KNX group address for stop commands. +- `position_address` (DPT 5.001): KNX group address for set position. +- `position_state_address` (DPT 5.001): KNX group address that receives HA position updates. +- `angle_address` (DPT 5.001): KNX group address for set tilt position. +- `angle_state_address` (DPT 5.001): KNX group address that receives HA tilt updates. + +If a group address is left empty, it is ignored. + +## Notes +- For DPT 1.008 (Up/Down), the bridge treats `0 = Up/Open` and `1 = Down/Close`. +- For DPT 5.001, values are interpreted as 0-100 percent where 0 = closed and 100 = open. + +## Roadmap +- Optional standalone KNX connection (without requiring the HA KNX integration). +- Additional entity types (light, switch, sensor, climate). +- Advanced DPT mapping options and inversion settings. diff --git a/custom_components/ha_knx_bridge/__init__.py b/custom_components/ha_knx_bridge/__init__.py new file mode 100644 index 0000000..f4fb611 --- /dev/null +++ b/custom_components/ha_knx_bridge/__init__.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryNotReady + +from .bridge import BridgeManager + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + manager = BridgeManager(hass, entry) + try: + await manager.async_setup() + except ConfigEntryNotReady: + await manager.async_unload() + raise + entry.runtime_data = manager + entry.async_on_unload(entry.add_update_listener(_update_listener)) + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + manager: BridgeManager | None = getattr(entry, "runtime_data", None) + if manager is None: + return True + await manager.async_unload() + return True + + +async def _update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None: + await hass.config_entries.async_reload(entry.entry_id) diff --git a/custom_components/ha_knx_bridge/bridge.py b/custom_components/ha_knx_bridge/bridge.py new file mode 100644 index 0000000..943e7a4 --- /dev/null +++ b/custom_components/ha_knx_bridge/bridge.py @@ -0,0 +1,307 @@ +from __future__ import annotations + +from dataclasses import dataclass +import logging +from typing import Any + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import Event, HomeAssistant, State +from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers import event as event_helper + +from .const import ( + CONF_ANGLE_ADDRESS, + CONF_ANGLE_STATE_ADDRESS, + CONF_MOVE_LONG_ADDRESS, + CONF_MOVE_SHORT_ADDRESS, + CONF_POSITION_ADDRESS, + CONF_POSITION_STATE_ADDRESS, + CONF_STATE_ADDRESS, + CONF_STOP_ADDRESS, +) + +_LOGGER = logging.getLogger(__name__) + +KNX_DOMAIN = "knx" + + +@dataclass(frozen=True) +class BinarySensorPort: + entity_id: str + state_address: str | None + + +@dataclass(frozen=True) +class CoverPort: + entity_id: str + move_long_address: str | None + move_short_address: str | None + stop_address: str | None + position_address: str | None + position_state_address: str | None + angle_address: str | None + angle_state_address: str | None + + +class BridgeManager: + def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: + self.hass = hass + self.entry = entry + self._unsub_listeners: list[callable] = [] + self._knx_event_unsub: callable | None = None + self._address_handlers: dict[str, callable[[Event], Any]] = {} + self._registered_addresses: list[tuple[str, str | None]] = [] + + async def async_setup(self) -> None: + if not self.hass.services.has_service(KNX_DOMAIN, "send"): + raise ConfigEntryNotReady("KNX integration services not available") + + binary_ports, cover_ports = self._load_ports() + self._register_outgoing(binary_ports, cover_ports) + await self._register_incoming(cover_ports) + + async def async_unload(self) -> None: + for unsub in self._unsub_listeners: + unsub() + self._unsub_listeners.clear() + + if self._knx_event_unsub is not None: + self._knx_event_unsub() + self._knx_event_unsub = None + + await self._unregister_knx_events() + + def _load_ports(self) -> tuple[list[BinarySensorPort], list[CoverPort]]: + binary_ports: list[BinarySensorPort] = [] + cover_ports: list[CoverPort] = [] + + subentries = getattr(self.entry, "subentries", []) + for subentry in subentries: + data = subentry.data + if subentry.type == "binary_sensor": + binary_ports.append( + BinarySensorPort( + entity_id=data["entity_id"], + state_address=_clean_address(data.get(CONF_STATE_ADDRESS)), + ) + ) + elif subentry.type == "cover": + cover_ports.append( + CoverPort( + entity_id=data["entity_id"], + move_long_address=_clean_address(data.get(CONF_MOVE_LONG_ADDRESS)), + move_short_address=_clean_address(data.get(CONF_MOVE_SHORT_ADDRESS)), + stop_address=_clean_address(data.get(CONF_STOP_ADDRESS)), + position_address=_clean_address(data.get(CONF_POSITION_ADDRESS)), + position_state_address=_clean_address( + data.get(CONF_POSITION_STATE_ADDRESS) + ), + angle_address=_clean_address(data.get(CONF_ANGLE_ADDRESS)), + angle_state_address=_clean_address( + data.get(CONF_ANGLE_STATE_ADDRESS) + ), + ) + ) + + return binary_ports, cover_ports + + def _register_outgoing( + self, binary_ports: list[BinarySensorPort], cover_ports: list[CoverPort] + ) -> None: + for port in binary_ports: + if not port.state_address: + continue + self._unsub_listeners.append( + event_helper.async_track_state_change_event( + self.hass, [port.entity_id], self._binary_sensor_changed(port) + ) + ) + + for port in cover_ports: + if not (port.position_state_address or port.angle_state_address): + continue + self._unsub_listeners.append( + event_helper.async_track_state_change_event( + self.hass, [port.entity_id], self._cover_changed(port) + ) + ) + + async def _register_incoming(self, cover_ports: list[CoverPort]) -> None: + for port in cover_ports: + self._register_knx_address(port.move_long_address, None, port, "move_long") + self._register_knx_address(port.move_short_address, None, port, "move_short") + self._register_knx_address(port.stop_address, None, port, "stop") + self._register_knx_address( + port.position_address, "percent", port, "position" + ) + self._register_knx_address( + port.angle_address, "percent", port, "angle" + ) + + if self._address_handlers: + self._knx_event_unsub = event_helper.async_track_event( + self.hass, "knx_event", self._handle_knx_event + ) + + await self._register_knx_events() + + def _register_knx_address( + self, + address: str | None, + value_type: str | None, + port: CoverPort, + action: str, + ) -> None: + if not address: + return + + self._address_handlers[address] = lambda event: self._handle_cover_action( + port, action, event + ) + self._registered_addresses.append((address, value_type)) + + async def _register_knx_events(self) -> None: + for address, value_type in self._registered_addresses: + data: dict[str, Any] = {"address": address} + if value_type: + data["type"] = value_type + await self.hass.services.async_call( + KNX_DOMAIN, "event_register", data, blocking=False + ) + + async def _unregister_knx_events(self) -> None: + for address, value_type in self._registered_addresses: + data: dict[str, Any] = {"address": address, "remove": True} + if value_type: + data["type"] = value_type + await self.hass.services.async_call( + KNX_DOMAIN, "event_register", data, blocking=False + ) + self._registered_addresses.clear() + + def _binary_sensor_changed(self, port: BinarySensorPort) -> callable[[Event], Any]: + async def _handler(event: Event) -> None: + new_state: State | None = event.data.get("new_state") + if new_state is None: + return + if new_state.state not in ("on", "off"): + return + payload = 1 if new_state.state == "on" else 0 + await self._knx_send_raw(port.state_address, payload) + + return _handler + + def _cover_changed(self, port: CoverPort) -> callable[[Event], Any]: + async def _handler(event: Event) -> None: + new_state: State | None = event.data.get("new_state") + if new_state is None: + return + + if port.position_state_address is not None: + position = new_state.attributes.get("current_position") + if position is not None: + await self._knx_send_percent(port.position_state_address, position) + + if port.angle_state_address is not None: + angle = new_state.attributes.get("current_tilt_position") + if angle is not None: + await self._knx_send_percent(port.angle_state_address, angle) + + return _handler + + async def _handle_knx_event(self, event: Event) -> None: + if event.data.get("direction") != "Incoming": + return + + destination = event.data.get("destination") + if not destination: + return + + handler = self._address_handlers.get(destination) + if handler is None: + return + + await handler(event) + + async def _handle_cover_action( + self, port: CoverPort, action: str, event: Event + ) -> None: + value = _extract_event_value(event) + if value is None: + return + + if action == "move_long": + if value == 0: + await self._call_cover_service(port.entity_id, "open_cover") + elif value == 1: + await self._call_cover_service(port.entity_id, "close_cover") + elif action == "move_short": + if value in (0, 1): + await self._call_cover_service(port.entity_id, "stop_cover") + elif action == "stop": + if value in (0, 1): + await self._call_cover_service(port.entity_id, "stop_cover") + elif action == "position": + await self._call_cover_service( + port.entity_id, "set_cover_position", {"position": value} + ) + elif action == "angle": + await self._call_cover_service( + port.entity_id, "set_cover_tilt_position", {"tilt_position": value} + ) + + async def _call_cover_service( + self, entity_id: str, service: str, service_data: dict[str, Any] | None = None + ) -> None: + data = {"entity_id": entity_id} + if service_data: + data.update(service_data) + await self.hass.services.async_call( + "cover", service, data, blocking=False + ) + + async def _knx_send_raw(self, address: str | None, payload: int) -> None: + if not address: + return + await self.hass.services.async_call( + KNX_DOMAIN, + "send", + {"address": address, "payload": payload, "payload_length": 0}, + blocking=False, + ) + + async def _knx_send_percent(self, address: str | None, value: int) -> None: + if not address: + return + await self.hass.services.async_call( + KNX_DOMAIN, + "send", + {"address": address, "payload": value, "type": "percent"}, + blocking=False, + ) + + +def _extract_event_value(event: Event) -> int | None: + if "value" in event.data: + try: + return int(event.data["value"]) + except (TypeError, ValueError): + return None + data = event.data.get("data") + if data is None: + return None + if isinstance(data, list) and data: + data = data[0] + try: + return int(data) & 1 + except (TypeError, ValueError): + return None + + +def _clean_address(address: Any) -> str | None: + if not address: + return None + if isinstance(address, str): + stripped = address.strip() + return stripped or None + return None diff --git a/custom_components/ha_knx_bridge/config_flow.py b/custom_components/ha_knx_bridge/config_flow.py new file mode 100644 index 0000000..8ec5a16 --- /dev/null +++ b/custom_components/ha_knx_bridge/config_flow.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import logging + +import voluptuous as vol + +from homeassistant import config_entries +from homeassistant.const import CONF_ENTITY_ID +from homeassistant.core import callback +from homeassistant.helpers import selector + +from .const import ( + CONF_ANGLE_ADDRESS, + CONF_ANGLE_STATE_ADDRESS, + CONF_KNX_ENTRY_ID, + CONF_MOVE_LONG_ADDRESS, + CONF_MOVE_SHORT_ADDRESS, + CONF_POSITION_ADDRESS, + CONF_POSITION_STATE_ADDRESS, + CONF_STATE_ADDRESS, + CONF_STOP_ADDRESS, + DOMAIN, +) + +_LOGGER = logging.getLogger(__name__) + +KNX_DOMAIN = "knx" + + +class HAKnxBridgeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + VERSION = 1 + + async def async_step_user(self, user_input: dict | None = None): + knx_entries = self.hass.config_entries.async_entries(KNX_DOMAIN) + if not knx_entries: + return self.async_abort(reason="knx_not_configured") + + if user_input is not None: + return self.async_create_entry( + title="HA KNX Bridge", + data={CONF_KNX_ENTRY_ID: user_input[CONF_KNX_ENTRY_ID]}, + ) + + options = [ + {"value": entry.entry_id, "label": entry.title or entry.entry_id} + for entry in knx_entries + ] + schema = vol.Schema( + { + vol.Required(CONF_KNX_ENTRY_ID): selector.SelectSelector( + selector.SelectSelectorConfig(options=options, mode="dropdown") + ) + } + ) + return self.async_show_form(step_id="user", data_schema=schema) + + @staticmethod + @callback + def async_get_supported_subentry_types(config_entry): + return { + "binary_sensor": config_entries.SubentryType( + name="Binary Sensor Port", flow_class=BinarySensorPortSubentryFlow + ), + "cover": config_entries.SubentryType( + name="Cover Port", flow_class=CoverPortSubentryFlow + ), + } + + +class BinarySensorPortSubentryFlow(config_entries.ConfigSubentryFlow): + VERSION = 1 + + async def async_step_user(self, user_input: dict | None = None): + if user_input is not None: + title = _entity_title(self.hass, user_input[CONF_ENTITY_ID]) + return self.async_create_entry(title=title, data=user_input) + + schema = vol.Schema( + { + vol.Required(CONF_ENTITY_ID): selector.EntitySelector( + selector.EntitySelectorConfig(domain=["binary_sensor"]) + ), + vol.Optional(CONF_STATE_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + } + ) + return self.async_show_form(step_id="user", data_schema=schema) + + +class CoverPortSubentryFlow(config_entries.ConfigSubentryFlow): + VERSION = 1 + + async def async_step_user(self, user_input: dict | None = None): + if user_input is not None: + title = _entity_title(self.hass, user_input[CONF_ENTITY_ID]) + return self.async_create_entry(title=title, data=user_input) + + schema = vol.Schema( + { + vol.Required(CONF_ENTITY_ID): selector.EntitySelector( + selector.EntitySelectorConfig(domain=["cover"]) + ), + vol.Optional(CONF_MOVE_LONG_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + vol.Optional(CONF_MOVE_SHORT_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + vol.Optional(CONF_STOP_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + vol.Optional(CONF_POSITION_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + vol.Optional(CONF_POSITION_STATE_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + vol.Optional(CONF_ANGLE_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + vol.Optional(CONF_ANGLE_STATE_ADDRESS): selector.TextSelector( + selector.TextSelectorConfig(type="text") + ), + } + ) + return self.async_show_form(step_id="user", data_schema=schema) + + +def _entity_title(hass, entity_id: str) -> str: + state = hass.states.get(entity_id) + if state is None: + return entity_id + return state.attributes.get("friendly_name", entity_id) diff --git a/custom_components/ha_knx_bridge/const.py b/custom_components/ha_knx_bridge/const.py new file mode 100644 index 0000000..c6fdeed --- /dev/null +++ b/custom_components/ha_knx_bridge/const.py @@ -0,0 +1,13 @@ +DOMAIN = "ha_knx_bridge" + +CONF_KNX_ENTRY_ID = "knx_entry_id" + +CONF_STATE_ADDRESS = "state_address" + +CONF_MOVE_LONG_ADDRESS = "move_long_address" +CONF_MOVE_SHORT_ADDRESS = "move_short_address" +CONF_STOP_ADDRESS = "stop_address" +CONF_POSITION_ADDRESS = "position_address" +CONF_POSITION_STATE_ADDRESS = "position_state_address" +CONF_ANGLE_ADDRESS = "angle_address" +CONF_ANGLE_STATE_ADDRESS = "angle_state_address" diff --git a/custom_components/ha_knx_bridge/manifest.json b/custom_components/ha_knx_bridge/manifest.json new file mode 100644 index 0000000..d617a0d --- /dev/null +++ b/custom_components/ha_knx_bridge/manifest.json @@ -0,0 +1,13 @@ +{ + "domain": "ha_knx_bridge", + "name": "HA KNX Bridge", + "version": "0.1.0", + "config_flow": true, + "documentation": "https://github.com/OWNER/REPO", + "issue_tracker": "https://github.com/OWNER/REPO/issues", + "codeowners": ["@your-github-handle"], + "dependencies": ["knx"], + "after_dependencies": ["knx"], + "homeassistant": "2025.12.0", + "iot_class": "local_push" +} diff --git a/custom_components/ha_knx_bridge/strings.json b/custom_components/ha_knx_bridge/strings.json new file mode 100644 index 0000000..265a7b4 --- /dev/null +++ b/custom_components/ha_knx_bridge/strings.json @@ -0,0 +1,46 @@ +{ + "config": { + "abort": { + "knx_not_configured": "Set up the Home Assistant KNX integration first." + }, + "step": { + "user": { + "title": "HA KNX Bridge", + "description": "Select the Home Assistant KNX integration entry to use.", + "data": { + "knx_entry_id": "KNX integration" + } + } + } + }, + "config_subentries": { + "binary_sensor": { + "step": { + "user": { + "title": "Binary Sensor Port", + "data": { + "entity_id": "Binary sensor entity", + "state_address": "State group address (DPT 1)" + } + } + } + }, + "cover": { + "step": { + "user": { + "title": "Cover Port", + "data": { + "entity_id": "Cover entity", + "move_long_address": "Move long (DPT 1.008 Up/Down)", + "move_short_address": "Move short (DPT 1.007 Step)", + "stop_address": "Stop (DPT 1)", + "position_address": "Set position (DPT 5.001)", + "position_state_address": "State position (DPT 5.001)", + "angle_address": "Set tilt (DPT 5.001)", + "angle_state_address": "State tilt (DPT 5.001)" + } + } + } + } + } +} diff --git a/hacs.json b/hacs.json new file mode 100644 index 0000000..1d6c4e2 --- /dev/null +++ b/hacs.json @@ -0,0 +1,5 @@ +{ + "name": "HA KNX Bridge", + "content_in_root": false, + "domains": ["ha_knx_bridge"] +}