Theory & Background

MQTT (Message Queuing Telemetry Transport): lightweight pub/sub protocol ideal for IoT (small packets, low bandwidth, unreliable networks). QoS levels: 0 (at most once), 1 (at least once), 2 (exactly once). Topic hierarchy: catb/site/{location}/device/{device_id}/sensor/{sensor_type}. Example: catb/site/lab/device/pi001/sensor/temperature. Wildcards: + (single level), # (multi-level). Subscribe to catb/site/lab/# to receive all lab device data. Retained messages: last message kept on broker for new subscribers.

Components & Requirements

10 components required for this project.

#	Component	Purpose	Qty
1	EMQX MQTT Broker (or Mosquitto)	IoT device message broker	x1
2	TimescaleDB (PostgreSQL extension)	Time-series IoT data storage	x1
3	Grafana	Real-time sensor dashboards	x1
4	Node-RED	Visual IoT flow programming	x1
5	Raspberry Pi sensors (temp/humidity)	IoT sensor nodes	x3
6	Python MQTT client (paho-mqtt)	Sensor data publishing	x1
7	Telegraf	MQTT → TimescaleDB pipeline	x1
8	Kafka (optional, high scale)	Message streaming for massive scale	x1
9	AWS IoT Core (optional)	Cloud-managed MQTT broker	x1
10	TensorFlow Lite (edge AI)	On-device inference	x1

Step-by-Step Implementation

Follow these 5 steps carefully.

1

MQTT Protocol and Topic Design

MQTT (Message Queuing Telemetry Transport): lightweight pub/sub protocol ideal for IoT (small packets, low bandwidth, unreliable networks). QoS levels: 0 (at most once), 1 (at least once), 2 (exactly once). Topic hierarchy: catb/site/{location}/device/{device_id}/sensor/{sensor_type}. Example: catb/site/lab/device/pi001/sensor/temperature. Wildcards: + (single level), # (multi-level). Subscribe to catb/site/lab/# to receive all lab device data. Retained messages: last message kept on broker for new subscribers.

2

EMQX Broker Setup and Security

EMQX: high-performance MQTT broker (1M connections per node). Configure authentication: API key for each device (X.509 client certificates for production). Authorization: each device can only publish to its own topic prefix (prevent device impersonation). Configure TLS on port 8883. EMQX Dashboard: monitor connections, message rates, subscription trees. Rule engine: filter messages, transform, forward to Kafka/HTTP/database without a separate consumer.

3

TimescaleDB Time-Series Storage

TimescaleDB extends PostgreSQL with automatic partitioning of time-series data (hypertables). Schema: CREATE TABLE sensor_data (time TIMESTAMPTZ NOT NULL, device_id TEXT, sensor_type TEXT, value DOUBLE PRECISION, tags JSONB). SELECT create_hypertable('sensor_data', 'time', chunk_time_interval => INTERVAL '1 day'). Auto-compression: compress chunks older than 7 days (10–20× compression). Continuous aggregates: pre-compute hourly/daily averages for fast historical queries.

4

Real-Time Grafana Dashboard

Connect Grafana to TimescaleDB (PostgreSQL data source). SQL query: SELECT time, device_id, AVG(value) FILTER (WHERE sensor_type='temperature') as temp FROM sensor_data WHERE time > NOW() - INTERVAL '1 hour' GROUP BY time_bucket('1 minute', time), device_id ORDER BY time. Panel types: time-series graph (temperature over time), gauge (current value), stat (min/max/avg), geomap (device locations with latest values), alert panel (devices with abnormal readings). Set 5-second refresh for near-real-time.

5

Edge Computing with Node-RED

Node-RED runs on Raspberry Pi for edge processing. MQTT-in node subscribes to sensor data. Function nodes: apply calibration corrections, filter outliers (reject values outside physical limits), compute rolling average. Alert node: if temperature > 35°C → send push notification (PushOver, Telegram). Local storage: SQLite for offline operation (sync to cloud when reconnected). ML inference: TensorFlow Lite node runs anomaly detection model locally — no cloud dependency for time-critical decisions.

Code & Implementation

Core code for iot_sensor.py:

iot_sensor.py
  Python

import paho.mqtt.client as mqtt import json, time, random, ssl  BROKER   = "mqtt.catb.in" PORT     = 8883 TOPIC    = "catb/site/lab/device/pi001/sensor" CLIENT_ID = "pi001"  client = mqtt.Client(client_id=CLIENT_ID, protocol=mqtt.MQTTv5) client.tls_set(ca_certs="ca.crt", certfile="pi001.crt", keyfile="pi001.key",                tls_version=ssl.PROTOCOL_TLS)  def publish_sensor_data():     # Read from actual sensors in production     temperature = 22.5 + random.gauss(0, 0.5)  # Simulate with noise     humidity    = 45.0 + random.gauss(0, 1.0)      for sensor_type, value in [("temperature", temperature), ("humidity", humidity)]:         payload = json.dumps({             "device_id": CLIENT_ID,             "sensor_type": sensor_type,             "value": round(value, 2),             "timestamp": time.time(),             "unit": "°C" if sensor_type == "temperature" else "%"         })         result = client.publish(f"{TOPIC}/{sensor_type}", payload, qos=1, retain=False)         print(f"Published {sensor_type}: {value:.2f} (rc={result.rc})")  client.connect(BROKER, PORT) client.loop_start() while True:     publish_sensor_data()     time.sleep(30)  # Publish every 30 seconds

Testing & Troubleshooting

Test IoT Data Platform by verifying each subsystem individually before full integration.

!

Troubleshooting Tips

Verify power voltages, check ground connections, use serial monitor for debug.

Real-World Applications

*Smart building energy management

*Industrial equipment monitoring

*Smart agriculture soil and climate sensing

*Fleet vehicle telemetry

*Wearable health monitor data collection

*Smart city sensor network

*Supply chain cold chain temperature monitoring

*Environmental air quality monitoring

Extensions & Next Steps

Add MQTT over WebSocket for browser-based real-time dashboards
Implement OTA (Over-The-Air) firmware update via MQTT
Build anomaly detection ML model trained on sensor data
Add digital twin simulation synchronized with physical sensors
Implement adaptive sampling: increase frequency when anomaly detected

Interactive Playground

Coming Soon

An interactive simulator will be available here — simulate circuits and run code in-browser without hardware.

Frequently Asked Questions

How does MQTT differ from HTTP for IoT applications?

HTTP: request-response model, client must initiate every request (polling wastes bandwidth and battery). Each request includes full headers (200–800 bytes overhead). No persistence between requests. MQTT: persistent connection, server can push messages without client requesting. Tiny overhead (2-byte header minimum vs 200+ bytes HTTP). QoS levels guarantee delivery on unreliable networks. Retained messages provide last-known-value without device being connected. MQTT is 93% more bandwidth-efficient than HTTP for typical IoT sensor reporting. HTTP is better for complex REST APIs and one-time requests.

Introduction