Real-Time OEE Monitoring with MQTT: Build a Live OEE Service

Real-time OEE monitoring replaces the end-of-shift spreadsheet with a number that updates every minute while there is still time to act. In this tutorial you will build the full pipeline: machines publish state and counts over MQTT, a Python service calculates availability, performance and quality for the current shift, and the result is published back to the broker for any dashboard to display. You can run every piece on a laptop against a live broker.

Why Real-Time OEE Beats End-of-Shift Reports

A shift report tells you what went wrong yesterday. By then the shift is over, the operators have gone home and the reasons are half forgotten. A live OEE figure shown at the line lets the supervisor see availability drop after a breakdown, notice performance sagging when speed is reduced, and react while the same people and conditions are still there.

Automatic calculation also removes arguments about the inputs. When states and counts come straight from the machine, the number is the same for operators, supervisors and managers, and discussions move from "is the number right?" to "what do we do about it?".

Architecture

[Machine / PLC gateway] --state, counts--> [MQTT broker] --> [OEE service] --oee--> [MQTT broker] --> [Dashboards, andon, historian]

Every component talks only to the broker. Machines do not know the OEE service exists, and dashboards do not know where the data came from. That decoupling is the main reason to use a broker; if the pattern is new to you, read about what an MQTT broker does first.

Topics for one machine:

States are RUNNING, DOWN, IDLE, SETUP and PLANNED_STOP. Planned stops are excluded from planned production time; every other non-running state counts as availability loss. The formula details and edge cases are covered in how to calculate OEE.

Step 1: Simulate a Machine

Use the public MQTT test broker for development. Pick a random ID for your topic prefix because the public area is visible to everyone.

# machine_sim.py
import json
import random
import time

import paho.mqtt.client as mqtt

BASE = "connectmq/demo/q8n3d1/plant1/line2/filler"
IDEAL_CYCLE_S = 1.0

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="filler-sim-q8n3d1")
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()

shift_start = time.time()
client.publish(f"{BASE}/config", json.dumps({"ideal_cycle_s": IDEAL_CYCLE_S, "shift_start": shift_start}), qos=1, retain=True)

state, total, reject = "RUNNING", 0, 0
client.publish(f"{BASE}/state", json.dumps({"state": state, "ts": time.time()}), qos=1, retain=True)

while True:
    # Random state changes
    if state == "RUNNING" and random.random() < 0.02:
        state = random.choice(["DOWN", "IDLE", "SETUP"])
        client.publish(f"{BASE}/state", json.dumps({"state": state, "ts": time.time()}), qos=1, retain=True)
    elif state != "RUNNING" and random.random() < 0.2:
        state = "RUNNING"
        client.publish(f"{BASE}/state", json.dumps({"state": state, "ts": time.time()}), qos=1, retain=True)

    if state == "RUNNING":
        made = int(10 / IDEAL_CYCLE_S * random.uniform(0.8, 0.98))
        total += made
        reject += sum(1 for _ in range(made) if random.random() < 0.02)

    client.publish(f"{BASE}/counts", json.dumps({"total": total, "reject": reject, "ts": time.time()}), qos=1)
    time.sleep(10)

Step 2: Build the Real-Time OEE Monitoring Service

The service subscribes to every machine under the line with wildcards, keeps the state history for the current shift and recalculates every 60 seconds. It handles several machines at once, because the machine name is taken from the topic.

# oee_service.py
import json
import threading
import time
from collections import defaultdict

import paho.mqtt.client as mqtt

LINE = "connectmq/demo/q8n3d1/plant1/line2"
INTERVAL_S = 60

machines = defaultdict(lambda: {
    "events": [],          # list of (ts, state)
    "total": 0, "reject": 0,
    "ideal_cycle_s": None, "shift_start": None,
})
lock = threading.Lock()


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)
    client.subscribe([(f"{LINE}/+/state", 1), (f"{LINE}/+/counts", 1), (f"{LINE}/+/config", 1)])


def on_message(client, userdata, msg):
    parts = msg.topic.split("/")
    machine, kind = parts[-2], parts[-1]
    data = json.loads(msg.payload)
    with lock:
        m = machines[machine]
        if kind == "config":
            m["ideal_cycle_s"] = data["ideal_cycle_s"]
            if m["shift_start"] != data["shift_start"]:
                m["shift_start"] = data["shift_start"]
                # Carry the current state into the new shift, drop older history
                m["events"] = [(data["shift_start"], m["events"][-1][1])] if m["events"] else []
        elif kind == "state":
            m["events"].append((data["ts"], data["state"]))
        elif kind == "counts":
            m["total"], m["reject"] = data["total"], data["reject"]


def calculate(m, now):
    start = m["shift_start"]
    events = sorted(e for e in m["events"] if e[0] <= now)
    run = planned_stop = 0.0
    for i, (ts, state) in enumerate(events):
        seg_start = max(ts, start)
        seg_end = events[i + 1][0] if i + 1 < len(events) else now
        if seg_end <= seg_start:
            continue
        if state == "RUNNING":
            run += seg_end - seg_start
        elif state == "PLANNED_STOP":
            planned_stop += seg_end - seg_start
    planned = (now - start) - planned_stop
    total, reject = m["total"], m["reject"]
    a = run / planned if planned > 0 else 0.0
    p = min(m["ideal_cycle_s"] * total / run, 1.0) if run > 0 else 0.0
    q = (total - reject) / total if total > 0 else 0.0
    return {
        "availability": round(a, 4), "performance": round(p, 4), "quality": round(q, 4),
        "oee": round(a * p * q, 4), "run_min": round(run / 60, 1),
        "planned_min": round(planned / 60, 1), "good": total - reject, "ts": now,
    }


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="oee-service-q8n3d1")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()

