Monitorización de procesos por lotes con MQTT: guía

La monitorización de procesos por lotes consiste en saber, en todo momento, en qué fase está cada lote, si sus variables de proceso se mantienen dentro de los límites de la receta y qué desviaciones necesitan una firma antes de la liberación. En este tutorial construirás un pipeline pequeño pero completo con MQTT: un reactor simulado publica fases y lecturas, y un monitor en Python las compara con las consignas y registra cada desviación.

Por qué la monitorización de procesos por lotes es distinta de la monitorización discreta

En la fabricación discreta cuentas piezas. En la industria de procesos (alimentos, bebidas, químicos, recubrimientos, farmacéutica, cosmética) produces una cantidad de material a través de una secuencia de fases: carga, calentamiento, mantenimiento, enfriamiento, trasvase. Una variable de proceso perfectamente correcta en una fase es una desviación en la siguiente. Una lectura de 75 °C es esperable durante el calentamiento, pero un problema durante una fase de enfriamiento.

Esto tiene tres consecuencias para la monitorización:

El contexto es obligatorio. Cada lectura debe llevar el ID de lote, la receta y la fase actual; de lo contrario no se puede evaluar.
Los límites cambian con el tiempo. El monitor debe cargar consignas y tolerancias por fase, no por tag.
El historial importa para la liberación. Los equipos de calidad necesitan un registro de lote: cuándo empezó y terminó cada fase, y cada desviación con su duración y su valor pico.

La norma ISA-88 aporta el vocabulario que la mayoría de las plantas ya utiliza: procedimientos, procedimientos de unidad, operaciones y fases que se ejecutan en unidades. No necesitas un motor de lotes ISA-88 completo para beneficiarte de una monitorización estructurada, pero adoptar su nomenclatura hace que tus datos resulten comprensibles para los ingenieros de procesos.

Diseñar topics MQTT para datos de lotes

MQTT funciona bien para datos de lotes porque un cambio de fase es un evento, y los eventos son justo lo que gestiona el modelo publish/subscribe. Un broker MQTT recibe cada evento una vez y lo distribuye a todos los consumidores interesados: el HMI, el historiador, el dashboard de calidad y el monitor que construirás a continuación.

Una estructura práctica para una unidad es esta:

Topic	Payload	Retenido	Propósito
`.../kettle1/batch/state`	ID de lote, receta, estado	Sí	Lote actual en la unidad
`.../kettle1/batch/phase`	nombre de fase, hora de inicio, consignas	Sí	Fase actual y sus límites
`.../kettle1/pv`	temperatura, velocidad del agitador, nivel	No	Variables de proceso, cada pocos segundos
`.../kettle1/batch/deviation`	parámetro, valor, límite, fase	No	Eventos de desviación para el registro de lote

Retener state y phase significa que cualquier cliente que se conecte a mitad de lote conoce el contexto de inmediato. Las variables de proceso no se retienen porque la siguiente lectura llega segundos después de todos modos.

En producción, antepón tu ruta ISA-95 (empresa, sitio, área, línea o celda) como se describe en topics MQTT ISA-95. En este tutorial usamos el broker público de pruebas, que admite hasta 8 niveles de topic, así que mantenemos un prefijo corto: connectmq/demo/<random-id>/kettle1/....

Paso 1: simular un lote en el broker público

El servidor MQTT público gratuito en broker.connectmq.com acepta conexiones anónimas con cualquier client ID único y no vacío. Todo lo que se publica allí puede leerlo cualquiera, así que usa un prefijo de topic aleatorio y nunca envíes datos reales de recetas: es para pruebas y prototipos, no para datos de producción.

Instala la librería cliente:

pip install "paho-mqtt>=2.0"

El simulador siguiente ejecuta una receta corta con cuatro fases, publica el estado del lote y la fase como mensajes retenidos y envía variables de proceso cada segundo. Sobrepasa deliberadamente la temperatura de mantenimiento para que el monitor tenga algo que detectar.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/batch-7f3a/kettle1"  # cambia 7f3a por tu propio id aleatorio
BATCH_ID = "B" + time.strftime("%Y%m%d-%H%M%S")

RECIPE = [
    {"phase": "CHARGE", "duration": 10, "temp_sp": 20, "temp_tol": 5,  "rpm_min": 0},
    {"phase": "HEAT",   "duration": 15, "temp_sp": None, "temp_tol": None, "rpm_min": 60},
    {"phase": "HOLD",   "duration": 20, "temp_sp": 72, "temp_tol": 2,  "rpm_min": 60},
    {"phase": "COOL",   "duration": 15, "temp_sp": None, "temp_tol": None, "rpm_min": 30},
]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"batch-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()

