Cómo construir un sistema de trazabilidad en fabricación con eventos MQTT

Un sistema de trazabilidad en fabricación responde rápidamente y con evidencias a dos preguntas: qué entró en este producto y cómo se fabricó, y qué otros productos comparten el mismo material o las mismas condiciones sospechosas. Esta guía muestra cómo diseñar eventos de trazabilidad, publicarlos por MQTT desde cada estación, almacenarlos en una base de datos y consultarlos cuando llega una reclamación de un cliente o una retirada de un proveedor.

Lo que la trazabilidad exige de verdad

La trazabilidad a menudo se reduce a "imprimimos un número de serie". Un sistema útil registra mucho más:

1. Identidad. Un número de serie único para cada unidad, o un ID de lote o partida cuando el seguimiento por unidad no es práctico.
2. Genealogía. Qué números de serie de componentes o lotes de material se consumieron en qué unidad. Es lo que permite la trazabilidad hacia atrás y hacia delante.
3. Ruta. Por qué estaciones pasó la unidad, cuándo y en qué línea.
4. Datos de proceso. Los parámetros vigentes mientras se procesaba la unidad: par y ángulo de un atornillado, temperatura y tiempo de un curado, presión de un ajuste a presión.
5. Resultados. Resultados de pruebas e inspecciones, valores medidos e historial de reprocesos.

Con estos cinco registros puedes rastrear hacia atrás (desde una unidad fallida hasta sus materiales, máquinas y parámetros) y hacia delante (desde un lote de material sospechoso o una mala ventana de tiempo hasta cada unidad afectada). La trazabilidad hacia delante es lo que convierte una retirada de "todo lo fabricado en marzo" en una lista precisa.

Por qué funciona la trazabilidad basada en eventos

La trazabilidad tradicional se construye a menudo haciendo que cada estación escriba directamente en una base de datos central o en el MES mediante una interfaz a medida. Funciona hasta que añades una estación, cambias de proveedor o necesitas los datos en otro lugar. Un enfoque basado en eventos desacopla productores y consumidores: cada estación publica lo que ha ocurrido como un mensaje pequeño, y cualquier número de sistemas se suscribe.

MQTT encaja bien aquí. Estaciones y gateways edge publican con un protocolo ligero, QoS 1 garantiza que cada evento se entrega al menos una vez, y se pueden añadir nuevos consumidores, como un dashboard de calidad o un servicio de SPC en tiempo real, sin tocar las estaciones. Si el patrón es nuevo para ti, lee primero qué hace un broker MQTT.

Hay una regla que importa por encima de todas: el broker transporta los eventos, la base de datos los almacena. Un broker MQTT no es un archivo, así que la base de datos de trazabilidad es un suscriptor que persiste cada evento.

Diseño de los eventos

Mantén un vocabulario de eventos pequeño y explícito. Estos seis tipos cubren la mayor parte de la fabricación discreta:

Cada evento lleva un event_id (un UUID generado en el origen) y una marca de tiempo de un reloj sincronizado. El ID de evento hace que el almacenamiento sea idempotente: QoS 1 puede entregar ocasionalmente un mensaje dos veces, y la base de datos simplemente ignora el duplicado.

Estructura de topics

Usa una jerarquía de estilo ISA-95 para que los eventos queden agrupados de forma natural por planta, área, línea y estación:

acme/plant1/assembly/line2/station30/trace/event

Un suscriptor que necesite todos los eventos de trazabilidad de la planta se suscribe a acme/plant1/+/+/+/trace/event. El artículo de topics MQTT según ISA-95 trata esta nomenclatura en profundidad, y el portal de ConnectMQ reconoce estos topics y agrupa automáticamente los dispositivos por nivel.

Payloads de ejemplo

{
  "event_id": "8f1c2e7a-5b9d-4c11-9e0f-2a7d61c3b4f0",
  "event": "component_consumed",
  "ts": "2026-10-16T07:52:13.118Z",
  "serial": "PMP-26-0041877",
  "station": "station20",
  "part_number": "MOT-0.75KW",
  "component_serial": "M2609-118422"
}

