BACnet para MQTT: Integrando Controladores de HVAC com um Gateway em Python

Um gateway de BACnet para MQTT lê pontos de controladores de HVAC, como os de um shopping center ou de um hospital, na rede BACnet/IP local e os republica como mensagens MQTT que dashboards e ferramentas de análise podem assinar de qualquer lugar. Neste tutorial você vai construir esse gateway em Python com BAC0 e paho-mqtt, testá-lo no servidor MQTT público e depois apontá-lo para um broker ConnectMQ privado via TLS.

Como o Gateway Funciona

Dispositivos BACnet/IP falam UDP na porta 47808 e esperam estar no mesmo domínio de broadcast que o cliente. É por isso que você não conecta um serviço em nuvem diretamente ao BACnet. Em vez disso, um pequeno gateway fica na rede do edifício:

1. Ele entra na rede BACnet/IP como um dispositivo cliente.
2. Ele lê o presentValue de uma lista configurada de objetos (entradas analógicas, valores analógicos, entradas binárias, valores multiestado).
3. Ele publica cada valor como JSON em um topic MQTT por uma conexão TLS de saída.
4. Opcionalmente, ele assina um pequeno conjunto de topics de comando e escreve setpoints aprovados de volta no controlador.

Como a conexão MQTT é apenas de saída, você não abre nenhuma porta de entrada no firewall. É o mesmo padrão descrito em nossa visão geral de arquiteturas de automação predial com MQTT.

O Que Você Precisa

• Uma máquina Linux, Raspberry Pi ou PC Windows na mesma sub-rede dos controladores BACnet/IP.
• Python 3.10 ou mais recente.
• A instância de dispositivo e o endereço IP de pelo menos um controlador, além dos tipos e instâncias de objetos que você deseja. O front-end do seu BMS ou uma ferramenta de exploração BACnet mostrará essas informações.
• Um broker MQTT. Para o primeiro teste, use o servidor MQTT público (gratuito) em broker.connectmq.com, feito para testes e protótipos. Para dados reais do edifício, use um broker privado.

Instale as bibliotecas:

python -m venv bacnet-gw
source bacnet-gw/bin/activate
pip install "BAC0>=2025.6" "paho-mqtt>=2.0"

As versões recentes do BAC0 são construídas sobre o bacpypes3 e usam asyncio, que é o alvo do código abaixo.

Passo 1: Descubra Dispositivos e Pontos

Antes de escrever um gateway, confirme que a máquina enxerga os controladores. O BAC0 precisa se vincular à interface da sub-rede BACnet, então passe o endereço dela com a máscara de sub-rede.

import asyncio
import BAC0

async def main():
    async with BAC0.start(ip="192.168.10.50/24") as bacnet:
        devices = await bacnet.who_is()
        for dev in devices:
            print(dev)
        value = await bacnet.read("192.168.10.21 analogInput 1 presentValue")
        print("AI1 =", value)

asyncio.run(main())

Se a descoberta não retornar nada, verifique se o gateway e os controladores estão na mesma sub-rede, se a UDP 47808 não está bloqueada por um firewall local e se nenhuma outra aplicação BACnet na mesma máquina já ocupa a porta. Se o controlador estiver atrás de um roteador BACnet em outra sub-rede, talvez você precise de um BBMD ou de registro como foreign device; consulte quem mantém o BMS.

Passo 2: Descreva os Pontos Que Você Quer

Mantenha a lista de pontos em um arquivo de configuração em vez de fixá-la no código. Cada entrada associa um objeto BACnet a um sufixo de topic MQTT, uma unidade e uma banda morta.

{
  "site": "ist01",
  "points": [
    {"addr": "192.168.10.21", "obj": "analogInput 1", "topic": "ahu-1/supply-air-temp", "unit": "degC", "deadband": 0.2},
    {"addr": "192.168.10.21", "obj": "analogInput 2", "topic": "ahu-1/return-air-temp", "unit": "degC", "deadband": 0.2},
    {"addr": "192.168.10.21", "obj": "analogInput 5", "topic": "ahu-1/filter-dp", "unit": "Pa", "deadband": 5},
    {"addr": "192.168.10.21", "obj": "binaryInput 1", "topic": "ahu-1/fan-status", "unit": "", "deadband": 0},
    {"addr": "192.168.10.22", "obj": "analogValue 3", "topic": "vav-301/zone-setpoint", "unit": "degC", "deadband": 0.1, "writable": true, "min": 18, "max": 26}
  ]
}

Salve como points.json.

Passo 3: O Script do Gateway de BACnet para MQTT

O script faz polling de cada ponto, publica apenas quando o valor muda mais do que sua banda morta (ou a cada cinco minutos, como heartbeat) e trata comandos de escrita para os pontos marcados como writable.

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

