EtherNet/IP vers MQTT : lire les équipements CIP et les tags Logix avec Python

Une passerelle EtherNet/IP vers MQTT permet de sortir les données des automates Rockwell Logix, des variateurs et des autres équipements CIP pour les envoyer vers des tableaux de bord, des historians et des applications cloud, sans toucher au programme de commande. Dans ce tutoriel, vous verrez la différence entre messagerie implicite et explicite, puis vous utiliserez la bibliothèque open source pycomm3 sur une passerelle pour lire des tags Logix et des attributs CIP génériques, et les publier via MQTT avec des payloads propres et autodescriptifs.

EtherNet/IP et CIP en cinq minutes

EtherNet/IP, c'est le Common Industrial Protocol (CIP) transporté sur Ethernet standard et TCP/UDP. CIP modélise chaque équipement comme un ensemble d'objets dotés de classes, d'instances et d'attributs. Un variateur, par exemple, possède un objet Identity (classe 0x01) qui décrit le fabricant, le code produit et le numéro de série, ainsi que des objets applicatifs pour la vitesse, le courant et l'état.

Le trafic CIP se présente sous deux formes :

Une passerelle de supervision doit utiliser la messagerie explicite. Elle ne perturbe pas les connexions d'E/S de l'automate, ne nécessite aucune modification du projet automate et peut être ajoutée ou retirée sans arrêt. La contrepartie : c'est un protocole d'interrogation (polling), c'est donc vous qui choisissez la fréquence de lecture.

Deux types de cibles

1. Automates Logix (ControlLogix, CompactLogix, et dans une certaine mesure Micro800). Ils prennent en charge l'accès symbolique aux tags : vous lisez Line1_Speed ou Conveyor[3].Status par leur nom. C'est de loin la source de données la plus pratique.
2. Équipements CIP génériques (variateurs, adaptateurs d'E/S, scanners, équipements tiers). Ils exposent des attributs d'objets que vous lisez par numéro de classe, d'instance et d'attribut, d'après la documentation de l'équipement ou son fichier EDS.

pycomm3 gère les deux : LogixDriver pour les automates à tags et CIPDriver pour les messages explicites génériques. pylogix est une autre option open source populaire pour les tags Logix. Siemens, Beckhoff et les autres plateformes ont leurs propres voies d'accès ; pour les sujets propres à Rockwell, comme la configuration de l'automate, consultez notre guide Allen-Bradley MQTT.

Prérequis

• Python 3.10 ou plus récent sur une passerelle capable d'atteindre l'automate en TCP 44818.
• pycomm3 et paho-mqtt 2.x.
• L'adresse IP de l'automate et, pour les automates en châssis, le numéro d'emplacement (par exemple 10.0.0.20/0 pour l'emplacement 0).
• Une liste de tags en lecture seule validée avec l'automaticien.

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

Étape 1 : explorer l'automate

Avant d'écrire la passerelle, vérifiez la connectivité et listez les tags disponibles. Ce script affiche les informations de l'automate et les premiers tags de portée contrôleur.

# explore_logix.py - vérifier la connectivité et lister les tags
from pycomm3 import LogixDriver

with LogixDriver("10.0.0.20/0") as plc:          # IP/emplacement de l'automate
    print(plc.info)                               # fabricant, produit, révision, nom
    tags = plc.get_tag_list()                     # tags de portée contrôleur
    for t in tags[:30]:
        print(t["tag_name"], t["data_type_name"])

get_tag_list() télécharge les définitions de tags, y compris les types définis par l'utilisateur, ce qui peut prendre un moment sur les gros projets. Les tags de portée programme se lisent avec la syntaxe Program:MainProgram.TagName.

Étape 2 : lire un attribut CIP générique

Pour un variateur ou tout équipement CIP sans tags symboliques, utilisez des messages explicites. L'exemple lit le nom du produit dans l'objet Identity, que tout équipement EtherNet/IP implémente.

# identity.py - lire l'attribut 7 de l'objet Identity (Product Name) sur n'importe quel équipement CIP
from pycomm3 import CIPDriver, Services, ClassCode, SHORT_STRING

with CIPDriver("10.0.0.31") as dev:
    resp = dev.generic_message(
        service=Services.get_attribute_single,
        class_code=ClassCode.identity_object,   # 0x01
        instance=1,
        attribute=7,                             # Product Name
        data_type=SHORT_STRING,
        connected=False,
        name="product_name",
    )
    print(resp.value if not resp.error else resp.error)

Remplacez la classe, l'instance, l'attribut et le type de données par les valeurs du manuel de l'équipement pour lire des données applicatives comme la fréquence de sortie ou le courant moteur.

