From Factory Motor Runtime to AI Analysis with MQTT and Raspberry Pi Pico
From Factory Motor Runtime to AI Analysis with MQTT and Raspberry Pi Pico
Our last factory-floor project followed one temperature reading all the way from an industrial motor to a Linux workstation.
A Raspberry Pi Pico W read the sensor, added context, converted the observation to JSON and published it through MQTT. Linux received the messages, Python and Pandas cleaned the data, and we produced CSV for people and JSON for software.
This time we'll keep that pipeline and change two things.
Instead of temperature, we'll monitor whether our factory motor is running or stopped.
Then we'll send the cleaned observations to an AI model and ask it to evaluate the motor's operating pattern.
Our new pipeline looks like this:
factory motor
↓
motor state sensor
↓
Pico W
↓
motor ID + state + timestamp
↓
Python dictionary
↓
JSON + MQTT publish
↓
MQTT broker
↓
Linux subscriber
↓
Python + Pandas
↓
cleaned data
├── CSV
└── JSON
↓
AI API
↓
AI evaluation JSON
The basic system hasn't changed.
We've changed what we're measuring and added another consumer at the end.
What We're Measuring
Our question is simple:
Is the motor running or stopped?
For our demonstration, the Pico W checks the motor state once every 60 seconds.
That's an interval we've chosen for this example, not a universal industrial sampling rule.
A single observation might begin as:
motor_01,running
The Pico adds the timestamp:
2026-11-03T19:00:00,motor_01,running
In MicroPython, we can represent the observation as a dictionary:
record = {
"timestamp": timestamp,
"motor_id": "motor_01",
"state": "running"
}
Then we serialize it as JSON:
{
"timestamp": "2026-11-03T19:00:00",
"motor_id": "motor_01",
"state": "running"
}
Just as before, the dictionary itself doesn't travel across the network.
The Pico serializes the record and publishes the resulting payload through MQTT.
MQTT uses a publish-and-subscribe system in which devices publish messages to topics and subscribers receive messages from topics they follow. Pico W can participate in that architecture using MicroPython.
Code on the Pico W
The Pico still has a small job.
Every 60 seconds:
- Read the motor state.
-
Associate it with
motor_01. - Add a timestamp.
- Build a Python dictionary.
- Serialize the dictionary as JSON.
- Publish it through MQTT.
- Wait for the next check.
Conceptually, our MicroPython program looks like this:
import time
import ujson
from umqtt.simple import MQTTClient
BROKER = "192.168.1.100"
TOPIC = b"factory/motor_01/state"
client = MQTTClient(
"pico-motor-01",
BROKER
)
client.connect()
while True:
state = read_motor_state()
timestamp = get_timestamp()
record = {
"timestamp": timestamp,
"motor_id": "motor_01",
"state": state
}
payload = ujson.dumps(record)
client.publish(
TOPIC,
payload.encode()
)
time.sleep(60)
The hardware-specific
read_motor_state()
and timestamp code are intentionally conceptual. Their implementation depends
on how the particular motor state is sensed and how time is supplied to the
Pico.
Our focus remains on the data:
motor
↓
running / stopped
↓
MicroPython dictionary
↓
JSON
↓
MQTT payload
↓
broker
The Linux Workstation
Our Linux program subscribes to:
factory/motor_01/state
As MQTT messages arrive, the subscriber turns them back into Python records.
Conceptually:
import json
record = json.loads(
message.payload.decode()
)
records.append(record)
After several minutes, our collection might look like this:
19:00 motor_01 running
19:01 motor_01 running
19:02 motor_01 running
19:03 motor_01 running
19:04 motor_01 stopped
19:05 motor_01 stopped
19:06 motor_01 running
19:07 motor_01 running
Something useful has happened.
We no longer have an isolated
running
or
stopped
signal.
We have motor activity over time.
And because each observation represents our 60-second sampling interval, that sequence can also help us describe observed runtime during the period covered by the data.
Cleaning the Motor Activity Data
Factory data can still arrive dirty.
Suppose our collected observations include:
2026-11-03T19:00:00 motor_01 running
2026-11-03T19:01:00 MOTOR_01 RUNNING
bad_timestamp motor_01 running
2026-11-03T19:03:00 motor_01 stopped
2026-11-03T19:03:00 motor_01 stopped
2026-11-03T19:04:00 motor_01
2026-11-03T19:05:00 motor_01 running
There's an invalid timestamp.
The motor ID and state use inconsistent capitalization.
One observation is duplicated.
Another has no state at all.
That's dirty data.
Our Linux program can clean it with Pandas:
import pandas as pd
# INPUT
df = pd.DataFrame(records)
# PROCESSING
df["timestamp"] = pd.to_datetime(
df["timestamp"],
errors="coerce"
)
df["motor_id"] = (
df["motor_id"]
.str.strip()
.str.lower()
)
df["state"] = (
df["state"]
.str.strip()
.str.lower()
)
# Remove invalid or incomplete observations
df = df.dropna(
subset=["timestamp", "motor_id", "state"]
)
# Keep only states our pipeline understands
df = df[
df["state"].isin(["running", "stopped"])
]
# Remove duplicate observations
df = df.drop_duplicates()
# Put observations in time order
df = df.sort_values("timestamp")
# Reset row numbers
df = df.reset_index(drop=True)
# OUTPUT
df.to_csv(
"motor_activity_clean.csv",
index=False
)
df.to_json(
"motor_activity_clean.json",
orient="records",
date_format="iso",
indent=2
)
print(df)
print("\nCreated motor_activity_clean.csv")
print("Created motor_activity_clean.json")
After cleaning, the useful records might look like this:
2026-11-03T19:00:00 motor_01 running
2026-11-03T19:01:00 motor_01 running
2026-11-03T19:03:00 motor_01 stopped
2026-11-03T19:05:00 motor_01 running
The bad timestamp is gone.
The missing state is gone.
The duplicate is gone.
And our motor ID and state are consistent.
Once again we have two useful artifacts:
motor_activity_clean.csv
motor_activity_clean.json
But this time we're going to do something else with the JSON.
Giving the Motor Activity to an AI Model
We'll send
motor_activity_clean.json
to an AI model and ask for an evaluation.
For this experiment we'll again use OpenRouter's Free Models Router.
openrouter/free
selects from available free models and is intended for experimentation,
learning, and low-volume use cases.
You'll need an API key:
export OPENROUTER_API_KEY="your-key-here"
We'll also use Requests:
python3 -m pip install requests
Now create:
nano analyze_motor_activity.py
Here's the program:
import json
import os
import requests
API_KEY = os.environ["OPENROUTER_API_KEY"]
with open(
"motor_activity_clean.json",
"r"
) as file:
motor_data = json.load(file)
prompt = f"""
You are evaluating a small set of cleaned factory
motor activity records.
The motor state was sampled once every 60 seconds.
Briefly describe the observed running and stopped pattern.
Identify any periods of stopped activity that may deserve
further investigation.
Do not invent mechanical causes, failures, or explanations
that are not supported by the supplied data.
Motor activity records:
{json.dumps(motor_data, indent=2)}
"""
response = requests.post(
"https://openrouter.ai/api/v1/responses",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
},
json={
"model": "openrouter/free",
"input": prompt,
},
timeout=60,
)
response.raise_for_status()
result = response.json()
message = next(
item
for item in reversed(result["output"])
if item["type"] == "message"
)
analysis = next(
part["text"]
for part in message["content"]
if part["type"] == "output_text"
)
evaluation = {
"model": result["model"],
"analysis": analysis
}
with open(
"motor_activity_ai_evaluation.json",
"w"
) as file:
json.dump(
evaluation,
file,
indent=2
)
print(analysis)
print("\nModel used:", result["model"])
print(
"Created motor_activity_ai_evaluation.json"
)
Run it:
python3 analyze_motor_activity.py
OpenRouter's Responses API accepts requests at
/api/v1/responses, and the Free Models Router uses
openrouter/free
as the model identifier. The returned response also identifies the model that
was actually selected.
Our program takes that response and does something important with it.
It saves the evaluation.
Example AI Evaluation
A response might look something like this:
The motor is recorded as running during most of the observed period, with a short stopped interval appearing between otherwise running observations. The stopped activity may be worth reviewing alongside production or maintenance records. The supplied data shows the operating-state pattern but does not establish why the motor stopped.
The exact response will vary depending on the model selected and how the model interprets the supplied records.
The important point isn't the precise wording.
The AI evaluation has become another artifact:
motor_activity_ai_evaluation.json
It doesn't disappear when the terminal window closes.
Python Cleans. AI Interprets.
We haven't handed the entire job to AI.
That's deliberate.
The Pico reads the motor state.
MQTT transports the observation.
The Linux subscriber collects it.
Pandas fixes predictable data problems.
Only after we've produced clean, structured motor activity do we ask an AI model a less deterministic question:
What operating pattern does this data appear to show?
And even then, we deliberately constrain the model.
We're asking it to describe what the records support, not diagnose the motor.
The log might establish that the motor stopped.
It doesn't necessarily establish why.
That's the division of labor:
Sensors observe. Pico packages. MQTT transports. Pandas cleans. Code validates. AI interprets.
One Pipeline, Three Artifacts
We've now produced three durable outputs.
motor_activity_clean.csv
is aimed primarily at people.
A manager or analyst can open it in Excel, inspect running and stopped observations, calculate summaries, and build a chart or timeline showing motor activity. Those results can become part of normal reporting in Excel and move into PowerPoint presentations or Word reports.
motor_activity_clean.json
contains the same underlying cleaned observations for developers and software.
Another Python program, API, monitoring application, automation process, or larger data pipeline can consume those records directly.
And now we have:
motor_activity_ai_evaluation.json
That artifact carries the AI model's interpretation.
A manager can use the assessment alongside the underlying data while preparing an operational report.
Developers and programs can consume the evaluation JSON programmatically while still retaining the cleaned records that produced it.
The AI evaluation doesn't replace the data.
It travels alongside it.
Follow the Whole Pipeline
Our observation starts on the factory floor:
motor_01,running
The Pico adds context:
timestamp + motor ID + state
MicroPython turns that into a dictionary.
The dictionary becomes JSON.
MQTT carries the serialized message to the broker.
Linux receives it and turns that message back into structured Python data.
Pandas cleans the accumulated observations.
Then the pipeline branches:
factory motor
↓
state sensor
↓
Pico W reads every 60 seconds
↓
timestamp + motor ID + state
↓
Python dictionary
↓
JSON
↓
MQTT publish
↓
MQTT broker
↓
Linux subscriber
↓
Python records
↓
Pandas
↓
cleaned motor activity
│
├── CSV → Excel → charts and reporting
│
├── JSON → developers and software pipelines
│
└── JSON → AI API
↓
AI evaluation JSON
↓
people + programs
That's a remarkably complete pipeline for a small Raspberry Pi project.
We began with one simple fact on a factory floor:
The motor is running.
By the end of the pipeline, that observation has become part of a cleaned historical dataset, a spreadsheet-friendly artifact, a software-friendly artifact, and an AI-generated evaluation that we've preserved as structured data.
The individual programs remain small.
But the system is becoming considerably more interesting.
We're not simply collecting factory data anymore. We're turning factory data into information that both people and software can use.
.jpeg)
