Batchprozess-Überwachung mit MQTT: Phasen, Sollwerte und Abweichungen

Batchprozess-Überwachung bedeutet, jederzeit zu wissen, in welcher Phase sich jede Charge befindet, ob ihre Prozesswerte innerhalb der Rezeptgrenzen bleiben und welche Abweichungen vor der Freigabe eine Unterschrift benötigen. In diesem Tutorial bauen Sie mit MQTT eine kleine, aber vollständige Pipeline: Ein simulierter Kessel veröffentlicht Phasen und Messwerte, und ein Python-Monitor vergleicht sie mit Sollwerten und erfasst jede Abweichung.

Warum sich die Batchprozess-Überwachung von der Überwachung diskreter Fertigung unterscheidet

In der diskreten Fertigung zählen Sie Teile. In der Prozessindustrie (Lebensmittel, Getränke, Chemie, Lacke, Pharma, Kosmetik) stellen Sie eine Materialmenge über eine Abfolge von Phasen her: Befüllen, Aufheizen, Halten, Abkühlen, Umpumpen. Ein Prozesswert, der in einer Phase völlig in Ordnung ist, ist in der nächsten eine Abweichung. 75 °C sind beim Aufheizen erwartet, in der Kühlphase aber ein Problem.

Daraus ergeben sich drei Konsequenzen für die Überwachung:

1. Kontext ist Pflicht. Jeder Messwert muss Chargen-ID, Rezept und aktuelle Phase tragen, sonst lässt er sich nicht beurteilen.
2. Grenzwerte ändern sich über die Zeit. Der Monitor muss Sollwerte und Toleranzen pro Phase laden, nicht pro Tag.
3. Historie zählt für die Freigabe. Qualitätsteams brauchen ein Chargenprotokoll: wann jede Phase begann und endete, und jede Abweichung mit Dauer und Spitzenwert.

Die Norm ISA-88 liefert das Vokabular, das die meisten Werke bereits nutzen: Prozeduren, Teilprozeduren, Operationen und Phasen, die auf Teilanlagen laufen. Sie brauchen keine vollständige ISA-88-Batch-Engine, um von strukturierter Überwachung zu profitieren, doch wenn Sie deren Benennung übernehmen, bleiben Ihre Daten für Verfahrenstechniker verständlich.

MQTT-Topics für Batchdaten entwerfen

MQTT eignet sich gut für Batchdaten, weil ein Phasenwechsel ein Ereignis ist, und Ereignisse sind genau das, wofür Publish/Subscribe gemacht ist. Ein MQTT-Broker empfängt jedes Ereignis einmal und verteilt es an alle interessierten Abnehmer: das HMI, den Historian, das Qualitäts-Dashboard und den Monitor, den Sie weiter unten bauen.

Ein praxistaugliches Layout für eine Teilanlage sieht so aus:

Werden state und phase retained gespeichert, kennt jeder Client, der sich mitten in einer Charge verbindet, sofort den Kontext. Prozesswerte werden nicht retained, weil der nächste Messwert ohnehin Sekunden später eintrifft.

In der Produktion stellen Sie diesen Topics Ihren ISA-95-Pfad (Unternehmen, Werk, Bereich, Linie oder Zelle) voran, wie in ISA-95-MQTT-Topics beschrieben. So wird die ISA-95-Hierarchie zur Topic-Struktur Ihres Unified Namespace (UNS), und Batchdaten stehen dort neben allen anderen Werksdaten, etwa unter acme/werk-ludwigshafen/ansatz/linie-2/kessel1/.... Für dieses Tutorial nutzen wir den öffentlichen MQTT-Broker, der bis zu 8 Topic-Ebenen erlaubt, und halten das Präfix daher kurz: connectmq/demo/<random-id>/kettle1/....

Schritt 1: Eine Charge auf dem öffentlichen MQTT-Broker simulieren

Der kostenlose öffentliche MQTT-Broker unter broker.connectmq.com akzeptiert anonyme Verbindungen mit jeder nicht leeren, eindeutigen Client-ID. Alles, was dort veröffentlicht wird, kann jeder mitlesen. Verwenden Sie daher ein zufälliges Topic-Präfix und senden Sie niemals echte Rezeptdaten.

Installieren Sie die Client-Bibliothek:

pip install "paho-mqtt>=2.0"

Der folgende Simulator fährt ein kurzes Rezept mit vier Phasen, veröffentlicht den retained Chargenstatus und die Phase und streamt jede Sekunde Prozesswerte. Er überschreitet absichtlich die Haltetemperatur, damit der Monitor etwas zu erkennen hat.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/batch-7f3a/kettle1"  # 7f3a durch eine eigene Zufalls-ID ersetzen
BATCH_ID = "B" + time.strftime("%Y%m%d-%H%M%S")

