Python MQTT JSON: Zaman Damgası ve Şema Sürümüyle Sensör Verisi Publish Etme

Python MQTT JSON payload'ları, sensör verisini cihazlar, paneller ve veritabanları arasında taşımanın varsayılan yoludur; ancak yerel zaman damgaları, NaN değerleri veya sürümlenmemiş alanlar gibi küçük hatalar ileride büyük baş ağrılarına yol açar. Bu eğitimde ölçümleri UTC zaman damgaları, sıra numaraları ve bir şema sürümüyle kompakt ve geçerli JSON'a serileştiren ve bunları MQTT 5 content type meta verisiyle broker.connectmq.com adresine gönderen bir Python publisher, ayrıca aldığını doğrulayan bir subscriber oluşturacaksınız. Sonunda projeniz büyüdükçe kullanmaya devam edebileceğiniz bir payload formatına sahip olacaksınız.

MQTT Payload'ları için Neden JSON?

MQTT bir mesajın içinde ne olduğuyla ilgilenmez; payload, broker'ın boyut sınırına kadar yalnızca baytlardan ibarettir. Bu özgürlük, bir format seçmeniz gerektiği anlamına gelir. JSON genellikle tercih edilir, çünkü:

• Node-RED ve Telegraf'tan JSON sütunlu veritabanlarına kadar her dil ve araç onu ayrıştırabilir;
• insan tarafından okunabilir, bu yüzden mosquitto_sub veya bir topic gezgini ile hata ayıklayabilirsiniz;
• kendini tanımlar, bu yüzden bir alan eklemek eski tüketicileri bozmaz.

Bedeli boyuttur. JSON alan adlarını her mesajda tekrarlar ve bu kısıtlı bağlantılarda önem taşır. Çoğu Wi-Fi ve Ethernet cihazı için bu maliyet önemsizdir ve hata ayıklama avantajları ağır basar. Daha sonra daha kompakt bir şeye ihtiyaç duyarsanız, Sparkplug B ve Unified Namespace rehberinde ele alındığı gibi Protobuf (Sparkplug B'nin kullandığı) gibi binary formatlar bir seçenektir.

Payload'ı Tasarlamak

İyi bir sensör payload'ı beş soruyu yanıtlar: bu formatın hangi sürümü, hangi cihaz gönderdi, ne zaman ölçüldü, dizideki kaçıncı mesaj ve birimleriyle birlikte değerler neler. Bu eğitimin kullandığı format şudur:

{
  "v": 1,
  "device": "room1",
  "ts": "2026-10-16T08:15:30.123Z",
  "seq": 42,
  "temperature_c": 21.4,
  "humidity_pct": 45.2,
  "battery_v": null
}

Her alanın arkasındaki tasarım kararları:

Birimleri alan adlarına koymak, her ölçüm için iç içe {"value": 21.4, "unit": "C"} nesnelerinden kaçınır; bu da payload'ları küçük, sorguları basit tutar. Hangi kuralı seçerseniz seçin, tüm cihazlarda tutarlı şekilde uygulayın.

Bir Python MQTT JSON Publisher Oluşturmak

paho-mqtt'yi kurun ve publisher'ı json_publisher.py olarak kaydedin. Ücretsiz herkese açık MQTT broker'ın diğer kullanıcılarından ayrı kalmak için k7f3q2 değerini kendi rastgele dizginizle değiştirin:

pip install "paho-mqtt>=2.0"

import json
import math
import random
import time
import uuid
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Optional

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BROKER = "broker.connectmq.com"
PORT = 1883
PREFIX = "connectmq/demo/k7f3q2"
DEVICE = "room1"
SCHEMA_VERSION = 1


@dataclass
class Reading:
    device: str
    ts: str
    seq: int
    temperature_c: Optional[float]
    humidity_pct: Optional[float]
    battery_v: Optional[float] = None
    v: int = SCHEMA_VERSION


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def clean(value: Optional[float], digits: int = 2) -> Optional[float]:
    """Float'ları yuvarlar ve NaN veya sonsuzluğu JSON null olacak None'a çevirir."""
    if value is None or math.isnan(value) or math.isinf(value):
        return None
    return round(value, digits)


def read_sensor() -> tuple:
    # Gerçek sensör koduyla değiştirin. Ara sıra başarısız bir okumayı simüle eder.
    if random.random() < 0.05:
        return float("nan"), float("nan")
    return random.uniform(20, 25), random.uniform(40, 55)


def to_payload(reading: Reading) -> bytes:
    return json.dumps(asdict(reading), separators=(",", ":"), allow_nan=False).encode("utf-8")


client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id=f"json-pub-{uuid.uuid4().hex[:8]}",
    protocol=mqtt.MQTTv5,
)
# Özel ConnectMQ broker: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); port 8883
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

