Build an MQTT Device Simulator in Python for Hundreds of Virtual Sensors

An MQTT device simulator lets you build dashboards, storage pipelines and alert rules before a single real sensor is installed. In this tutorial you will write a Python simulator that models hundreds of virtual sensors with believable noise, drift, faults and dropouts, and publishes them to the public broker at broker.connectmq.com without breaking its rate limits.

Why Simulate Devices Before You Have Hardware

Hardware is usually the slowest part of an IoT project. Boards are on order, firmware is half written and the installer has not visited the site yet, while the people building the dashboard and the database already need data. A simulator removes that dependency. It gives every downstream component a realistic stream to work against, and it lets you rehearse things that are awkward to provoke with real equipment: a sensor that spikes to an impossible value, a device that goes silent for ten minutes, or a whole line coming online at once.

There is a second benefit that teams discover later. Once the real fleet exists, the same simulator becomes your regression tool. You can run it against a staging broker after every change to your ingestion code and see immediately if a parser or dashboard broke.

What separates a useful simulator from a toy is realism. A loop that publishes random.uniform(20, 25) every second produces data that looks nothing like a real temperature probe. Real signals change slowly, follow daily cycles, carry measurement noise, drift over weeks and occasionally fail. The simulator below models each of those behaviours with a few lines of code.

Design of the MQTT Device Simulator

Before writing code, decide on the topic structure and payload format, because everything you build downstream will depend on them. This tutorial uses one topic per device and a compact JSON payload:

connectmq/demo/<run-id>/<line>/<device>/telemetry

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

The <run-id> is a random string generated at start-up. The public broker is readable by everyone, so a unique prefix keeps your stream separate from other people's experiments and makes it easy to subscribe to exactly your data. The structure stays within six topic levels, comfortably under the public broker's limit of eight.

The simulator has three parts:

1. A sensor model that holds per-device state (base value, drift, phase of the daily cycle, fault flag) and returns a new reading when asked.
2. A scheduler that decides which device publishes next. Devices are spread evenly over the interval instead of all firing at the same moment.
3. A rate limiter that keeps the publishing client below the broker's per-client message limit.

Respect the Broker's Rate Limits

The public test broker allows 10 messages per second per client, 10 subscriptions per client and 64 KB per message. Hundreds of devices publishing every second would exceed that within moments, so the simulator has to do some arithmetic first. The aggregate rate is simply the number of devices divided by the publish interval.

The code caps itself at 8 messages per second to leave headroom for retries and for subscribing tools you run alongside it. If you need a higher rate, you have two honest options: spread devices across several client connections, each with its own ID and its own budget, or move to a private broker where the limits are set by your plan rather than by the shared public service.

The Complete Python Simulator

Install the only dependency:

pip install "paho-mqtt>=2.0"

Save the following as simulator.py. It uses the paho-mqtt 2.x callback API and only the standard library otherwise.

import argparse
import heapq
import json
import math
import random
import time
import uuid

import paho.mqtt.client as mqtt


class VirtualSensor:
    """One simulated temperature and humidity sensor with persistent state."""

    def __init__(self, index: int, fault_rate: float):
        self.device_id = f"sensor-{index:04d}"
        self.line = f"line{index % 4 + 1}"
        self.base_temp = random.uniform(19.0, 24.0)
        self.humidity = random.uniform(35.0, 55.0)
        self.phase = random.uniform(0, 2 * math.pi)
        self.drift = 0.0
        self.faulty = random.random() < fault_rate

    def read(self, now: float) -> dict:
        # Slow daily cycle of +/- 2.5 degrees, shifted per device
        daily = 2.5 * math.sin(2 * math.pi * (now % 86400) / 86400 + self.phase)
        # Bounded random walk to mimic calibration drift
        self.drift = max(-3.0, min(3.0, self.drift + random.gauss(0, 0.05)))
        # Gaussian measurement noise
        temp = self.base_temp + daily + self.drift + random.gauss(0, 0.15)
        self.humidity = max(10.0, min(95.0, self.humidity + random.gauss(0, 0.3)))
        # Faulty devices occasionally emit physically implausible spikes
        if self.faulty and random.random() < 0.05:
            temp += random.choice([-40.0, 60.0])
        return {
            "device": self.device_id,
            "ts": int(now * 1000),
            "temperature": round(temp, 2),
            "humidity": round(self.humidity, 1),
        }


