MQTT Architecture:

  Publisher                Broker               Subscriber
  ──────────               ──────               ──────────
  [ESP32 Sensor] ─PUBLISH─► [Mosquitto] ─PUSH─► [Dashboard]
  [Raspberry Pi] ─PUBLISH─►     │         ─PUSH─► [Mobile App]
                                │         ─PUSH─► [Data Logger]
  [Dashboard]   ◄─PUSH──────────┘
    (also subscriber to commands)

Key roles:
  Publisher  — sends messages to topics
  Subscriber — receives messages from topics
  Broker     — routes messages between publishers and subscribers
  Client     — any device that connects to the broker (pub AND/OR sub)

# Install Mosquitto on Linux/Raspberry Pi
sudo apt install mosquitto mosquitto-clients -y
sudo systemctl enable mosquitto
sudo systemctl start mosquitto

# Test with command-line clients (two terminal windows)
# Terminal 1 — subscribe
mosquitto_sub -h localhost -t "test/topic" -v

# Terminal 2 — publish
mosquitto_pub -h localhost -t "test/topic" -m "Hello MQTT!"

# Terminal 1 should show:
# test/topic Hello MQTT!

# /etc/mosquitto/mosquitto.conf

listener 1883            # Standard MQTT port
listener 8883            # MQTT over TLS
cafile /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key

listener 9001            # WebSocket support (for browser clients)
protocol websockets

allow_anonymous false    # Require authentication
password_file /etc/mosquitto/passwd

# Python — full MQTT client with all lifecycle callbacks
import paho.mqtt.client as mqtt
import json
import time

class IoTMQTTClient:
    def __init__(self, broker, port, client_id, username=None, password=None):
        self.broker    = broker
        self.port      = port
        self.client    = mqtt.Client(client_id=client_id)
        self.connected = False

        # Set credentials if provided
        if username:
            self.client.username_pw_set(username, password)

        # Register callbacks
        self.client.on_connect    = self._on_connect
        self.client.on_disconnect = self._on_disconnect
        self.client.on_message    = self._on_message
        self.client.on_publish    = self._on_publish

        # Last Will and Testament (sent if client disconnects uncleanly)
        lwt_payload = json.dumps({"status": "offline", "client_id": client_id})
        self.client.will_set(
            topic=f"clients/{client_id}/status",
            payload=lwt_payload,
            qos=1,
            retain=True  # Retained so new subscribers see it immediately
        )

    def _on_connect(self, client, userdata, flags, rc):
        rc_messages = {
            0: "Connection accepted",
            1: "Wrong protocol version",
            2: "Client ID rejected",
            3: "Broker unavailable",
            4: "Bad username or password",
            5: "Not authorized",
        }
        print(f"Connect: {rc_messages.get(rc, f'Unknown ({rc})')}")
        if rc == 0:
            self.connected = True
            # Announce online status
            self.client.publish(
                f"clients/{self.client._client_id.decode()}/status",
                json.dumps({"status": "online"}),
                qos=1, retain=True
            )

    def _on_disconnect(self, client, userdata, rc):
        self.connected = False
        if rc != 0:
            print(f"Unexpected disconnect (rc={rc}). Will auto-reconnect.")

    def _on_message(self, client, userdata, msg):
        print(f"Received [{msg.topic}] QoS={msg.qos}: {msg.payload.decode()}")

    def _on_publish(self, client, userdata, mid):
        print(f"Message {mid} confirmed by broker")

    def connect(self):
        self.client.connect(self.broker, self.port, keepalive=60)
        self.client.loop_start()  # Background thread for MQTT

    def publish(self, topic, payload, qos=1, retain=False):
        if isinstance(payload, dict):
            payload = json.dumps(payload)
        result = self.client.publish(topic, payload, qos=qos, retain=retain)
        return result.mid

    def subscribe(self, topic, qos=1):
        self.client.subscribe(topic, qos)
        print(f"Subscribed to: {topic}")

    def disconnect(self):
        self.client.loop_stop()
        self.client.disconnect()


# Usage
client = IoTMQTTClient(
    broker="broker.hivemq.com",
    port=1883,
    client_id="factory-node-01"
)
client.connect()
time.sleep(1)  # Wait for connection

client.subscribe("factory/+/commands")
client.publish("factory/node-01/temperature", {"value": 24.5, "unit": "C"})

