Surveillance des procédés batch avec MQTT : phases, consignes et écarts

La surveillance des procédés batch consiste à savoir, à tout instant, dans quelle phase se trouve chaque lot, si ses valeurs de procédé restent dans les limites de la recette et quels écarts nécessitent une signature avant libération. Dans ce tutoriel, vous allez construire un pipeline MQTT modeste mais complet : une cuve simulée publie ses phases et ses mesures, et un moniteur Python les compare aux consignes et enregistre chaque écart.

Pourquoi la surveillance des procédés batch diffère de la surveillance discrète

Dans l'industrie manufacturière discrète, on compte des pièces. Dans l'industrie de procédés (agroalimentaire, boissons, chimie, revêtements, pharmacie, cosmétique), on produit une quantité de matière à travers une séquence de phases : chargement, chauffe, maintien, refroidissement, transfert. Une valeur de procédé parfaitement normale dans une phase devient un écart dans la suivante. Une mesure de 75 °C est attendue pendant une montée en température, mais pose problème pendant une phase de refroidissement.

Cela a trois conséquences pour la surveillance :

1. Le contexte est indispensable. Chaque mesure doit porter l'identifiant du lot, la recette et la phase en cours, sinon elle ne peut pas être évaluée.
2. Les limites évoluent dans le temps. Le moniteur doit charger les consignes et tolérances par phase, et non par tag.
3. L'historique compte pour la libération. Les équipes qualité ont besoin d'un dossier de lot : début et fin de chaque phase, et chaque écart avec sa durée et sa valeur crête.

La norme ISA-88 fournit le vocabulaire que la plupart des usines utilisent déjà : procédures, procédures d'unité, opérations et phases exécutées sur des unités. Pas besoin d'un moteur batch ISA-88 complet pour profiter d'une surveillance structurée, mais reprendre sa terminologie rend vos données compréhensibles pour les ingénieurs procédés.

Concevoir les topics MQTT pour les données batch

MQTT convient bien aux données batch, car un changement de phase est un événement, et les événements sont précisément ce que gère le modèle publish/subscribe. Un broker MQTT reçoit chaque événement une seule fois et le diffuse à tous les consommateurs intéressés : l'IHM, l'historien, le tableau de bord qualité et le moniteur que vous allez construire ci-dessous.

Voici une organisation pratique pour une unité :

Mettre state et phase en retained permet à tout client qui se connecte en cours de lot de connaître immédiatement le contexte. Les valeurs de procédé ne sont pas retained, puisque la mesure suivante arrive de toute façon quelques secondes plus tard.

En production, préfixez ces topics par votre chemin ISA-95 (entreprise, site, zone, ligne ou cellule), comme décrit dans Topics MQTT ISA-95. Pour ce tutoriel, nous utilisons le serveur MQTT public gratuit, qui autorise jusqu'à 8 niveaux de topic ; nous gardons donc un préfixe court : connectmq/demo/<random-id>/kettle1/....

Étape 1 : simuler un lot sur le serveur public

Le serveur MQTT public gratuit broker.connectmq.com accepte les connexions anonymes avec n'importe quel client ID unique et non vide. Tout ce qui y est publié est lisible par n'importe qui : utilisez donc un préfixe de topic aléatoire et n'envoyez jamais de vraies données de recette.

Installez la bibliothèque client :

pip install "paho-mqtt>=2.0"

Le simulateur ci-dessous exécute une courte recette en quatre phases, publie l'état du lot et la phase en retained, et diffuse les valeurs de procédé chaque seconde. Il dépasse volontairement la température de maintien pour que le moniteur ait quelque chose à détecter.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/batch-7f3a/kettle1"  # remplacez 7f3a par votre propre identifiant aléatoire
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  # dépassement injecté
        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()

Étape 2 : construire le service de surveillance des procédés batch

Le moniteur s'abonne à l'unité, mémorise la phase en cours et ses limites, vérifie chaque valeur de procédé et publie un événement d'écart lorsqu'une valeur sort de sa plage. Il suit également la durée et la crête de l'écart, ce dont un dossier de lot a réellement besoin.

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

