9.Implement MQTT communication using both an online MQTT broker (HiveMQ) and a local offline MQTT broker (Eclipse Mosquitto)  to publish, receive, and display IoT sensor data.

pip install paho-mqtt

mosquitto -h

mosquitto -v

mqtt_broker_demo/
│
├── publisher.py
└── subscriber.py

publisher.py:


import paho.mqtt.client as mqtt
import random
import json
import time
import threading

# Choose: "online" or "offline"
MODE = "online"

if MODE == "online":
    BROKER = "broker.hivemq.com"
elif MODE == "offline":
    BROKER = "127.0.0.1"
else:
    raise ValueError("MODE must be 'online' or 'offline'")

PORT = 1883
TOPIC = "iot/environment"

connected = threading.Event()


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        connected.clear()
        print("Connection failed:", reason_code)
    else:
        connected.set()
        print("Connected to MQTT Broker!")


def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
    connected.clear()
    print("Disconnected:", reason_code)


client = mqtt.Client(
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    client_id="mqtt_publisher",
    protocol=mqtt.MQTTv311
)

client.on_connect = on_connect
client.on_disconnect = on_disconnect

try:
    print("Connecting to:", BROKER)
    client.connect(BROKER, PORT, 60)
    client.loop_start()

    while True:
        if not connected.wait(timeout=2):
            print("Waiting for MQTT connection...")
            continue

        temperature = round(random.uniform(20, 40), 2)
        humidity = round(random.uniform(40, 90), 2)

        data = {
            "temperature": temperature,
            "humidity": humidity
        }

        payload = json.dumps(data)

        result = client.publish(TOPIC, payload, qos=1)

        if result.rc == mqtt.MQTT_ERR_SUCCESS:
            try:
                result.wait_for_publish(timeout=10)

                if result.is_published():
                    print("Published:", payload)
                else:
                    print("Publishing timed out")

            except RuntimeError as e:
                print("Publishing error:", e)
        else:
            print("Publish failed:", mqtt.error_string(result.rc))

        time.sleep(5)

except KeyboardInterrupt:
    print("\nPublisher stopped.")

except Exception as e:
    print("Error:", e)

finally:
    client.loop_stop()
    client.disconnect()
    print("Disconnected from MQTT Broker.")


subscriber.py:


import paho.mqtt.client as mqtt
import json

# Choose: "online" or "offline"
MODE = "online"

if MODE == "online":
    BROKER = "broker.hivemq.com"
else:
    BROKER = "127.0.0.1"

PORT = 1883

TOPIC = "iot/environment"


def on_connect(client, userdata, flags, reason_code, properties):

    if reason_code.is_failure:
        print("Connection failed:", reason_code)
        return

    print("Connected to MQTT Broker!")

    result, mid = client.subscribe(TOPIC)

    if result == mqtt.MQTT_ERR_SUCCESS:
        print("Subscribed to:", TOPIC)
    else:
        print("Subscription failed:", mqtt.error_string(result))


def on_message(client, userdata, msg):

    try:
        data = json.loads(msg.payload.decode())

        temperature = float(data["temperature"])
        humidity = float(data["humidity"])

        print("\nReceived Sensor Data")
        print("Temperature:", temperature, "°C")
        print("Humidity:", humidity, "%")
        print("------------------------")

    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as e:
        print("Invalid MQTT message:", e)


client = mqtt.Client(
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
    client_id="mqtt_subscriber",
    protocol=mqtt.MQTTv311
)

client.on_connect = on_connect
client.on_message = on_message

print("Connecting to:", BROKER)

client.connect(BROKER, PORT, 60)

print("Waiting for messages...")

client.loop_forever()
