HiveMQ Edge-to-Cloud AI Pipeline

Anomaly Detection using HiveMQ Cloud, Python (PyOD), and Flask

This article explains the full data pipeline shown in the diagram starting from a PLC on the shopfloor, sending data through HiveMQ Edge and Cloud, into an AI model built with Python + PyOD, and finally sending the anomaly result back to the HiveMQ Edge for visualization.

Kindly refer to last article to learn how to send factory data to HiveMQ Cloud. This article continues from there to further send data to ML for anomaly detection.

System Overview

This architecture demonstrates how vibration data from a machine is collected at the OT (Operational Technology) level, transported securely to the Cloud, evaluated using an AI model, and returned to the shop-floor for further action.

It consists of:

Input Layer – Reading vibration data from a PLC (via OPC UA)
Transport Layer – Sending data from HiveMQ Edge → HiveMQ Cloud (via MQTT)
AI Layer – ML server running Python + Flask + PyOD locally on the computer
Output Layer – Returning anomaly alerts via MQTT → HiveMQ Edge → PLC (optional)

Step-by-Step Breakdown of the Pipeline

1. Input Layer — PLC to HiveMQ Edge (OPC UA)

A Siemens S7-1500 PLC measures vibration (simulated) from the machine.
This data is exposed through OPC UA.
HiveMQ Edge connects to the PLC's OPC UA server.
It maps the OPC UA tag (e.g., vibration) to an MQTT topic:

machine/s71500/rVib

Industrial data becomes IIoT data.

2. Transport Layer — HiveMQ Edge to HiveMQ Cloud (MQTT)

HiveMQ Edge publishes the MQTT message (machine/s71500/vib) to HiveMQ Cloud, which acts as a central cloud broker.

Benefits:

Low latency
Very small message size → suitable for industrial IoT
Secure TLS transport and easy to scale

Learn more about that in our earlier posts on HiveMQ where you will learn the complete workflow: PLC to Cloud via HiveMQ (Edge + Cloud) — OPC UA → MQTT → Confluent

3. AI Layer — ML server running Python + Flask + PyOD locally on the compute

This is where the anomaly detection happens. Let's first understand what is Anomaly detection and why we need it

What is Anomaly detection?

Anomaly detection is the process of identifying data points or behavior that deviate from what’s considered normal. In simple terms, it spots when something unusual is happening.

In Machine Learning, anomaly detection helps models understand patterns and flag unexpected behaviour without needing labels. In IIoT, it's essential because machines generate huge amounts of real-time data, and even small deviations can signal issues like equipment failure, quality defects, cyber-attacks, or unsafe operating conditions.

Why we need it:

Detect problems early before they become costly
Improve uptime and reliability
Enhance safety and Reduce maintenance costs

🧠 PyOD — Python Outlier Detection Library

PyOD is a widely-used ML library that provides 40+ algorithms for anomaly detection, such as:

Isolation Forest (IForest)
AutoEncoders
One-Class SVM
LOF (Local Outlier Factor)

In the example, Isolation Forest (IForest) is used. It learns what “normal vibration” looks like from training data:

train = np.array([[0.23], [0.25], [0.21], [0.29]])
model = IForest()
model.fit(train)

When new data comes, PyOD predicts:

0 → Normal
1 → Anomaly

Learn more about Pyod here: https://pyod.readthedocs.io/en/latest/

Installing Pyod and testing Anomaly detection

In our example, the Pyod has been installed in the windows computer. The following are the steps:

Install pyod on Windows PC

Make sure Python is installed in your computer. In our case, we are using Python 3. To install Python visit here: https://www.python.org/downloads/

pip3 install pyod

Create a python file for testing:

test_pyod.py

Add the following code in that

from pyod.models.iforest import IForest
import numpy as np

# training data: only normal vibration history
train = np.array([[0.23], [0.25], [0.21], [0.29]])

model = IForest()
model.fit(train)

# real-time vibration input
value = 0.45 #change this value to see different result.
prediction = model.predict([[value]])[0]

if prediction == 1:
    print("Anomaly detected!")
else:
    print("Normal")

Run the Python file

python test_pyod.py

Validate the output

‘Anomaly detected’ or ‘Normal’ based on the value sent in the code (0.45)

Anomaly detection using Pyod and Node-RED