def build_client(args, run_id: str) -> mqtt.Client:
    client_id = args.client_id or f"sim-{run_id}"
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if args.username:
        client.username_pw_set(args.username, args.password)
    if args.tls:
        client.tls_set()  # system CA store; the broker uses a Let's Encrypt certificate

    def on_connect(client, userdata, flags, reason_code, properties):
        print(f"connected: {reason_code}")

    def on_disconnect(client, userdata, flags, reason_code, properties):
        print(f"disconnected: {reason_code}")

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    return client


def main():
    parser = argparse.ArgumentParser(description="MQTT device simulator")
    parser.add_argument("--host", default="broker.connectmq.com")
    parser.add_argument("--port", type=int, default=1883)
    parser.add_argument("--tls", action="store_true")
    parser.add_argument("--username")
    parser.add_argument("--password")
    parser.add_argument("--client-id")
    parser.add_argument("--prefix", help="topic prefix, default connectmq/demo/<run-id>")
    parser.add_argument("--devices", type=int, default=200)
    parser.add_argument("--interval", type=float, default=30.0)
    parser.add_argument("--max-rate", type=float, default=8.0)
    parser.add_argument("--fault-rate", type=float, default=0.02)
    parser.add_argument("--dropout", type=float, default=0.01)
    args = parser.parse_args()

    run_id = uuid.uuid4().hex[:8]
    prefix = args.prefix or f"connectmq/demo/{run_id}"
    expected = args.devices / args.interval
    if expected > args.max_rate:
        raise SystemExit(f"{expected:.1f} msg/s exceeds --max-rate {args.max_rate}")
    print(f"publishing to {prefix}/# at about {expected:.1f} msg/s")

    sensors = [VirtualSensor(i, args.fault_rate) for i in range(args.devices)]
    client = build_client(args, run_id)
    client.connect_async(args.host, args.port, keepalive=30)
    client.loop_start()

    start = time.time()
    # Spread first readings evenly over one interval
    schedule = [(start + i * args.interval / args.devices, i) for i in range(args.devices)]
    heapq.heapify(schedule)
    min_gap = 1.0 / args.max_rate
    last_send = 0.0

    try:
        while True:
            due, i = heapq.heappop(schedule)
            now = time.time()
            wait = max(due - now, last_send + min_gap - now)
            if wait > 0:
                time.sleep(wait)
            sensor = sensors[i]
            now = time.time()
            if random.random() >= args.dropout and client.is_connected():
                topic = f"{prefix}/{sensor.line}/{sensor.device_id}/telemetry"
                client.publish(topic, json.dumps(sensor.read(now)), qos=0)
                last_send = time.time()
            heapq.heappush(schedule, (due + args.interval, i))
    except KeyboardInterrupt:
        print("stopping")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Run it with the defaults:

python simulator.py --devices 200 --interval 30

The script prints the prefix it chose, for example connectmq/demo/3f9a1c2e/#. Open a second terminal and watch the stream:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-3f9a1c2e -t "connectmq/demo/3f9a1c2e/#" -v

How the Realism Works

Each reading is the sum of four components, and each one exists to exercise a specific part of your pipeline.

• Daily cycle. A sine wave with a 24-hour period and a random phase per device. Your charts get a visible shape, and aggregation queries like hourly averages return meaningful variation instead of a flat line.
• Bounded drift. A random walk clamped to plus or minus three degrees. It mimics sensors slowly losing calibration, which is exactly what drift-detection logic needs to see.
• Measurement noise. Gaussian noise with a small standard deviation. Smoothing, rolling averages and deadband filters only make sense when the input is noisy.
• Faults and dropouts. About 2 percent of devices are marked faulty and occasionally emit spikes of plus 60 or minus 40 degrees. Separately, 1 percent of scheduled readings are simply skipped. Spikes test validation and alert thresholds; gaps test staleness detection and how your charts render missing data.