Lancez d'abord le moniteur, puis le simulateur dans un second terminal. Vous verrez les transitions de phase, puis un écart qui s'ouvre environ huit secondes après le début de HOLD et se referme quatre secondes plus tard environ, avec sa crête et sa durée. Observez en parallèle le trafic brut avec la CLI Mosquitto :

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

Étape 3 : transformer les événements en dossier de lot

Le topic des écarts est la matière première de votre dossier de lot électronique. Un abonné distinct (un petit service Python, un flux Node-RED ou une instance Telegraf) peut écrire les changements de phase et les écarts dans une base de données indexée par ID de lot. Au moment de la libération, une requête restitue toute l'histoire : heures de début et de fin des phases, consignes en vigueur, et chaque excursion avec sa durée et sa crête.

Gardez ces règles en tête lors de la conception :

1. Enregistrez les consignes avec chaque événement de phase. Les recettes évoluent ; le dossier doit montrer les limites qui s'appliquaient à ce lot, pas les valeurs du jour.
2. Utilisez la QoS 1 pour les événements de phase et d'écart. Perdre un changement de phase fausse toutes les évaluations qui suivent. Les valeurs de procédé en QoS 0 conviennent généralement.
3. Horodatez à la source. Le champ ts provient de l'automate ou de la passerelle, pas de l'abonné, afin que la latence réseau ne décale pas les limites de phase.
4. Considérez les doublons comme normaux. La QoS 1 garantit une livraison au moins une fois : rendez l'insertion en base idempotente sur l'ID de lot, la phase et l'horodatage.

Pour les tendances statistiques sur de nombreux lots, les mêmes données alimentent la MSP en temps réel, et relier les ID de lot aux lots de matières premières constitue l'épine dorsale d'un système de traçabilité industrielle.

Dépannage des pipelines de surveillance batch

Passer en production sur ConnectMQ

Le serveur public sert aux expérimentations. Pour de vrais lots, il vous faut des identifiants privés, TLS et un broker qui n'est pas partagé avec le monde entier. ConnectMQ vous fournit un broker MQTT managé compatible MQTT 3.1.1 et 5.0, avec TLS sur le port 8883 et un nom d'utilisateur, un mot de passe et un client ID générés pour chaque broker. Modifiez les lignes de connexion dans les deux scripts :

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # utilise le magasin de certificats du système ; le broker présente un certificat Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=30)

Dans le portail ConnectMQ, l'explorateur de topics en direct affiche l'état retained du lot par unité, et les règles d'alerte sur seuil peuvent signaler une valeur de procédé qui franchit une limite fixe, en seconde ligne de défense à côté de votre moniteur sensible aux phases. La place des données batch dans l'ensemble de l'usine est détaillée dans le guide de l'industrie intelligente.

Commencez à surveiller vos lots sur ConnectMQ

Créez un broker sur app.connectmq.com avec l'offre Serverless gratuite, sans carte bancaire, collez les identifiants dans les scripts ci-dessus et lancez votre premier lot surveillé en quelques minutes.

Questions fréquentes

Faut-il un moteur batch ISA-88 pour la surveillance des procédés batch avec MQTT ?

Non. Le moteur batch ou l'automate doit seulement publier la phase en cours et les consignes lorsqu'elles changent. Tout automate capable de le faire, directement ou via une passerelle, peut alimenter le moniteur.

Quel niveau de QoS utiliser pour les événements batch ?

Utilisez la QoS 1 pour les changements de phase, l'état du lot et les écarts afin qu'ils soient livrés au moins une fois, et rendez les consommateurs idempotents. Les valeurs de procédé à haute fréquence peuvent utiliser la QoS 0 lorsqu'un échantillon perdu de temps en temps n'a pas d'importance.

L'offre gratuite peut-elle gérer une vraie ligne batch ?

L'offre Serverless gratuite prend en charge jusqu'à 25 connexions, une QoS jusqu'à 1 et des messages de 256 Ko, ce qui suffit pour un pilote sur quelques unités. Les usines plus importantes passent aux offres Serverless payante ou Dedicated.

Où stocker le dossier de lot ?

Dans une base de données gérée par votre système qualité. Le broker transporte les événements en temps réel ; un service abonné les écrit dans un stockage durable pour la revue et la libération.