Gerçekçi MQTT test verisi üretmek için her sensörü bir taban değer artı günlük döngü, ölçüm gürültüsü, yavaş kayma ve ara sıra bir sıçrama olarak modelleyin; ardından ölçümü cihazlarınızın yaptığı gibi kodlayın — JSON veya paketlenmiş bir binary struct — ve aynı topic'lere publish edin. Böylece dashboard'lar, kurallar ve depolama katmanı, gerçek donanım ortada yokken bile inandırıcı bir veriyle çalışır.
Bu rehber bu üreticiyi Python ile kuruyor, bir binary kodlama ve negatif testler için kasıtlı olarak bozulmuş payload'lar ekliyor ve kompakt bir Node.js sürümü gösteriyor.
Sensör verisini gerçekçi yapan nedir
random.uniform(15, 30) hiçbir yapısı olmadan zıplayan değerler üretir. Gerçek sensörler farklı davranır ve bu farklar tam olarak veriyi tüketen kodu bozan şeylerdir:
| Özellik | Gerçek hayattan örnek | Neyi test eder |
|---|---|---|
| Taban değer | Bir oda 21 °C civarında durur | Varsayılan grafik aralıkları, makul eşikler |
| Günlük döngü | Öğleden sonra daha sıcak, gece daha serin | Saatlik/günlük toplama, eğilim çizgileri |
| Gürültü | Örnekler arasında ±0,1–0,2 °C oynama | Yumuşatma, ölü bantlar, değişim tespiti |
| Kayma (drift) | Kalibrasyon haftalar içinde yavaşça kayar | Anomali tespiti, yeniden kalibrasyon mantığı |
| Sıçramalar | Bir kapı açılır, sensör anlık hata verir | Aykırı değer filtreleri, alarm tekilleştirme |
| Nicemleme | Sensör 0,1 °C veya %0,5 adımlarla ölçer | Eşitlik karşılaştırmaları, yuvarlama hataları |
| Boşluklar | Zayıf bir bağlantıda kaybolan ölçümler | Eksik veri yönetimi, sıra kontrolleri |
| Saat sapması | Cihaz zaman damgaları birkaç yüz ms kayar | Sıralama, zaman penceresi birleştirmeleri |
Python'da bir sensör modeli
Aşağıdaki model her cihaz için ayrı bir rastgele sayı üreteci kullanır, böylece aynı seed ile yapılan her çalıştırma birebir aynı veriyi üretir. Kayma, ortalamaya dönen bir rastgele yürüyüştür: dolaşır ama çok günlük bir testte sonsuza kaçmaz.
import math, random
from dataclasses import dataclass
@dataclass
class Sensor:
base: float # tipik değer
daily_amp: float # gece/gündüz döngüsünün genliği
noise: float # ölçüm gürültüsünün standart sapması
drift_step: float # örnek başına rastgele yürüyüş adımı
lo: float # fiziksel alt sınır
hi: float # fiziksel üst sınır
resolution: float # nicemleme adımı, örn. 0.1
rng: random.Random
spike_prob: float = 0.002
drift: float = 0.0
def read(self, t: float) -> float:
# tepe noktası 15:00 UTC, en düşük noktası 03:00 UTC olan sinüs dalgası
day = math.sin(2 * math.pi * ((t % 86400) / 86400 - 0.375))
self.drift = 0.999 * self.drift + self.rng.gauss(0, self.drift_step)
v = self.base + self.daily_amp * day + self.drift + self.rng.gauss(0, self.noise)
if self.rng.random() < self.spike_prob:
v += self.rng.choice((-1, 1)) * (self.hi - self.lo) * 0.2
v = min(max(v, self.lo), self.hi)
return round(round(v / self.resolution) * self.resolution, 3)
def make_device(seed: int) -> dict:
rng = random.Random(seed)
return {
"rng": rng,
"battery_mv": rng.uniform(4000, 4200),
"temperature": Sensor(rng.uniform(19, 24), 2.5, 0.15, 0.02, -40, 85, 0.1, rng),
"humidity": Sensor(rng.uniform(40, 55), -6.0, 0.8, 0.05, 0, 100, 0.5, rng),
}Nem negatif genlik kullanıyor, çünkü bir oda ısındıkça bağıl nem genellikle düşer — küçük bir ayrıntı, ama analizlerinizdeki korelasyonların gerçek hayattaki gibi davranmasını sağlar.
JSON telemetri publish etmek
Publisher birkaç cihaz oluşturur, her aralıkta cihaz başına bir ölçüm üretir, kayıplı bir bağlantıyı simüle etmek için ölçümlerin yaklaşık %1'ini atlar ve zaman damgalarına biraz saat sapması ekler. SPEEDUP zamanı sıkıştırır; böylece geliştirme sırasında tam bir gece/gündüz döngüsü birkaç dakikada geçer.
import json, os, time
import paho.mqtt.client as mqtt
from sensors import make_device
HOST, PORT = "broker.connectmq.com", 1883
N_DEVICES, INTERVAL, SPEEDUP = 5, 2.0, 360 # 360x: her 4 dakikada bir simüle gün
devices = {f"sim-{i:03d}": make_device(seed=i) for i in range(N_DEVICES)}
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"datagen-{os.getpid()}")
client.username_pw_set(os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASS"))
client.connect(HOST, PORT)
client.loop_start()
start, seq = time.time(), 0
while True:
sim_t = start + (time.time() - start) * SPEEDUP
for dev_id, d in devices.items():
rng = d["rng"]
if rng.random() < 0.01: # kaybolan ölçüm
continue
d["battery_mv"] -= rng.uniform(0, 0.05)
reading = {
"ts": int((sim_t + rng.uniform(-0.3, 0.3)) * 1000),
"seq": seq,
"temperature": d["temperature"].read(sim_t),
"humidity": d["humidity"].read(sim_t),
"battery_mv": int(d["battery_mv"]),
}
client.publish(f"testmqtt/datagen/{dev_id}/telemetry", json.dumps(reading), qos=1)
seq += 1
time.sleep(INTERVAL)Akışı mosquitto_sub -h broker.connectmq.com -t 'testmqtt/datagen/+/telemetry' -v ile izleyin ya da çevrimiçi MQTT istemcisini açıp testmqtt/datagen/# topic'ine subscribe olun. Alan adlarını ve birimleri cihazlar sahaya çıkmadan önce doğru belirlemek önemlidir; MQTT payload tasarımı en iyi uygulamaları artıları ve eksileri ele alıyor.
struct ile binary kodlama
Birçok kısıtlı cihaz JSON yerine paketlenmiş binary gönderir. Aynı ölçüm yaklaşık 90 bayt yerine 14 bayta sığar:
| Alan | Tip | Kodlama |
|---|---|---|
| version | uint8 | Format sürümü, 1'den başlar |
| ts | uint32 | Saniye cinsinden Unix zamanı |
| seq | uint16 | 65.535'ten sonra başa döner |
| temperature | int16 | °C × 100 |
| humidity | uint16 | % × 100 |
| battery_mv | uint16 | Milivolt |
| flags | uint8 | Bit alanı (düşük pil, sensör hatası…) |
import math, struct
FMT = "<BIHhHHB" # little-endian, dolgu yok: 14 bayt
def encode(r: dict, flags: int = 0) -> bytes:
return struct.pack(FMT, 1, r["ts"] // 1000, r["seq"] & 0xFFFF,
round(r["temperature"] * 100), round(r["humidity"] * 100),
r["battery_mv"], flags)
def decode(b: bytes) -> dict:
v, ts, seq, t, h, bat, flags = struct.unpack(FMT, b)
return {"v": v, "ts": ts, "seq": seq, "temperature": t / 100,
"humidity": h / 100, "battery_mv": bat, "flags": flags}
r = {"ts": 1760700000123, "seq": 42, "temperature": 21.4, "humidity": 47.5, "battery_mv": 4093}
packed = encode(r)
assert len(packed) == struct.calcsize(FMT) == 14
assert math.isclose(decode(packed)["temperature"], r["temperature"])
print(packed.hex())datagen.py içinde encode(reading) çıktısını testmqtt/datagen/sim-001/bin gibi kardeş bir topic'e publish edin ve her payload'ı hex olarak gösteren mosquitto_sub -t 'testmqtt/datagen/+/bin' -F '%t %x' ile hat üzerinde kontrol edin. Bayt sırasını her zaman açıkça belirtin (burada <): ESP32 ve çoğu ARM çipi little-endian'dır, ancak ağ bayt sırasını varsayan bir decoder sessizce anlamsız sonuçlar üretir. Elle yazılmış bir struct yerine kendini tanımlayan bir binary format istiyorsanız CBOR ve MessagePack karşılaştırmasına bakın.
MQTT paket boyutu hesaplayıcı, sabit başlık ve topic dahil edildiğinde iki kodlamanın mesaj başına maliyetini gösterir.
Negatif testler için bozuk payload'lar
Gerçekçi veri mutlu yolu test eder. Tüketicilerin, gerçek filoların er ya da geç gönderdiği payload'lara da dayanması gerekir — bir firmware hatası, kesilmiş bir mesaj, yanlış kodlama. Bilinen hatalı payload'lardan oluşan bir set publish edin ve subscriber'ınızın bunları loglayıp çalışmaya devam ettiğini doğrulayın:
import json, os
import paho.mqtt.client as mqtt
BAD = {
"empty": b"",
"truncated": b'{"temperature": 21.',
"string-number": b'{"temperature": "21.4"}',
"nan": b'{"temperature": NaN}', # Python'un json modülü kabul eder, JSON.parse etmez
"huge-number": b'{"temperature": 1e999}',
"not-utf8": '{"site": "café"}'.encode("latin-1"),
"wrong-format": bytes.fromhex("01a0e4f1680a2a005c08"), # JSON beklenen yerde binary
"oversized": json.dumps({"temperature": 21.4, "pad": "x" * 200_000}).encode(),
}
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
c.username_pw_set(os.environ.get("MQTT_USER"), os.environ.get("MQTT_PASS"))
c.connect("broker.connectmq.com", 1883)
c.loop_start()
for name, payload in BAD.items():
c.publish("testmqtt/datagen/sim-bad/telemetry", payload, qos=1).wait_for_publish()
print("sent", name, len(payload), "bytes")
c.loop_stop()
c.disconnect()Her durum gerçek bir arıza biçimine karşılık gelir: parser exception fırlatır, sayısal bir sütuna string sızar, NaN bir veritabanı insert'ünü bozar ya da büyük bir payload broker'ın maksimum paket boyutunu aşar. MQTT JSON parse hataları her birinin tüketici tarafında nasıl ele alınacağını anlatıyor.
Aynı fikir Node.js ile
MQTT.js 5 ile üretici birkaç satırdan ibarettir. Box–Muller yöntemi Math.random() değerini Gauss gürültüsüne dönüştürür:
import mqtt from "mqtt";
const client = await mqtt.connectAsync("mqtt://broker.connectmq.com:1883", {
username: process.env.MQTT_USER,
password: process.env.MQTT_PASS,
});
const gauss = () => Math.sqrt(-2 * Math.log(1 - Math.random())) * Math.cos(2 * Math.PI * Math.random());
let drift = 0, seq = 0;
setInterval(async () => {
const t = Date.now() / 1000;
drift = 0.999 * drift + gauss() * 0.02;
const day = Math.sin(2 * Math.PI * ((t % 86400) / 86400 - 0.375));
const temperature = Math.round((21 + 2.5 * day + drift + gauss() * 0.15) * 10) / 10;
const payload = JSON.stringify({ ts: Date.now(), seq: seq++, temperature });
await client.publishAsync("testmqtt/datagen/js-001/telemetry", payload, { qos: 1 });
}, 2000);Sentetik veri ve kaydedilmiş veri
Sentetik veri başlangıçta ve kolayca üretemeyeceğiniz senaryolar için idealdir; örneğin 10.000 cihaz ya da gece 3'te arızalanan bir sensör. Pilot bir filo çalışmaya başladığında gerçek trafiğini kaydedip onu da tekrar oynatın — MQTT replay testi nasıl yapılacağını gösteriyor. En iyi test paketleri ikisini birlikte kullanır: gerçekçilik için kaydedilmiş veri, ölçek ve uç durumlar için üretilmiş veri. Bu üreticiyi her biri kendi bağlantısına ve Last Will'ine sahip çok sayıda bağımsız sanal cihaza dönüştürmek için Python ile MQTT cihaz simülatörü yazısına bakın.
Üretilen telemetriyi canlı izleyin
İyi bir test verisi üreticisi bir öğleden sonranıza mal olur ve kurduğunuz her dashboard'da, kuralda ve veritabanı şemasında kendini geri öder. Sinyali dürüstçe modelleyin, seed kullanın, tam olarak cihaz gibi kodlayın ve yanında bozuk payload'lardan oluşan bir klasör tutun.
Sıkça sorulan sorular
Test verisi olarak neden rastgele sayılar publish etmiyoruz?
Düzgün dağılımlı rastgele değerlerin eğilimi, günlük döngüsü ve örnekler arası korelasyonu yoktur; bu yüzden asıl test etmeniz gereken mantığı asla tetiklemezler: eşik değerleri, yumuşatma, anomali tespiti, ölü bantlar veya grafik ölçekleme. Taban değer, döngü, gürültü, kayma ve ara sıra sıçramalardan oluşan bir model gerçek bir sensöre benzer ve bu mantığı çalıştırır.
MQTT test verisi JSON mu yoksa binary mi olmalı?
Cihazlarınızın gerçekten gönderdiği formatı kullanın. JSON okunması ve hata ayıklaması en kolay olandır; paketlenmiş bir binary struct ise birkaç kat daha küçüktür ve kısıtlı veya hücresel cihazlarda yaygındır. İyi bir üretici aynı ölçümden ikisini de üretir, böylece iki decoder'ı da test edebilirsiniz.
Üretilen test verisini nasıl tekrarlanabilir yaparım?
Her cihazın rastgele sayı üretecine bir seed verin, örneğin cihazın indeksini. Aynı seed aynı ölçüm dizisini üretir; böylece başarısız bir test birebir tekrar oynatılabilir ve bir benchmark'ın iki çalıştırması aynı koşulları karşılaştırır.