RECIPE = [
    {"phase": "CHARGE", "duration": 10, "temp_sp": 20, "temp_tol": 5,  "rpm_min": 0},
    {"phase": "HEAT",   "duration": 15, "temp_sp": None, "temp_tol": None, "rpm_min": 60},
    {"phase": "HOLD",   "duration": 20, "temp_sp": 72, "temp_tol": 2,  "rpm_min": 60},
    {"phase": "COOL",   "duration": 15, "temp_sp": None, "temp_tol": None, "rpm_min": 30},
]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"batch-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()

def publish(subtopic, payload, retain=False):
    client.publish(f"{PREFIX}/{subtopic}", json.dumps(payload), qos=1, retain=retain)

publish("batch/state", {"batchId": BATCH_ID, "recipe": "SAUCE-A", "status": "RUNNING"}, retain=True)

temp = 18.0
for step in RECIPE:
    publish("batch/phase", {"batchId": BATCH_ID, "ts": time.time(), **step}, retain=True)
    for second in range(step["duration"]):
        if step["phase"] == "HEAT":
            temp += 3.5
        elif step["phase"] == "HOLD":
            temp = 72 + random.uniform(-1.5, 1.5)
            if 8 <= second <= 11:
                temp = 75.2  # eingebaute Überschreitung
        elif step["phase"] == "COOL":
            temp -= 3.0
        rpm = 0 if step["phase"] == "CHARGE" else 80 + random.uniform(-5, 5)
        publish("pv", {"batchId": BATCH_ID, "ts": time.time(),
                       "temp": round(temp, 2), "rpm": round(rpm, 1)})
        time.sleep(1)

publish("batch/state", {"batchId": BATCH_ID, "recipe": "SAUCE-A", "status": "COMPLETE"}, retain=True)
time.sleep(1)
client.loop_stop()
client.disconnect()

Schritt 2: Den Dienst für die Batchprozess-Überwachung bauen

Der Monitor abonniert die Teilanlage, merkt sich die aktuelle Phase und ihre Grenzwerte, prüft jeden Prozesswert und veröffentlicht ein Abweichungsereignis, sobald ein Wert sein Band verlässt. Außerdem verfolgt er Dauer und Spitzenwert der Abweichung, also genau das, was ein Chargenprotokoll tatsächlich braucht.

import json
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/batch-7f3a/kettle1"
current_phase = {}
open_deviation = None

def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)
    client.subscribe(f"{PREFIX}/batch/#", qos=1)
    client.subscribe(f"{PREFIX}/pv", qos=1)

def check(client, pv):
    global open_deviation
    sp, tol = current_phase.get("temp_sp"), current_phase.get("temp_tol")
    out_of_spec = sp is not None and abs(pv["temp"] - sp) > tol
    if out_of_spec and open_deviation is None:
        open_deviation = {"batchId": pv["batchId"], "phase": current_phase["phase"],
                          "parameter": "temp", "limit": f"{sp}+/-{tol}",
                          "start": pv["ts"], "peak": pv["temp"]}
        print("DEVIATION START", open_deviation)
    elif out_of_spec:
        open_deviation["peak"] = max(open_deviation["peak"], pv["temp"], key=lambda v: abs(v - sp))
    elif open_deviation is not None:
        open_deviation["end"] = pv["ts"]
        open_deviation["durationS"] = round(pv["ts"] - open_deviation["start"], 1)
        client.publish(f"{PREFIX}/batch/deviation", json.dumps(open_deviation), qos=1)
        print("DEVIATION CLOSED", open_deviation)
        open_deviation = None

def on_message(client, userdata, msg):
    global current_phase
    data = json.loads(msg.payload)
    if msg.topic.endswith("/batch/phase"):
        current_phase = data
        print(f"Phase -> {data['phase']} (batch {data['batchId']})")
    elif msg.topic.endswith("/batch/state"):
        print(f"Batch {data['batchId']} is {data['status']}")
    elif msg.topic.endswith("/pv") and current_phase:
        if data["rpm"] < current_phase.get("rpm_min", 0):
            print(f"Agitator below minimum in {current_phase['phase']}: {data['rpm']} rpm")
        check(client, data)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"batch-mon-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_forever()

Starten Sie zuerst den Monitor und dann den Simulator in einem zweiten Terminal. Sie sehen die Phasenwechsel und anschließend eine Abweichung, die etwa acht Sekunden nach Beginn von HOLD beginnt und rund vier Sekunden später mit Spitzenwert und Dauer geschlossen wird. Verfolgen Sie parallel den Rohverkehr mit dem Mosquitto-CLI:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watcher-7f3a -t "connectmq/demo/batch-7f3a/kettle1/#" -v