props = Properties(PacketTypes.PUBLISH)
props.ContentType = "application/json"
props.PayloadFormatIndicator = 1  # payload UTF-8 metindir

topic = f"{PREFIX}/sensors/{DEVICE}/telemetry"
seq = 0
try:
    while True:
        temperature, humidity = read_sensor()
        reading = Reading(
            device=DEVICE,
            ts=utc_now_iso(),
            seq=seq,
            temperature_c=clean(temperature),
            humidity_pct=clean(humidity, 1),
        )
        payload = to_payload(reading)
        client.publish(topic, payload, qos=1, properties=props)
        print(f"{len(payload):>4} bytes  {payload.decode()}")
        seq += 1
        time.sleep(2)
except KeyboardInterrupt:
    pass
finally:
    client.disconnect()
    client.loop_stop()

python json_publisher.py komutunu çalıştırın. Her iki saniyede bir buna benzer bir satır yazdırılır ve yaklaşık yirmi ölçümden biri, simüle edilen sensörün başarısız olduğu yerlerde null değerler gösterir:

 116 bytes  {"device":"room1","ts":"2026-10-16T08:15:30.123Z","seq":42,"temperature_c":21.4,"humidity_pct":45.2,"battery_v":null,"v":1}

Her Ayrıntı Neden Önemli

Kompakt ayırıcılar

separators=(",", ":"), json.dumps fonksiyonunun varsayılan olarak eklediği boşlukları kaldırır. Bu boyuttaki bir payload'da bu, bedavaya yaklaşık yüzde on tasarruf sağlar. Okunaklı biçimlendirme hata ayıklama araçlarınıza aittir, hatta değil.

allow_nan=False ve NaN tuzağı

Python'un json.dumps fonksiyonu varsayılan olarak NaN ve Infinity değerlerini seve seve yazar, ancak bunlar geçerli JSON değildir. JavaScript, Go ve veritabanlarındaki katı ayrıştırıcılar dahil pek çok tüketici mesajın tamamını reddeder. Sensörler bir okuma başarısız olduğunda sürekli NaN üretir; bu yüzden clean() yardımcısı bunları None değerine çevirir ve allow_nan=False, gözden kaçan herhangi bir değerin aşağı akıştaki veriyi bozmak yerine sizin tarafınızda hata fırlatmasını sağlar.

Milisaniyeli UTC zaman damgaları

Saat dilimi olmayan zaman damgaları belirsizdir ve yerel saatler yaz saati uygulaması nedeniyle yılda iki kez sıçrar. Z sonekli UTC'de ISO 8601 bir dizgi olarak doğru sıralanır ve her yerde ayrıştırılır. Cihazınızın gerçek zamanlı saati yoksa publish etmeden önce onu NTP ile senkronize edin veya ts alanını çıkarıp alınma zamanını tüketicinin kaydetmesine izin verin.

Sıra numaraları

seq birkaç bayta mal olur ve tüketicilere eksik mesajları (41'den 43'e bir sıçrama) ve tekrarları (QoS 1 bir mesajı iki kez iletebilir) fark etmenin bir yolunu sunar. Yeniden başlatmada sıfırlayın ve tüketicilerin sıfıra düşüşü bir yeniden başlatma olarak değerlendirmesine izin verin.

MQTT 5 content type

ContentType ve PayloadFormatIndicator properties, herhangi bir MQTT 5 tüketicisine payload'ın tahmin gerektirmeden UTF-8 JSON olduğunu söyler. İsteğe bağlıdırlar, ancak genel araçları ve karışık formatlı topic ağaçlarını yönetmeyi kolaylaştırırlar. MQTT 5 özellikleri yazımız diğer properties'i ele alır.

Şemanızı Sürümlemek

Payload'ınız değişecek: yeni sensörler, yeniden adlandırılan alanlar, farklı birimler. Bunu ilk mesajdan itibaren planlayın:

1. Eklemeli değişiklikler (yeni bir isteğe bağlı alan) aynı v değerini korur. İyi yazılmış tüketiciler bilmedikleri alanları yok sayar.
2. Kırıcı değişiklikler (yeniden adlandırma, kaldırma, birim veya tür değiştirme) v değerini 2'ye yükseltir.
3. Bir geçiş sırasında her iki sürümü de publish edin veya tüketicilerin v değerine göre ayrım yaparak her ikisini de okuyabilmesini sağlayın.
4. Bir sürüm numarasını asla farklı bir format için yeniden kullanmayın.
5. Herkesin aynı sözleşmeyi kullanması için her sürümü topic yapınızın yanında belgeleyin.

Doğrulama Yapan Bir Subscriber

Tüketiciler payload'lara asla körü körüne güvenmemelidir. Bu subscriber ayrıştırır, sürümü ve zorunlu alanları kontrol eder ve diğer her şeyi çökmeden reddeder:

import json
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/k7f3q2"
REQUIRED = {"v", "device", "ts", "seq"}


def parse_reading(raw: bytes):
    try:
        data = json.loads(raw)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"not valid JSON: {exc}")
    if not isinstance(data, dict) or not REQUIRED <= data.keys():
        raise ValueError("missing required fields")
    if data["v"] != 1:
        raise ValueError(f"unsupported schema version {data['v']}")
    return data


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/sensors/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    try:
        r = parse_reading(msg.payload)
    except ValueError as exc:
        print(f"Rejected message on {msg.topic}: {exc}")
        return
    print(f"{r['device']} #{r['seq']} at {r['ts']}: {r.get('temperature_c')} C, {r.get('humidity_pct')} %")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"json-sub-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

JSON Mesajlarını İncelemek

Herkese açık broker'a karşı hızlı kontroller için komut satırı istemcisinin çıktısını okunaklı ve renkli bir görünüm için jq aracına yönlendirin:

mosquitto_sub -h broker.connectmq.com -p 1883 -i json-inspect-k7f3q2 -t 'connectmq/demo/k7f3q2/sensors/+/telemetry' | jq .

Özel bir broker'da ConnectMQ portalının canlı topic gezgini, ağacınızdaki her topic'i en son payload'ıyla birlikte gösterir; böylece bir tüketici yazmadan alan adlarını, null işleme biçimini ve zaman damgalarını doğrulayabilirsiniz.

Boyut Sınırları ve Toplu Gönderim

Mesaj boyutunu takip edin. Herkese açık broker mesaj başına 64 KB'a, ücretsiz Serverless planı ise 256 KB'a kadar kabul eder. Tekil ölçümler bunun çok altındadır, ancak toplu gönderim yaparsanız, örneğin bir dakikalık örnekleri bir dizi olarak gönderirseniz, publish etmeden önce len(payload) değerini kontrol edin ve büyük toplu işleri bölün. Trafiği azaltmanın bir başka yolu, yalnızca bir değer belirlenen eşikten fazla değiştiğinde publish etmek ve tüketicilerin sessiz bir sensörü ölü bir sensörden ayırt edebilmesi için periyodik bir heartbeat eklemektir. Bu yaklaşım, oda sıcaklığı gibi yavaş değişen değerler için anlamlı hiçbir bilgiyi kaybetmeden mesaj hacmini çoğu zaman önemli ölçüde azaltır.

Sırada Ne Var

Bu payload'ları, json_extract sorguları bu formatla doğrudan çalışan Python ile MQTT'den SQLite'a eğitimiyle saklayın veya Streamlit MQTT paneli ile canlı grafiğe dökün. Asenkron servisler için aynı payload kodu bir aiomqtt asyncio istemcisine doğrudan eklenebilir. MQTT istemci kütüphanelerine genel bakış diğer dillerin aynı JSON'u nasıl ayrıştırdığını gösterir; Python json modülü dokümantasyonu ise her encoder seçeneğini ele alır.

ConnectMQ ile Başlayın

ConnectMQ; content type gibi properties için MQTT 5 desteği, JSON payload'larını incelemek için canlı bir topic gezgini ve 8883 portunda TLS sunan barındırılan bir MQTT broker hizmetidir. Prototipinizi broker.connectmq.com üzerinde geliştirin, ardından cihazlarınızı özel bir bulut MQTT broker'ına taşıyın. Ücretsiz Serverless planına kaydolun, kredi kartı gerekmez.

Sık Sorulan Sorular

Python'da MQTT üzerinden JSON nasıl gönderilir?

Verilerinizi json.dumps() ile serileştirin, UTF-8 baytlarına kodlayın ve client.publish() fonksiyonuna payload olarak verin. Alıcı tarafta json.loads(msg.payload) çağırın.

Zaman damgaları payload'da mı olmalı, yoksa tüketici tarafından mı eklenmeli?

Cihazın güvenilir bir saati varsa payload'a UTC ölçüm zaman damgası ekleyin. Yoksa alınma zamanını tüketicinin kaydetmesine izin verin ve ts alanını çıkarın.

JSON payload'ım neden NaN içeriyor?

Python'un json.dumps fonksiyonu varsayılan olarak float NaN değerleri için geçersiz JSON olan NaN yazar. Kodlamadan önce NaN'ı None değerine çevirin ve kalanları yakalamak için allow_nan=False ayarlayın.

Şema sürümü nereye konmalı, payload'a mı topic'e mi?

İkisi de olur. Payload'daki bir v alanı daha basittir ve topic'leri sabit tutar; topic'teki bir sürüm segmenti ise tüketicilerin yalnızca anladıkları sürümlere subscribe olmasını sağlar.