Batch Process Monitoring with MQTT: Phases, Setpoints and Deviations

Batch process monitoring means knowing, at every moment, which phase each batch is in, whether its process values stay inside the recipe limits and which deviations need a signature before release. In this tutorial you will build a small but complete pipeline with MQTT: a simulated kettle publishes phases and readings, and a Python monitor compares them to setpoints and records every deviation.

Why Batch Process Monitoring Is Different from Discrete Monitoring

In discrete manufacturing you count parts. In process manufacturing (food, beverages, chemicals, coatings, pharmaceuticals, cosmetics) you produce a quantity of material through a sequence of phases: charge, heat, hold, cool, transfer. A process value that is perfectly fine in one phase is a deviation in the next. A 75 °C reading is expected during a heat-up but a problem during a cooling phase.

That has three consequences for monitoring:

1. Context is mandatory. Every reading must carry the batch ID, the recipe and the current phase, or it cannot be judged.
2. Limits change over time. The monitor must load setpoints and tolerances per phase, not per tag.
3. History matters for release. Quality teams need a batch record: when each phase started and ended, and every deviation with its duration and peak value.

The ISA-88 standard gives the vocabulary most plants already use: procedures, unit procedures, operations and phases running on units. You do not need a full ISA-88 batch engine to benefit from structured monitoring, but borrowing its naming keeps your data understandable to process engineers.

Designing MQTT Topics for Batch Data

MQTT works well for batch data because a phase change is an event, and events are exactly what publish/subscribe handles. An MQTT broker receives each event once and fans it out to every interested consumer: the HMI, the historian, the quality dashboard and the monitor you build below.

A practical layout for one unit looks like this:

Retaining state and phase means any client that connects mid-batch immediately knows the context. Process values are not retained because the next reading arrives seconds later anyway.

In production, prefix these with your ISA-95 path (enterprise, site, area, line or cell) as described in ISA-95 MQTT topics. For this tutorial we use the public test broker, which allows up to 8 topic levels, so we keep the prefix short: connectmq/demo/<random-id>/kettle1/....

Step 1: Simulate a Batch on the Public Broker

The free public MQTT test broker at broker.connectmq.com accepts anonymous connections with any non-empty unique client ID. Everything published there is readable by anyone, so use a random topic prefix and never send real recipe data.

Install the client library:

pip install "paho-mqtt>=2.0"

The simulator below runs a short recipe with four phases, publishes the retained batch state and phase, and streams process values every second. It deliberately overshoots the hold temperature so the monitor has something to catch.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/batch-7f3a/kettle1"  # change 7f3a to your own random id
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  # injected overshoot
        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()

Step 2: Build the Batch Process Monitoring Service

The monitor subscribes to the unit, remembers the current phase and its limits, checks every process value and publishes a deviation event when a value leaves its band. It also tracks the deviation's duration and peak, which is what a batch record actually needs.

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

Start the monitor first, then the simulator in a second terminal. You will see the phase transitions, then a deviation that opens around eight seconds into HOLD and closes about four seconds later with its peak and duration. Watch the raw traffic at the same time with the Mosquitto CLI:

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

Step 3: Turn Events into a Batch Record

The deviation topic is the raw material for your electronic batch record. A separate subscriber (a small Python service, Node-RED flow or Telegraf instance) can write phase changes and deviations into a database keyed by batch ID. At release time, a query returns the full story: phase start and end times, the setpoints in force, and each excursion with its duration and peak.

Keep these rules in mind when you design it:

1. Write the setpoints with each phase event. Recipes change; the record must show the limits that applied to this batch, not today's values.
2. Use QoS 1 for phase and deviation events. Losing a phase change corrupts every judgment that follows. Process values at QoS 0 are usually fine.
3. Timestamp at the source. The ts field comes from the controller or gateway, not from the subscriber, so network delay does not shift phase boundaries.
4. Treat duplicates as normal. QoS 1 is at-least-once, so make the database insert idempotent on batch ID, phase and timestamp.

For statistical trends across many batches, the same data feeds real-time SPC, and linking batch IDs to raw-material lots gives you the backbone of a manufacturing traceability system.

Troubleshooting Batch Monitoring Pipelines

Moving to Production on ConnectMQ

The public broker is for experiments. For real batches you want private credentials, TLS and a broker that is not shared with the world. ConnectMQ gives you a managed MQTT broker with MQTT 3.1.1 and 5.0, TLS on port 8883 and generated username, password and client ID per broker. Change the connection lines in both scripts:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # uses the system CA store; the broker presents a Let's Encrypt certificate
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

In the ConnectMQ portal, the live topic explorer shows retained batch state per unit, and threshold alert rules can flag a process value that crosses a fixed limit as a second line of defense next to your phase-aware monitor. The broader picture of where batch data fits in a plant is covered in the smart manufacturing guide.

Start Monitoring Batches on ConnectMQ

Create a broker at app.connectmq.com on the free Serverless plan with no credit card, paste the credentials into the scripts above and run your first monitored batch in minutes.

Frequently Asked Questions

Do I need an ISA-88 batch engine to use MQTT batch process monitoring?

No. The batch engine or PLC only has to publish the current phase and setpoints when they change. Any controller that can do that, directly or through a gateway, can feed the monitor.

Which QoS level should batch events use?

Use QoS 1 for phase changes, batch state and deviations so they are delivered at least once, and make consumers idempotent. High-rate process values can use QoS 0 when an occasional lost sample does not matter.

Can the free plan handle a real batch line?

The free Serverless plan supports up to 25 connections, QoS up to 1 and 256 KB messages, which is enough for a pilot on a few units. Larger plants move to the paid Serverless or Dedicated plans.

Where should the batch record be stored?

In a database owned by your quality system. The broker moves events in real time; a subscriber service writes them into durable storage for review and release.