KNX MQTT Integration: Gruppenadressen auf Topics abbilden

Eine KNX-MQTT-Bridge wandelt Telegramme auf einem KNX-Bus in MQTT-Nachrichten um und MQTT-Befehle wieder in KNX-Gruppenschreibvorgänge. In diesem Tutorial bauen Sie diese Bridge in Python mit der Open-Source-Bibliothek xknx, bilden Gruppenadressen auf lesbare Topics ab, dekodieren Datenpunkttypen in JSON und verbinden alles mit einem ConnectMQ-Broker, damit Dashboards und andere Gebäude Ihre KNX-Daten nutzen können.

Warum KNX mit MQTT verbinden

KNX ist hervorragend für die Automation auf Raumebene und in Deutschland, Österreich und der Schweiz in Büro-, Verwaltungs- und Industriegebäuden weit verbreitet. Taster, Dimmer, Jalousieaktoren und Präsenzmelder kommunizieren über Gruppenadressen direkt miteinander, sodass das Licht auch dann funktioniert, wenn jeder Server im Gebäude ausgeschaltet ist. Dieses dezentrale Konzept ist zugleich der Grund, warum KNX-Daten das Gebäude selten verlassen: Der Bus spricht Gruppentelegramme mit kompakten binären Datenpunkttypen (DPTs), und KNX/IP-Routing nutzt Multicast, das nicht über das Internet geht.

Eine MQTT-Bridge ergänzt eine nordgerichtete Schnittstelle, ohne die Funktionsweise der Installation zu verändern. Jede relevante Gruppenadresse wird zu einem Topic, Werte werden zu JSON, und jeder mit dem MQTT-Broker verbundene Client kann sie lesen. Es ist dieselbe Integrationsschicht, die in unserem Leitfaden zur Gebäudeautomation mit MQTT beschrieben wird. Auf einem Werksgelände können die KNX-Topics so neben Maschinendaten im Unified Namespace (UNS) liegen, etwa für Energiemonitoring von Beleuchtung und Verschattung pro Halle.

Was Sie benötigen

Eine KNX/IP-Schnittstelle oder einen KNX/IP-Router, im LAN erreichbar. Tunneling zu einer Schnittstelle ist der häufigste Aufbau.
Die Gruppenadressliste aus dem ETS-Projekt mit dem DPT jeder Adresse. Bitten Sie den KNX-Integrator um einen Export, falls Sie kein ETS haben.
Einen Linux-Rechner, Raspberry Pi oder PC im selben Netzwerk.
Python 3.10 oder neuer.

python -m venv knx-gw
source knx-gw/bin/activate
pip install xknx "paho-mqtt>=2.0"

Die meisten KNX/IP-Schnittstellen unterstützen nur eine begrenzte Anzahl gleichzeitiger Tunnel, oft zwischen einem und einer Handvoll. Wenn sich ETS, eine Visualisierung und Ihre Bridge gleichzeitig verbinden, kann eine davon abgewiesen werden. Prüfen Sie vor dem Start, wie viele Tunnel Ihre Schnittstelle bietet.

Schritt 1: Gruppenadressen zuordnen

KNX-Gruppenadressen sehen aus wie 1/2/3. Sie direkt als MQTT-Topic-Ebenen zu verwenden, erzeugt tiefe Hierarchien und verwirrende Topic-Namen. Ordnen Sie daher jeder Adresse einen aussagekräftigen Namen und einen DPT zu:

Gruppenadresse	Name	DPT	Bedeutung	Richtung
1/1/1	f2/office-201/light/switch	1.001	Schalten ein/aus	MQTT zu KNX
1/1/2	f2/office-201/light/state	1.001	Schaltstatus	KNX zu MQTT
1/1/5	f2/office-201/light/brightness-state	5.001	Dimmwert 0-100 %	KNX zu MQTT
3/1/1	f2/office-201/temperature	9.001	Raumtemperatur in degC	KNX zu MQTT
4/1/1	f2/office-201/presence	1.018	Belegt / nicht belegt	KNX zu MQTT

Übertragen Sie dieselbe Zuordnung als Python-Dictionary in die Bridge. Status- und Befehlsadressen getrennt zu halten, ist gängige KNX-Praxis und passt gut zu MQTT: State-Topics veröffentlicht die Bridge, Befehls-Topics abonniert sie.

Schritt 2: Die KNX-MQTT-Bridge

Die Bridge verbindet sich per Tunneling mit KNX, hört alle Telegramme mit, dekodiert die in der Zuordnung enthaltenen und veröffentlicht sie. Außerdem abonniert sie .../set-Topics und wandelt diese in Gruppenschreibvorgänge um.

