Passerelle OPC UA vers MQTT : faire remonter les données d'automates vers le cloud

Une passerelle OPC UA vers MQTT lit les tags des automates et des serveurs OPC UA du réseau de l'usine et les publie sous forme de messages MQTT vers un broker auquel tableaux de bord, historians, jumeaux numériques et applications cloud peuvent s'abonner. Dans ce tutoriel, vous allez en construire une en Python avec des bibliothèques open source : elle s'abonne à des nœuds OPC UA, les associe à des topics de style ISA-95 et publie un JSON propre vers un broker cloud. L'ensemble tourne sur un ordinateur portable en une quinzaine de minutes.

Pourquoi relier OPC UA à MQTT ?

OPC UA excelle dans l'atelier. Il dispose d'un modèle d'information riche, d'une sécurité intégrée et d'une large prise en charge chez les fabricants d'automates. Mais c'est un protocole client/serveur : chaque consommateur ouvre sa propre session vers chaque serveur. Cela fonctionne pour un SCADA qui dialogue avec dix automates, et devient malcommode quand vingt applications, dont certaines dans le cloud, veulent les mêmes données.

MQTT inverse cette relation. La passerelle maintient une session OPC UA par serveur et une connexion MQTT sortante, et n'importe quel nombre de consommateurs s'abonne au broker. Le pare-feu de l'usine n'a besoin que d'une connexion sortante sur le port 8883, et les automates ne voient jamais le trafic cloud. Si vous comparez plus largement les deux protocoles, notre comparatif OPC UA vs MQTT détaille les compromis ; en résumé, la plupart des architectures utilisent les deux.

Architecture de la passerelle

 PLC / OPC UA server            Edge gateway (this tutorial)              MQTT broker
 opc.tcp://plc:4840   ---->  asyncua subscription  ---> paho-mqtt  ---->  broker.connectmq.com:8883
 (tags, monitored items)     map node -> ISA-95 topic   JSON payload       dashboards, historian, apps

La passerelle a quatre responsabilités :

1. S'abonner, ne pas interroger. Les abonnements OPC UA permettent au serveur de pousser les changements de valeur à l'intervalle d'échantillonnage de votre choix, ce qui est bien plus efficace que de lire chaque tag à intervalle fixe.
2. Associer les nœuds aux topics. Des identifiants de nœud comme ns=2;s=Line1.CNC1.SpindleTemp ne signifient rien pour un développeur cloud. La passerelle les traduit en une hiérarchie lisible.
3. Normaliser les payloads. Chaque message contient la valeur, le statut OPC UA, un horodatage source et une unité physique.
4. Survivre aux déconnexions. Les deux côtés finiront par décrocher ; la passerelle doit se reconnecter sans perdre sa configuration.

Prérequis

• Python 3.10 ou plus récent.
• asyncua (la bibliothèque cliente open source FreeOpcUa) et paho-mqtt 2.x.
• Un serveur OPC UA. Si vous n'avez pas d'automate sous la main, le tutoriel inclut un petit serveur simulé.
• Un broker MQTT. Pour les tests, utilisez le serveur MQTT public gratuit sur broker.connectmq.com ; pour de vraies données d'usine, utilisez des identifiants privés issus d'un broker MQTT managé.

python -m venv .venv
source .venv/bin/activate        # sous Windows : .venv\Scripts\activate
pip install asyncua "paho-mqtt>=2.0"

Étape 1 : un serveur OPC UA simulé (facultatif)

Si vous disposez déjà d'un serveur OPC UA, passez cette étape et notez son URL d'endpoint et ses identifiants de nœuds. Sinon, ce script expose trois tags variables sur opc.tcp://localhost:4840.

# sim_server.py - un petit serveur OPC UA avec des valeurs CNC variables
import asyncio
import math
import random
from asyncua import Server, ua

async def main():
    server = Server()
    await server.init()
    server.set_endpoint("opc.tcp://0.0.0.0:4840/sim/")
    idx = await server.register_namespace("urn:demo:cnc")

    cnc = await server.nodes.objects.add_object(ua.NodeId("Line1.CNC1", idx), "CNC1")
    temp = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleTemp", idx), "SpindleTemp", 40.0)
    speed = await cnc.add_variable(ua.NodeId("Line1.CNC1.SpindleSpeed", idx), "SpindleSpeed", 0.0)
    count = await cnc.add_variable(ua.NodeId("Line1.CNC1.PartCount", idx), "PartCount", 0)

    async with server:
        t = 0
        while True:
            t += 1
            await temp.write_value(round(45 + 5 * math.sin(t / 20) + random.random(), 2))
            await speed.write_value(float(random.choice([0, 8000, 12000])))
            if t % 10 == 0:
                await count.write_value(await count.read_value() + 1)
            await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

