Yüzlerce Sanal Sensör için Python ile MQTT Cihaz Simülatörü Oluşturun

Bir MQTT cihaz simülatörü, tek bir gerçek sensör kurulmadan önce paneller, depolama hatları ve alarm kuralları oluşturmanızı sağlar. Bu eğitimde inandırıcı gürültü, kayma, arıza ve kesintilerle yüzlerce sanal sensörü modelleyen ve bunları hız sınırlarını aşmadan broker.connectmq.com adresindeki herkese açık broker'a publish eden bir Python simülatörü yazacaksınız.

Donanımınız Olmadan Neden Cihaz Simüle Etmelisiniz

Donanım genellikle bir IoT projesinin en yavaş kısmıdır. Kartlar sipariş aşamasındadır, firmware yarı yazılmıştır ve kurulumcu henüz sahayı ziyaret etmemiştir; oysa paneli ve veritabanını oluşturan kişilerin şimdiden veriye ihtiyacı vardır. Bir simülatör bu bağımlılığı ortadan kaldırır. Aşağı akıştaki her bileşene üzerinde çalışabileceği gerçekçi bir akış sağlar ve gerçek ekipmanla tetiklemesi zahmetli olan durumları prova etmenizi mümkün kılar: imkânsız bir değere sıçrayan bir sensör, on dakika boyunca sessizleşen bir cihaz veya bir anda devreye giren koca bir hat.

Ekiplerin daha sonra keşfettiği ikinci bir fayda da vardır. Gerçek filo oluştuğunda aynı simülatör regresyon aracınız olur. Veri alım kodunuzdaki her değişiklikten sonra onu bir staging broker'ına karşı çalıştırabilir ve bir ayrıştırıcının veya panelin bozulup bozulmadığını hemen görebilirsiniz.

Faydalı bir simülatörü bir oyuncaktan ayıran şey gerçekçiliktir. Her saniye random.uniform(20, 25) publish eden bir döngü, gerçek bir sıcaklık probuna hiç benzemeyen veri üretir. Gerçek sinyaller yavaş değişir, günlük döngüleri izler, ölçüm gürültüsü taşır, haftalar içinde kayar ve ara sıra arızalanır. Aşağıdaki simülatör bu davranışların her birini birkaç satır kodla modeller.

MQTT Cihaz Simülatörünün Tasarımı

Kod yazmadan önce topic yapısına ve payload formatına karar verin, çünkü aşağı akışta oluşturacağınız her şey bunlara bağlı olacaktır. Bu eğitim cihaz başına bir topic ve kompakt bir JSON payload kullanır:

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

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

<run-id> başlangıçta üretilen rastgele bir dizgidir. Herkese açık broker'ı herkes okuyabilir; bu nedenle benzersiz bir önek, akışınızı başkalarının denemelerinden ayrı tutar ve tam olarak kendi verinize subscribe olmayı kolaylaştırır. Yapı altı topic seviyesi içinde kalır; bu da herkese açık broker'ın sekiz seviyelik sınırının rahatça altındadır.

Simülatörün üç bölümü vardır:

1. Cihaz başına durumu (temel değer, kayma, günlük döngünün fazı, arıza bayrağı) tutan ve istendiğinde yeni bir ölçüm döndüren bir sensör modeli.
2. Bir sonraki publish'i hangi cihazın yapacağına karar veren bir zamanlayıcı. Cihazlar hepsi aynı anda tetiklenmek yerine aralık boyunca eşit şekilde dağıtılır.
3. Publish yapan istemciyi broker'ın istemci başına mesaj sınırının altında tutan bir hız sınırlayıcı.

Broker'ın Hız Sınırlarına Uyun

Herkese açık test broker'ı istemci başına saniyede 10 mesaja, istemci başına 10 aboneliğe ve mesaj başına 64 KB'a izin verir. Her saniye publish yapan yüzlerce cihaz bunu kısa sürede aşar; bu yüzden simülatörün önce biraz hesap yapması gerekir. Toplam hız, cihaz sayısının publish aralığına bölünmesidir.

Kod, yeniden denemeler ve yanında çalıştırdığınız subscribe araçları için pay bırakmak üzere kendini saniyede 8 mesajla sınırlar. Daha yüksek bir hıza ihtiyacınız varsa iki dürüst seçeneğiniz vardır: cihazları her biri kendi ID'si ve kendi bütçesi olan birkaç istemci bağlantısına dağıtmak veya sınırların paylaşılan herkese açık hizmet yerine planınıza göre belirlendiği özel bir broker'a geçmek.

Eksiksiz Python Simülatörü

Tek bağımlılığı kurun:

pip install "paho-mqtt>=2.0"

