BACnet vers MQTT : intégrer des contrôleurs CVC avec une passerelle Python

Une passerelle BACnet vers MQTT lit les points des contrôleurs CVC sur le réseau BACnet/IP local et les republie sous forme de messages MQTT auxquels tableaux de bord, outils d'analyse et jumeaux numériques peuvent s'abonner depuis n'importe où. Dans ce tutoriel, vous allez construire cette passerelle en Python avec BAC0 et paho-mqtt, la tester sur le serveur MQTT public gratuit, puis la pointer vers un broker ConnectMQ privé via TLS.

Comment fonctionne la passerelle

Les équipements BACnet/IP communiquent en UDP sur le port 47808 et s'attendent à se trouver dans le même domaine de broadcast que le client. C'est pourquoi on ne connecte pas directement un service cloud à BACnet. À la place, une petite passerelle est installée sur le réseau du bâtiment :

1. Elle rejoint le réseau BACnet/IP en tant qu'équipement client.
2. Elle lit la presentValue d'une liste configurée d'objets (entrées analogiques, valeurs analogiques, entrées binaires, valeurs multi-états).
3. Elle publie chaque valeur en JSON sur un topic MQTT via une connexion TLS sortante.
4. En option, elle s'abonne à un petit ensemble de topics de commande et écrit les consignes approuvées dans le contrôleur.

Comme la connexion MQTT est uniquement sortante, vous n'ouvrez aucun port entrant sur le pare-feu. C'est le même schéma que celui décrit dans notre présentation des architectures MQTT de gestion technique du bâtiment.

Ce dont vous avez besoin

• Une machine Linux, un Raspberry Pi ou un PC Windows sur le même sous-réseau que les contrôleurs BACnet/IP.
• Python 3.10 ou plus récent.
• L'instance d'équipement et l'adresse IP d'au moins un contrôleur, ainsi que les types et instances d'objets souhaités. Votre supervision GTB ou un outil d'exploration BACnet vous les indiquera.
• Un broker MQTT. Pour le premier test, utilisez le serveur MQTT public gratuit sur broker.connectmq.com, réservé aux tests. Pour de vraies données de bâtiment, utilisez un broker privé.

Installez les bibliothèques :

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

Les versions récentes de BAC0 reposent sur bacpypes3 et utilisent asyncio, ce que cible le code ci-dessous.

Étape 1 : découvrir les équipements et les points

Avant d'écrire une passerelle, vérifiez que la machine voit les contrôleurs. BAC0 doit se lier à l'interface du sous-réseau BACnet ; passez donc son adresse avec le masque de sous-réseau.

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())

Si la découverte ne renvoie rien, vérifiez que la passerelle et les contrôleurs sont sur le même sous-réseau, que le port UDP 47808 n'est pas bloqué par un pare-feu local et qu'aucune autre application BACnet sur la même machine n'occupe déjà le port. Si le contrôleur se trouve derrière un routeur BACnet sur un autre sous-réseau, vous aurez peut-être besoin d'un BBMD ou d'un enregistrement en tant que foreign device ; renseignez-vous auprès de la personne qui maintient la GTB.

Étape 2 : décrire les points souhaités

Conservez la liste des points dans un fichier de configuration plutôt que de la coder en dur. Chaque entrée associe un objet BACnet à un suffixe de topic MQTT, une unité et une bande morte.

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

Enregistrez-la sous points.json.

Étape 3 : le script de la passerelle BACnet vers MQTT

Le script interroge chaque point, ne publie que lorsque la valeur varie de plus que sa bande morte (ou toutes les cinq minutes en guise de heartbeat) et traite les commandes d'écriture pour les points marqués 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 with TLS for a private broker
USERNAME = None                  # "YOUR_USERNAME" for a private broker
PASSWORD = None                  # "YOUR_PASSWORD"
CLIENT_ID = f"bacnet-gw-{uuid.uuid4().hex[:8]}"   # "YOUR_CLIENT_ID" for a private broker
PREFIX = "connectmq/demo/bacnet-7k2q"             # unique prefix on the public broker
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
    # paho runs callbacks in its own thread: hand the job to the asyncio loop safely
    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, unknown object, device 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"   # priority 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())

Lancez-le avec python gateway.py. Chaque point apparaît sous connectmq/demo/bacnet-7k2q/ist01/....

Associer les objets BACnet aux topics MQTT

BACnet nomme les objets par type et numéro d'instance, par exemple analogInput 3 sur l'équipement 1201. Ces identifiants ne signifient rien pour un gestionnaire d'installations qui parcourt les topics, et ils changent si un contrôleur est remplacé ou reprogrammé. C'est pourquoi la liste des points associe chaque objet à un topic descriptif comme ahu-1/supply-air-temp, tandis que la référence d'origine voyage dans le payload sous source.

Quelques conventions permettent de garder cette correspondance maintenable à mesure que le bâtiment évolue :

