Allen-Bradley MQTT : publier les tags ControlLogix et CompactLogix

L'intégration Allen-Bradley MQTT est le moyen le plus rapide d'envoyer les données de tags ControlLogix et CompactLogix vers des tableaux de bord, des historians et des outils d'analyse cloud sans charger votre serveur SCADA. Dans ce tutoriel, vous allez construire une petite passerelle Python qui lit les tags Logix via EtherNet/IP avec pycomm3, les associe à des topics ISA-95 et ne publie que les changements significatifs vers un broker MQTT. Vous testerez d'abord sur le serveur MQTT public gratuit, puis passerez à un broker MQTT managé privé avec TLS.

Comment les automates Allen-Bradley s'intègrent à MQTT

Les automates Logix communiquent nativement via EtherNet/IP en utilisant CIP (Common Industrial Protocol). Les projets Studio 5000 organisent les données en tags nommés, de portée contrôleur ou programme, avec des types atomiques (BOOL, DINT, REAL), des tableaux et des types définis par l'utilisateur (UDT). Ce modèle de tags nommés est une aubaine pour l'intégration MQTT : contrairement à Modbus, aucune table de registres n'est nécessaire, car le nom du tag porte déjà le sens.

Il existe trois façons courantes de faire passer ces tags dans MQTT :

Ce tutoriel suit la première approche, car elle montre chaque rouage. La même conception de topics et de payloads s'applique si vous passez ensuite à un driver commercial ou à Ignition avec MQTT.

Prérequis

• Un automate ControlLogix ou CompactLogix joignable depuis la passerelle sur le port TCP 44818.
• Une machine Linux ou Windows avec Python 3.9 ou plus récent.
• Quelques tags à lire, par exemple Line1_Temp (REAL), Line1_Speed (DINT), Line1_Running (BOOL) et un tag de portée programme Program:MainProgram.Pump1_Pressure (REAL).
• Les clients mosquitto pour les tests.

Installez les bibliothèques :

python -m venv abmqtt
source abmqtt/bin/activate
pip install pycomm3 "paho-mqtt>=2.0"

Étape 1 : découvrir la liste des tags

Avant d'écrire la passerelle, vérifiez la connectivité et regardez précisément comment les tags sont nommés. pycomm3 télécharge la base de tags de l'automate à la connexion :

from pycomm3 import LogixDriver

# CompactLogix : l'IP suffit. ControlLogix avec la CPU dans l'emplacement 3 : "192.168.1.20/3"
PLC_PATH = "192.168.1.20"

with LogixDriver(PLC_PATH) as plc:
    print(plc.info)                       # nom de l'automate, révision, numéro de série
    for name, definition in plc.tags.items():
        print(name, definition["data_type_name"], definition.get("dim", 0))

Les tags de portée programme ne figurent pas par défaut dans la liste de portée contrôleur ; adressez-les sous la forme Program:MainProgram.TagName lors de la lecture. Pour les châssis ControlLogix, le chemin après la barre oblique est l'emplacement de l'automate dans le fond de panier. Si vous vous trompez, la connexion s'ouvre vers le module Ethernet mais les lectures de tags échouent.

La documentation de pycomm3 détaille la syntaxe des chemins, les tableaux et la gestion des UDT.

Étape 2 : concevoir des topics ISA-95

Associez chaque tag à un topic qui décrit l'équipement, pas l'automate. Une simple table de correspondance dans la passerelle tient les conventions de nommage automate à l'écart de votre espace de noms MQTT :

Avec un préfixe de site et de zone, un topic complet devient plant1/assembly/line1/oven/temperature. Ce sont des chemins cohérents comme celui-ci qui font fonctionner un espace de noms à l'échelle de l'usine ; consultez topics MQTT ISA-95 pour une analyse approfondie des niveaux et du nommage.

Étape 3 : la passerelle Allen-Bradley MQTT

Chaque seconde, la passerelle lit tous les tags associés en une seule requête multi-tags, applique des bandes mortes par tag, publie les changements en QoS 1, rafraîchit l'ensemble toutes les cinq minutes et entretient un statut en ligne/hors ligne retenu avec un Last Will.

import json
import time
import uuid
from datetime import datetime, timezone

from pycomm3 import LogixDriver
import paho.mqtt.client as mqtt

PLC_PATH = "192.168.1.20"

# D'abord le broker de test public. L'espace public est lisible par tous : données de test uniquement.
BROKER, PORT, USE_TLS = "broker.connectmq.com", 1883, False
USERNAME, PASSWORD = None, None                  # production : "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"ab-gw-{uuid.uuid4().hex[:8]}"      # production : "YOUR_CLIENT_ID"
BASE = f"connectmq/demo/{uuid.uuid4().hex[:6]}/plant1/assembly/line1"

