MQTT-Gerätesimulator in Python bauen

import asyncio, json, os, random, socket, sys, time, uuid
from collections import Counter
import aiomqtt

HOST, PORT = os.environ.get("MQTT_HOST", "broker.connectmq.com"), 1883
USER, PASS = os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASS")
N_DEVICES = int(os.environ.get("DEVICES", 100))
RAMP_S = 60              # spread device boot over one minute
CRASH_PROB = 0.002       # chance of a power loss after each publish
RUN = uuid.uuid4().hex[:6]
stats = Counter()

class PowerLoss(Exception):
    pass

async def commands(client, base, state):
    try:
        async for msg in client.messages:
            try:
                cmd = dict(json.loads(msg.payload))
                if "interval" in cmd:
                    state["interval"] = max(1.0, float(cmd["interval"]))
            except (ValueError, TypeError):
                continue                                   # ignore malformed commands
            await client.publish(f"{base}/cmd/ack", json.dumps({"ok": True, **cmd}), qos=1)
            stats["commands"] += 1
    except aiomqtt.MqttError:
        pass                                               # connection lost; device() reconnects

async def device(i: int):
    dev_id = f"sim-{RUN}-{i:04d}"
    base = f"testmqtt/sim/{dev_id}"
    rng = random.Random(i)
    state = {"interval": rng.choice([5, 10, 30])}
    temp, boots, attempt = rng.uniform(19, 24), 0, 0
    await asyncio.sleep(rng.uniform(0, RAMP_S))          # staggered boot

    while True:
        try:
            will = aiomqtt.Will(f"{base}/status", b"offline", qos=1, retain=True)
            async with aiomqtt.Client(HOST, PORT, identifier=dev_id, username=USER,
                                      password=PASS, keepalive=30, will=will) as client:
                attempt, boots, started = 0, boots + 1, time.monotonic()
                stats["online"] += 1
                try:
                    await client.publish(f"{base}/status", b"online", qos=1, retain=True)
                    await client.subscribe(f"{base}/cmd", qos=1)
                    listener = asyncio.create_task(commands(client, base, state))
                    try:
                        while True:
                            temp += rng.gauss(0, 0.05)
                            reading = {"temp": round(temp, 2), "boots": boots,
                                       "uptime": int(time.monotonic() - started), "fw": "1.4.2"}
                            await client.publish(f"{base}/telemetry", json.dumps(reading), qos=1)
                            stats["published"] += 1
                            if rng.random() < CRASH_PROB:
                                # power loss: drop TCP without DISCONNECT so the broker fires the will
                                # (aiomqtt has no public API for this, so reach the paho client)
                                client._client.socket().shutdown(socket.SHUT_RDWR)
                                await asyncio.sleep(1)
                                raise PowerLoss
                            await asyncio.sleep(state["interval"] * rng.uniform(0.9, 1.1))
                    finally:
                        listener.cancel()
                finally:
                    stats["online"] -= 1
        except PowerLoss:
            stats["power_losses"] += 1
            await asyncio.sleep(rng.uniform(5, 30))       # time to reboot
        except aiomqtt.MqttError:
            stats["conn_errors"] += 1
            attempt += 1
            await asyncio.sleep(min(60, 2 ** attempt) * rng.uniform(0.5, 1.0))

async def report():
    while True:
        await asyncio.sleep(10)
        print(time.strftime("%H:%M:%S"), dict(stats), flush=True)

async def main():
    print(f"run {RUN}: starting {N_DEVICES} devices under testmqtt/sim/sim-{RUN}-*")
    await asyncio.gather(report(), *(device(i) for i in range(N_DEVICES)))

if __name__ == "__main__":
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())

ulimit -n 4096                     # one socket per device, plus headroom
export MQTT_HOST=your-broker-host    # your own broker instance for the full fleet
export MQTT_USER=username MQTT_PASS=password
DEVICES=1000 python simulator.py

mosquitto_sub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/sim/+/status' -v -W 5 | grep -c ' online$'

mosquitto_sub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/sim/sim-3fa2c1-0042/cmd/ack' -C 1 -v &
mosquitto_pub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t 'testmqtt/sim/sim-3fa2c1-0042/cmd' -q 1 -m '{"interval": 2}'