Comme les identifiants de nœuds utilisent des chaînes dans le namespace enregistré (index 2 ici), ils apparaîtront sous la forme ns=2;s=Line1.CNC1.SpindleTemp, et ainsi de suite. Sur un vrai automate, parcourez l'espace d'adressage avec un outil client comme UaExpert pour trouver les identifiants exacts.

Étape 2 : la table de correspondance des tags

Gardez la correspondance dans la configuration, pas dans le code. Chaque entrée associe un nœud OPC UA au chemin de topic et à l'unité que les consommateurs doivent voir. L'organisation des topics suit une hiérarchie de style ISA-95 (site, zone, ligne, cellule), expliquée en détail dans notre guide sur les topics MQTT ISA-95.

{
  "opcua_endpoint": "opc.tcp://localhost:4840/sim/",
  "topic_root": "connectmq/demo/a91c/plant1",
  "tags": [
    { "node": "ns=2;s=Line1.CNC1.SpindleTemp",  "path": "machining/line1/cnc1/spindle_temp",  "unit": "degC" },
    { "node": "ns=2;s=Line1.CNC1.SpindleSpeed", "path": "machining/line1/cnc1/spindle_speed", "unit": "rpm" },
    { "node": "ns=2;s=Line1.CNC1.PartCount",    "path": "machining/line1/cnc1/part_count",    "unit": "count" }
  ]
}

Enregistrez-le sous tags.json. Remplacez a91c par votre propre chaîne aléatoire pour que vos topics n'entrent pas en collision avec ceux d'autres utilisateurs du serveur public. Les topics obtenus ont huit niveaux, le maximum autorisé par le serveur MQTT public gratuit ; sur un broker privé, vous avez plus de marge.

Étape 3 : le code de la passerelle OPC UA vers MQTT

# gateway.py - abonnement OPC UA -> publication MQTT
import asyncio
import json
import logging
import ssl
import uuid
from datetime import timezone

import paho.mqtt.client as mqtt
from asyncua import Client

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

CONFIG = json.load(open("tags.json"))
MQTT_HOST = "broker.connectmq.com"
MQTT_PORT = 8883
MQTT_USER = None          # production: "YOUR_USERNAME"
MQTT_PASS = None          # production: "YOUR_PASSWORD"
CLIENT_ID = f"opcua-gw-{uuid.uuid4().hex[:8]}"   # production: "YOUR_CLIENT_ID"
STATUS_TOPIC = f"{CONFIG['topic_root']}/gateway/status"