{
  "event_id": "c3a90b5e-0d2f-47b8-8d6e-91f4a7e02b1c",
  "event": "station_result",
  "ts": "2026-10-16T07:55:40.902Z",
  "serial": "PMP-26-0041877",
  "station": "station30",
  "result": "pass",
  "values": {"torque_nm": 12.4, "angle_deg": 186.0, "leak_rate_ml_min": 0.8},
  "program": "SCREW-P7",
  "operator": "OP114"
}

Construir el suscriptor de trazabilidad

El servicio de Python siguiente se suscribe a todos los eventos de trazabilidad bajo un prefijo, los valida y los escribe en SQLite. Guarda el evento completo como JSON y extrae a columnas indexadas los campos que más se consultan. SQLite mantiene el ejemplo autocontenido; el mismo esquema funciona en PostgreSQL para producción. Usa paho-mqtt 2.x y el servidor MQTT público gratuito de ConnectMQ, pensado para pruebas y prototipos, no para datos de producción.

import json
import sqlite3
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a41b06"
db = sqlite3.connect("traceability.db", check_same_thread=False)
db.executescript("""
CREATE TABLE IF NOT EXISTS events (
    event_id TEXT PRIMARY KEY,
    event TEXT NOT NULL,
    ts TEXT NOT NULL,
    serial TEXT NOT NULL,
    station TEXT,
    topic TEXT,
    payload TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_serial ON events(serial);
CREATE INDEX IF NOT EXISTS ix_station_ts ON events(station, ts);
CREATE TABLE IF NOT EXISTS genealogy (
    serial TEXT NOT NULL,
    component TEXT NOT NULL,
    part_number TEXT,
    PRIMARY KEY (serial, component)
);
CREATE INDEX IF NOT EXISTS ix_component ON genealogy(component);
""")

REQUIRED = {"event_id", "event", "ts", "serial"}

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

def on_message(client, userdata, msg):
    try:
        e = json.loads(msg.payload)
    except json.JSONDecodeError:
        print("invalid JSON on", msg.topic)
        return
    if not REQUIRED.issubset(e):
        print("missing fields", e)
        return
    with db:
        db.execute("INSERT OR IGNORE INTO events VALUES (?,?,?,?,?,?,?)",
                   (e["event_id"], e["event"], e["ts"], e["serial"],
                    e.get("station"), msg.topic, json.dumps(e)))
        if e["event"] == "component_consumed":
            component = e.get("component_serial") or e.get("lot")
            db.execute("INSERT OR IGNORE INTO genealogy VALUES (?,?,?)",
                       (e["serial"], component, e.get("part_number")))

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="trace-writer-a41b06",
                     clean_session=False)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Dos ajustes son deliberados. clean_session=False con un client ID fijo pide al broker que mantenga la suscripción y encole los mensajes QoS 1 mientras el escritor está brevemente desconectado, dentro de los límites de sesión del broker. INSERT OR IGNORE sobre el ID de evento convierte las entregas duplicadas en operaciones sin efecto.

Publica un evento de prueba desde un terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i station30-a41b06 -q 1 \
  -t "connectmq/demo/a41b06/line2/station30/trace/event" \
  -m '{"event_id":"c3a90b5e-0d2f-47b8-8d6e-91f4a7e02b1c","event":"station_result","ts":"2026-10-16T07:55:40Z","serial":"PMP-26-0041877","station":"station30","result":"pass"}'

Consultar los datos de trazabilidad

El valor del sistema se ve cuando alguien hace una pregunta difícil. Estas tres consultas cubren la mayoría de las investigaciones.

Trazabilidad hacia atrás: todo lo que le ocurrió a una unidad.

SELECT ts, station, event, payload FROM events
WHERE serial = 'PMP-26-0041877' ORDER BY ts;

Trazabilidad hacia delante: todas las unidades que consumieron un lote de componente sospechoso.

SELECT serial FROM genealogy WHERE component = 'LOT-SEAL-2609-07';

