BACnet zu MQTT: HLK-Controller mit einem Python-Gateway integrieren

Ein BACnet-zu-MQTT-Gateway liest Datenpunkte aus HLK-Controllern im lokalen BACnet/IP-Netz und veröffentlicht sie erneut als MQTT-Nachrichten, die Dashboards und Analysewerkzeuge von überall abonnieren können. In diesem Tutorial bauen Sie dieses Gateway in Python mit BAC0 und paho-mqtt, testen es gegen den öffentlichen MQTT-Broker und verbinden es anschließend über TLS mit einem privaten ConnectMQ-Broker.

So funktioniert das Gateway

BACnet/IP-Geräte kommunizieren per UDP auf Port 47808 und erwarten, sich in derselben Broadcast-Domäne wie der Client zu befinden. Deshalb verbinden Sie einen Cloud-Dienst nicht direkt mit BACnet. Stattdessen sitzt ein kleines Gateway im Gebäudenetz:

1. Es tritt dem BACnet/IP-Netz als Client-Gerät bei.
2. Es liest den presentValue einer konfigurierten Liste von Objekten (Analogeingänge, Analogwerte, Binäreingänge, Mehrstufenwerte).
3. Es veröffentlicht jeden Wert als JSON über eine ausgehende TLS-Verbindung auf einem MQTT-Topic.
4. Optional abonniert es eine kleine Menge von Befehls-Topics und schreibt freigegebene Sollwerte zurück in den Controller.

Da die MQTT-Verbindung ausschließlich ausgehend ist, öffnen Sie keine eingehenden Firewall-Ports. Es ist dasselbe Muster, das in unserer Übersicht zu MQTT-Architekturen für die Gebäudeautomation beschrieben wird.

Was Sie benötigen

• Einen Linux-Rechner, Raspberry Pi oder Windows-PC im selben Subnetz wie die BACnet/IP-Controller.
• Python 3.10 oder neuer.
• Die Geräteinstanz und IP-Adresse mindestens eines Controllers sowie die gewünschten Objekttypen und -instanzen. Ihre GLT-Oberfläche oder ein BACnet-Explorer-Werkzeug zeigt diese an.
• Einen MQTT-Broker. Für den ersten Test verwenden Sie den kostenlosen öffentlichen MQTT-Broker unter broker.connectmq.com. Für echte Gebäudedaten verwenden Sie einen privaten Broker.

Installieren Sie die Bibliotheken:

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

Aktuelle BAC0-Versionen basieren auf bacpypes3 und nutzen asyncio, worauf der folgende Code ausgelegt ist.

Schritt 1: Geräte und Datenpunkte erkennen

Bevor Sie ein Gateway schreiben, prüfen Sie, ob der Rechner die Controller sieht. BAC0 muss sich an die Schnittstelle im BACnet-Subnetz binden, übergeben Sie daher deren Adresse mit Subnetzmaske.

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

Wenn die Erkennung nichts liefert, prüfen Sie, ob Gateway und Controller im selben Subnetz liegen, ob UDP 47808 nicht durch eine Host-Firewall blockiert wird und ob keine andere BACnet-Anwendung auf demselben Rechner den Port bereits belegt. Sitzt der Controller hinter einem BACnet-Router in einem anderen Subnetz, benötigen Sie möglicherweise ein BBMD oder eine Foreign-Device-Registrierung; fragen Sie die Person, die die GLT betreut.

Schritt 2: Die gewünschten Datenpunkte beschreiben

Halten Sie die Datenpunktliste in einer Konfigurationsdatei, statt sie fest im Code zu hinterlegen. Jeder Eintrag ordnet ein BACnet-Objekt einem MQTT-Topic-Suffix, einer Einheit und einem Totband zu.

{
  "site": "stg01",
  "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}
  ]
}

Speichern Sie sie als points.json.

Schritt 3: Das BACnet-zu-MQTT-Gateway-Skript