Because the state lives in each VirtualSensor object, a device's values are continuous over time. That matters when you test anything that compares the current value with the previous one, such as rate-of-change alerts.

Common Pitfalls and How to Avoid Them

Bursty publishing. If every device publishes at second zero of each interval, the broker sees a burst of hundreds of messages followed by silence. Real fleets are rarely synchronised like that, and a burst will hit per-client limits. The scheduler staggers the first reading of each device across the interval, which keeps the rate smooth.

Clock confusion. The payload carries its own millisecond timestamp. Downstream tools such as Telegraf or a Python bridge can use it instead of the arrival time, which keeps charts correct even if messages are delayed. Make sure every consumer agrees on the unit: milliseconds here, not seconds.

Reusing client IDs. If you start two copies of the simulator with the same --client-id, the broker disconnects the older session every time the newer one connects and the two will fight. Let the script generate a random ID on the public broker, or use the ID issued for your credentials on a private one.

Sending sensitive data to a public broker. Everything on the public broker can be read by anyone who guesses or wildcards the topic. Simulated data is fine there; real customer or plant data is not.

Publishing while disconnected. The loop checks client.is_connected() and skips readings while the connection is down, just as a real sensor without buffering would. If you want to test buffering behaviour, read the paho MQTT reconnect guide and add an offline queue.

Pointing the Simulator at Your Own Broker

The public broker is ideal for a quick start, but for sustained load prototyping you want an isolated environment. With a managed MQTT broker from ConnectMQ you get your own credentials and a TLS endpoint:

python simulator.py \
  --host broker.connectmq.com --port 8883 --tls \
  --username YOUR_USERNAME --password YOUR_PASSWORD --client-id YOUR_CLIENT_ID \
  --prefix connectmq/factory1/production \
  --devices 500 --interval 60 --max-rate 9

Port 8883 uses TLS with a Let's Encrypt certificate, so tls_set() with the system trust store is enough. On a managed broker, client IDs are bound to the credentials you are issued, so pass the ID from the portal rather than inventing one. The portal's in-browser client and live topic explorer are a quick way to confirm that the simulated devices appear under the expected topics. The Serverless (Free) plan supports up to 25 connections at QoS 0 or 1, which is enough to split a large simulated fleet across several processes if each one uses its own issued credentials.

Where to Send the Simulated Data Next

A simulator is only interesting once something consumes it. Natural next steps:

1. Parse the payloads in Python as shown in working with MQTT JSON in Python and compute statistics in pandas.
2. Persist the stream with Telegraf and InfluxDB using the TIG stack with Docker Compose, then build Grafana panels on it.
3. Compare this script with dedicated benchmarking utilities in our overview of MQTT load testing tools when you need to measure broker throughput rather than generate realistic data.

For a broader view of the Python and other client options, the MQTT client libraries guide compares the main libraries by language, and the paho-mqtt documentation covers every client option used above.

Get Started with ConnectMQ

Start with the free public MQTT broker at broker.connectmq.com to see your simulated fleet in minutes. When you are ready for private credentials, TLS and your own quotas, ConnectMQ runs a fully managed broker in the cloud. Create a free Serverless broker at app.connectmq.com with no credit card, copy the generated username, password and client ID, and point the simulator at it. If you are still comparing hosting options, our guide to the cloud MQTT broker explains what to look for.

Frequently Asked Questions

How many devices can an MQTT device simulator publish from one client?

It depends on the interval and the broker's per-client limit. On the public broker, keep devices divided by interval under 10 messages per second, for example 200 devices every 30 seconds or 500 devices every 60 seconds.

Should the simulator use QoS 0 or QoS 1?

QoS 0 is fine for most telemetry simulations and mirrors what many real sensors do. Use QoS 1 when you specifically want to test acknowledgement handling or duplicate detection downstream.

Can I simulate devices going offline?

Yes. Increase --dropout for random gaps, or add a Last Will message per client so subscribers receive an offline status when a simulator process is killed.

Is it safe to run the simulator on the public broker?

It is safe as long as the data is synthetic. The public broker is readable by anyone, so never publish real customer, plant or personal data there.