A machine on a factory floor starts running hot at 2:14 PM. If someone finds out at 2:15, it's a quick fix. If they find out in tomorrow's Excel report, you may be paying for a new spindle.
That gap between "something happened" and "someone knows about it" is what most Industry 4.0 work comes down to. At Prism Infoways we build these systems end to end, from sensor hardware and firmware to cloud dashboards. The core pattern we use again and again is small enough to fit in a blog post.
In this tutorial we'll build it:
- 🏭 Simulated machines publish temperature and vibration readings over MQTT
- ⚡ A FastAPI service subscribes, validates every message with Pydantic and checks alert thresholds
- 📡 Readings are pushed to the browser over WebSockets, with no polling
- 📊 A zero-dependency HTML dashboard shows each machine live and turns red when something's wrong
Then we'll cover what changes when you take this to a real plant with hundreds of devices.
The architecture
┌─────────────────────────┐
│ Sensors / PLC gateways │
└────────────┬────────────┘
│
│ MQTT publish
▼
┌─────────────────────────┐
│ Mosquitto (broker) │
└────────────┬────────────┘
│
│ MQTT subscribe
▼
┌─────────────────────────┐
│ FastAPI service │
│ validate + alert │
└────────────┬────────────┘
│
│ WebSocket push
▼
┌─────────────────────────┐
│ Dashboard (browser) │
└─────────────────────────┘
Why MQTT and not plain HTTP from the devices? Field devices sit on flaky networks, have little memory and send small payloads very often. MQTT was designed for exactly that. It keeps one long-lived connection, has a tiny packet overhead, offers delivery guarantees (QoS levels) and supports a "last will" message that tells you when a device drops off. HTTP makes every reading pay for a full request/response cycle.
Why WebSockets to the browser? Polling a REST endpoint every second from 50 open dashboards is 50 requests per second doing nothing useful most of the time. With a WebSocket, the server pushes a reading only when there is one.
Step 0: Setup
You'll need Python 3.10+ and Docker (for the broker).
mkdir iot-pipeline && cd iot-pipeline
python -m venv .venv && source .venv/bin/activate
pip install fastapi "uvicorn[standard]" aiomqtt
aiomqtt is an asyncio wrapper around the well-known paho-mqtt client, so it fits naturally into FastAPI's event loop.
Windows users: aiomqtt needs the selector event loop on Windows. See the aiomqtt README for the one-line fix.
Step 1: Run an MQTT broker
We'll use Eclipse Mosquitto, the most widely used open-source broker.
mosquitto.conf:
listener 1883
allow_anonymous true # local development ONLY — see the production section
docker-compose.yml:
services:
mosquitto:
image: eclipse-mosquitto:2
ports:
- "1883:1883"
volumes:
- ./mosquitto.conf:/mosquitto/config/mosquitto.conf
docker compose up -d
Mosquitto 2.x refuses outside connections unless you declare a listener. That's why the config file is needed even for a local setup.
Step 2: Simulate the machines
In a real plant this would be a PLC gateway, an ESP32 or an industrial edge box. For now, a script will stand in for three machines.
sensor.py:
import asyncio
import json
import random
import time
import aiomqtt
MACHINES = ["press-01", "press-02", "cnc-07"]
async def main():
async with aiomqtt.Client("localhost") as client:
while True:
for machine in MACHINES:
payload = {
"machine_id": machine,
"temperature": round(random.gauss(65, 8), 1), # °C
"vibration": round(abs(random.gauss(2.0, 0.8)), 2), # mm/s RMS
"ts": time.time(),
}
await client.publish(
f"factory/{machine}/telemetry",
json.dumps(payload),
qos=1,
)
await asyncio.sleep(1)
asyncio.run(main())
Look at the topic structure: factory/{machine}/telemetry. Topic design is the MQTT equivalent of URL design, and it's very hard to change later. A hierarchy lets consumers subscribe with wildcards: factory/+/telemetry matches every machine, and factory/# matches everything under factory.
Step 3: The FastAPI ingestion service
This is the core of the system. One background task consumes MQTT, and a small hub fans each reading out to every connected browser.
main.py:
import asyncio
import contextlib
from contextlib import asynccontextmanager
import aiomqtt
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse
from pydantic import BaseModel, ValidationError
BROKER_HOST = "localhost"
TOPIC = "factory/+/telemetry"
TEMP_LIMIT = 80.0 # °C
VIBRATION_LIMIT = 4.0 # mm/s
class Reading(BaseModel):
machine_id: str
temperature: float
vibration: float
ts: float
def check_alerts(r: Reading) -> list[str]:
alerts = []
if r.temperature > TEMP_LIMIT:
alerts.append(f"High temperature ({r.temperature}°C)")
if r.vibration > VIBRATION_LIMIT:
alerts.append(f"High vibration ({r.vibration} mm/s)")
return alerts
class Hub:
"""Tracks connected dashboards and the latest reading per machine."""
def __init__(self):
self.clients: set[WebSocket] = set()
self.latest: dict[str, dict] = {}
async def connect(self, ws: WebSocket):
await ws.accept()
self.clients.add(ws)
# New dashboards get the current state immediately instead of a blank screen
await ws.send_json({"type": "snapshot", "data": list(self.latest.values())})
def disconnect(self, ws: WebSocket):
self.clients.discard(ws)
async def broadcast(self, message: dict):
# Iterate over a copy: clients may connect/disconnect while we await
for ws in list(self.clients):
try:
await ws.send_json(message)
except Exception:
self.disconnect(ws)
hub = Hub()
async def consume_mqtt():
while True:
try:
async with aiomqtt.Client(BROKER_HOST) as client:
await client.subscribe(TOPIC, qos=1)
print(f"Subscribed to {TOPIC}")
async for message in client.messages:
try:
reading = Reading.model_validate_json(message.payload)
except ValidationError as e:
# Never let one malformed device take down the pipeline
print(f"Dropped bad payload on {message.topic}: {e}")
continue
data = reading.model_dump()
data["alerts"] = check_alerts(reading)
hub.latest[reading.machine_id] = data
await hub.broadcast({"type": "reading", "data": data})
except aiomqtt.MqttError:
print("Lost connection to broker — retrying in 3s")
await asyncio.sleep(3)
@asynccontextmanager
async def lifespan(app: FastAPI):
task = asyncio.create_task(consume_mqtt())
yield
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def dashboard():
return FileResponse("dashboard.html")
@app.get("/machines")
async def machines():
"""REST snapshot, handy for scripts and health checks."""
return list(hub.latest.values())
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
await hub.connect(ws)
try:
while True:
await ws.receive_text() # we only push; this just detects disconnects
except WebSocketDisconnect:
hub.disconnect(ws)
A few decisions here are worth pointing out:
-
The
lifespanhandler starts the MQTT consumer when the app boots and cancels it cleanly on shutdown. This is the modern replacement for the deprecated@app.on_event("startup"). -
The reconnect loop. Brokers restart and networks blip. Without the outer
while Trueandexcept aiomqtt.MqttError, one hiccup would silently kill your ingestion until someone restarts the service. -
Validate at the edge of your system.
Reading.model_validate_jsonrejects a payload where a firmware bug sends"temperature": "ERR". You log it and move on, and the bad data never reaches your database or your users. - Send a snapshot on connect. When someone opens the dashboard, they see the current state right away instead of waiting for the next reading.
Step 4: The live dashboard
No React, no build step: one HTML file.
dashboard.html:
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Factory Floor — Live</title>
<style>
body { font-family: system-ui, sans-serif; margin: 2rem; background: #0f172a; color: #e2e8f0; }
table { border-collapse: collapse; width: 100%; max-width: 820px; }
th, td { padding: .6rem 1rem; border-bottom: 1px solid #334155; text-align: left; }
tr.alert { background: #7f1d1d; }
#status { color: #94a3b8; }
</style>
</head>
<body>
<h1>Factory Floor — Live</h1>
<p id="status">Connecting…</p>
<table>
<thead>
<tr><th>Machine</th><th>Temp (°C)</th><th>Vibration (mm/s)</th><th>Alerts</th><th>Updated</th></tr>
</thead>
<tbody id="rows"></tbody>
</table>
<script>
const rows = document.getElementById("rows");
const statusEl = document.getElementById("status");
function render(r) {
let tr = document.getElementById(r.machine_id);
if (!tr) {
tr = document.createElement("tr");
tr.id = r.machine_id;
rows.appendChild(tr);
}
tr.className = r.alerts.length ? "alert" : "";
const cells = [
r.machine_id,
r.temperature,
r.vibration,
r.alerts.join(", ") || "—",
new Date(r.ts * 1000).toLocaleTimeString(),
].map((value) => {
const td = document.createElement("td");
td.textContent = value; // textContent, not innerHTML: device data is untrusted input
return td;
});
tr.replaceChildren(...cells);
}
function connect() {
const proto = location.protocol === "https:" ? "wss" : "ws";
const ws = new WebSocket(`${proto}://${location.host}/ws`);
ws.onopen = () => (statusEl.textContent = "● Live");
ws.onmessage = (event) => {
const msg = JSON.parse(event.data);
if (msg.type === "snapshot") msg.data.forEach(render);
else render(msg.data);
};
ws.onclose = () => {
statusEl.textContent = "Disconnected — retrying…";
setTimeout(connect, 2000);
};
}
connect();
</script>
</body>
</html>
The textContent line is easy to skip past, and it matters. Anything that comes from a device, including the machine_id, should be treated as untrusted input. A compromised or misconfigured sensor that publishes <img src=x onerror=...> as its ID shouldn't be able to run JavaScript in your operators' browsers.
Step 5: Run it
Three terminals:
# 1. broker (if not already running)
docker compose up -d
# 2. the API + dashboard
uvicorn main:app --reload
# 3. the "machines"
python sensor.py
Open http://localhost:8000. Three rows appear and update every second, and whenever a simulated machine crosses a threshold, its row turns red. Stop sensor.py and watch the timestamps stop moving. Restart Mosquitto and the service reconnects without any help.
That's a working sensor-to-screen pipeline in about 150 lines.
Taking it to production: what we learned the hard way
The demo above is the right shape. These are the things that change once you're dealing with a real plant, real networks and real people relying on the screen.
1. Lock down the broker
allow_anonymous true on a public IP is the IoT version of an open S3 bucket. In production:
- Use TLS (port 8883) for every connection
- Give each device its own credentials (or client certificates)
- Use ACLs so
press-01can only publish tofactory/press-01/#and can't impersonate other machines
2. Pick QoS per message type, not globally
- QoS 0 (at most once): high-frequency telemetry, where missing one reading out of 3,600 an hour doesn't matter
- QoS 1 (at least once): alarms, state changes and commands. Your consumer must handle duplicates, so make processing idempotent.
- QoS 2 (exactly once): rarely worth the overhead
3. Use Last Will and retained messages
Have each device register a Last Will message like {"status": "offline"} on factory/press-01/status with retain=True. If the device disappears without a clean disconnect, the broker publishes that message for it. You get offline detection for free, and any new subscriber immediately learns the last known status.
4. Persist to a time-series database, in batches
Readings need to go somewhere for trends, reports and predictive maintenance. TimescaleDB (if your team already knows Postgres) or InfluxDB are good fits. Don't INSERT one row per message: buffer readings and write a batch every second or every few hundred rows.
5. Scaling past one process
This is where people get caught out. If you run uvicorn --workers 4, every worker subscribes to the same topic and processes every message four times. Two fixes:
-
MQTT 5 shared subscriptions: subscribe to
$share/ingest/factory/+/telemetryand the broker load-balances messages across the workers in theingestgroup - Redis Pub/Sub (or similar) for the WebSocket fan-out: a browser connected to worker 2 still needs readings that worker 3 processed
6. Don't let one slow client block everyone
Our broadcast sends to clients one at a time. A dashboard on a bad 3G connection on the shop floor slows down everyone else. In production, give each client its own bounded asyncio.Queue and sender task, and drop the oldest messages (or the client) when the queue fills up.
7. Don't trust device clocks
Cheap hardware drifts, resets to 1970 after a power cut, or ignores time zones. Store the server receive time next to the device timestamp, and use it whenever the two disagree by more than a few seconds.
Where to go from here
- Smarter alerts: swap fixed thresholds for rolling averages or rate-of-change checks. A temperature climbing 2°C a minute is often a better warning than one that has already crossed 80°C.
- Predictive maintenance: once a few months of vibration data are stored, even simple ML models (isolation forests, for example) can flag bearing wear weeks before it fails.
- Device management: an admin UI for registering machines, thresholds and credentials. If you're on FastAPI, our team's free admin panel, Tungsten, is built for exactly this.
-
Commands back to devices: publish to
factory/{machine}/commandsto change sample rates or trigger a safe shutdown. Lock that topic down even more tightly than telemetry.
Wrapping up
The pattern is simple: devices publish to a broker, a service validates and enriches, and browsers receive pushes. Most of the real engineering is in the details: reconnecting, validating, securing the broker and scaling without processing every message twice.
At Prism Infoways, Industry 4.0 and IoT are a core part of what we do, from hardware prototyping and embedded firmware to cloud platforms, dashboards and AI-driven analytics. If you're planning a smart-factory, remote-monitoring or connected-product project, we'd be glad to talk it through. Book a free consultation →
Have questions, or a war story about MQTT in production? Leave it in the comments — we read every one. 👇
Top comments (0)