import BAC0
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883                      # use 8883 com TLS para um broker privado
USERNAME = None                  # "YOUR_USERNAME" para um broker privado
PASSWORD = None                  # "YOUR_PASSWORD"
CLIENT_ID = f"bacnet-gw-{uuid.uuid4().hex[:8]}"   # "YOUR_CLIENT_ID" para um broker privado
PREFIX = "connectmq/demo/bacnet-7k2q"             # prefixo único no broker público
POLL_SECONDS = 15
HEARTBEAT_SECONDS = 300

with open("points.json", encoding="utf-8") as f:
    CONFIG = json.load(f)

POINTS = {p["topic"]: p for p in CONFIG["points"]}
last_sent = {}              # topic -> (value, timestamp)
write_queue: asyncio.Queue = asyncio.Queue()
loop: asyncio.AbstractEventLoop | None = None


def now_iso():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.subscribe(f"{PREFIX}/{CONFIG['site']}/cmd/#", qos=1)
    client.publish(f"{PREFIX}/{CONFIG['site']}/gateway/status", "online", qos=1, retain=True)


def on_message(client, userdata, msg):
    suffix = msg.topic.split("/cmd/", 1)[1]
    point = POINTS.get(suffix)
    if not point or not point.get("writable"):
        print("Rejected write to", suffix)
        return
    try:
        value = float(json.loads(msg.payload)["value"])
    except (ValueError, KeyError, TypeError):
        print("Bad command payload:", msg.payload)
        return
    if not point["min"] <= value <= point["max"]:
        print("Out of range:", value)
        return
    # o paho executa callbacks em sua própria thread: entregue a tarefa ao loop asyncio com segurança
    loop.call_soon_threadsafe(write_queue.put_nowait, (point, value))


def make_client():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
        client.tls_set()
    client.will_set(f"{PREFIX}/{CONFIG['site']}/gateway/status", "offline", qos=1, retain=True)
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)
    client.connect(BROKER, PORT, keepalive=60)
    client.loop_start()
    return client


def should_publish(topic, value, deadband):
    prev = last_sent.get(topic)
    if prev is None:
        return True
    old_value, ts = prev
    if time.time() - ts > HEARTBEAT_SECONDS:
        return True
    if isinstance(value, (int, float)) and isinstance(old_value, (int, float)):
        return abs(value - old_value) >= deadband
    return value != old_value


async def poll(bacnet, client):
    for topic, p in POINTS.items():
        ref = f"{p['addr']} {p['obj']} presentValue"
        try:
            value = await bacnet.read(ref)
            quality = "good"
        except Exception as exc:          # timeouts, objeto desconhecido, dispositivo offline
            print("Read failed", ref, exc)
            value, quality = None, "bad"
        if isinstance(value, float):
            value = round(value, 2)
        if quality == "bad" or should_publish(topic, value, p["deadband"]):
            payload = {"ts": now_iso(), "value": value, "unit": p["unit"],
                       "quality": quality, "source": f"{p['addr']}/{p['obj']}"}
            client.publish(f"{PREFIX}/{CONFIG['site']}/{topic}", json.dumps(payload), qos=0)
            if quality == "good":
                last_sent[topic] = (value, time.time())


async def handle_writes(bacnet, client):
    while True:
        point, value = await write_queue.get()
        ref = f"{point['addr']} {point['obj']} presentValue {value} - 8"   # prioridade 8
        try:
            await bacnet.write(ref)
            print("Wrote", ref)
            client.publish(f"{PREFIX}/{CONFIG['site']}/{point['topic']}/ack",
                           json.dumps({"ts": now_iso(), "value": value, "result": "ok"}), qos=1)
        except Exception as exc:
            client.publish(f"{PREFIX}/{CONFIG['site']}/{point['topic']}/ack",
                           json.dumps({"ts": now_iso(), "value": value, "result": str(exc)}), qos=1)


async def main():
    global loop
    loop = asyncio.get_running_loop()
    client = make_client()
    async with BAC0.start(ip="192.168.10.50/24") as bacnet:
        asyncio.create_task(handle_writes(bacnet, client))
        while True:
            await poll(bacnet, client)
            await asyncio.sleep(POLL_SECONDS)

asyncio.run(main())

Execute-o com python gateway.py. Cada ponto aparece sob connectmq/demo/bacnet-7k2q/ist01/....

Mapeando Objetos BACnet para Topics MQTT

O BACnet nomeia objetos por tipo e número de instância, por exemplo analogInput 3 no dispositivo 1201. Esses identificadores não significam nada para um gestor de facilities navegando pelos topics ou para quem monta um gêmeo digital do edifício, e mudam se um controlador for substituído ou reprogramado. É por isso que a lista de pontos associa cada objeto a um topic descritivo como ahu-1/supply-air-temp, enquanto a referência original viaja dentro do payload como source.

Algumas convenções mantêm o mapeamento sustentável conforme o edifício cresce:

• Nomeie equipamentos, não controladores. Um controlador pode atender a duas unidades de tratamento de ar, e uma unidade pode abranger dois controladores. Os topics devem seguir o equipamento.
• Use um único sistema de unidades. Converta Fahrenheit ou polegadas de coluna d'água no gateway se um site misturar fabricantes, e coloque a unidade em todo payload.
• Traduza valores multiestado. Um multiStateValue retorna um inteiro; publique o texto do estado (por exemplo occupied, unoccupied, standby) junto com o número, usando a lista de textos de estado do objeto.
• Trate objetos binários como booleanos. O BACnet retorna active ou inactive; publique true ou false para que dashboards e regras de alerta possam compará-los facilmente.
• Mantenha a lista sob controle de versão. Quando uma empresa de instalações renumera objetos, uma alteração revisada no points.json é muito mais fácil de auditar do que um script editado à mão.

Passo 4: Verifique os Dados

Assine a partir de qualquer máquina:

mosquitto_sub -h broker.connectmq.com -p 1883 -i viewer-7k2q -t "connectmq/demo/bacnet-7k2q/#" -v

Depois envie uma alteração de setpoint de teste para o ponto VAV gravável:

mosquitto_pub -h broker.connectmq.com -p 1883 -i cmd-7k2q -q 1 \
  -t "connectmq/demo/bacnet-7k2q/ist01/cmd/vav-301/zone-setpoint" -m '{"value": 22.5}'

O gateway escreve o valor na prioridade 8 e publica uma confirmação no topic /ack. Lembre-se de que tudo no servidor MQTT público pode ser lido por qualquer pessoa, então faça isso apenas com um controlador de teste ou um simulador. Um topic de comando no servidor MQTT público nunca deve chegar a um edifício real.

Passo 5: Migre para um Broker Privado com TLS

Para produção, crie um broker no portal do ConnectMQ e copie o usuário, a senha e o client ID gerados. Altere quatro constantes:

BROKER = "broker.connectmq.com"   # ou o host exibido para o seu broker no portal
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"

O client.tls_set() usa o repositório de CAs do sistema, que já confia no certificado Let's Encrypt apresentado pelo broker. Use uma raiz de topic como acme/ist01/... em vez do prefixo de demonstração. Um broker MQTT gerenciado mantém este gateway simples: nenhum certificado para renovar do seu lado e nenhum broker para atualizar.

Checklist de Solução de Problemas

Dicas para Produção

• Prefira COV quando houver suporte. As assinaturas change-of-value do BACnet reduzem a carga de polling em controladores antigos. Comece com polling pela simplicidade e depois migre os pontos mais movimentados para COV.
• Não faça polling agressivo demais. Controladores MS/TP mais antigos atrás de roteadores respondem devagar; 15 a 60 segundos costumam bastar para HVAC.
• Publique metadados como retidos. Uma mensagem retida com unidades, nomes de objetos e mín./máx. ajuda os dashboards a renderizar corretamente desde o primeiro segundo.
• Monitore o próprio gateway. A mensagem last will muda gateway/status para offline se o processo morrer, e uma regra de alerta por limite do ConnectMQ sobre a telemetria principal pode sinalizar sensores que saem da faixa.

Para um caso de uso prático construído sobre esses pontos, veja monitoramento de HVAC com sensores IoT. Se o mesmo edifício também usa KNX para a iluminação, a bridge KNX para MQTT segue a mesma estrutura. Mais projetos de automação estão reunidos em nosso guia de automação IoT, e a documentação oficial do BAC0 fica em bac0.readthedocs.io.

Leve Seus Dados BACnet ao ConnectMQ

Cadastre-se em app.connectmq.com e crie um broker Serverless gratuito, sem cartão de crédito. Aponte o gateway para ele, abra o explorador de topics ao vivo e você verá cada ponto de HVAC chegar em tempo real. O ConnectMQ cuida de TLS, autenticação e disponibilidade para que sua equipe possa se concentrar no edifício.

Perguntas frequentes

Um serviço em nuvem pode falar BACnet diretamente?

Não de forma confiável. O BACnet/IP depende de broadcasts UDP em uma rede local, então você precisa de um gateway no local que converta os pontos BACnet para MQTT e se conecte ao broker por uma conexão de saída.

Quais objetos BACnet devo publicar primeiro?

Comece pelas temperaturas do ar de insuflamento e de retorno, pela pressão diferencial dos filtros, pelo status dos ventiladores e pelas principais temperaturas de zona. Eles oferecem o maior valor de diagnóstico com o menor número de pontos.

É seguro escrever setpoints via MQTT?

Pode ser, se o gateway permitir apenas pontos específicos, validar as faixas, escrever em uma prioridade combinada e usar uma conexão TLS autenticada com um broker privado.

Que intervalo de polling devo usar?

Para HVAC, 15 a 60 segundos com uma banda morta é o típico. Polling mais rápido raramente acrescenta informação e pode sobrecarregar controladores mais antigos.