Eclipse Tahu: implementaciones de referencia de Sparkplug en Java, Python y C

Eclipse Tahu es la implementación de referencia de código abierto de la especificación Sparkplug, con bibliotecas y ejemplos en Java, Python y C. En esta guía aprenderás qué hace cada parte del repositorio de Tahu, cómo compilarlo, los escollos con los que se tropieza la gente la primera vez y cómo conectar un edge node basado en Tahu a un broker MQTT real y ver llegar sus births y sus datos.

Qué es Eclipse Tahu y qué no es

Sparkplug se gestiona en la Eclipse Foundation, y Tahu es el proyecto que convierte la especificación en código funcional. Ofrece tres cosas:

• La definición Protobuf de los payloads Sparkplug B, sparkplug_b.proto, junto con el código generado a partir de ella para cada lenguaje.
• Bibliotecas auxiliares que construyen y analizan payloads: crear payloads BIRTH y DEATH, añadir métricas con los tipos de datos correctos, gestionar los números de secuencia.
• Ejemplos y herramientas de compatibilidad que muestran un edge node y una aplicación host mínimos, y ayudan a los fabricantes a comprobar sus implementaciones frente a la especificación.

Lo que Eclipse Tahu no es: no es un broker MQTT, ni un sistema SCADA, ni un producto gateway. Todos los ejemplos de Tahu se conectan como clientes MQTT corrientes a un broker que proporcionas tú. Esa separación es precisamente la esencia de Sparkplug. El edge node y el host hablan Sparkplug, y el broker MQTT que hay en medio solo tiene que ser un servidor MQTT que cumpla el estándar.

Tahu es además código de referencia. Está escrito para ser claro y correcto, no para incorporarse sin cambios a un gateway de producción. Muchos equipos usan directamente las definiciones Protobuf y las funciones auxiliares de payload y escriben su propia gestión de sesión a su alrededor, que es exactamente lo que haremos más abajo.

Un recorrido por el repositorio

El código fuente está en el repositorio de Eclipse Tahu en GitHub. Clónalo para explorarlo:

git clone https://github.com/eclipse-tahu/tahu.git
cd tahu
ls

La estructura está organizada por lenguaje. Los nombres de las carpetas han cambiado entre versiones, así que toma esta tabla como un mapa y no como un listado exacto, y consulta el README de la rama que clones.

Java

La implementación en Java es la más completa. Contiene clases de modelo para payloads y métricas, codificadores y decodificadores, y ejemplos de alto nivel de edge node y host. Se compila con Maven:

cd java
mvn clean install -DskipTests

Los artefactos resultantes pueden usarse desde tu propio proyecto Maven o Gradle. Java es la opción natural si estás escribiendo una aplicación host o integrando con un SCADA o un MES basado en la JVM.

Python

La carpeta de Python contiene dos archivos que usarás constantemente. sparkplug_b_pb2.py es el módulo Protobuf generado, y sparkplug_b.py añade funciones auxiliares como getNodeBirthPayload, getDeviceBirthPayload, getDdataPayload, getNodeDeathPayload y addMetric, además de una enumeración MetricDataType. La carpeta de ejemplos muestra un edge node sencillo que publica births, envía datos y responde a comandos.

C

La implementación en C codifica y decodifica payloads Sparkplug usando nanopb, un pequeño runtime de Protobuf diseñado para microcontroladores. Es el punto de partida si necesitas Sparkplug en un gateway con Linux embebido o en un firmware con poca memoria. El cliente de ejemplo combina el codificador con una biblioteca MQTT; se compila con el Makefile proporcionado después de instalar las cabeceras de desarrollo de esa biblioteca.

Escollos habituales con Eclipse Tahu

La mayoría de los problemas del primer día con Tahu son problemas del entorno, no de Sparkplug.