import asyncio
import json
import uuid
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from xknx import XKNX
from xknx.dpt import DPTArray, DPTBinary, DPTScaling, DPTTemperature
from xknx.io import ConnectionConfig, ConnectionType
from xknx.telegram import GroupAddress, Telegram
from xknx.telegram.apci import GroupValueResponse, GroupValueWrite

KNX_GATEWAY_IP = "192.168.20.10"
BROKER = "broker.connectmq.com"
PORT = 1883                         # 8883 + TLS für einen privaten Broker
USERNAME = None                     # "YOUR_USERNAME"
PASSWORD = None                     # "YOUR_PASSWORD"
CLIENT_ID = f"knx-gw-{uuid.uuid4().hex[:8]}"   # "YOUR_CLIENT_ID"
PREFIX = "connectmq/demo/knx-4mq8"

# Gruppenadresse -> (Topic-Suffix, DPT-Art)
STATE_MAP = {
    "1/1/2": ("f2/office-201/light/state", "switch"),
    "1/1/5": ("f2/office-201/light/brightness-state", "percent"),
    "3/1/1": ("f2/office-201/temperature", "temperature"),
    "4/1/1": ("f2/office-201/presence", "switch"),
}
# Topic-Suffix -> (Gruppenadresse, DPT-Art)
COMMAND_MAP = {
    "f2/office-201/light/switch": ("1/1/1", "switch"),
    "f2/office-201/light/brightness": ("1/1/4", "percent"),
}

loop: asyncio.AbstractEventLoop | None = None
xknx: XKNX | None = None
mqttc: mqtt.Client | None = None


def decode(kind, value):
    if kind == "switch" and isinstance(value, DPTBinary):
        return bool(value.value)
    if kind == "percent" and isinstance(value, DPTArray):
        return DPTScaling.from_knx(value)
    if kind == "temperature" and isinstance(value, DPTArray):
        return round(DPTTemperature.from_knx(value), 2)
    return None


def encode(kind, value):
    if kind == "switch":
        return DPTBinary(1 if value in (True, 1, "ON", "on", "true") else 0)
    if kind == "percent":
        return DPTScaling.to_knx(max(0, min(100, float(value))))
    raise ValueError(kind)


async def telegram_received(telegram: Telegram):
    ga = str(telegram.destination_address)
    if ga not in STATE_MAP:
        return
    if not isinstance(telegram.payload, (GroupValueWrite, GroupValueResponse)):
        return
    suffix, kind = STATE_MAP[ga]
    value = decode(kind, telegram.payload.value)
    if value is None:
        return
    payload = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
               "value": value, "ga": ga, "source": str(telegram.source_address)}
    # retained, damit Dashboards den letzten Zustand sofort sehen
    mqttc.publish(f"{PREFIX}/{suffix}", json.dumps(payload), qos=1, retain=True)


async def send_knx(ga, kind, value):
    telegram = Telegram(destination_address=GroupAddress(ga),
                        payload=GroupValueWrite(encode(kind, value)))
    await xknx.telegrams.put(telegram)


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/+/+/+/set", qos=1)
    client.subscribe(f"{PREFIX}/+/+/+/set", qos=1)
    client.publish(f"{PREFIX}/gateway/status", "online", qos=1, retain=True)


def on_message(client, userdata, msg):
    suffix = msg.topic[len(PREFIX) + 1:].removesuffix("/set")
    if suffix not in COMMAND_MAP:
        print("Ignored command for", suffix)
        return
    ga, kind = COMMAND_MAP[suffix]
    try:
        value = json.loads(msg.payload)
    except json.JSONDecodeError:
        value = msg.payload.decode()
    if isinstance(value, dict):
        value = value.get("value")
    # paho-Callbacks laufen in einem anderen Thread; KNX-Schreibvorgang in der asyncio-Schleife einplanen
    asyncio.run_coroutine_threadsafe(send_knx(ga, kind, value), loop)


async def main():
    global loop, xknx, mqttc
    loop = asyncio.get_running_loop()

    mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mqttc.username_pw_set(USERNAME, PASSWORD)
        mqttc.tls_set()
    mqttc.will_set(f"{PREFIX}/gateway/status", "offline", qos=1, retain=True)
    mqttc.on_connect = on_connect
    mqttc.on_message = on_message
    mqttc.connect(BROKER, PORT, keepalive=60)
    mqttc.loop_start()

    xknx = XKNX(
        connection_config=ConnectionConfig(
            connection_type=ConnectionType.TUNNELING,
            gateway_ip=KNX_GATEWAY_IP,
        ),
        telegram_received_cb=telegram_received,
    )
    await xknx.start()
    print("KNX connected")
    try:
        await asyncio.Event().wait()   # endlos laufen
    finally:
        await xknx.stop()
        mqttc.loop_stop()

