Monitoramento de Processos em Batelada com MQTT: Fases, Setpoints e Desvios

Monitoramento de processos em batelada significa saber, a cada momento, em qual fase cada lote está, se suas variáveis de processo permanecem dentro dos limites da receita e quais desvios precisam de assinatura antes da liberação. Neste tutorial você vai montar um pipeline pequeno, mas completo, com MQTT: um tacho simulado publica fases e leituras, e um monitor em Python as compara com os setpoints e registra cada desvio.

Por que o monitoramento de processos em batelada é diferente do monitoramento discreto

Na manufatura discreta você conta peças. Na indústria de processo (alimentos, bebidas, químicos, tintas, farmacêuticos, cosméticos) você produz uma quantidade de material por meio de uma sequência de fases: carga, aquecimento, patamar, resfriamento, transferência. Uma variável de processo perfeitamente normal em uma fase é um desvio na seguinte. Uma leitura de 75 °C é esperada durante o aquecimento, mas é um problema durante uma fase de resfriamento.

Isso tem três consequências para o monitoramento:

1. O contexto é obrigatório. Toda leitura precisa carregar o ID do lote, a receita e a fase atual, ou não pode ser avaliada.
2. Os limites mudam ao longo do tempo. O monitor precisa carregar setpoints e tolerâncias por fase, não por tag.
3. O histórico importa para a liberação. As equipes de qualidade precisam de um registro de lote: quando cada fase começou e terminou, e cada desvio com sua duração e valor de pico.

A norma ISA-88 fornece o vocabulário que a maioria das plantas já usa: procedimentos, procedimentos de unidade, operações e fases executadas em unidades. Você não precisa de um motor de batelada ISA-88 completo para se beneficiar de um monitoramento estruturado, mas adotar sua nomenclatura mantém os dados compreensíveis para os engenheiros de processo.

Projetando topics MQTT para dados de batelada

O MQTT funciona bem para dados de batelada porque uma mudança de fase é um evento, e eventos são exatamente o que o modelo publish/subscribe trata. Um broker MQTT recebe cada evento uma vez e o distribui a todos os consumidores interessados: a IHM, o historiador, o dashboard de qualidade e o monitor que você vai construir abaixo.

Um layout prático para uma unidade fica assim:

Reter state e phase significa que qualquer client que se conecte no meio do lote conhece imediatamente o contexto. As variáveis de processo não são retidas porque a próxima leitura chega segundos depois de qualquer forma.

Em produção, prefixe esses topics com o seu caminho ISA-95 (empresa, site, área, linha ou célula), como descrito em topics MQTT ISA-95. Neste tutorial usamos o servidor MQTT público, que permite até 8 níveis de topic, então mantemos o prefixo curto: connectmq/demo/<random-id>/kettle1/....

Passo 1: simule um lote no servidor público

O servidor MQTT público gratuito em broker.connectmq.com aceita conexões anônimas com qualquer client ID único e não vazio. Tudo o que é publicado ali pode ser lido por qualquer pessoa, então use um prefixo de topic aleatório e nunca envie dados reais de receita. Ele serve para testes e protótipos, não para dados de produção.

Instale a biblioteca cliente:

pip install "paho-mqtt>=2.0"

O simulador abaixo executa uma receita curta com quatro fases, publica o estado do lote e a fase como mensagens retidas e transmite variáveis de processo a cada segundo. Ele ultrapassa propositalmente a temperatura de patamar para que o monitor tenha algo a detectar.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/batch-7f3a/kettle1"  # troque 7f3a pelo seu próprio id aleatório
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  # sobressinal injetado
        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()

Passo 2: construa o serviço de monitoramento de processos em batelada

O monitor assina a unidade, memoriza a fase atual e seus limites, verifica cada variável de processo e publica um evento de desvio quando um valor sai da faixa. Ele também acompanha a duração e o pico do desvio, que é o que um registro de lote realmente precisa.

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