Aşağıdakini simulator.py olarak kaydedin. paho-mqtt 2.x callback API'sini, bunun dışında yalnızca standart kütüphaneyi kullanır.

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

import paho.mqtt.client as mqtt


class VirtualSensor:
    """Kalıcı duruma sahip, simüle edilmiş tek bir sıcaklık ve nem sensörü."""

    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:
        # Cihaz başına kaydırılmış, +/- 2,5 derecelik yavaş günlük döngü
        daily = 2.5 * math.sin(2 * math.pi * (now % 86400) / 86400 + self.phase)
        # Kalibrasyon kaymasını taklit eden sınırlı rastgele yürüyüş
        self.drift = max(-3.0, min(3.0, self.drift + random.gauss(0, 0.05)))
        # Gauss ölçüm gürültüsü
        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)))
        # Arızalı cihazlar ara sıra fiziksel olarak mantıksız sıçramalar üretir
        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()  # sistem CA deposu; broker bir Let's Encrypt sertifikası kullanır

    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()
    # İlk ölçümleri bir aralık boyunca eşit şekilde dağıt
    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()

Varsayılan ayarlarla çalıştırın:

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

Betik seçtiği öneki yazdırır, örneğin connectmq/demo/3f9a1c2e/#. İkinci bir terminal açın ve akışı izleyin:

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

Gerçekçilik Nasıl Sağlanıyor

Her ölçüm dört bileşenin toplamıdır ve her biri hattınızın belirli bir bölümünü sınamak için vardır.

• Günlük döngü. 24 saatlik periyoda ve cihaz başına rastgele bir faza sahip bir sinüs dalgası. Grafikleriniz görünür bir şekil kazanır ve saatlik ortalamalar gibi toplama sorguları düz bir çizgi yerine anlamlı bir değişim döndürür.
• Sınırlı kayma. Artı eksi üç dereceyle sınırlandırılmış bir rastgele yürüyüş. Sensörlerin kalibrasyonlarını yavaşça kaybetmesini taklit eder; kayma tespit mantığının görmesi gereken de tam olarak budur.
• Ölçüm gürültüsü. Küçük standart sapmalı Gauss gürültüsü. Yumuşatma, hareketli ortalamalar ve deadband filtreleri yalnızca girdi gürültülü olduğunda anlam taşır.
• Arızalar ve kesintiler. Cihazların yaklaşık yüzde 2'si arızalı olarak işaretlenir ve ara sıra artı 60 veya eksi 40 derecelik sıçramalar üretir. Bundan ayrı olarak, zamanlanmış ölçümlerin yüzde 1'i basitçe atlanır. Sıçramalar doğrulamayı ve alarm eşiklerini, boşluklar ise bayatlık tespitini ve grafiklerinizin eksik veriyi nasıl gösterdiğini test eder.

Durum her VirtualSensor nesnesinde tutulduğu için bir cihazın değerleri zaman içinde süreklidir. Bu, değişim hızı alarmları gibi mevcut değeri bir öncekiyle karşılaştıran her şeyi test ederken önemlidir.

Yaygın Tuzaklar ve Bunlardan Kaçınma Yolları

Patlamalı publish. Her cihaz her aralığın sıfırıncı saniyesinde publish yaparsa broker yüzlerce mesajlık bir patlamanın ardından sessizlik görür. Gerçek filolar nadiren bu şekilde senkronizedir ve bir patlama istemci başına sınırlara takılır. Zamanlayıcı her cihazın ilk ölçümünü aralık boyunca kaydırarak hızı dengeli tutar.

Saat karışıklığı. Payload kendi milisaniye zaman damgasını taşır. Telegraf veya bir Python köprüsü gibi aşağı akış araçları varış zamanı yerine bunu kullanabilir; bu da mesajlar gecikse bile grafiklerin doğru kalmasını sağlar. Her tüketicinin birim konusunda hemfikir olduğundan emin olun: burada saniye değil, milisaniye.

Client ID'leri yeniden kullanmak. Simülatörün iki kopyasını aynı --client-id ile başlatırsanız, broker yenisi her bağlandığında eski oturumun bağlantısını keser ve ikisi sürekli çekişir. Herkese açık broker'da betiğin rastgele bir ID üretmesine izin verin veya özel bir broker'da kimlik bilgileriniz için verilen ID'yi kullanın.

Hassas veriyi herkese açık bir broker'a göndermek. Herkese açık broker'daki her şey, topic'i tahmin eden veya wildcard kullanan herkes tarafından okunabilir. Simüle edilmiş veri orada sorun değildir; gerçek müşteri veya tesis verisi ise sorundur.