• Nommez les équipements, pas les contrôleurs. Un contrôleur peut desservir deux centrales de traitement d'air, et une centrale peut s'étendre sur deux contrôleurs. Les topics doivent suivre l'équipement.
• Utilisez un seul système d'unités. Convertissez les degrés Fahrenheit ou les pouces de colonne d'eau dans la passerelle si un site mélange les fabricants, et indiquez l'unité dans chaque payload.
• Traduisez les valeurs multi-états. Une multiStateValue renvoie un entier ; publiez le libellé de l'état (par exemple occupied, unoccupied, standby) à côté du nombre, en utilisant la liste des libellés d'état de l'objet.
• Traitez les objets binaires comme des booléens. BACnet renvoie active ou inactive ; publiez true ou false pour que tableaux de bord et règles d'alerte puissent les comparer facilement.
• Versionnez la liste. Quand un installateur CVC renumérote les objets, une modification relue de points.json est bien plus facile à auditer qu'un script modifié à la main.

Étape 4 : vérifier les données

Abonnez-vous depuis n'importe quelle machine :

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

Envoyez ensuite un changement de consigne de test au point VAV accessible en écriture :

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

La passerelle écrit la valeur en priorité 8 et publie un acquittement sur le topic /ack. N'oubliez pas que tout ce qui transite par le serveur MQTT public gratuit est lisible par n'importe qui ; ne faites donc cela qu'avec un contrôleur de test ou un simulateur. Un topic de commande sur un serveur public ne doit jamais atteindre un vrai bâtiment.

Étape 5 : passer à un broker privé via TLS

Pour la production, créez un broker dans le portail ConnectMQ et copiez le nom d'utilisateur, le mot de passe et le client ID générés. Modifiez quatre constantes :

BROKER = "broker.connectmq.com"   # or the host shown for your broker in the portal
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"

client.tls_set() utilise le magasin de certificats d'autorité du système, qui fait déjà confiance au certificat Let's Encrypt présenté par le broker. Utilisez une racine de topic comme acme/ist01/... au lieu du préfixe de démonstration. Un broker MQTT managé garde cette passerelle simple : aucun certificat à renouveler de votre côté et aucun broker à mettre à jour.

Checklist de dépannage

Conseils pour la production

• Privilégiez le COV lorsqu'il est pris en charge. Les abonnements BACnet change-of-value réduisent la charge d'interrogation sur les anciens contrôleurs. Commencez par l'interrogation pour plus de simplicité, puis passez les points les plus actifs en COV.
• N'interrogez pas trop agressivement. Les anciens contrôleurs MS/TP derrière des routeurs répondent lentement ; 15 à 60 secondes suffisent généralement pour le CVC.
• Publiez les métadonnées en retenu. Un message retenu avec les unités, les noms d'objets et les min/max aide les tableaux de bord à s'afficher correctement dès la première seconde.
• Surveillez la passerelle elle-même. Le message de dernière volonté (last will) bascule gateway/status à offline si le processus s'arrête, et une règle d'alerte à seuil ConnectMQ sur la télémétrie clé peut signaler les capteurs qui sortent de leur plage.

Pour un cas d'usage concret construit sur ces points, consultez la surveillance CVC avec des capteurs IoT. Si le même bâtiment utilise aussi KNX pour l'éclairage, le pont KNX vers MQTT suit la même structure. D'autres projets d'automatisation sont réunis dans notre guide de l'automatisation IoT, et la documentation officielle de BAC0 se trouve sur bac0.readthedocs.io.

Amenez vos données BACnet sur ConnectMQ

Inscrivez-vous sur app.connectmq.com et créez un broker Serverless gratuit, sans carte bancaire. Pointez la passerelle vers celui-ci, ouvrez l'explorateur de topics en direct et vous verrez chaque point CVC arriver en temps réel. ConnectMQ prend en charge TLS, l'authentification et la disponibilité pour que votre équipe puisse se concentrer sur le bâtiment.

Questions fréquentes

Un service cloud peut-il parler BACnet directement ?

Pas de manière fiable. BACnet/IP dépend des broadcasts UDP sur un réseau local ; il vous faut donc une passerelle sur site qui convertit les points BACnet en MQTT et se connecte en sortie au broker.

Quels objets BACnet publier en premier ?

Commencez par les températures d'air soufflé et de reprise, la pression différentielle des filtres, l'état des ventilateurs et les principales températures de zone. Ils offrent la plus grande valeur de diagnostic pour le plus petit nombre de points.

Est-il sûr d'écrire des consignes via MQTT ?

Oui, à condition que la passerelle n'autorise que des points précis, valide les plages, écrive à une priorité convenue et utilise une connexion TLS authentifiée vers un broker privé.

Quel intervalle d'interrogation choisir ?

Pour le CVC, 15 à 60 secondes avec une bande morte est la norme. Une interrogation plus rapide apporte rarement plus d'informations et peut surcharger les anciens contrôleurs.