Das Skript fragt jeden Datenpunkt ab, veröffentlicht nur, wenn sich der Wert um mehr als sein Totband ändert (oder alle fünf Minuten als Heartbeat), und verarbeitet Schreibbefehle für Datenpunkte, die als writable markiert sind.

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                      # für einen privaten Broker 8883 mit TLS verwenden
USERNAME = None                  # "YOUR_USERNAME" für einen privaten Broker
PASSWORD = None                  # "YOUR_PASSWORD"
CLIENT_ID = f"bacnet-gw-{uuid.uuid4().hex[:8]}"   # "YOUR_CLIENT_ID" für einen privaten Broker
PREFIX = "connectmq/demo/bacnet-7k2q"             # eindeutiges Präfix auf dem öffentlichen MQTT-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 führt Callbacks in einem eigenen Thread aus: Auftrag sicher an die asyncio-Schleife übergeben
    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, unbekanntes Objekt, Gerät 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"   # Priorität 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())

Starten Sie es mit python gateway.py. Jeder Datenpunkt erscheint unter connectmq/demo/bacnet-7k2q/stg01/....

BACnet-Objekte auf MQTT-Topics abbilden

BACnet benennt Objekte nach Typ und Instanznummer, zum Beispiel analogInput 3 auf Gerät 1201. Diese Bezeichner sagen einem Facility Manager, der Topics durchsucht, nichts, und sie ändern sich, wenn ein Controller ausgetauscht oder neu programmiert wird. Deshalb ordnet die Datenpunktliste jedes Objekt einem sprechenden Topic wie ahu-1/supply-air-temp zu, während die ursprüngliche Referenz im Payload als source mitreist. Betreiben Sie neben dem Gebäude auch eine Produktion, lohnt sich ein gemeinsamer Unified Namespace (UNS): Die ISA-95-Hierarchie (Enterprise → Site → Area → Line → Cell) bildet die Topic-Struktur, und die Gebäudetechnik hängt als eigener Bereich darunter, etwa acme/werk-stuttgart/gebaeudetechnik/ahu-1/supply-air-temp neben acme/werk-stuttgart/montage/linie-1/.... So lassen sich Energieverbrauch von Lüftung und Produktion in einem Energiemonitoring zusammenführen.

Einige Konventionen halten die Zuordnung wartbar, wenn das Gebäude wächst:

• Anlagen benennen, nicht Controller. Ein Controller kann zwei Lüftungsgeräte bedienen, und ein Lüftungsgerät kann sich über zwei Controller erstrecken. Topics sollten der Anlage folgen.
• Ein einheitliches Einheitensystem verwenden. Rechnen Sie Fahrenheit oder Zoll Wassersäule im Gateway um, wenn ein Standort Hersteller mischt, und nehmen Sie die Einheit in jeden Payload auf.
• Mehrstufenwerte übersetzen. Ein multiStateValue liefert eine Ganzzahl; veröffentlichen Sie den Zustandstext (zum Beispiel occupied, unoccupied, standby) zusammen mit der Zahl, anhand der State-Text-Liste des Objekts.
• Binäre Objekte als Boolesche Werte behandeln. BACnet liefert active oder inactive; veröffentlichen Sie true oder false, damit Dashboards und Alarmregeln sie leicht vergleichen können.
• Die Liste unter Versionskontrolle halten. Wenn ein Anlagenbauer Objekte neu nummeriert, lässt sich eine geprüfte Änderung an points.json weit leichter nachvollziehen als ein von Hand bearbeitetes Skript.

Schritt 4: Die Daten prüfen

Abonnieren Sie von einem beliebigen Rechner aus:

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

Senden Sie dann eine Test-Sollwertänderung an den schreibbaren VVS-Datenpunkt:

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

Das Gateway schreibt den Wert mit Priorität 8 und veröffentlicht eine Bestätigung auf dem Topic /ack. Denken Sie daran, dass alles auf dem öffentlichen MQTT-Broker für jeden lesbar ist; tun Sie das daher nur mit einem Test-Controller oder einem Simulator. Ein Befehls-Topic auf dem öffentlichen MQTT-Broker darf niemals ein echtes Gebäude erreichen.

Schritt 5: Wechsel zu einem privaten Broker mit TLS

Für den Produktivbetrieb erstellen Sie im ConnectMQ-Portal einen Broker und kopieren den generierten Benutzernamen, das Passwort und die Client-ID. Ändern Sie vier Konstanten:

BROKER = "broker.connectmq.com"   # oder der im Portal für Ihren Broker angezeigte Host
PORT = 8883
USERNAME = "YOUR_USERNAME"
PASSWORD = "YOUR_PASSWORD"
CLIENT_ID = "YOUR_CLIENT_ID"

client.tls_set() verwendet den CA-Speicher des Systems, der dem vom Broker präsentierten Let's-Encrypt-Zertifikat bereits vertraut. Verwenden Sie eine Topic-Wurzel wie acme/stg01/... anstelle des Demo-Präfixes, passend zur UNS-Struktur Ihres Unternehmens. Ein verwalteter MQTT-Broker hält dieses Gateway einfach: keine Zertifikate, die Sie erneuern müssen, und kein Broker, den Sie patchen müssen.

Checkliste zur Fehlersuche

Tipps für den Produktivbetrieb

• COV bevorzugen, wo unterstützt. BACnet-Change-of-Value-Abonnements verringern die Polling-Last auf älteren Controllern. Beginnen Sie der Einfachheit halber mit Polling und stellen Sie stark genutzte Datenpunkte dann auf COV um.
• Nicht zu aggressiv pollen. Ältere MS/TP-Controller hinter Routern antworten langsam; 15 bis 60 Sekunden reichen für HLK meist völlig aus.
• Metadaten retained veröffentlichen. Eine retained Nachricht mit Einheiten, Objektnamen und Min-/Max-Werten hilft Dashboards, ab der ersten Sekunde korrekt darzustellen.
• Das Gateway selbst überwachen. Die Last-Will-Nachricht setzt gateway/status auf offline, wenn der Prozess abstürzt, und eine ConnectMQ-Schwellenwert-Alarmregel auf wichtiger Telemetrie kann Sensoren melden, die den zulässigen Bereich verlassen.

Einen praktischen Anwendungsfall auf Basis dieser Datenpunkte finden Sie unter HLK-Überwachung mit IoT-Sensoren. Nutzt dasselbe Gebäude KNX für die Beleuchtung, folgt die KNX-zu-MQTT-Bridge derselben Struktur. Weitere Automatisierungsprojekte sind in unserem Leitfaden zur IoT-Automatisierung gesammelt, und die offizielle BAC0-Dokumentation finden Sie unter bac0.readthedocs.io.

Bringen Sie Ihre BACnet-Daten zu ConnectMQ

Registrieren Sie sich unter app.connectmq.com und erstellen Sie einen kostenlosen Serverless-Broker, ganz ohne Kreditkarte. Richten Sie das Gateway darauf aus, öffnen Sie den Live-Topic-Explorer, und Sie sehen jeden HLK-Datenpunkt in Echtzeit eintreffen. ConnectMQ kümmert sich um TLS, Authentifizierung und Verfügbarkeit, damit sich Ihr Team auf das Gebäude konzentrieren kann.

Häufig gestellte Fragen

Kann ein Cloud-Dienst direkt BACnet sprechen?

Nicht zuverlässig. BACnet/IP ist auf UDP-Broadcasts in einem lokalen Netz angewiesen, daher benötigen Sie vor Ort ein Gateway, das BACnet-Datenpunkte in MQTT umwandelt und sich ausgehend mit dem Broker verbindet.

Welche BACnet-Objekte sollte ich zuerst veröffentlichen?

Beginnen Sie mit Zu- und Ablufttemperaturen, Filterdifferenzdruck, Ventilatorstatus und den wichtigsten Zonentemperaturen. Sie liefern mit den wenigsten Datenpunkten den größten Diagnosewert.

Ist es sicher, Sollwerte über MQTT zu schreiben?

Das kann es sein, wenn das Gateway nur bestimmte Datenpunkte zulässt, Wertebereiche prüft, mit einer vereinbarten Priorität schreibt und eine authentifizierte TLS-Verbindung zu einem privaten Broker nutzt.

Welches Polling-Intervall sollte ich verwenden?

Für HLK sind 15 bis 60 Sekunden mit Totband üblich. Schnelleres Polling bringt selten zusätzliche Erkenntnisse und kann ältere Controller überlasten.