Bağlantı yokken publish etmek. Döngü client.is_connected() değerini kontrol eder ve bağlantı koptuğunda, tamponlaması olmayan gerçek bir sensörün yapacağı gibi ölçümleri atlar. Tamponlama davranışını test etmek istiyorsanız paho MQTT yeniden bağlanma rehberini okuyun ve bir çevrimdışı kuyruk ekleyin.

Simülatörü Kendi Broker'ınıza Yönlendirmek

Herkese açık broker hızlı bir başlangıç için idealdir, ancak sürekli yük prototiplemesi için izole bir ortam istersiniz. ConnectMQ'nun yönetilen MQTT broker hizmetiyle kendi kimlik bilgilerinizi ve bir TLS uç noktasını edinirsiniz:

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

8883 portu bir Let's Encrypt sertifikasıyla TLS kullanır; bu yüzden sistem güven deposuyla tls_set() yeterlidir. Yönetilen bir broker'da client ID'ler size verilen kimlik bilgilerine bağlıdır; dolayısıyla bir tane uydurmak yerine portaldaki ID'yi geçin. Portalın tarayıcı içi istemcisi ve canlı topic gezgini, simüle edilen cihazların beklenen topic'ler altında göründüğünü doğrulamanın hızlı bir yoludur. Serverless (Free) planı QoS 0 veya 1 ile 25 bağlantıya kadar destekler; her biri kendisine verilen kimlik bilgilerini kullandığı sürece bu, büyük bir simüle filoyu birkaç işleme bölmek için yeterlidir.

Simüle Edilen Veriyi Sırada Nereye Göndermeli

Bir simülatör ancak bir şey onu tükettiğinde ilginç hale gelir. Doğal sonraki adımlar:

1. Payload'ları Python'da MQTT JSON ile çalışma yazısında gösterildiği gibi Python'da ayrıştırın ve pandas ile istatistik hesaplayın.
2. Akışı Docker Compose ile TIG stack kullanarak Telegraf ve InfluxDB ile kalıcı hale getirin, ardından üzerine Grafana panelleri kurun.
3. Gerçekçi veri üretmek yerine broker verimini ölçmeniz gerektiğinde bu betiği MQTT yük testi araçlarına genel bakışımızdaki özel kıyaslama araçlarıyla karşılaştırın.

Python ve diğer istemci seçeneklerine daha geniş bir bakış için MQTT istemci kütüphaneleri rehberi başlıca kütüphaneleri dile göre karşılaştırır; paho-mqtt dokümantasyonu ise yukarıda kullanılan her istemci seçeneğini ele alır.

ConnectMQ ile Başlayın

Simüle filonuzu dakikalar içinde görmek için broker.connectmq.com adresindeki ücretsiz herkese açık MQTT broker ile başlayın. Özel kimlik bilgilerine, TLS'e ve kendi kotalarınıza hazır olduğunuzda ConnectMQ bulutta tam yönetilen bir broker çalıştırır. Kredi kartı olmadan app.connectmq.com adresinde ücretsiz bir Serverless broker oluşturun, üretilen kullanıcı adı, parola ve client ID'yi kopyalayın ve simülatörü ona yönlendirin. Hâlâ barındırma seçeneklerini karşılaştırıyorsanız, bulut MQTT broker rehberimiz nelere dikkat etmeniz gerektiğini açıklar.

Sık Sorulan Sorular

Bir MQTT cihaz simülatörü tek bir istemciden kaç cihaz publish edebilir?

Bu, aralığa ve broker'ın istemci başına sınırına bağlıdır. Herkese açık broker'da cihaz sayısının aralığa bölümünü saniyede 10 mesajın altında tutun; örneğin her 30 saniyede 200 cihaz veya her 60 saniyede 500 cihaz.

Simülatör QoS 0 mı yoksa QoS 1 mi kullanmalı?

QoS 0 çoğu telemetri simülasyonu için yeterlidir ve birçok gerçek sensörün yaptığını yansıtır. Aşağı akışta özellikle onay işlemeyi veya tekrar tespitini test etmek istediğinizde QoS 1 kullanın.

Cihazların çevrimdışı olmasını simüle edebilir miyim?

Evet. Rastgele boşluklar için --dropout değerini artırın veya istemci başına bir Last Will mesajı ekleyin; böylece bir simülatör işlemi sonlandırıldığında subscriber'lar çevrimdışı durumu alır.

Simülatörü herkese açık broker'da çalıştırmak güvenli mi?

Veriler sentetik olduğu sürece güvenlidir. Herkese açık broker'ı herkes okuyabilir; bu yüzden oraya asla gerçek müşteri, tesis veya kişisel veri publish etmeyin.