1. paho-mqtt 2.x rompe los ejemplos de Python antiguos. Los ejemplos de Python de Tahu se escribieron para paho-mqtt 1.x. Con paho-mqtt 2.x, mqtt.Client() exige una versión de la API de callbacks como primer argumento. O bien fija pip install "paho-mqtt<2" para los ejemplos, o bien crea el cliente como mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, client_id=...) para conservar las firmas de callback antiguas.
2. Incompatibilidad del runtime de Protobuf. Un sparkplug_b_pb2.py generado por un protoc antiguo puede fallar con un paquete protobuf nuevo con el mensaje "Descriptors cannot be created directly". Regenéralo con protoc --python_out=. sparkplug_b.proto usando un compilador actual.
3. Rutas de importación. Los ejemplos importan las funciones auxiliares principales insertando una ruta relativa. Si copias un ejemplo a otro lugar, copia sparkplug_b.py y sparkplug_b_pb2.py junto a él.
4. Parámetros de conexión fijos en el código. Los ejemplos apuntan a localhost por defecto. Edita el host del broker, el puerto, el group ID, el edge node ID y las credenciales al principio del archivo.
5. Group IDs compartidos en brokers públicos. Los topics Sparkplug siempre empiezan por spBv1.0, así que en un broker público usa un group ID aleatorio para no chocar con las pruebas de otras personas.

Conectar un edge node Tahu a un broker MQTT

El siguiente script usa las funciones auxiliares de Python de Tahu con paho-mqtt 2.x. Copia sparkplug_b.py y sparkplug_b_pb2.py de la carpeta core de Python de Tahu en un directorio vacío, añade este archivo junto a ellos e instala las dependencias con pip install "paho-mqtt>=2.0" protobuf.

import random
import time

import paho.mqtt.client as mqtt
from sparkplug_b import (
    MetricDataType,
    addMetric,
    getDdataPayload,
    getDeviceBirthPayload,
    getNodeBirthPayload,
    getNodeDeathPayload,
)

BROKER = "broker.connectmq.com"
PORT = 1883
GROUP = "tahu-demo-5mz8"       # random group ID for the public broker
NODE = "tahu-edge-01"
DEVICE = "pump-01"
CLIENT_ID = f"tahu-edge-{random.randint(1000, 9999)}"

def t(msg_type, device=None):
    base = f"spBv1.0/{GROUP}/{msg_type}/{NODE}"
    return f"{base}/{device}" if device else base

def publish_births(client):
    nbirth = getNodeBirthPayload()
    addMetric(nbirth, "Node Control/Rebirth", None, MetricDataType.Boolean, False)
    client.publish(t("NBIRTH"), bytearray(nbirth.SerializeToString()), 0, False)

    dbirth = getDeviceBirthPayload()
    addMetric(dbirth, "Flow", None, MetricDataType.Float, 12.5)
    addMetric(dbirth, "Pressure", None, MetricDataType.Float, 3.1)
    addMetric(dbirth, "Running", None, MetricDataType.Boolean, True)
    client.publish(t("DBIRTH", DEVICE), bytearray(dbirth.SerializeToString()), 0, False)
    print("NBIRTH and DBIRTH published")

def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe(t("NCMD"))
    publish_births(client)

def on_message(client, userdata, msg):
    print("command received on", msg.topic)
    publish_births(client)   # treat any NCMD as a rebirth request in this demo

# The death payload must be built before the birth so both carry the same bdSeq
death = getNodeDeathPayload()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_message = on_message
client.will_set(t("NDEATH"), bytearray(death.SerializeToString()), 1, False)
client.connect(BROKER, PORT, 60)
client.loop_start()

try:
    while True:
        time.sleep(3)
        ddata = getDdataPayload()
        addMetric(ddata, "Flow", None, MetricDataType.Float, round(random.uniform(11, 14), 2))
        addMetric(ddata, "Pressure", None, MetricDataType.Float, round(random.uniform(2.8, 3.4), 2))
        client.publish(t("DDATA", DEVICE), bytearray(ddata.SerializeToString()), 0, False)
        print("DDATA published")
except KeyboardInterrupt:
    client.loop_stop()
    client.disconnect()

El orden de getNodeDeathPayload() y getNodeBirthPayload() importa porque las funciones auxiliares de Tahu gestionan internamente el contador bdSeq, y el NDEATH registrado como will debe llevar el mismo valor que el NBIRTH que le sigue. Si tu copia de las funciones auxiliares se comporta de otra forma, inspecciona la métrica bdSeq en ambos payloads y ajústalo.

