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
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:
In our example, the Pyod has been installed in the windows computer. The following are the steps:
1
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/
2
Create a python file for testing:
test_pyod.py
3
Add the following code in that
4
Run the Python file
5
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.
Once, the Node-RED is installed proceed with the following steps:
1
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:
2
Install Flask
3
Run your Python Server
4
Validate
You should see the following:
Your anomaly detection API is now LIVE at: http://localhost:5000/predict
5
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:
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.
1
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.
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.
train = np.array([[0.23], [0.25], [0.21], [0.29]])
model = IForest()
model.fit(train)
pip3 install pyod
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")
python test_pyod.py
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)