Now, we are going to test the code using Node-RED. In this case, we setup a flask server in Node-RED so that we can execute the Python code using API. This makes it easy to send data to the Python file and get anomaly results as feedback.

Make sure latest version of Node-RED and Python is installed in your system.

To install Node-RED visit here: https://nodered.org/
To install Python visit here: https://www.python.org/downloads/

Once, the Node-RED is installed proceed with the following steps:

Convert Your Python Code into a Simple API Server.

We will not run Python file directly from Node-RED, we will create a Python web API for data exchange.

Create a new Python file named 'server.py' with the following code:

from flask import Flask, request, jsonify
from pyod.models.iforest import IForest
import numpy as np

app = Flask(__name__)

# Train model once at startup
train = np.array([[0.23], [0.25], [0.21], [0.29]])
model = IForest()
model.fit(train)

@app.route("/predict", methods=["POST"])
def predict():
    data = request.json
    value = float(data["value"])  # vibration input

    prediction = model.predict([[value]])[0]

    return jsonify({
        "value": value,
        "prediction": int(prediction),
        "status": "Anomaly" if prediction == 1 else "Normal"
    })

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Install Flask

pip install flask

Run your Python Server

python server.py

Validate

You should see the following:

Your anomaly detection API is now LIVE at: http://localhost:5000/predict

Start Node-RED and test the Anomaly detection

You can use the 'http request node' in Node-RED to send sample data and fetch anomaly results as shown below.

Kindly check the reference Node-RED flow below:

[
    {
        "id": "53bf0f485d74978b",
        "type": "tab",
        "label": "Flow 3",
        "disabled": false,
        "info": "",
        "env": []
    },
    {
        "id": "05ded4d2341631a0",
        "type": "http request",
        "z": "53bf0f485d74978b",
        "name": "",
        "method": "POST",
        "ret": "obj",
        "paytoqs": "ignore",
        "url": "http://localhost:5000/predict",
        "tls": "",
        "persist": false,
        "proxy": "",
        "insecureHTTPParser": false,
        "authType": "",
        "senderr": false,
        "headers": [],
        "x": 530,
        "y": 80,
        "wires": [
            [
                "7b1ef153885b1894"
            ]
        ]
    },
    {
        "id": "4abe4b4dca5e520e",
        "type": "function",
        "z": "53bf0f485d74978b",
        "name": "function 3",
        "func": "msg.headers = { \"Content-Type\": \"application/json\" };\nmsg.payload = { value: msg.payload };\nreturn msg;\n",
        "outputs": 1,
        "timeout": 0,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "libs": [],
        "x": 340,
        "y": 80,
        "wires": [
            [
                "05ded4d2341631a0"
            ]
        ]
    },
    {
        "id": "7b1ef153885b1894",
        "type": "debug",
        "z": "53bf0f485d74978b",
        "name": "debug 1",
        "active": true,
        "tosidebar": true,
        "console": false,
        "tostatus": false,
        "complete": "false",
        "statusVal": "",
        "statusType": "auto",
        "x": 700,
        "y": 80,
        "wires": []
    },
    {
        "id": "301488acacb31208",
        "type": "inject",
        "z": "53bf0f485d74978b",
        "name": "",
        "props": [
            {
                "p": "payload"
            },
            {
                "p": "topic",
                "vt": "str"
            }
        ],
        "repeat": "",
        "crontab": "",
        "once": false,
        "onceDelay": 0.1,
        "topic": "",
        "payload": "0.45",
        "payloadType": "num",
        "x": 150,
        "y": 80,
        "wires": [
            [
                "4abe4b4dca5e520e"
            ]
        ]
    },
    {
        "id": "0f591249346fc88a",
        "type": "inject",
        "z": "53bf0f485d74978b",
        "name": "",
        "props": [
            {
                "p": "payload"
            },
            {
                "p": "topic",
                "vt": "str"
            }
        ],
        "repeat": "",
        "crontab": "",
        "once": false,
        "onceDelay": 0.1,
        "topic": "",
        "payload": "0.25",
        "payloadType": "num",
        "x": 150,
        "y": 120,
        "wires": [
            [
                "4abe4b4dca5e520e"
            ]
        ]
    }
]

Testing Anomaly detection with PLC Data