def publish(subtopic, payload, retain=False):
    client.publish(f"{PREFIX}/{subtopic}", json.dumps(payload), qos=1, retain=retain)

publish("batch/state", {"batchId": BATCH_ID, "recipe": "SAUCE-A", "status": "RUNNING"}, retain=True)

temp = 18.0
for step in RECIPE:
    publish("batch/phase", {"batchId": BATCH_ID, "ts": time.time(), **step}, retain=True)
    for second in range(step["duration"]):
        if step["phase"] == "HEAT":
            temp += 3.5
        elif step["phase"] == "HOLD":
            temp = 72 + random.uniform(-1.5, 1.5)
            if 8 <= second <= 11:
                temp = 75.2  # sobrepaso inyectado
        elif step["phase"] == "COOL":
            temp -= 3.0
        rpm = 0 if step["phase"] == "CHARGE" else 80 + random.uniform(-5, 5)
        publish("pv", {"batchId": BATCH_ID, "ts": time.time(),
                       "temp": round(temp, 2), "rpm": round(rpm, 1)})
        time.sleep(1)

publish("batch/state", {"batchId": BATCH_ID, "recipe": "SAUCE-A", "status": "COMPLETE"}, retain=True)
time.sleep(1)
client.loop_stop()
client.disconnect()

Paso 2: construir el servicio de monitorización de procesos por lotes

El monitor se suscribe a la unidad, recuerda la fase actual y sus límites, comprueba cada variable de proceso y publica un evento de desviación cuando un valor sale de su banda. También registra la duración y el pico de la desviación, que es lo que realmente necesita un registro de lote.

import json
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/batch-7f3a/kettle1"
current_phase = {}
open_deviation = None

def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)
    client.subscribe(f"{PREFIX}/batch/#", qos=1)
    client.subscribe(f"{PREFIX}/pv", qos=1)

def check(client, pv):
    global open_deviation
    sp, tol = current_phase.get("temp_sp"), current_phase.get("temp_tol")
    out_of_spec = sp is not None and abs(pv["temp"] - sp) > tol
    if out_of_spec and open_deviation is None:
        open_deviation = {"batchId": pv["batchId"], "phase": current_phase["phase"],
                          "parameter": "temp", "limit": f"{sp}+/-{tol}",
                          "start": pv["ts"], "peak": pv["temp"]}
        print("DEVIATION START", open_deviation)
    elif out_of_spec:
        open_deviation["peak"] = max(open_deviation["peak"], pv["temp"], key=lambda v: abs(v - sp))
    elif open_deviation is not None:
        open_deviation["end"] = pv["ts"]
        open_deviation["durationS"] = round(pv["ts"] - open_deviation["start"], 1)
        client.publish(f"{PREFIX}/batch/deviation", json.dumps(open_deviation), qos=1)
        print("DEVIATION CLOSED", open_deviation)
        open_deviation = None

def on_message(client, userdata, msg):
    global current_phase
    data = json.loads(msg.payload)
    if msg.topic.endswith("/batch/phase"):
        current_phase = data
        print(f"Phase -> {data['phase']} (batch {data['batchId']})")
    elif msg.topic.endswith("/batch/state"):
        print(f"Batch {data['batchId']} is {data['status']}")
    elif msg.topic.endswith("/pv") and current_phase:
        if data["rpm"] < current_phase.get("rpm_min", 0):
            print(f"Agitator below minimum in {current_phase['phase']}: {data['rpm']} rpm")
        check(client, data)

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

Inicia primero el monitor y después el simulador en un segundo terminal. Verás las transiciones de fase y, a continuación, una desviación que se abre unos ocho segundos después de entrar en HOLD y se cierra unos cuatro segundos más tarde con su pico y su duración. Observa al mismo tiempo el tráfico en bruto con la CLI de Mosquitto:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watcher-7f3a -t "connectmq/demo/batch-7f3a/kettle1/#" -v

Paso 3: convertir los eventos en un registro de lote

