Integración KNX MQTT: cómo puentear direcciones de grupo a topics MQTT

Un puente KNX MQTT convierte los telegramas de un bus KNX en mensajes MQTT y transforma las órdenes MQTT de nuevo en escrituras de grupo KNX. En este tutorial construirás ese puente en Python con la librería de código abierto xknx, asociarás las direcciones de grupo con topics legibles, decodificarás los tipos de punto de datos a JSON y lo conectarás a un broker de ConnectMQ para que los paneles y otros edificios puedan usar tus datos KNX.

Por qué puentear KNX a MQTT

KNX es excelente en la automatización a nivel de estancia. Pulsadores, reguladores, actuadores de persianas y detectores de presencia hablan directamente entre sí mediante direcciones de grupo, de modo que las luces siguen funcionando aunque todos los servidores del edificio estén apagados. Ese diseño descentralizado es también la razón por la que los datos KNX rara vez salen del edificio: el bus habla con telegramas de grupo con tipos de punto de datos (DPT) binarios compactos, y el enrutamiento KNX/IP usa multicast, que no atraviesa internet.

Un puente MQTT añade una interfaz hacia el norte sin cambiar el funcionamiento de la instalación. Cada dirección de grupo de interés se convierte en un topic, los valores pasan a JSON y cualquier cliente conectado al broker MQTT puede leerlos. Es la misma capa de integración descrita en nuestra guía de sistemas de automatización de edificios con MQTT.

Qué necesitas

• Una interfaz o un router KNX/IP accesible en la LAN. El tunneling hacia una interfaz es la configuración más habitual.
• La lista de direcciones de grupo del proyecto ETS, con el DPT de cada dirección. Pide una exportación al integrador KNX si no tienes ETS.
• Un equipo Linux, una Raspberry Pi o un PC en la misma red.
• Python 3.10 o posterior.

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

La mayoría de las interfaces KNX/IP solo admiten un número limitado de túneles simultáneos, a menudo entre uno y unos pocos. Si ETS, una visualización y tu puente se conectan a la vez, alguno puede ser rechazado. Comprueba cuántos túneles ofrece tu interfaz antes de empezar.

Paso 1: asociar las direcciones de grupo

Las direcciones de grupo KNX tienen el aspecto 1/2/3. Usarlas directamente como niveles de topic MQTT crea jerarquías profundas y nombres de topic confusos, así que asocia cada una con un nombre significativo y un DPT:

Escribe la misma asociación en el puente como un diccionario de Python. Mantener separadas las direcciones de estado y de orden es práctica estándar en KNX y encaja muy bien con MQTT: el puente publica los topics de estado y se suscribe a los topics de órdenes.

Paso 2: el puente KNX MQTT

El puente se conecta a KNX mediante tunneling, escucha todos los telegramas, decodifica los que figuran en la asociación y los publica. También se suscribe a los topics .../set y los convierte en escrituras de grupo.

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 para un broker privado
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"

# dirección de grupo -> (sufijo de topic, tipo de dpt)
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"),
}
# sufijo de topic -> (dirección de grupo, tipo de dpt)
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)}
    # retenido para que los paneles vean el último estado de inmediato
    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")
    # los callbacks de paho se ejecutan en otro hilo; programa la escritura KNX en el bucle asyncio
    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()   # ejecutar indefinidamente
    finally:
        await xknx.stop()
        mqttc.loop_stop()

asyncio.run(main())

Aquí importan algunos detalles. Las dos suscripciones cubren topics de órdenes a cuatro y a tres niveles por debajo del prefijo, lo que coincide con los nombres de COMMAND_MAP. El puente ignora además cualquier dirección de grupo que no conozca, de modo que un error en un topic nunca puede escribir en una dirección arbitraria. Por último, run_coroutine_threadsafe es necesario porque paho entrega los mensajes en su hilo de red, mientras que xknx se ejecuta dentro de asyncio.

Leer los estados iniciales al arrancar

Los dispositivos KNX solo envían un telegrama cuando algo cambia, así que un puente recién arrancado no sabe nada del estado actual del edificio. Los mensajes MQTT retenidos lo disimulan para los paneles, pero tras una interrupción larga los valores retenidos pueden estar desfasados. La solución es enviar un GroupValueRead a cada dirección de estado en cuanto la conexión KNX esté activa. Los actuadores con el indicador de lectura activado en ETS responden con un GroupValueResponse, que el puente ya gestiona porque telegram_received acepta tanto escrituras como respuestas.

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)   # trata el bus con cuidado