Étape 3 : la passerelle EtherNet/IP vers MQTT

À chaque cycle, la passerelle lit une liste de tags en une seule requête multi-tags, publie les valeurs qui ont changé (ou qui franchissent une bande morte) et signale les erreurs de lecture comme une qualité « bad » au lieu de conserver silencieusement les anciennes valeurs.

# eip_mqtt_bridge.py - interroger des tags Logix via EtherNet/IP, publier vers MQTT
import json, logging, time, uuid
import paho.mqtt.client as mqtt
from pycomm3 import LogixDriver, CommError

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("eip")

PLC_PATH = "10.0.0.20/0"
POLL_SECONDS = 1.0
TOPIC_ROOT = "connectmq/demo/9e6d/plant1/packaging/line2"   # votre propre identifiant aléatoire
MQTT_HOST, MQTT_PORT = "broker.connectmq.com", 1883         # production : 8883 avec TLS
MQTT_USER, MQTT_PASS = None, None                           # production : "YOUR_USERNAME", "YOUR_PASSWORD"
CLIENT_ID = f"eip-bridge-{uuid.uuid4().hex[:8]}"            # production : "YOUR_CLIENT_ID"

# tag automate -> (feuille du topic, unité, bande morte)
TAGS = {
    "Line2_Speed":          ("speed", "ppm", 1.0),
    "Line2_Running":        ("running", "", 0),
    "Line2_GoodCount":      ("good_count", "count", 0),
    "Line2_RejectCount":    ("reject_count", "count", 0),
    "Filler.MotorCurrent":  ("filler_current", "A", 0.2),
}


def mqtt_client():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if MQTT_USER:
        c.username_pw_set(MQTT_USER, MQTT_PASS)
        c.tls_set()
    c.will_set(f"{TOPIC_ROOT}/bridge", "offline", qos=1, retain=True)
    c.on_connect = lambda cl, u, f, rc, p: cl.publish(f"{TOPIC_ROOT}/bridge", "online", qos=1, retain=True)
    c.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    c.loop_start()
    return c


def changed(old, new, deadband):
    if old is None:
        return True
    if isinstance(new, (int, float)) and not isinstance(new, bool) and deadband:
        return abs(new - old) >= deadband
    return new != old


def main():
    mq = mqtt_client()
    last = {}
    while True:
        try:
            with LogixDriver(PLC_PATH) as plc:
                log.info("connected to %s", plc.info.get("product_name"))
                while True:
                    results = plc.read(*TAGS.keys())
                    if not isinstance(results, list):
                        results = [results]
                    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
                    for r in results:
                        leaf, unit, deadband = TAGS[r.tag]
                        topic = f"{TOPIC_ROOT}/{leaf}"
                        if r.error:
                            if last.get(r.tag, {}).get("quality") != "bad":
                                mq.publish(topic, json.dumps({"value": None, "quality": "bad",
                                                              "error": r.error, "ts": now}), qos=1, retain=True)
                                last[r.tag] = {"value": None, "quality": "bad"}
                            continue
                        prev = last.get(r.tag, {})
                        if prev.get("quality") == "bad" or changed(prev.get("value"), r.value, deadband):
                            payload = {"value": r.value, "unit": unit, "type": r.type,
                                       "quality": "good", "ts": now}
                            mq.publish(topic, json.dumps(payload), qos=1, retain=True)
                            last[r.tag] = {"value": r.value, "quality": "good"}
                    time.sleep(POLL_SECONDS)
        except CommError as exc:
            log.error("EtherNet/IP connection lost: %s - retrying in 5 s", exc)
            time.sleep(5)


if __name__ == "__main__":
    main()

Appelée avec plusieurs noms de tags, plc.read() les regroupe dans aussi peu de requêtes que la taille de connexion le permet, ce qui sollicite bien moins l'automate qu'une requête par tag. Chaque résultat est un Tag avec les champs tag, value, type et error, ce qui permet à la passerelle de publier le type de données avec la valeur.

Étape 4 : exécuter et vérifier

python eip_mqtt_bridge.py
mosquitto_sub -h broker.connectmq.com -p 1883 -i eip-viewer-9e6d -t "connectmq/demo/9e6d/#" -v

