Python MQTT JSON: publicar datos de sensores con marcas de tiempo y versiones de esquema

Los payloads Python MQTT JSON son la forma habitual de mover datos de sensores entre dispositivos, dashboards y bases de datos, pero pequeños errores como marcas de tiempo locales, valores NaN o campos sin versionar provocan grandes quebraderos de cabeza más adelante. En este tutorial construirás un publicador en Python que serializa las lecturas a JSON compacto y válido con marcas de tiempo UTC, números de secuencia y una versión de esquema, y las envía a broker.connectmq.com con metadatos content type de MQTT 5, además de un suscriptor que valida lo que recibe. Terminarás con un formato de payload que podrás seguir usando a medida que crezca tu proyecto.

¿Por qué JSON para los payloads MQTT?

A MQTT no le importa qué hay dentro de un mensaje; el payload son solo bytes, hasta el límite de tamaño del broker. Esa libertad implica que debes elegir un formato. JSON es la opción habitual porque:

• cualquier lenguaje y herramienta puede interpretarlo, desde Node-RED y Telegraf hasta bases de datos con columnas JSON;
• es legible por humanos, así que puedes depurar con mosquitto_sub o con un explorador de topics;
• es autodescriptivo, así que añadir un campo no rompe a los consumidores antiguos.

La contrapartida es el tamaño. JSON repite los nombres de los campos en cada mensaje, lo que importa en enlaces limitados. Para la mayoría de dispositivos Wi-Fi y Ethernet ese coste es despreciable y ganan las ventajas de depuración. Si más adelante necesitas algo más compacto, los formatos binarios como Protobuf (el que usa Sparkplug B) son una opción, como se explica en la guía de Sparkplug B y Unified Namespace.

Diseño del payload

Un buen payload de sensor responde a cinco preguntas: qué versión del formato es, qué dispositivo lo envió, cuándo se midió, qué posición ocupa en la secuencia y cuáles son los valores con sus unidades. Este es el formato que usa el tutorial:

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

Las decisiones de diseño detrás de cada campo:

Poner las unidades en los nombres de los campos evita objetos anidados {"value": 21.4, "unit": "C"} para cada lectura, lo que mantiene los payloads pequeños y las consultas sencillas. Elijas la convención que elijas, aplícala de forma coherente en todos los dispositivos.

Construir un publicador Python MQTT JSON

Instala paho-mqtt y guarda el publicador como json_publisher.py. Cambia k7f3q2 por tu propia cadena aleatoria para no mezclarte con otros usuarios del servidor MQTT público gratuito, pensado para pruebas y no para datos de producción:

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]:
    """Redondea floats y convierte NaN o infinito en None, que se convierte en null en JSON."""
    if value is None or math.isnan(value) or math.isinf(value):
        return None
    return round(value, digits)


def read_sensor() -> tuple:
    # Sustitúyelo por código real del sensor. De vez en cuando simula una lectura fallida.
    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,
)
# Broker privado de ConnectMQ: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); puerto 8883
client.connect(BROKER, PORT, keepalive=60)
client.loop_start()

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

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()

Ejecuta python json_publisher.py. Cada dos segundos se imprime una línea como esta, y aproximadamente una de cada veinte lecturas muestra valores null donde falló el sensor simulado:

 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}

Por qué importa cada detalle

Separadores compactos

separators=(",", ":") elimina los espacios que json.dumps añade por defecto. En un payload de este tamaño eso ahorra en torno a un diez por ciento, gratis. El formateo bonito es cosa de tus herramientas de depuración, no de lo que viaja por la red.

allow_nan=False y la trampa de NaN

json.dumps de Python escribe sin problemas NaN e Infinity por defecto, pero no son JSON válido. Muchos consumidores, incluidos los parsers estrictos de JavaScript, Go y las bases de datos, rechazan el mensaje entero. Los sensores producen NaN continuamente cuando falla una lectura, así que el helper clean() los convierte en None, y allow_nan=False hace que cualquier valor que se cuele lance un error en tu lado en lugar de corromper los datos aguas abajo.

Marcas de tiempo UTC con milisegundos

Las marcas de tiempo sin zona horaria son ambiguas, y las horas locales saltan dos veces al año con el cambio de horario. ISO 8601 en UTC con sufijo Z se ordena correctamente como cadena y se interpreta en todas partes. Si tu dispositivo no tiene reloj de tiempo real, sincronízalo con NTP antes de publicar, o deja fuera ts y que el consumidor registre la hora de recepción.

Números de secuencia