Llama a await read_all_states() justo después de await xknx.start(). La breve pausa entre lecturas es importante: una línea KNX de par trenzado tiene un ancho de banda limitado, y lanzar decenas de lecturas a la vez puede retrasar los telegramas reales de los pulsadores. Si una dirección de estado nunca responde, probablemente el indicador de lectura no está activado en el objeto de comunicación del actuador; pide al integrador que lo active en ETS en lugar de buscar un rodeo en el puente.

Paso 3: probar el puente

Observa los estados publicados:

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

Pulsa el interruptor de pared de la oficina 201. Deberías ver:

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

Ahora enciende la luz desde MQTT y regúlala al 40 por ciento:

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

El servidor MQTT público gratuito sirve para esta primera ronda de pruebas, pero cualquiera puede leer y publicar en cualquier topic público. Usa un kit KNX de laboratorio, no una oficina real, mientras pruebas en público.

Paso 4: ejecutarlo contra un broker privado

Para una instalación real, crea un broker en el portal de ConnectMQ y usa sus credenciales:

BROKER = "broker.connectmq.com"   # o el host que se muestra para tu broker
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"
PREFIX = "acme/ist01/hq"

Después instala el puente como servicio systemd para que se reinicie tras cada arranque:

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

Con un broker MQTT gestionado, el único proceso que mantienes in situ es el propio puente. Los mismos topics de estado pueden alimentar después un gemelo digital 3D del edificio, ya sea una oficina en Bogotá o en Barcelona.

Mantener seguro el puente

El puente mantiene una conexión con la instalación KNX, así que trátalo como cualquier otro controlador del edificio. Ejecútalo en un dispositivo o una VLAN dedicados, mantén actualizado su sistema operativo y guarda la contraseña MQTT en un archivo de entorno legible solo por el usuario del servicio, no en el propio script. Si tu instalación KNX usa KNX Secure, xknx puede emplear tunneling seguro con el keyring exportado desde ETS; pide las credenciales al integrador en lugar de desactivar la seguridad de la interfaz.

Resolución de problemas

1. Conexión de túnel rechazada. Todos los túneles de la interfaz están en uso. Cierra ETS u otra visualización, o usa una interfaz con más túneles.
2. No llega ningún telegrama. La interfaz puede estar filtrando los telegramas de grupo. Revisa su tabla de filtros en ETS o conéctate a un router KNX/IP.
3. Los valores se decodifican sin sentido. El DPT de tu asociación es incorrecto. Un flotante de 2 bytes (DPT 9) decodificado como valor de escala (DPT 5) produce números sin significado.
4. Faltan estados tras reiniciar el puente. KNX solo envía telegramas cuando hay cambios. Envía un GroupValueRead a cada dirección de estado al arrancar o confía en los mensajes MQTT retenidos.
5. Órdenes duplicadas. QoS 1 puede entregar un mensaje dos veces; encender una luz dos veces es inofensivo, pero evita la semántica de conmutación (toggle) y envía siempre valores absolutos.

Alternativas que conviene conocer

Si prefieres una herramienta visual, Node-RED con un paquete de nodos KNX puede hacer la misma asociación, y Home Assistant tiene una integración KNX nativa que puede publicar en MQTT mediante automatizaciones. El puente en Python anterior es la opción más transparente y es fácil de mantener en control de versiones. La documentación oficial de xknx en xknx.github.io enumera todos los DPT compatibles.

Para escenas de iluminación sobre este puente, consulta iluminación inteligente con MQTT, y si el mismo edificio tiene climatización BACnet, combínalo con el gateway de BACnet a MQTT. Nuestra guía de automatización IoT reúne proyectos relacionados.

Conecta tu instalación KNX a ConnectMQ

Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y tendrás credenciales para el puente en pocos minutos. El explorador de topics en vivo de ConnectMQ muestra cada dirección de grupo a medida que cambia, lo que agiliza mucho la puesta en marcha y la localización de averías.

Preguntas frecuentes

¿Necesito un router KNX/IP o basta con una interfaz?

Una interfaz IP con tunneling basta para un solo puente. Un router ayuda cuando varias aplicaciones necesitan acceso simultáneo, porque el enrutamiento usa multicast en lugar de un número limitado de túneles.

¿Seguirá funcionando la instalación KNX si MQTT cae?

Sí. Los dispositivos KNX se comunican directamente mediante direcciones de grupo, así que el puente solo añade visibilidad y control remoto; la conmutación local nunca depende de él.

¿Deben retenerse los topics de estado?

Sí. KNX solo envía valores cuando cambian, así que los mensajes MQTT retenidos permiten que un panel muestre el estado actual nada más conectarse.

¿Puedo enviar cualquier dirección de grupo desde MQTT?

No deberías. Mantén en el puente una asociación de órdenes explícita para que solo puedan escribirse desde MQTT las direcciones aprobadas.