Schritt 3: Aus Ereignissen ein Chargenprotokoll machen

Das Abweichungs-Topic ist das Rohmaterial für Ihr elektronisches Chargenprotokoll. Ein separater Subscriber (ein kleiner Python-Dienst, ein Node-RED-Flow oder eine Telegraf-Instanz) kann Phasenwechsel und Abweichungen in eine Datenbank schreiben, die nach Chargen-ID geschlüsselt ist. Bei der Freigabe liefert eine Abfrage die ganze Geschichte: Start- und Endzeiten der Phasen, die gültigen Sollwerte und jede Exkursion mit Dauer und Spitzenwert.

Beachten Sie beim Entwurf diese Regeln:

1. Schreiben Sie die Sollwerte mit jedem Phasenereignis mit. Rezepte ändern sich; das Protokoll muss die Grenzwerte zeigen, die für diese Charge galten, nicht die heutigen.
2. Verwenden Sie QoS 1 für Phasen- und Abweichungsereignisse. Ein verlorener Phasenwechsel verfälscht jede nachfolgende Beurteilung. Für Prozesswerte genügt meist QoS 0.
3. Zeitstempel an der Quelle setzen. Das Feld ts stammt von der Steuerung oder dem Gateway, nicht vom Subscriber, damit Netzwerkverzögerungen die Phasengrenzen nicht verschieben.
4. Duplikate als normal behandeln. QoS 1 bedeutet „mindestens einmal“, machen Sie den Datenbank-Insert daher idempotent auf Chargen-ID, Phase und Zeitstempel.

Für statistische Trends über viele Chargen speisen dieselben Daten eine Echtzeit-SPC, und die Verknüpfung von Chargen-IDs mit Rohstofflosen bildet das Rückgrat eines Rückverfolgbarkeitssystems für die Fertigung.

Fehlerbehebung in Pipelines zur Batchüberwachung

Mit ConnectMQ in die Produktion

Der öffentliche MQTT-Broker ist für Experimente gedacht. Für echte Chargen brauchen Sie private Zugangsdaten, TLS und einen Broker, den Sie nicht mit der ganzen Welt teilen. ConnectMQ bietet Ihnen einen verwalteten MQTT-Broker mit MQTT 3.1.1 und 5.0, TLS auf Port 8883 sowie generiertem Benutzernamen, Passwort und Client-ID pro Broker. Ändern Sie die Verbindungszeilen in beiden Skripten:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # nutzt den CA-Speicher des Systems; der Broker präsentiert ein Let's-Encrypt-Zertifikat
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

Im ConnectMQ-Portal zeigt der Live-Topic-Explorer den retained Chargenstatus pro Teilanlage, und Schwellenwert-Alarmregeln können einen Prozesswert melden, der einen festen Grenzwert überschreitet, als zweite Verteidigungslinie neben Ihrem phasenbewussten Monitor. Das Gesamtbild, wo Batchdaten in einem Werk ihren Platz haben, beschreibt der Smart-Manufacturing-Leitfaden.

Starten Sie die Batchüberwachung mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen Broker im kostenlosen Serverless-Plan ohne Kreditkarte, fügen Sie die Zugangsdaten in die obigen Skripte ein und fahren Sie in wenigen Minuten Ihre erste überwachte Charge.

Häufig gestellte Fragen

Brauche ich eine ISA-88-Batch-Engine für die Batchprozess-Überwachung mit MQTT?

Nein. Die Batch-Engine oder SPS muss lediglich die aktuelle Phase und die Sollwerte veröffentlichen, wenn sie sich ändern. Jede Steuerung, die das direkt oder über ein Gateway kann, kann den Monitor speisen.

Welche QoS-Stufe sollten Batchereignisse verwenden?

Verwenden Sie QoS 1 für Phasenwechsel, Chargenstatus und Abweichungen, damit sie mindestens einmal zugestellt werden, und machen Sie die Abnehmer idempotent. Hochfrequente Prozesswerte können QoS 0 nutzen, wenn ein gelegentlich verlorener Messwert keine Rolle spielt.

Reicht der kostenlose Plan für eine echte Batchlinie?

Der kostenlose Serverless-Plan unterstützt bis zu 25 Verbindungen, QoS bis 1 und Nachrichten bis 256 KB, was für ein Pilotprojekt mit einigen Teilanlagen genügt. Größere Werke wechseln zu den Plänen Serverless (Paid) oder Dedicated.

Wo sollte das Chargenprotokoll gespeichert werden?

In einer Datenbank, die Ihrem Qualitätssystem gehört. Der Broker transportiert Ereignisse in Echtzeit; ein Subscriber-Dienst schreibt sie in einen dauerhaften Speicher für Prüfung und Freigabe.