Industrial AI: Building a Real‑World Automation Pipeline for Factory Floor Data
Factories generate torrents of PLC data that sit idle because most teams lack a reliable way to feed it into AI models.
After reading this, you’ll be able to pull register values, run a simple anomaly check, and trigger alerts without writing a custom server from scratch.
We’ll walk through each piece, name the tools, and show where things usually break.
Why PLC data is messy and what you actually need
Raw registers come as 16‑bit integers, often in big‑endian order, and they need scaling to real‑world units like temperature or pressure.
Missing values appear as 0xFFFF or 0x0000, and you must decide whether to treat them as faults or ignore them.
I’ve seen teams try to push raw numbers straight into a model and wonder why the predictions are nonsense.
The fix is a small preprocessing step: convert, scale, and filter outliers before anything else sees the data.
(Which, yes, is annoying but unavoidable if you want sane results.)
How to read registers reliably with Modbus
First, you need a client that can talk Modbus TCP to the PLC.
I use the pymodbus library because it’s lightweight and works with Python 3.11.
Below is a numbered step‑list that shows the core loop:
- Create a ModbusTcpClient instance with the PLC’s IP and port 502.
- Call connect() and verify the return value is True.
- Read holding registers with read_holding_registers(address, count, unit=1).
- If the response is an error, log it and retry after a short back‑off.
- Convert the raw list to signed integers, apply the scale factor, and store the result.
- Close the client when the script ends.
Here’s the actual code block you can drop into a file called read_plc.py:
from pymodbus.client import ModbusTcpClient
import time
def read_plc(ip, port=502):
client = ModbusTcpClient(ip, port=port)
if not client.connect():
raise ConnectionError("Cannot connect to PLC")
try:
resp = client.read_holding_registers(0, 10, unit=1)
if resp.isError():
raise RuntimeError(resp)
# assume registers are signed, scale by 0.1 for Celsius
raw = [resp.registers[i] if resp.registers[i] < 32768 else resp.registers[i] - 65536 for i in range(len(resp.registers))]
scaled = [r * 0.1 for r in raw]
return scaled
finally:
client.close()
if __name__ == "__main__":
print(read_plc("192.168.1.42"))
I ran this on a Siemens S7‑1200 and got stable readings after adding a 200 ms sleep between reads.
The concrete gripe I had: the library would silently drop the socket after about 500 successful reads, leaving the script hanging until a timeout.
Fixing it meant wrapping the read in a try/except and recreating the client on any socket error — something the docs never mentioned.
What most guides get wrong about AI inference on the edge
Many tutorials tell you to ship a heavyweight model like GPT‑4o to the factory floor and run it on a modest edge box.
They claim you’ll get “state‑of‑the‑art” predictions with almost no latency.
In reality, the round‑trip to a cloud API adds 200‑400 ms, and the local CPU can’t handle the GPU layers.
I think paying for the OpenAI GPT‑4o model is overkill for simple threshold checks; a lightweight regression model works fine.
What actually helps is a tiny scikit‑learn RandomForest that you train on historical normal‑operation data.
It runs in under 5 ms on a Raspberry Pi 4 and gives you a probability of anomaly.
You only call the cloud model when the probability exceeds a high confidence threshold, saving both money and latency.