Now, let's use the above example and sends the PLC data to the Python server to get anomaly results.

For the sake of demonstration, we are considering the normal values within the range $0.0∼5.0$ and anomalies outside this range. So, in our Python code, we have created a dataset of $500$ samples in the range of $0.0$ to $5.0$ . Our model will learn from this dataset, as shown in the code below.

Update your Python code in the server.py file

The code trains an anomaly detection model, exposes it through a REST API, and lets you retrain the model using real machine data.

from flask import Flask, request, jsonify
from pyod.models.iforest import IForest
from sklearn.preprocessing import StandardScaler
import numpy as np

app = Flask(__name__)

# =========================================================
# 1. TRAIN MODEL ON REALISTIC NORMAL VIBRATION DISTRIBUTION
# =========================================================

# Simulate "normal" vibration: around 2.0 + noise(-3 to +3)
# That gives ~ 0 to 5 mm/s range
normal_values = np.random.uniform(0.0, 5.0, 500)   # 500 samples
normal_values = normal_values.reshape(-1, 1)

# Scale the training data (important for broader vibration ranges)
scaler = StandardScaler()
normal_scaled = scaler.fit_transform(normal_values)

# Train Isolation Forest
model = IForest(contamination=0.05)  # 5% anomaly expected
model.fit(normal_scaled)


# =========================================================
# 2. API ENDPOINT: PREDICT ANOMALY
# =========================================================
@app.route("/predict", methods=["POST"])
def predict():
    data = request.json
    value = float(data["value"])

    # Scale incoming data
    scaled_value = scaler.transform([[value]])

    # Raw prediction: 1 = anomaly, 0 = normal
    prediction = int(model.predict(scaled_value)[0])

    # Anomaly score (higher → more anomaly)
    score = float(model.decision_function(scaled_value)[0])

    return jsonify({
        "input_value": value,
        "scaled_value": scaled_value[0][0],
        "anomaly_score": score,
        "prediction": prediction,
        "status": "Anomaly Detected" if prediction == 1 else "Normal Behaviour"
    })


# =========================================================
# 3. OPTIONAL — RETRAIN USING DATA FROM CLIENT (Node-RED)
# =========================================================
@app.route("/retrain", methods=["POST"])
def retrain():
    data = request.json
    new_samples = np.array(data["samples"]).reshape(-1, 1)

    global scaler, model

    scaler = StandardScaler()
    scaled = scaler.fit_transform(new_samples)

    model = IForest(contamination=0.05)
    model.fit(scaled)

    return jsonify({"message": "Model retrained successfully", "samples_used": len(new_samples)})


# =========================================================
# 4. RUN SERVER
# =========================================================
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

This script creates a small AI-powered anomaly detection API using Flask and the PyOD Isolation Forest model.

It generates normal vibration data (0–5 mm/s) and uses it to train an Isolation Forest model.
It scales all vibration values using StandardScaler so the model can understand them correctly.
It exposes a /predict API endpoint where you send one vibration value, and the server returns:
- Normal or Anomaly
- Anomaly score
- Scaled value
It provides a /retrain endpoint that allows you to send new vibration samples (from Node-RED or a PLC) and retrain the model on the fly.
Flask runs the server on port 5000, making it easy to integrate with IIoT systems.

Learn more about this code here: Anomaly detection Code Explanation

Update the Node-RED flow

Now we need to subscribe the following HiveMQ Cloud topic to get the LIVE vibration data:

hivemq-edge/machine/s71500/rVib

Every time vibration data arrives from HiveMQ Cloud → the ML server receives it, runs inference, and give the result back to Node-RED

4. Output Layer – Returning anomaly alerts via MQTT → HiveMQ Edge → PLC (optional)

Now we need to send the result back to HiveMQ Edge, visualize it on the dashboard or sends to the PLC for further action.

Add MQTT out node in the flow

We will take another MQTT topic to filter and publish the result

hivemq-edge/machine/s71500/anomaly

The following flow can be used as a reference.

