Integração KNX MQTT: Endereços de Grupo para Topics

Uma bridge KNX MQTT transforma telegramas de um barramento KNX em mensagens MQTT e transforma comandos MQTT de volta em escritas de grupo KNX. Neste tutorial você vai construir essa bridge em Python com a biblioteca open source xknx, mapear endereços de grupo para topics legíveis, decodificar tipos de datapoint em JSON e conectá-la a um broker ConnectMQ para que dashboards e outros edifícios possam usar seus dados KNX.

Por Que Fazer a Bridge de KNX para MQTT

O KNX é excelente em automação no nível do ambiente. Botões, dimmers, atuadores de persianas e detectores de presença conversam diretamente entre si por meio de endereços de grupo, de modo que as luzes continuam funcionando mesmo que todos os servidores do edifício estejam desligados. Esse design descentralizado também é o motivo pelo qual os dados KNX raramente saem do edifício: o barramento fala telegramas de grupo com tipos de datapoint (DPTs) binários compactos, e o roteamento KNX/IP usa multicast, que não atravessa a internet.

Uma bridge MQTT adiciona uma interface northbound sem mudar o funcionamento da instalação. Cada endereço de grupo de interesse vira um topic, os valores viram JSON e qualquer cliente conectado ao broker MQTT pode lê-los. É a mesma camada de integração descrita em nosso guia de automação predial com MQTT. Em uma torre corporativa em São Paulo, por exemplo, esses mesmos topics podem alimentar um gêmeo digital do edifício, sem tocar na programação KNX existente.

O Que Você Precisa

Uma interface ou roteador KNX/IP acessível na LAN. O tunneling para uma interface é a configuração mais comum.
A lista de endereços de grupo do projeto ETS, com o DPT de cada endereço. Peça uma exportação ao integrador KNX se você não tiver o ETS.
Uma máquina Linux, Raspberry Pi ou PC na mesma rede.
Python 3.10 ou mais recente.

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

A maioria das interfaces KNX/IP suporta apenas um número limitado de túneis simultâneos, muitas vezes entre um e alguns poucos. Se o ETS, uma visualização e a sua bridge se conectarem ao mesmo tempo, um deles pode ser recusado. Verifique quantos túneis a sua interface oferece antes de começar.

Passo 1: Mapeie os Endereços de Grupo

Os endereços de grupo KNX têm o formato 1/2/3. Usá-los diretamente como níveis de topic MQTT cria hierarquias profundas e nomes de topics confusos, então associe cada um a um nome significativo e a um DPT:

Endereço de grupo	Nome	DPT	Significado	Direção
1/1/1	f2/office-201/light/switch	1.001	Ligar/desligar	MQTT para KNX
1/1/2	f2/office-201/light/state	1.001	Status do interruptor	KNX para MQTT
1/1/5	f2/office-201/light/brightness-state	5.001	Valor do dimmer 0-100 %	KNX para MQTT
3/1/1	f2/office-201/temperature	9.001	Temperatura ambiente em degC	KNX para MQTT
4/1/1	f2/office-201/presence	1.018	Ocupado / não ocupado	KNX para MQTT

Escreva o mesmo mapeamento na bridge como um dicionário Python. Manter endereços de status e de comando separados é uma prática padrão no KNX e combina muito bem com o MQTT: os topics de estado são publicados pela bridge, os topics de comando são assinados.

Passo 2: A Bridge KNX MQTT

A bridge se conecta ao KNX por tunneling, escuta todos os telegramas, decodifica os que estão no mapeamento e os publica. Ela também assina os topics .../set e os converte em escritas 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 um 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"

# endereço de grupo -> (sufixo do 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"),
}
# sufixo do topic -> (endereço 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)}
    # retido para que os dashboards vejam o último estado imediatamente
    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")
    # os callbacks do paho rodam em outra thread; agende a escrita KNX no loop 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()   # roda para sempre
    finally:
        await xknx.stop()
        mqttc.loop_stop()

asyncio.run(main())

Alguns detalhes importam aqui. As duas assinaturas cobrem topics de comando com quatro e com três níveis abaixo do prefixo, o que corresponde aos nomes em COMMAND_MAP. A bridge também ignora qualquer endereço de grupo que não conheça, então um erro em um topic nunca pode escrever em um endereço arbitrário. Por fim, o run_coroutine_threadsafe é necessário porque o paho entrega as mensagens em sua thread de rede, enquanto o xknx roda dentro do asyncio.

Lendo os Estados Iniciais na Inicialização

Dispositivos KNX só enviam um telegrama quando algo muda, então uma bridge recém-iniciada não sabe nada sobre o estado atual do edifício. As mensagens MQTT retidas escondem isso dos dashboards, mas após uma longa interrupção os valores retidos podem estar desatualizados. A solução é enviar um GroupValueRead para cada endereço de status assim que a conexão KNX estiver ativa. Atuadores com a flag de leitura ativada no ETS respondem com um GroupValueResponse, que a bridge já trata, porque telegram_received aceita tanto escritas quanto respostas.

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)   # seja gentil com o barramento