Trazabilidad por ventana de tiempo: todas las unidades procesadas en una estación mientras estaba fuera de control.

SELECT DISTINCT serial FROM events
WHERE station = 'station30'
  AND ts BETWEEN '2026-10-16T07:30:00Z' AND '2026-10-16T08:15:00Z';

Combina las consultas hacia delante y por ventana de tiempo con los eventos de expedición y podrás decir a atención al cliente exactamente qué palés retener.

Lista de implementación

1. Define el punto de serialización. Decide dónde se crean los números de serie y haz que esa estación publique unit_created.
2. Escanea cada componente trazable. La genealogía solo es tan completa como la disciplina de escaneo en el montaje.
3. Sincroniza los relojes. Usa NTP en cada gateway y PC de estación; las consultas por ventana de tiempo dependen de ello.
4. Genera los ID de evento en el origen. Nunca en la base de datos, o los duplicados serán indistinguibles.
5. Publica con QoS 1. Usa sesiones persistentes para el escritor.
6. Almacena en búfer en el edge. Las estaciones deben encolar los eventos localmente mientras la red está caída y publicarlos al reconectarse.
7. Valida y vigila. Cuenta los eventos por estación y hora y alerta cuando una estación deja de publicar.
8. Define la retención. Acuerda cuánto tiempo deben conservarse los registros para tu sector y tus clientes, y planifica el almacenamiento de la base de datos en consecuencia.
9. Protege la canalización con credenciales privadas y TLS, siguiendo las buenas prácticas de seguridad MQTT.

Conectar la trazabilidad con la calidad

La trazabilidad es más potente cuando se vincula con datos de calidad en vivo. Cuando el control de calidad IoT en fabricación detecta un fallo, la base de datos de trazabilidad te dice qué lotes y máquinas anteriores intervinieron. Cuando el SPC marca un periodo fuera de control, la consulta por ventana de tiempo enumera todas las unidades que hay que poner en cuarentena. La guía de fabricación inteligente muestra cómo estos sistemas comparten una misma columna vertebral de datos, que también puede alimentar un gemelo digital 3D de la línea para seguir cada unidad estación por estación.

Alojamiento en producción

Los eventos de trazabilidad son registros de negocio, así que deben ir a un broker privado y autenticado, nunca a un servidor público de pruebas. Un broker MQTT gestionado proporciona endpoints TLS y credenciales emitidas para cada estación sin que tengas que operar infraestructura de broker. Las plantas con requisitos de aislamiento más estrictos pueden usar un clúster dedicado o ejecutar el broker en su propia cuenta en la nube con BYOC sin dejar de tenerlo gestionado.

Construye tu columna vertebral de trazabilidad en ConnectMQ

ConnectMQ ofrece MQTT 3.1.1 y 5.0 con TLS, sesiones persistentes, QoS 1 y 2 en los planes de pago, detección de topics ISA-95 y un explorador de topics en vivo para verificar los eventos de cada estación. Crea un broker en el plan gratuito Serverless en app.connectmq.com, sin tarjeta de crédito, y conecta hoy tu primera estación.

Preguntas frecuentes

¿Qué es un sistema de trazabilidad en fabricación?

Es un sistema que registra la identidad, los componentes, la ruta, los parámetros de proceso y los resultados de cada unidad o lote, para que puedas rastrear tanto hacia atrás desde un producto como hacia delante desde un material o un evento.

¿La trazabilidad debe ser por número de serie o por lote?

Usa números de serie por unidad cuando los productos son de alto valor o relevantes para la seguridad, e ID de lote o partida cuando el marcado por unidad no es práctico. Muchas plantas combinan ambos, con unidades serializadas que consumen materiales trazados por lote.

¿Puede el broker MQTT almacenar el historial de trazabilidad?

No. El broker transporta los eventos. Un suscriptor debe escribir cada evento en una base de datos, que es el sistema de registro.

¿Cómo evito registros duplicados con QoS 1?

Genera un ID de evento único en el origen y haz que la inserción en la base de datos sea idempotente sobre ese ID, para que los mensajes reentregados se ignoren.