asyncio.run(main())

Ein paar Details sind hier wichtig. Die beiden Abonnements decken Befehls-Topics sowohl vier als auch drei Ebenen unterhalb des Präfixes ab, was zu den Namen in COMMAND_MAP passt. Außerdem ignoriert die Bridge jede Gruppenadresse, die sie nicht kennt, sodass ein Fehler in einem Topic niemals auf eine beliebige Adresse schreiben kann. Schließlich ist run_coroutine_threadsafe erforderlich, weil paho Nachrichten in seinem Netzwerk-Thread zustellt, während xknx innerhalb von asyncio läuft.

Anfangszustände beim Start lesen

KNX-Geräte senden nur dann ein Telegramm, wenn sich etwas ändert, daher weiß eine frisch gestartete Bridge nichts über den aktuellen Zustand des Gebäudes. Retained MQTT-Nachrichten verbergen das für Dashboards, doch nach einem längeren Ausfall können die retained Werte veraltet sein. Die Lösung besteht darin, nach dem Aufbau der KNX-Verbindung an jede Statusadresse einmal einen GroupValueRead zu senden. Aktoren, bei denen in ETS das Lesen-Flag gesetzt ist, antworten mit einem GroupValueResponse, den die Bridge bereits verarbeitet, weil telegram_received sowohl Schreibvorgänge als auch Antworten akzeptiert.

from xknx.telegram.apci import GroupValueRead

async def read_all_states():
    for ga in STATE_MAP:
        await xknx.telegrams.put(Telegram(destination_address=GroupAddress(ga),
                                          payload=GroupValueRead()))
        await asyncio.sleep(0.2)   # den Bus schonen

Rufen Sie await read_all_states() direkt nach await xknx.start() auf. Die kurze Pause zwischen den Lesevorgängen ist wichtig: Eine KNX-Twisted-Pair-Linie hat begrenzte Bandbreite, und Dutzende Lesevorgänge auf einmal können echte Telegramme von Tastern verzögern. Antwortet eine Statusadresse nie, ist vermutlich das Lesen-Flag am Kommunikationsobjekt des Aktors nicht gesetzt; bitten Sie den Integrator, es in ETS zu aktivieren, statt das Problem in der Bridge zu umgehen.

Schritt 3: Die Bridge testen

Beobachten Sie die veröffentlichten Zustände:

mosquitto_sub -h broker.connectmq.com -p 1883 -i knx-view-4mq8 -t "connectmq/demo/knx-4mq8/#" -v

Drücken Sie den Wandtaster in Büro 201. Sie sollten Folgendes sehen:

connectmq/demo/knx-4mq8/f2/office-201/light/state {"ts": "2026-10-16T09:02:11+00:00", "value": true, "ga": "1/1/2", "source": "1.1.12"}

Schalten Sie das Licht nun per MQTT ein und dimmen Sie es auf 40 Prozent:

mosquitto_pub -h broker.connectmq.com -p 1883 -i knx-cmd-4mq8 -q 1 \
  -t "connectmq/demo/knx-4mq8/f2/office-201/light/switch/set" -m '{"value": "ON"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i knx-cmd-4mq8 -q 1 \
  -t "connectmq/demo/knx-4mq8/f2/office-201/light/brightness/set" -m '{"value": 40}'

Der öffentliche MQTT-Broker genügt für diese erste Runde, doch jeder kann unter jedem öffentlichen Topic lesen und veröffentlichen; er ist für Tests und Prototypen gedacht, nicht für Produktionsdaten. Verwenden Sie beim öffentlichen Testen einen KNX-Laboraufbau, kein echtes Büro.

Schritt 4: Betrieb mit einem privaten Broker

Für eine echte Installation erstellen Sie im ConnectMQ-Portal einen Broker und verwenden dessen Zugangsdaten:

BROKER = "broker.connectmq.com"   # oder der für Ihren Broker angezeigte Host
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"
PREFIX = "acme/muc01/hq"

Installieren Sie die Bridge anschließend als systemd-Dienst, damit sie nach Neustarts wieder startet:

sudo tee /etc/systemd/system/knx-mqtt.service > /dev/null <<'EOF'
[Unit]
Description=KNX MQTT bridge
After=network-online.target
Wants=network-online.target

[Service]
WorkingDirectory=/opt/knx-gw
ExecStart=/opt/knx-gw/bin/python /opt/knx-gw/bridge.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF
sudo systemctl daemon-reload
sudo systemctl enable --now knx-mqtt