time.sleep(5)
client.disconnect()

# Publish a retained message (e.g. device configuration)
client.publish(
    topic  = "devices/sensor-01/config",
    payload = json.dumps({"interval": 60, "unit": "celsius"}),
    qos    = 1,
    retain = True   # Broker keeps this for new subscribers
)

# Clear a retained message by publishing empty payload
client.publish(
    topic   = "devices/sensor-01/config",
    payload = "",   # Empty payload
    retain  = True  # + retain=True clears the stored message
)

# Set LWT before connecting
client.will_set(
    topic   = "fleet/truck-007/status",
    payload = json.dumps({"status": "OFFLINE", "reason": "unexpected_disconnect"}),
    qos     = 1,
    retain  = True
)
client.connect(broker, 1883, keepalive=60)

# When connection is alive, publish "online" status
client.publish(
    "fleet/truck-007/status",
    json.dumps({"status": "ONLINE", "firmware": "1.4.2"}),
    qos=1, retain=True
)
# If the client crashes or loses power, the broker
# automatically publishes the LWT after the keepalive timeout

# Python — MQTT with TLS encryption
import paho.mqtt.client as mqtt
import ssl

client = mqtt.Client(client_id="secure-node")

# TLS configuration
client.tls_set(
    ca_certs   = "/path/to/ca.crt",       # Server certificate authority
    certfile   = "/path/to/client.crt",   # Client certificate
    keyfile    = "/path/to/client.key",   # Client private key
    tls_version = ssl.PROTOCOL_TLSv1_2
)
client.tls_insecure_set(False)  # Never set to True in production

client.username_pw_set("device-01", "strong-password")
client.connect("your-broker.example.com", 8883, 60)  # Note: port 8883 for TLS

# MQTT wildcard subscriptions
import paho.mqtt.client as mqtt

SUBSCRIPTIONS = [
    # + matches exactly one level
    ("factory/+/temperature", 1),    # factory/line-1/temperature ✓
                                      # factory/line-2/temperature ✓
                                      # factory/line-1/humidity   ✗

    # # matches zero or more levels (must be last)
    ("factory/#", 1),                # factory/line-1/temperature ✓
                                      # factory/line-2/humidity    ✓
                                      # factory/alarms/critical    ✓

    # Exact match
    ("factory/line-1/status", 2),    # Exactly this topic only

    # System topics (Mosquitto $SYS)
    ("$SYS/broker/clients/connected", 0),  # Connected client count
    ("$SYS/broker/messages/received", 0),  # Total messages received
]

def on_connect(client, userdata, flags, rc):
    for topic, qos in SUBSCRIPTIONS:
        client.subscribe(topic, qos)
        print(f"Subscribed: {topic} (QoS {qos})")

# Complete pipeline: ESP32 sensor → MQTT → Python logger
# This is the server-side logger

import paho.mqtt.client as mqtt
import json
import csv
import os
from datetime import datetime

LOG_FILE = "sensor_log.csv"

# Initialize CSV file with header
if not os.path.exists(LOG_FILE):
    with open(LOG_FILE, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(["timestamp", "device_id", "temperature", "humidity", "battery"])

def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload.decode())
        timestamp = datetime.utcnow().isoformat()

        # Log to CSV
        with open(LOG_FILE, 'a', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([
                timestamp,
                data.get("device_id", "unknown"),
                data.get("temperature"),
                data.get("humidity"),
                data.get("battery")
            ])

        print(f"[{timestamp}] {data.get('device_id')}: "
              f"T={data.get('temperature')}°C "
              f"H={data.get('humidity')}% "
              f"Bat={data.get('battery')}%")

        # Trigger alert if temperature is too high
        if data.get("temperature", 0) > 35:
            alert_payload = json.dumps({
                "type": "HIGH_TEMP",
                "device": data.get("device_id"),
                "value": data.get("temperature"),
                "time": timestamp
            })
            client.publish("alerts/temperature", alert_payload, qos=2)

    except (json.JSONDecodeError, KeyError) as e:
        print(f"Parse error on {msg.topic}: {e}")

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Logger connected to broker")
        client.subscribe("sensors/#", qos=1)
        print("Logging all sensor data to sensor_log.csv")

client = mqtt.Client(client_id="data-logger")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.hivemq.com", 1883, 60)
client.loop_forever()