Inicie primeiro o monitor e depois o simulador em um segundo terminal. Você verá as transições de fase e, em seguida, um desvio que se abre por volta de oito segundos após o início do HOLD e se fecha cerca de quatro segundos depois, com seu pico e duração. Acompanhe ao mesmo tempo o tráfego bruto com a CLI do Mosquitto:

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

Passo 3: transforme eventos em um registro de lote

O topic de desvios é a matéria-prima do seu registro eletrônico de lote. Um assinante separado (um pequeno serviço em Python, um fluxo do Node-RED ou uma instância do Telegraf) pode gravar mudanças de fase e desvios em um banco de dados indexado pelo ID do lote. Na hora da liberação, uma consulta devolve a história completa: horários de início e fim das fases, os setpoints vigentes e cada excursão com sua duração e pico.

Tenha estas regras em mente ao projetá-lo:

1. Grave os setpoints junto com cada evento de fase. Receitas mudam; o registro precisa mostrar os limites aplicados a este lote, não os valores de hoje.
2. Use QoS 1 para eventos de fase e de desvio. Perder uma mudança de fase corrompe todos os julgamentos seguintes. Variáveis de processo em QoS 0 geralmente bastam.
3. Registre o timestamp na origem. O campo ts vem do controlador ou do gateway, não do assinante, para que o atraso de rede não desloque os limites das fases.
4. Trate duplicatas como algo normal. QoS 1 é entrega "pelo menos uma vez", então torne a inserção no banco idempotente por ID do lote, fase e timestamp.

Para tendências estatísticas entre muitos lotes, os mesmos dados alimentam o CEP em tempo real, e vincular IDs de lote a lotes de matéria-prima forma a espinha dorsal de um sistema de rastreabilidade na manufatura.

Solução de problemas em pipelines de monitoramento de batelada

Indo para produção na ConnectMQ

O servidor público é para experimentos. Para lotes reais você quer credenciais privadas, TLS e um broker que não seja compartilhado com o mundo. A ConnectMQ oferece um broker MQTT gerenciado com MQTT 3.1.1 e 5.0, TLS na porta 8883 e usuário, senha e client ID gerados por broker. Altere as linhas de conexão nos dois scripts:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # usa o repositório de CAs do sistema; o broker apresenta um certificado Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

No portal da ConnectMQ, o explorador de topics ao vivo mostra o estado retido do lote por unidade, e regras de alerta por limite podem sinalizar uma variável de processo que cruza um limite fixo, como segunda linha de defesa ao lado do seu monitor sensível à fase. O panorama de onde os dados de batelada se encaixam em uma planta está no guia de manufatura inteligente.

Comece a monitorar lotes na ConnectMQ

Crie um broker em app.connectmq.com no plano Serverless gratuito, sem cartão de crédito, cole as credenciais nos scripts acima e execute seu primeiro lote monitorado em minutos.

Perguntas frequentes

Preciso de um motor de batelada ISA-88 para usar monitoramento de processos em batelada com MQTT?

Não. O motor de batelada ou o CLP só precisa publicar a fase atual e os setpoints quando eles mudam. Qualquer controlador que faça isso, diretamente ou por meio de um gateway, pode alimentar o monitor.

Qual nível de QoS os eventos de batelada devem usar?

Use QoS 1 para mudanças de fase, estado do lote e desvios, para que sejam entregues pelo menos uma vez, e torne os consumidores idempotentes. Variáveis de processo de alta frequência podem usar QoS 0 quando a perda ocasional de uma amostra não importa.

O plano gratuito aguenta uma linha de batelada real?

O plano Serverless gratuito suporta até 25 conexões, QoS até 1 e mensagens de 256 KB, o que é suficiente para um piloto em algumas unidades. Plantas maiores migram para os planos Serverless pago ou Dedicated.

Onde o registro de lote deve ser armazenado?

Em um banco de dados pertencente ao seu sistema de qualidade. O broker movimenta os eventos em tempo real; um serviço assinante os grava em armazenamento durável para revisão e liberação.