TAGS = {
    "Line1_Temp": ("oven/temperature", "degC", 0.5),
    "Line1_Speed": ("conveyor/speed", "rpm", 5),
    "Line1_Running": ("line/running", None, 0),
    "Program:MainProgram.Pump1_Pressure": ("pump1/pressure", "bar", 0.05),
}


def on_connect(client, userdata, flags, reason_code, properties):
    print("MQTT connected:", reason_code)
    client.publish(f"{BASE}/gateway/status", "online", qos=1, retain=True)


def should_publish(tag, value, last, deadband):
    if tag not in last:
        return True
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return value != last[tag]
    return abs(value - last[tag]) >= deadband


def main():
    mq = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        mq.username_pw_set(USERNAME, PASSWORD)
    if USE_TLS:
        mq.tls_set()
    mq.will_set(f"{BASE}/gateway/status", "offline", qos=1, retain=True)
    mq.on_connect = on_connect
    mq.connect(BROKER, PORT, keepalive=30)
    mq.loop_start()
    print("Publishing under", BASE)

    plc = LogixDriver(PLC_PATH)
    last, last_full = {}, 0.0

    while True:
        try:
            if not plc.connected:
                plc.open()
            results = plc.read(*TAGS.keys())
            now = time.time()
            force = now - last_full > 300
            ts = datetime.now(timezone.utc).isoformat(timespec="seconds")

            for result in results:
                suffix, unit, deadband = TAGS[result.tag]
                if result.error:
                    mq.publish(f"{BASE}/{suffix}", json.dumps(
                        {"value": None, "quality": "bad", "error": result.error, "ts": ts}), qos=1)
                    continue
                value = result.value
                if isinstance(value, float):
                    value = round(value, 3)
                if force or should_publish(result.tag, value, last, deadband):
                    payload = {"value": value, "quality": "good", "ts": ts}
                    if unit:
                        payload["unit"] = unit
                    mq.publish(f"{BASE}/{suffix}", json.dumps(payload), qos=1)
                    last[result.tag] = value
            if force:
                last_full = now
        except Exception as exc:
            print("PLC communication failed:", exc)
            try:
                plc.close()
            except Exception:
                pass
            time.sleep(5)
        time.sleep(1)


if __name__ == "__main__":
    main()

Quelques choix de conception méritent d'être soulignés :

• Une seule lecture multi-tags. pycomm3 regroupe plusieurs tags dans aussi peu de requêtes CIP que possible, ce qui ménage bien plus l'automate que la lecture de chaque tag séparément.
• Erreurs par tag. Une faute de frappe dans un nom de tag renvoie une erreur pour ce seul tag. Publier quality: bad rend le problème visible au lieu de figer silencieusement une valeur.
• Les UDT fonctionnent aussi. La lecture d'un UDT renvoie un dict Python, que json.dumps transforme en payload structuré. Gardez les payloads d'UDT compacts et évitez de publier de grands tableaux chaque seconde.

Étape 4 : vérifier sur le broker public

Démarrez la passerelle et abonnez-vous à son préfixe :

mosquitto_sub -h broker.connectmq.com -p 1883 -i ab-watch-01 \
  -t 'connectmq/demo/+/plant1/assembly/line1/#' -v

Vous devriez voir immédiatement le statut retenu de la passerelle et le premier jeu complet de valeurs. Basculez Line1_Running depuis Studio 5000 : le changement arrive en moins d'une seconde. Utiliser le serveur MQTT public gratuit pour cette étape ne pose pas de problème tant que les valeurs ne sont pas sensibles.

Étape 5 : la production sur un broker privé

Pour la production, créez un broker privé dans le portail ConnectMQ et modifiez quatre lignes : le port 8883, USE_TLS = True, le nom d'utilisateur et le mot de passe fournis, et le client ID fourni. Supprimez le préfixe connectmq/demo/... pour que les topics commencent au niveau de l'entreprise ou du site. Comme les client IDs sont liés à leurs identifiants, attribuez à chaque passerelle son propre jeu plutôt que d'en partager un entre plusieurs sites.

Exécutez la passerelle en tant que service pour qu'elle redémarre automatiquement :

[Unit]
Description=Allen-Bradley MQTT gateway
After=network-online.target

[Service]
ExecStart=/opt/abmqtt/bin/python /opt/abmqtt/gateway.py
Restart=always
RestartSec=5
User=abmqtt

[Install]
WantedBy=multi-user.target