Mit einem verwalteten MQTT-Broker ist der einzige Prozess, den Sie vor Ort pflegen, die Bridge selbst.

Die Bridge absichern

Die Bridge hält eine Verbindung in die KNX-Installation, behandeln Sie sie daher wie jeden anderen Gebäudecontroller. Betreiben Sie sie auf einem eigenen Gerät oder in einem eigenen VLAN, halten Sie ihr Betriebssystem aktuell und speichern Sie das MQTT-Passwort in einer Umgebungsdatei, die nur der Dienstbenutzer lesen kann, statt im Skript selbst. Nutzt Ihre KNX-Installation KNX Secure, kann xknx sicheres Tunneling mit dem aus ETS exportierten Keyring verwenden; bitten Sie den Integrator um die Zugangsdaten, statt die Sicherheit an der Schnittstelle zu deaktivieren.

Fehlersuche

Tunnelverbindung abgewiesen. Alle Tunnel der Schnittstelle sind belegt. Schließen Sie ETS oder eine andere Visualisierung oder verwenden Sie eine Schnittstelle mit mehr Tunneln.
Es kommen keine Telegramme an. Die Schnittstelle filtert möglicherweise Gruppentelegramme. Prüfen Sie ihre Filtertabelle in ETS oder verbinden Sie sich stattdessen mit einem KNX/IP-Router.
Werte werden unsinnig dekodiert. Der DPT in Ihrer Zuordnung ist falsch. Ein 2-Byte-Gleitkommawert (DPT 9), der als Skalierungswert (DPT 5) dekodiert wird, ergibt bedeutungslose Zahlen.
Zustände fehlen nach einem Neustart der Bridge. KNX sendet Telegramme nur bei Änderungen. Senden Sie beim Start einen GroupValueRead für jede Statusadresse oder verlassen Sie sich auf die retained MQTT-Nachrichten.
Doppelte Befehle. QoS 1 kann eine Nachricht zweimal zustellen; ein Licht zweimal einzuschalten, ist harmlos, aber vermeiden Sie Umschalt-Semantik und senden Sie immer absolute Werte.

Alternativen, die Sie kennen sollten

Wenn Sie ein visuelles Werkzeug bevorzugen, kann Node-RED mit einem KNX-Node-Paket dieselbe Zuordnung übernehmen, und Home Assistant hat eine native KNX-Integration, die über Automationen an MQTT veröffentlichen kann. Die obige Python-Bridge ist die transparenteste Option und lässt sich leicht versionieren. Die offizielle xknx-Dokumentation unter xknx.github.io listet jeden unterstützten DPT auf.

Für Lichtszenen auf Basis dieser Bridge siehe intelligente Beleuchtung mit MQTT, und wenn im selben Gebäude BACnet-HLK läuft, kombinieren Sie sie mit dem BACnet-zu-MQTT-Gateway. Unser Leitfaden zur IoT-Automatisierung sammelt verwandte Projekte.

Verbinden Sie Ihre KNX-Installation mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ganz ohne Kreditkarte, und Sie haben in wenigen Minuten Zugangsdaten für die Bridge. Der Live-Topic-Explorer in ConnectMQ zeigt jede Gruppenadresse, sobald sie sich ändert, was Inbetriebnahme und Fehlersuche deutlich beschleunigt.

Häufig gestellte Fragen

Brauche ich einen KNX/IP-Router oder reicht eine Schnittstelle?

Für eine einzelne Bridge reicht eine IP-Schnittstelle mit Tunneling. Ein Router hilft, wenn mehrere Anwendungen gleichzeitig Zugriff benötigen, weil Routing Multicast statt einer begrenzten Anzahl von Tunneln verwendet.

Funktioniert die KNX-Installation weiter, wenn MQTT ausfällt?

Ja. KNX-Geräte kommunizieren direkt über Gruppenadressen, die Bridge ergänzt also nur Transparenz und Fernsteuerung; das lokale Schalten hängt nie von ihr ab.

Sollten State-Topics retained sein?

Ja. KNX sendet Werte nur bei Änderungen, daher ermöglichen retained MQTT-Nachrichten einem Dashboard, den aktuellen Zustand direkt nach dem Verbinden anzuzeigen.

Kann ich jede beliebige Gruppenadresse per MQTT ansprechen?

Das sollten Sie nicht. Führen Sie in der Bridge eine explizite Befehlszuordnung, damit nur freigegebene Adressen per MQTT beschrieben werden können.

KNX-MQTT-Integration: Gruppenadressen auf MQTT-Topics abbilden