Intégration KNX MQTT : adresses de groupe vers topics

Un pont KNX MQTT transforme les télégrammes d'un bus KNX en messages MQTT, et les commandes MQTT en écritures de groupe KNX. Dans ce tutoriel, vous allez construire ce pont en Python avec la bibliothèque open source xknx, associer les adresses de groupe à des topics lisibles, décoder les types de points de données en JSON et le connecter à un broker MQTT managé ConnectMQ pour que tableaux de bord et autres bâtiments puissent exploiter vos données KNX.

Pourquoi relier KNX à MQTT

KNX excelle dans l'automatisation à l'échelle de la pièce. Boutons-poussoirs, variateurs, actionneurs de stores et détecteurs de présence dialoguent directement entre eux via des adresses de groupe, si bien que l'éclairage continue de fonctionner même si tous les serveurs du bâtiment sont éteints. Cette conception décentralisée explique aussi pourquoi les données KNX quittent rarement le bâtiment : le bus parle en télégrammes de groupe avec des types de points de données (DPT) binaires compacts, et le routage KNX/IP utilise un multicast qui ne traverse pas Internet.

Un pont MQTT ajoute une interface nord sans changer le fonctionnement de l'installation. Chaque adresse de groupe utile devient un topic, les valeurs deviennent du JSON, et tout client connecté au broker MQTT peut les lire. C'est la même couche d'intégration que celle décrite dans notre guide sur la gestion technique du bâtiment avec MQTT.

Ce dont vous avez besoin

Une interface ou un routeur KNX/IP joignable sur le LAN. Le tunneling vers une interface est la configuration la plus courante.
La liste des adresses de groupe issue du projet ETS, avec le DPT de chaque adresse. Demandez un export à l'intégrateur KNX si vous n'avez pas ETS.
Une machine Linux, un Raspberry Pi ou un PC sur le même réseau.
Python 3.10 ou plus récent.

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

La plupart des interfaces KNX/IP ne prennent en charge qu'un nombre limité de tunnels simultanés, souvent entre un et quelques-uns. Si ETS, une visualisation et votre pont se connectent en même temps, l'un d'eux peut être refusé. Vérifiez combien de tunnels propose votre interface avant de commencer.

Étape 1 : associer les adresses de groupe

Les adresses de groupe KNX ressemblent à 1/2/3. Les utiliser directement comme niveaux de topic MQTT crée des hiérarchies profondes et des noms de topics déroutants ; associez donc chacune à un nom explicite et à un DPT :

Adresse de groupe	Nom	DPT	Signification	Sens
1/1/1	f2/office-201/light/switch	1.001	Marche/arrêt	MQTT vers KNX
1/1/2	f2/office-201/light/state	1.001	État marche/arrêt	KNX vers MQTT
1/1/5	f2/office-201/light/brightness-state	5.001	Valeur de gradation 0-100 %	KNX vers MQTT
3/1/1	f2/office-201/temperature	9.001	Température de la pièce en degC	KNX vers MQTT
4/1/1	f2/office-201/presence	1.018	Occupé / inoccupé	KNX vers MQTT

Reportez la même correspondance dans le pont sous forme de dictionnaire Python. Séparer les adresses d'état et de commande est une pratique KNX standard qui s'accorde bien avec MQTT : les topics d'état sont publiés par le pont, les topics de commande sont souscrits.

Étape 2 : le pont KNX MQTT

Le pont se connecte à KNX en tunneling, écoute tous les télégrammes, décode ceux qui figurent dans la correspondance et les publie. Il s'abonne également aux topics .../set et les convertit en écritures de groupe.

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 for a private 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"

# group address -> (topic suffix, dpt kind)
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 -> (group address, dpt kind)
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 so dashboards see the last state immediately
    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 run in another thread; schedule the KNX write on the asyncio loop
    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()   # run forever
    finally:
        await xknx.stop()
        mqttc.loop_stop()

asyncio.run(main())

Quelques détails comptent ici. Les deux abonnements couvrent les topics de commande situés à quatre et à trois niveaux sous le préfixe, ce qui correspond aux noms de COMMAND_MAP. Le pont ignore aussi toute adresse de groupe qu'il ne connaît pas, si bien qu'une erreur dans un topic ne peut jamais écrire sur une adresse arbitraire. Enfin, run_coroutine_threadsafe est indispensable, car paho délivre les messages sur son propre thread réseau, alors que xknx s'exécute dans asyncio.

Lire les états initiaux au démarrage

Les équipements KNX n'envoient un télégramme que lorsque quelque chose change ; un pont fraîchement démarré ne sait donc rien de l'état actuel du bâtiment. Les messages MQTT retenus masquent ce problème pour les tableaux de bord, mais après une longue coupure, les valeurs retenues peuvent être périmées. La solution consiste à envoyer un GroupValueRead à chaque adresse d'état une fois la connexion KNX établie. Les actionneurs dont le drapeau de lecture est activé dans ETS répondent par un GroupValueResponse, que le pont gère déjà puisque telegram_received accepte à la fois les écritures et les réponses.

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)   # be gentle with the bus