while True:
    time.sleep(INTERVAL_S)
    now = time.time()
    with lock:
        ready = {k: v for k, v in machines.items() if v["shift_start"] and v["ideal_cycle_s"]}
        results = {k: calculate(v, now) for k, v in ready.items()}
    for machine, result in results.items():
        client.publish(f"{LINE}/{machine}/oee", json.dumps(result), qos=1, retain=True)
        print(machine, result)

Run the simulator in one terminal and the service in another. After a minute you will see OEE values printed and published. Watch them from a third terminal:

mosquitto_sub -h broker.connectmq.com -p 1883 -i oee-viewer-q8n3d1 -t "connectmq/demo/q8n3d1/plant1/line2/+/oee" -v

How the Design Handles Real-World Problems

Restarts of the service. Because state and config are retained, a restarted service immediately receives the current state and shift start for every machine. It loses the history of earlier state changes, so availability is approximate until the next shift. In production, persist events to a small database or rebuild them from a historian on startup.

Lost messages. Counts are cumulative, so a lost count message only delays the update. State messages use QoS 1 so the broker acknowledges them.

Shift change. The machine gateway (or an MES) publishes a new config with the new shift_start. The service resets that machine's history automatically. Counters should reset at the same moment.

Clock skew. All timestamps come from the edge. Keep gateways on NTP; otherwise a gateway with a clock a minute fast will produce negative or inflated durations.

Performance over 100 percent. The cap hides a bad ideal cycle time. Log a warning when it triggers and fix the master data.

Displaying Live OEE

Any MQTT-capable display can subscribe to the oee topics. Options that work well:

1. Node-RED Dashboard 2.0 with a gauge for OEE and three bars for A, P and Q. The Node-RED OEE dashboard guide walks through it.
2. Grafana fed by Telegraf and InfluxDB for shift history alongside the live value, as in the production line monitoring dashboard tutorial.
3. A browser page using MQTT.js over wss://broker.connectmq.com:8084/mqtt, which needs no backend at all.

Show the three factors, not just OEE. A supervisor who sees availability at 95 percent and performance at 70 percent knows to look at speed and micro-stops, not breakdowns. Pair the live number with machine downtime tracking so long stops get a reason while they are fresh.

Adding Targets and Alerts

A live OEE number becomes more useful when it is compared with something. Two additions are worth making early.

Shift target pacing. Publish a target count for the shift to the config topic and let the service add expected_now and gap to each result: the count the line should have reached at this point in the shift and the difference from actual. Operators react much more to "we are 340 units behind" than to "OEE is 71 percent", and both numbers come from the same data.

Threshold alerts. Decide which conditions need a person to act, for example availability below 80 percent for the last hour or a machine in DOWN for more than ten minutes. There are two straightforward ways to implement them:

1. Let the OEE service publish an alert message when a rule trips, and have a Node-RED flow, andon light or messaging integration subscribe to it.
2. Use the threshold alert rules in the ConnectMQ portal on the published oee topics, which record each trigger in the alerts log without extra code.

Keep the rule list short. Alerts that fire several times a shift without anyone acting on them should be tuned or removed, otherwise people stop reading all of them.

Rolling windows. Besides the shift-to-date number, calculate OEE for the last 60 minutes. Shift-to-date values move slowly late in the shift and can hide a fresh problem; a rolling hour shows it immediately. The same calculate function works if you pass a window start one hour ago instead of the shift start.

Going to Production

Move off the public broker before connecting real machines. With ConnectMQ you create a private broker in the portal and receive a username, password and client ID for each client. Change the connection 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()  # system CA store validates the Let's Encrypt certificate
client.connect("broker.connectmq.com", 8883, keepalive=30)

A fully managed MQTT service removes broker maintenance from the plant team, and the portal's live topic explorer shows whether every machine is publishing before you debug the service. For the bigger picture, read the smart manufacturing guide.

Run Your OEE Service on ConnectMQ

Create a free broker at app.connectmq.com on the Serverless plan with no credit card, point your gateways and the OEE service at it, and have live OEE on the line screen today.

Frequently Asked Questions

How often should real-time OEE be recalculated?

Once a minute is a good default. It is responsive enough for the floor and avoids a number that jumps around with every count message.

Where should the OEE calculation run?

In one service that publishes the result to the broker, so every dashboard shows the same value. Avoid recalculating in each dashboard with slightly different logic.

Can this handle several machines and lines?

Yes. The service uses wildcard subscriptions and keys everything by machine name from the topic. Add a level to the wildcard to cover several lines.

What happens if a machine stops publishing?

Its last retained state stays on the broker. Add a Last Will message on the gateway that publishes OFFLINE so the service can treat silence as unknown rather than running.