Chame await read_all_states() logo após await xknx.start(). A pequena pausa entre as leituras importa: uma linha KNX de par trançado tem largura de banda limitada, e disparar dezenas de leituras de uma vez pode atrasar telegramas reais vindos dos botões. Se um endereço de status nunca responder, provavelmente a flag de leitura não está ativada no objeto de comunicação do atuador; peça ao integrador para ativá-la no ETS em vez de contornar o problema na bridge.

Passo 3: Teste a Bridge

Observe os estados publicados:

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

Pressione o interruptor de parede no escritório 201. Você deverá 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"}

Agora acenda a luz via MQTT e reduza a intensidade para 40 por cento:

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

O servidor MQTT público serve para esta primeira rodada, mas qualquer pessoa pode ler e publicar em qualquer topic público. Use um kit KNX de laboratório, não um escritório real, enquanto testa em público.

Passo 4: Rode com um Broker Privado

Para uma instalação real, crie um broker MQTT gerenciado no portal do ConnectMQ e use as credenciais dele:

BROKER = "broker.connectmq.com"   # ou o host exibido para o seu broker
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"
PREFIX = "acme/ist01/hq"

Depois instale a bridge como serviço systemd para que ela reinicie após reboots:

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

Com um broker MQTT gerenciado, o único processo que você mantém no local é a própria bridge.

Mantendo a Bridge Segura

A bridge mantém uma conexão com a instalação KNX, então trate-a como qualquer outro controlador do edifício. Rode-a em um dispositivo ou VLAN dedicados, mantenha o sistema operacional atualizado e guarde a senha do MQTT em um arquivo de ambiente legível apenas pelo usuário do serviço, e não no próprio script. Se a sua instalação KNX usa KNX Secure, o xknx pode usar tunneling seguro com o keyring exportado do ETS; peça as credenciais ao integrador em vez de desativar a segurança na interface.

Solução de Problemas

Conexão de túnel recusada. Todos os túneis da interface estão em uso. Feche o ETS ou outra visualização, ou use uma interface com mais túneis.
Nenhum telegrama chega. A interface pode estar filtrando telegramas de grupo. Verifique a tabela de filtros no ETS ou conecte-se a um roteador KNX/IP.
Os valores decodificados não fazem sentido. O DPT no seu mapeamento está errado. Um float de 2 bytes (DPT 9) decodificado como valor de escala (DPT 5) gera números sem significado.
Estados ausentes após reiniciar a bridge. O KNX só envia telegramas quando há mudança. Envie um GroupValueRead para cada endereço de estado na inicialização ou confie nas mensagens MQTT retidas.
Comandos duplicados. O QoS 1 pode entregar uma mensagem duas vezes; ligar uma luz duas vezes é inofensivo, mas evite a semântica de alternância (toggle) e sempre envie valores absolutos.

Alternativas Que Vale Conhecer

Se você prefere uma ferramenta visual, o Node-RED com um pacote de nós KNX pode fazer o mesmo mapeamento, e o Home Assistant tem uma integração KNX nativa que pode publicar no MQTT por meio de automações. A bridge Python acima é a opção mais transparente e fácil de manter sob controle de versão. A documentação oficial do xknx em xknx.github.io lista todos os DPTs suportados.

Para cenas de iluminação sobre esta bridge, veja iluminação inteligente com MQTT e, se o mesmo edifício roda HVAC em BACnet, combine-a com o gateway de BACnet para MQTT. Nosso guia de automação IoT reúne projetos relacionados.

Conecte Sua Instalação KNX ao ConnectMQ

Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, e você terá as credenciais para a bridge em poucos minutos. O explorador de topics ao vivo do ConnectMQ mostra cada endereço de grupo conforme ele muda, o que torna o comissionamento e a busca de falhas muito mais rápidos.

Perguntas frequentes

Preciso de um roteador KNX/IP ou uma interface basta?

Uma interface IP com tunneling basta para uma única bridge. Um roteador ajuda quando várias aplicações precisam de acesso simultâneo, porque o roteamento usa multicast em vez de um número limitado de túneis.

A instalação KNX continua funcionando se o MQTT cair?

Sim. Os dispositivos KNX se comunicam diretamente por endereços de grupo, então a bridge apenas adiciona visibilidade e controle remoto; o acionamento local nunca depende dela.

Os topics de estado devem ser retidos?

Sim. O KNX só envia valores quando há mudança, então mensagens MQTT retidas permitem que um dashboard mostre o estado atual logo após se conectar.

Posso enviar qualquer endereço de grupo a partir do MQTT?

Não deveria. Mantenha um mapa de comandos explícito na bridge para que apenas endereços aprovados possam ser escritos via MQTT.

Integração KNX MQTT: Bridge de Endereços de Grupo para Topics MQTT