def build_mqtt() -> mqtt.Client:
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                         protocol=mqtt.MQTTv5)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
    client.will_set(STATUS_TOPIC, json.dumps({"online": False}), qos=1, retain=True)

    def on_connect(client, userdata, flags, reason_code, properties):
        log.info("MQTT connected: %s", reason_code)
        client.publish(STATUS_TOPIC, json.dumps({"online": True}), qos=1, retain=True)

    def on_disconnect(client, userdata, flags, reason_code, properties):
        log.warning("MQTT disconnected: %s", reason_code)

    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)
    client.connect(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()
    return client


class DataChangeHandler:
    """asyncua appelle datachange_notification à chaque changement d'un élément surveillé."""

    def __init__(self, mqtt_client, node_map):
        self.mqtt = mqtt_client
        self.node_map = node_map

    def datachange_notification(self, node, val, data):
        tag = self.node_map[node.nodeid.to_string()]
        dv = data.monitored_item.Value
        ts = dv.SourceTimestamp or dv.ServerTimestamp
        payload = {
            "value": val,
            "unit": tag["unit"],
            "quality": dv.StatusCode.name,
            "ts": ts.replace(tzinfo=timezone.utc).isoformat() if ts else None,
        }
        topic = f"{CONFIG['topic_root']}/{tag['path']}"
        self.mqtt.publish(topic, json.dumps(payload), qos=1, retain=True)
        log.info("%s -> %s", topic, payload["value"])


async def run_opcua(mqtt_client):
    node_map = {t["node"]: t for t in CONFIG["tags"]}
    while True:
        try:
            async with Client(url=CONFIG["opcua_endpoint"]) as opc:
                log.info("OPC UA connected to %s", CONFIG["opcua_endpoint"])
                handler = DataChangeHandler(mqtt_client, node_map)
                sub = await opc.create_subscription(500, handler)  # intervalle de publication en ms
                nodes = [opc.get_node(n) for n in node_map]
                await sub.subscribe_data_change(nodes)
                while True:                       # maintenir la session, détecter les pannes
                    await asyncio.sleep(5)
                    await opc.check_connection()
        except Exception as exc:                  # redémarrage du serveur, perte réseau, etc.
            log.error("OPC UA error: %s - retrying in 5 s", exc)
            await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(run_opcua(build_mqtt()))

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

• Les retained messages permettent à un tableau de bord qui se connecte plus tard de voir immédiatement la dernière valeur de chaque tag.
• Le Last Will sur le topic de statut passe à {"online": false} si la passerelle meurt sans déconnexion propre, pour que les consommateurs distinguent les données périmées des données en temps réel.
• Le code de statut OPC UA accompagne chaque valeur. Une mesure Good et une mesure BadCommunicationError ne doivent jamais se ressembler en aval.
• check_connection() lève une exception quand la session est perdue, ce qui nous fait entrer dans la boucle de nouvelle tentative et reconstruit l'abonnement.

Étape 4 : exécuter et vérifier

Lancez le serveur simulé, puis la passerelle, dans deux terminaux :

python sim_server.py
python gateway.py

Dans un troisième terminal, abonnez-vous à tout ce qui se trouve sous votre racine :

mosquitto_sub -h broker.connectmq.com -p 1883 -i opcua-viewer-a91c \
  -t "connectmq/demo/a91c/plant1/#" -v

Vous devriez voir des lignes comme celles-ci :

connectmq/demo/a91c/plant1/machining/line1/cnc1/spindle_temp {"value": 47.31, "unit": "degC", "quality": "Good", "ts": "2026-10-16T09:12:04.118000+00:00"}
connectmq/demo/a91c/plant1/gateway/status {"online": true}

Passer en production

Le serveur MQTT public gratuit est lisible par n'importe qui : il est réservé aux données d'exemple. Pour de vraies données d'automates, créez un broker privé, puis renseignez MQTT_USER, MQTT_PASS et CLIENT_ID avec les identifiants du portail. ConnectMQ lie les client IDs aux identifiants délivrés : utilisez donc le client ID généré plutôt qu'un identifiant aléatoire.

Liste de contrôle avant d'installer la passerelle sur un réseau d'usine :

Sur ConnectMQ, l'explorateur de topics en temps réel du portail affiche votre arborescence ISA-95 au fur et à mesure qu'elle se remplit, et la vue des appareils regroupe les machines par niveau ISA-95, ce qui permet de vérifier rapidement que votre correspondance de topics est cohérente d'une ligne à l'autre. Des règles d'alerte à seuil peuvent ensuite signaler, par exemple, une température de broche au-delà d'une limite, sans écrire de code supplémentaire. Pour situer cette passerelle dans la vue d'ensemble, consultez notre guide de l'IoT industriel avec MQTT.

Dépannage

• BadIdentityTokenRejected ou BadUserAccessDenied : le serveur n'autorise pas l'accès anonyme. Fournissez un utilisateur avec Client.set_user() et set_password(), ou configurez des certificats.
• BadNodeIdUnknown : la chaîne d'identifiant de nœud est incorrecte, souvent à cause de l'index de namespace. Les index de namespace peuvent changer d'une version de serveur à l'autre ; résolvez-les par URI de namespace avec get_namespace_index() au démarrage.
• Les valeurs arrivent mais les horodatages sont nuls : certains serveurs ne renseignent pas d'horodatage source. Le code se rabat alors sur l'horodatage serveur.
• MQTT se connecte puis se déconnecte aussitôt : sur le serveur public, vérifiez que votre client ID est unique et non vide. Sur un broker privé, vérifiez que le client ID correspond aux identifiants délivrés.
• Topics refusés sur le serveur public : vous avez dépassé huit niveaux de topic. Raccourcissez la racine ou le chemin.

Exécutez votre passerelle OPC UA sur ConnectMQ

Créez gratuitement un broker sur app.connectmq.com avec l'offre Serverless, sans carte bancaire. Copiez le nom d'utilisateur, le mot de passe et le client ID générés dans gateway.py, et en quelques minutes vos tags d'automates apparaissent dans un broker MQTT cloud privé, protégé par TLS et prêt pour les tableaux de bord et les historians.

Questions fréquentes

Faut-il un produit commercial pour une passerelle OPC UA vers MQTT ?

Non. Des bibliothèques open source comme asyncua et paho-mqtt suffisent pour de nombreux projets. Les passerelles commerciales ajoutent des interfaces graphiques, le store and forward et le support éditeur, ce qui peut compter à grande échelle.

La passerelle doit-elle interroger les tags OPC UA ou utiliser des abonnements ?

Utilisez des abonnements. Le serveur échantillonne les valeurs et ne pousse que les changements, ce qui réduit la charge sur l'automate et le réseau par rapport à des lectures cycliques.

La passerelle peut-elle écrire des valeurs dans l'automate ?

Techniquement oui, en s'abonnant à un topic de commande et en appelant write_value, mais traitez-le comme une fonctionnalité distincte et soigneusement sécurisée. La plupart des déploiements gardent la passerelle en lecture seule.

Quelle QoS MQTT utiliser pour les données de tags d'automates ?

La QoS 1 est le choix habituel : la livraison est garantie au moins une fois et le surcoût est faible. Associez-la aux retained messages pour que les nouveaux consommateurs obtiennent immédiatement l'état actuel.