Ejecuta el script y suscríbete al grupo en otra terminal:

mosquitto_sub -h broker.connectmq.com -p 1883 -i tahu-watch-5mz8 -t "spBv1.0/tahu-demo-5mz8/#" -v

Verás llegar los topics con payloads binarios. Para leer las métricas, abre el tráfico en un portal de ConnectMQ: el explorador de topics en vivo decodifica los payloads Sparkplug B, ya sean Protobuf o JSON, en métricas con nombre, tipo de datos y valor. Así confirmas rápidamente que Tahu ha generado lo que esperabas antes de que intervenga un host SCADA. El broker que acabas de usar es el servidor MQTT público gratuito; es para pruebas y prototipos y es legible por cualquiera, así que mantén fuera de él los nombres y valores de tags reales.

Del código de referencia a un edge node real

Una vez que el ejemplo funciona, un edge node de producción necesita más de lo que Tahu ofrece de serie:

• bdSeq persistente. Guarda el contador en disco para que se incremente entre reinicios del proceso.
• Report by exception. Publica solo los valores que cambian, con bandas muertas para las señales analógicas ruidosas.
• Store and forward. Almacena los datos en búfer durante las caídas del broker y reenvíalos con el indicador de histórico.
• Alias. Activa los alias de métricas de Sparkplug en enlaces limitados.
• Conciencia del host primario. En Sparkplug 3.0, los edge nodes pueden esperar el mensaje STATE del host primario antes de publicar sus births.
• TLS y credenciales. Cambia al puerto 8883 con credenciales por cliente.

Si quieres un script totalmente autónomo sin las funciones auxiliares de Tahu, nuestro simulador de edge node Sparkplug B construye los payloads directamente a partir del módulo Protobuf generado y gestiona de forma explícita los números de secuencia y los rebirths. Para los detalles internos del payload, consulta tipos de datos de Sparkplug B. La especificación la publica el grupo de trabajo Eclipse Sparkplug.

Para producción, conecta los nodos basados en Tahu a un broker privado. En ConnectMQ, crea un broker en el portal, copia los YOUR_USERNAME, YOUR_PASSWORD y YOUR_CLIENT_ID generados, establece el client ID en el script, añade client.username_pw_set(...) y client.tls_set() y conéctate al puerto 8883. ConnectMQ puede entonces republicar los datos Sparkplug B decodificados en topics ISA-95 como connectmq/factory1/production/line1/cnc1/telemetry, que los consumidores que no usan Sparkplug pueden leer como datos simples. La arquitectura general se trata en nuestra guía de MQTT para IoT industrial.

Conecta Eclipse Tahu a ConnectMQ

Tahu te da payloads correctos; aun así necesitas un broker fiable entre el edge y el host. ConnectMQ es un broker MQTT gestionado con MQTT 3.1.1 y 5.0, TLS en el puerto 8883, decodificación de Sparkplug B y republicación ISA-95. Crea un broker en el plan Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y apunta tu edge node Tahu a él en pocos minutos.

Preguntas frecuentes

¿Eclipse Tahu es un broker?

No. Eclipse Tahu proporciona bibliotecas y ejemplos de Sparkplug. Los edge nodes y hosts construidos con él se conectan como clientes MQTT a cualquier broker MQTT que cumpla el estándar.

¿Qué lenguaje de Tahu debería usar?

Usa Java para aplicaciones host y gateways en la JVM, Python para scripts, simuladores y pruebas, y C para gateways embebidos con poca memoria.

¿Por qué fallan los ejemplos de Python de Tahu con paho-mqtt 2.x?

paho-mqtt 2.x exige una versión de la API de callbacks al crear el cliente. Fija paho-mqtt por debajo de 2.0 para los ejemplos o pasa mqtt.CallbackAPIVersion.VERSION1 para conservar sus firmas de callback.

¿Tahu está listo para producción?

Es código de referencia centrado en la corrección. Muchos equipos usan en producción sus definiciones Protobuf y sus funciones auxiliares de payload, añadiendo su propia gestión de sesión, almacenamiento en búfer y seguridad.