Une fois les données en circulation, la vue des équipements du portail ConnectMQ regroupe les topics ISA-95 par site, zone et ligne, et des règles d'alerte à seuil peuvent signaler une température ou une pression hors limites. Historians, tableaux de bord et jumeaux numériques s'abonnent comme n'importe quel autre client du broker MQTT.

Régler les fréquences de lecture et les bandes mortes

L'intervalle de lecture et les bandes mortes déterminent ensemble le volume de trafic que génère la passerelle, sur le réseau d'automatisme comme sur le broker. Une façon pratique de les régler :

• Partez du consommateur. Un tableau de bord consulté par un opérateur toutes les quelques secondes n'a pas besoin de données à 100 ms. Un historian utilisé pour des rapports énergétiques peut se contenter de moyennes à la minute.
• Dimensionnez les bandes mortes d'après le bruit du capteur. Observez une valeur pendant quelques minutes dans la vue de tendance de Studio 5000. Une bande morte légèrement supérieure à la bande de bruit supprime les fluctuations sans masquer les vrais changements.
• N'appliquez jamais de bande morte aux états et aux alarmes. Les booléens, mots de mode et bits d'alarme doivent être publiés à chaque changement, et leurs topics sont de bons candidats aux messages retenus, pour qu'un nouvel abonné voie immédiatement l'état courant.
• Conservez le rafraîchissement périodique. Sans lui, une valeur réellement constante pendant des heures ne se distingue pas d'une passerelle qui s'est arrêtée silencieusement.

Si une seule passerelle ne suffit plus, faites-en tourner une par zone ou par ligne plutôt qu'un processus géant pour toute l'usine. Les pannes restent circonscrites et chaque instance a une liste de tags courte et facile à relire.

Considérations de sécurité

Dans la plupart des installations, CIP sur EtherNet/IP ne comporte aucune authentification : la passerelle est donc de fait un client d'ingénierie de confiance. Placez-la sur le réseau d'automatisme derrière un pare-feu, n'autorisez depuis elle que le TLS sortant vers le broker, et ne laissez jamais des commandes MQTT écrire dans les tags de l'automate sans liste blanche explicite, contrôle des plages et verrouillage dans la logique automate. La lecture peut être automatisée sans risque ; l'écriture mérite une revue de conception. Le guide EtherNet/IP vers MQTT approfondit la segmentation réseau pour le trafic CIP, et le guide de l'IoT industriel avec MQTT explique la place des passerelles dans une architecture complète.

Dépannage

1. La connexion expire. Vérifiez le routage et l'ouverture du port TCP 44818 entre la passerelle et l'automate.
2. Connecté, mais tous les tags échouent. Mauvais emplacement dans un chemin ControlLogix, ou tag de portée programme qui nécessite le préfixe Program:.
3. Valeurs REAL avec de longues décimales. C'est la représentation normale des flottants ; arrondissez avant de publier.
4. MQTT se déconnecte toutes les quelques secondes. Un autre client utilise le même client ID.
5. Charge de communication élevée sur l'automate. Augmentez l'intervalle de lecture ou réduisez la liste de tags ; la plupart des tableaux de bord n'ont pas besoin de données à la sous-seconde.

Connectez vos automates Logix à ConnectMQ

Votre passerelle est prête ; donnez-lui un broker MQTT cloud sécurisé vers lequel publier. Inscrivez-vous sur app.connectmq.com pour un plan Serverless gratuit sans carte bancaire, collez les identifiants générés dans le script et regardez vos tags Allen-Bradley apparaître en direct.

Questions fréquentes

Un automate ControlLogix peut-il publier nativement en MQTT ?

Les automates Logix n'intègrent pas de client MQTT généraliste dans le contrôleur lui-même. L'approche habituelle consiste à utiliser une passerelle edge, un serveur de connectivité ou un module SCADA qui lit les tags via EtherNet/IP et les publie.

pycomm3 fonctionne-t-il avec les Micro800 ou les SLC 500 ?

Le LogixDriver de pycomm3 cible ControlLogix, CompactLogix et Micro800 avec certaines limitations ; les anciens PLC-5 et SLC 500 utilisent un driver distinct. Consultez la documentation de la bibliothèque pour les fonctions prises en charge par famille.

À quelle fréquence lire les tags ?

Une seconde suffit largement pour la plupart des besoins de supervision. Utilisez des bandes mortes pour ne publier que les vrais changements, plus un rafraîchissement complet périodique.

Faut-il utiliser Sparkplug B plutôt que JSON ?

Sparkplug B ajoute des certificats de naissance et de mort ainsi qu'un encodage Protobuf compact, utile pour les grands nombres de tags et l'intégration SCADA. Le JSON simple est plus facile pour démarrer et à inspecter.