seq cuesta unos pocos bytes y da a los consumidores una forma de detectar mensajes perdidos (un salto de 41 a 43) y duplicados (QoS 1 puede entregar un mensaje dos veces). Reinícialo al arrancar y deja que los consumidores interpreten una caída a cero como un reinicio.

Content type de MQTT 5

Las propiedades ContentType y PayloadFormatIndicator indican a cualquier consumidor MQTT 5 que el payload es JSON en UTF-8 sin tener que adivinarlo. Son opcionales, pero facilitan el manejo de herramientas genéricas y árboles de topics con formatos mixtos. Nuestro artículo sobre las funciones de MQTT 5 cubre el resto de propiedades.

Versionar tu esquema

Tu payload cambiará: nuevos sensores, campos renombrados, unidades distintas. Planifícalo desde el primer mensaje:

1. Los cambios aditivos (un nuevo campo opcional) mantienen el mismo v. Los consumidores bien escritos ignoran los campos que no conocen.
2. Los cambios incompatibles (renombrar, eliminar, cambiar unidades o tipos) incrementan v a 2.
3. Durante una migración, publica ambas versiones o mantén a los consumidores capaces de leer las dos, distinguiendo por v.
4. Nunca reutilices un número de versión para un formato distinto.
5. Documenta cada versión junto a tu estructura de topics para que todos usen el mismo contrato.

Un suscriptor que valida

Los consumidores nunca deben confiar ciegamente en los payloads. Este suscriptor interpreta el mensaje, comprueba la versión y los campos obligatorios y rechaza todo lo demás sin caerse:

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()

Inspeccionar mensajes JSON

Para comprobaciones rápidas contra el broker público, canaliza el cliente de línea de comandos a jq y obtendrás una salida formateada y con colores:

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

En un broker privado, el explorador de topics en vivo del portal de ConnectMQ muestra cada topic de tu árbol con su último payload, así que puedes confirmar nombres de campos, manejo de nulos y marcas de tiempo sin escribir un consumidor.

Límites de tamaño y agrupación

Vigila el tamaño de los mensajes. El broker público acepta hasta 64 KB por mensaje y el plan Serverless gratuito hasta 256 KB. Las lecturas individuales quedan muy por debajo, pero si agrupas, por ejemplo enviando un minuto de muestras como array, comprueba len(payload) antes de publicar y divide los lotes grandes. Otra forma de reducir el tráfico es publicar solo cuando un valor cambia más de un umbral fijado, más un latido periódico para que los consumidores distingan un sensor tranquilo de uno muerto. Ese enfoque suele reducir drásticamente el volumen de mensajes para valores que cambian despacio, como la temperatura de una habitación, sin perder información relevante.

Siguientes pasos

Guarda estos payloads con el tutorial de MQTT a SQLite en Python, cuyas consultas json_extract funcionan directamente con este formato, o represéntalos en vivo con el dashboard MQTT en Streamlit. Para servicios asíncronos, el mismo código de payload encaja en un cliente aiomqtt con asyncio. El resumen de bibliotecas cliente MQTT muestra cómo interpretan el mismo JSON otros lenguajes, y la documentación del módulo json de Python cubre todas las opciones del codificador.

Empieza con ConnectMQ

ConnectMQ es un broker MQTT gestionado con soporte de MQTT 5 para propiedades como el content type, un explorador de topics en vivo para inspeccionar payloads JSON y TLS en el puerto 8883. Crea tu prototipo en broker.connectmq.com y pasa después tus dispositivos a un broker MQTT en la nube privado. Regístrate en el plan Serverless gratuito, sin tarjeta de crédito.

Preguntas frecuentes

¿Cómo envío JSON por MQTT en Python?

Serializa tus datos con json.dumps(), codifícalos a bytes UTF-8 y pásalos como payload a client.publish(). En el lado receptor, llama a json.loads(msg.payload).

¿La marca de tiempo debe ir en el payload o la añade el consumidor?

Incluye una marca de tiempo UTC de la medición en el payload cuando el dispositivo tenga un reloj fiable. Si no lo tiene, deja que el consumidor registre la hora de recepción y omite ts.

¿Por qué mi payload JSON contiene NaN?

json.dumps de Python escribe NaN para los valores float NaN por defecto, lo cual es JSON no válido. Convierte NaN en None antes de codificar y fija allow_nan=False para detectar los que queden.

¿Dónde debe ir la versión del esquema, en el payload o en el topic?

Ambas opciones funcionan. Un campo v en el payload es más sencillo y mantiene estables los topics; un segmento de versión en el topic permite a los consumidores suscribirse solo a las versiones que entienden.