[
    {
        "id": "1ad2c5724dad60ec",
        "type": "tab",
        "label": "Flow 1",
        "disabled": false,
        "info": "",
        "env": []
    },
    {
        "id": "2e10e2177a83dbbf",
        "type": "http request",
        "z": "1ad2c5724dad60ec",
        "name": "",
        "method": "POST",
        "ret": "obj",
        "paytoqs": "ignore",
        "url": "http://localhost:5000/predict",
        "tls": "",
        "persist": false,
        "proxy": "",
        "insecureHTTPParser": false,
        "authType": "",
        "senderr": false,
        "headers": [],
        "x": 570,
        "y": 80,
        "wires": [
            [
                "c3746a81c58e797a"
            ]
        ]
    },
    {
        "id": "1d57a237256e69e7",
        "type": "function",
        "z": "1ad2c5724dad60ec",
        "name": "function 1",
        "func": "msg.headers = { \"Content-Type\": \"application/json\" };\nmsg.payload = { value: msg.payload.value };\nreturn msg;\n",
        "outputs": 1,
        "timeout": 0,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "libs": [],
        "x": 400,
        "y": 80,
        "wires": [
            [
                "2e10e2177a83dbbf"
            ]
        ]
    },
    {
        "id": "ecd705d7eabc8086",
        "type": "mqtt in",
        "z": "1ad2c5724dad60ec",
        "name": "",
        "topic": "hivemq-edge/machine/s71500/rVib",
        "qos": "2",
        "datatype": "auto-detect",
        "broker": "08f8856327f4fe37",
        "nl": false,
        "rap": true,
        "rh": 0,
        "inputs": 0,
        "x": 160,
        "y": 80,
        "wires": [
            [
                "1d57a237256e69e7"
            ]
        ]
    },
    {
        "id": "67d96c79df053896",
        "type": "mqtt out",
        "z": "1ad2c5724dad60ec",
        "name": "",
        "topic": "hivemq-edge/machine/s71500/anomaly",
        "qos": "",
        "retain": "",
        "respTopic": "",
        "contentType": "",
        "userProps": "",
        "correl": "",
        "expiry": "",
        "broker": "08f8856327f4fe37",
        "x": 600,
        "y": 160,
        "wires": []
    },
    {
        "id": "c3746a81c58e797a",
        "type": "function",
        "z": "1ad2c5724dad60ec",
        "name": "function 2",
        "func": "msg.payload = msg.payload.status;\nreturn msg;",
        "outputs": 1,
        "timeout": 0,
        "noerr": 0,
        "initialize": "",
        "finalize": "",
        "libs": [],
        "x": 300,
        "y": 160,
        "wires": [
            [
                "67d96c79df053896"
            ]
        ]
    },
    {
        "id": "08f8856327f4fe37",
        "type": "mqtt-broker",
        "name": "",
        "broker": "1eba33276f3f4d2b9852f78bf6ab2880.s1.eu.hivemq.cloud",
        "port": "8883",
        "tls": "",
        "clientid": "",
        "autoConnect": true,
        "usetls": true,
        "protocolVersion": 4,
        "keepalive": 60,
        "cleansession": true,
        "autoUnsubscribe": true,
        "birthTopic": "",
        "birthQos": "0",
        "birthRetain": "false",
        "birthPayload": "",
        "birthMsg": {},
        "closeTopic": "",
        "closeQos": "0",
        "closeRetain": "false",
        "closePayload": "",
        "closeMsg": {},
        "willTopic": "",
        "willQos": "0",
        "willRetain": "false",
        "willPayload": "",
        "willMsg": {},
        "userProps": "",
        "sessionExpiry": ""
    }
]

Bridge the HiveMQ Cloud MQTT topic with HiveMQ Edge MQTT topic

Go to MQTT bridge in HiveMQ Edge and add a remote subscription in your current bridge as shown below:

Now navigate to Broker Configuration and 'Enable loop prevention'. This will prevent looping of data flow

Visualize Anomaly in HiveMQ Edge

Now that the anomaly arrives to MQTT Edge, you can visualize it with tools like Node-RED as shown below:

You can further sends the anomaly results back to PLC via OPC UA.

Complete Workflow

♥️ Work With Me

I regularly test industrial automation and IIoT devices. If you’d like me to review your product or showcase it in my courses and YouTube channel:

📧 Email: [email protected] or drop me a message on LinkedIn

PreviousMulti-Site PLC-to-Cloud Flow using HiveMQ Edge + Cloud NextAnomaly detection Code Explanation

Last updated 1 month ago

Was this helpful?