connectmq/demo/9e6d/plant1/packaging/line2/speed {"value": 118.0, "unit": "ppm", "type": "REAL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}
connectmq/demo/9e6d/plant1/packaging/line2/running {"value": true, "unit": "", "type": "BOOL", "quality": "good", "ts": "2026-10-16T12:00:01Z"}

L'arborescence des topics suit une organisation de type ISA-95 (site, zone, ligne, point), de sorte que les mêmes consommateurs, d'un tableau de bord à un jumeau numérique de la ligne, fonctionnent que les données proviennent d'EtherNet/IP, de Modbus ou d'OPC UA. Notre guide sur les topics MQTT ISA-95 détaille ces conventions.

Choisir les tags et les fréquences d'interrogation

Tous les tags d'un projet Logix ne méritent pas un topic. Partez des questions que poseront les consommateurs : la ligne tourne-t-elle, à quelle cadence, combien d'unités bonnes et rejetées, quelles alarmes sont actives, combien d'énergie consomment les principaux variateurs. Cela représente généralement quelques dizaines de valeurs par ligne, pas des milliers. Regroupez-les selon leur vitesse de variation. Les états et les compteurs peuvent être lus chaque seconde et publiés sur changement ; les grandeurs analogiques lentes, comme les niveaux de cuve ou les températures, peuvent être lues toutes les cinq ou dix secondes avec une bande morte. Si une valeur n'a d'intérêt qu'en total de poste, laissez plutôt l'automate la cumuler et publiez-la sur un cycle plus lent.

Bonnes pratiques pour la production

1. Lecture seule par conception. N'ajoutez pas de fonction d'écriture à une passerelle de supervision. Si vous avez un jour besoin d'envoyer des commandes, développez un service séparé avec une logique d'approbation explicite.
2. Utilisez un jeu de tags d'interface. Demandez à l'automaticien de copier les valeurs nécessaires dans un UDT ou un tableau dédié. Vous lisez alors une seule structure au lieu de dizaines de tags dispersés, et les modifications du programme ne cassent pas la passerelle.
3. Surveillez les limites de connexions. Les automates acceptent un nombre limité de connexions CIP, partagées avec les IHM, le SCADA et les outils d'ingénierie. Une connexion longue durée par passerelle est idéale ; évitez de vous reconnecter à chaque cycle.
4. Interrogez à la vitesse nécessaire. Une seconde suffit largement pour les comptages de production et les états. Une interrogation plus rapide ajoute de la charge sans apporter d'information.
5. Segmentation réseau. Placez la passerelle dans le réseau OT ou dans une DMZ, avec un accès limité au port 44818 des automates et au port 8883 sortant vers le broker. Notre article sur la DMZ MQTT décrit cette architecture.
6. Broker privé pour les données réelles. Le serveur MQTT public gratuit est lisible par tout le monde. Passez à un broker MQTT managé avec TLS et des identifiants propres à chaque broker avant de connecter une vraie ligne.

Dépannage

La place de ConnectMQ

ConnectMQ constitue la partie cloud de cette chaîne : un broker MQTT managé avec TLS sur le port 8883 et des WebSockets pour les tableaux de bord dans le navigateur. Il détecte les structures de topics ISA-95, si bien que la vue des équipements regroupe automatiquement les lignes et les machines, et des règles d'alerte à seuil peuvent signaler un nombre de rebuts ou un courant moteur hors limites. Si vous découvrez les brokers managés, voyez comment un broker MQTT cloud se compare à l'auto-hébergement et, pour l'architecture complète, lisez le guide MQTT pour l'IoT industriel.

Démarrer avec ConnectMQ

Connectez votre première ligne EtherNet/IP dès aujourd'hui. Inscrivez-vous sur app.connectmq.com, choisissez le plan gratuit Serverless sans carte bancaire, créez un broker et collez dans la passerelle le nom d'utilisateur, le mot de passe et le client ID générés. Vos tags Logix apparaîtront dans l'explorateur de topics ConnectMQ en quelques secondes.

Questions fréquentes

La lecture de tags via EtherNet/IP affecte-t-elle le programme de l'automate ?

La messagerie explicite utilise les ressources de communication de l'automate, pas ses connexions d'E/S, et ne modifie pas le programme. Gardez une fréquence d'interrogation modérée et peu de connexions pour ne pas pénaliser les IHM et le SCADA.

Puis-je utiliser la messagerie d'E/S implicite pour une passerelle MQTT ?

C'est possible si la passerelle est configurée comme adaptateur dans l'arborescence d'E/S de l'automate, mais cela exige une modification du projet. Les lectures explicites de tags sont plus simples et non intrusives pour la supervision.

pycomm3 fonctionne-t-il avec les Micro800 et les anciens automates PLC-5 ou SLC ?

pycomm3 cible les automates Logix et offre une prise en charge partielle des Micro800 ; les anciennes familles PLC-5 et SLC utilisent un adressage différent. Consultez la documentation de la bibliothèque pour votre automate.

Quel port utilise la messagerie explicite EtherNet/IP ?

Le port TCP 44818. Le trafic d'E/S implicite utilise le port UDP 2222.