Appelez await read_all_states() juste après await xknx.start(). La courte pause entre les lectures est importante : une ligne KNX à paire torsadée a une bande passante limitée, et envoyer des dizaines de lectures d'un coup peut retarder les vrais télégrammes des boutons-poussoirs. Si une adresse d'état ne répond jamais, le drapeau de lecture n'est probablement pas activé sur l'objet de communication de l'actionneur ; demandez à l'intégrateur de l'activer dans ETS plutôt que de contourner le problème dans le pont.

Étape 3 : tester le pont

Observez les états publiés :

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

Appuyez sur l'interrupteur mural du bureau 201. Vous devriez voir :

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"}

Allumez maintenant la lumière depuis MQTT et réglez-la à 40 % :

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}'

Le serveur MQTT public gratuit convient pour ce premier essai et pour les prototypes, mais n'importe qui peut lire et publier sous n'importe quel topic public. Utilisez un kit KNX de laboratoire, pas un vrai bureau ni un vrai logement, pendant vos tests en public.

Étape 4 : l'exécuter sur un broker privé

Pour une installation en service, créez un broker dans le portail ConnectMQ et utilisez ses identifiants :

BROKER = "broker.connectmq.com"   # or the host shown for your broker
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"
PREFIX = "acme/ist01/hq"

Installez ensuite le pont en tant que service systemd pour qu'il redémarre après un reboot :

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

Avec un broker MQTT managé, le seul processus que vous maintenez sur site est le pont lui-même.

Sécuriser le pont

Le pont maintient une connexion vers l'installation KNX ; traitez-le donc comme n'importe quel autre contrôleur du bâtiment. Exécutez-le sur un équipement ou un VLAN dédié, maintenez son système d'exploitation à jour et stockez le mot de passe MQTT dans un fichier d'environnement lisible uniquement par l'utilisateur du service plutôt que dans le script lui-même. Si votre installation KNX utilise KNX Secure, xknx peut utiliser le tunneling sécurisé avec le keyring exporté depuis ETS ; demandez les identifiants à l'intégrateur au lieu de désactiver la sécurité sur l'interface.

Dépannage

Connexion tunnel refusée. Tous les tunnels de l'interface sont occupés. Fermez ETS ou une autre visualisation, ou utilisez une interface disposant de plus de tunnels.
Aucun télégramme n'arrive. L'interface filtre peut-être les télégrammes de groupe. Vérifiez sa table de filtrage dans ETS, ou connectez-vous plutôt à un routeur KNX/IP.
Les valeurs décodées n'ont aucun sens. Le DPT de votre correspondance est erroné. Un flottant sur 2 octets (DPT 9) décodé comme une valeur de mise à l'échelle (DPT 5) produit des nombres aberrants.
États manquants après un redémarrage du pont. KNX n'envoie des télégrammes qu'en cas de changement. Envoyez un GroupValueRead pour chaque adresse d'état au démarrage, ou appuyez-vous sur les messages MQTT retenus.
Commandes en double. QoS 1 peut délivrer un message deux fois ; allumer une lumière deux fois est sans conséquence, mais évitez la sémantique de bascule (toggle) et envoyez toujours des valeurs absolues.

Les alternatives à connaître

Si vous préférez un outil visuel, Node-RED avec un paquet de nœuds KNX peut réaliser la même correspondance, et Home Assistant dispose d'une intégration KNX native capable de publier en MQTT via des automatisations. Dans une maison connectée, Jeedom et Domoticz savent aussi communiquer en MQTT et peuvent donc consommer les topics publiés par ce pont. Le pont Python ci-dessus est l'option la plus transparente et se versionne facilement. La documentation officielle de xknx sur xknx.github.io liste tous les DPT pris en charge.

Pour des scènes d'éclairage au-dessus de ce pont, consultez l'éclairage intelligent avec MQTT, et si le même bâtiment exploite du CVC en BACnet, associez-le à la passerelle BACnet vers MQTT. Notre guide de l'automatisation IoT rassemble des projets connexes.

Connectez votre installation KNX à ConnectMQ

Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, et vous disposerez des identifiants du pont en quelques minutes. L'explorateur de topics en direct de ConnectMQ affiche chaque adresse de groupe au moment où elle change, ce qui accélère nettement la mise en service et la recherche de pannes.

Questions fréquentes

Ai-je besoin d'un routeur KNX/IP ou une interface suffit-elle ?

Une interface IP avec tunneling suffit pour un seul pont. Un routeur est utile lorsque plusieurs applications ont besoin d'un accès simultané, car le routage utilise le multicast plutôt qu'un nombre limité de tunnels.

L'installation KNX fonctionnera-t-elle encore si MQTT tombe ?

Oui. Les équipements KNX communiquent directement via les adresses de groupe ; le pont n'ajoute que de la visibilité et du contrôle à distance, et la commande locale n'en dépend jamais.

Les topics d'état doivent-ils être retenus ?

Oui. KNX n'envoie des valeurs qu'en cas de changement ; les messages MQTT retenus permettent donc à un tableau de bord d'afficher l'état actuel dès sa connexion.

Puis-je envoyer n'importe quelle adresse de groupe depuis MQTT ?

Vous ne devriez pas. Conservez une table de commandes explicite dans le pont pour que seules les adresses approuvées puissent être écrites depuis MQTT.

Intégration KNX MQTT : relier les adresses de groupe à des topics MQTT