El topic de desviaciones es la materia prima de tu registro electrónico de lote. Un suscriptor independiente (un pequeño servicio en Python, un flujo de Node-RED o una instancia de Telegraf) puede escribir los cambios de fase y las desviaciones en una base de datos indexada por ID de lote. En el momento de la liberación, una consulta devuelve la historia completa: horas de inicio y fin de cada fase, las consignas vigentes y cada desviación con su duración y su pico.

Ten en cuenta estas reglas al diseñarlo:

Escribe las consignas con cada evento de fase. Las recetas cambian; el registro debe mostrar los límites que se aplicaron a este lote, no los valores de hoy.
Usa QoS 1 para los eventos de fase y de desviación. Perder un cambio de fase corrompe todas las evaluaciones posteriores. Las variables de proceso con QoS 0 suelen ser suficientes.
Pon la marca de tiempo en el origen. El campo ts procede del controlador o del gateway, no del suscriptor, para que el retardo de red no desplace los límites de las fases.
Trata los duplicados como algo normal. QoS 1 garantiza al menos una entrega, así que haz que la inserción en la base de datos sea idempotente sobre ID de lote, fase y marca de tiempo.

Para tendencias estadísticas entre muchos lotes, los mismos datos alimentan el SPC en tiempo real, y vincular los ID de lote con los lotes de materia prima te da la columna vertebral de un sistema de trazabilidad en fabricación.

Resolución de problemas en pipelines de monitorización de lotes

Síntoma	Causa probable	Solución
El monitor ignora las primeras lecturas	Se conectó después de publicarse la fase, sin fase retenida	Publica `batch/phase` con `retain=True`
Las desviaciones se abren y cierran sin parar	El valor oscila justo en el límite	Añade una banda muerta o exige N muestras consecutivas
Los límites de fase parecen incorrectos	Marcas de tiempo del suscriptor en lugar de las del origen	Pon `ts` en el controlador o el gateway
Filas de desviación duplicadas	Reentrega QoS 1 tras una reconexión	Inserción idempotente sobre una clave natural
El cliente se desconecta continuamente	Dos clientes comparten un mismo client ID	Genera un ID único por proceso

Pasar a producción con ConnectMQ

El broker público es para experimentos. Para lotes reales necesitas credenciales privadas, TLS y un broker que no se comparta con todo el mundo. ConnectMQ te ofrece un broker MQTT gestionado con MQTT 3.1.1 y 5.0, TLS en el puerto 8883 y usuario, contraseña y client ID generados por broker. Cambia las líneas de conexión en ambos scripts:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # usa el almacén de CA del sistema; el broker presenta un certificado de Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

En el portal de ConnectMQ, el explorador de topics en directo muestra el estado retenido de los lotes por unidad, y las reglas de alerta por umbral pueden señalar una variable de proceso que supera un límite fijo como segunda línea de defensa junto a tu monitor sensible a las fases. La visión general de dónde encajan los datos de lotes en una planta se trata en la guía de fabricación inteligente; esos mismos datos pueden mostrarse en un gemelo digital 3D de la planta.

Empieza a monitorizar lotes con ConnectMQ

Crea un broker en app.connectmq.com con el plan gratuito Serverless, sin tarjeta de crédito, pega las credenciales en los scripts anteriores y ejecuta tu primer lote monitorizado en minutos.

Preguntas frecuentes

¿Necesito un motor de lotes ISA-88 para la monitorización de procesos por lotes con MQTT?

No. El motor de lotes o el PLC solo tiene que publicar la fase actual y las consignas cuando cambian. Cualquier controlador capaz de hacerlo, directamente o a través de un gateway, puede alimentar el monitor.

¿Qué nivel de QoS deben usar los eventos de lote?

Usa QoS 1 para cambios de fase, estado del lote y desviaciones, de modo que se entreguen al menos una vez, y haz que los consumidores sean idempotentes. Las variables de proceso de alta frecuencia pueden usar QoS 0 cuando perder alguna muestra ocasional no importa.

¿Puede el plan gratuito gestionar una línea de lotes real?

El plan gratuito Serverless admite hasta 25 conexiones, QoS hasta 1 y mensajes de 256 KB, suficiente para un piloto con unas pocas unidades. Las plantas más grandes pasan a los planes Serverless de pago o Dedicated.

¿Dónde debe almacenarse el registro de lote?

En una base de datos propiedad de tu sistema de calidad. El broker mueve los eventos en tiempo real; un servicio suscriptor los escribe en un almacenamiento duradero para su revisión y liberación.

Monitorización de procesos por lotes con MQTT: fases, consignas y desviaciones