Séries temporelles industrielles : échantillonnage, compression et stockage

Les séries temporelles industrielles sont le flux de mesures horodatées que machines, capteurs et contrôleurs produisent à chaque seconde de chaque poste. Bien les gérer implique de choisir à quelle fréquence échantillonner, ce qu'il faut écarter, comment horodater et où conserver les données. Cet article passe en revue chacune de ces décisions et vous fournit du code fonctionnel pour appliquer une compression deadband aux données MQTT avant qu'elles n'atteignent un historian.

Ce qui rend les séries temporelles industrielles différentes

Toutes les séries temporelles ont la même forme : un horodatage, un identifiant et une valeur. Les données industrielles ajoutent plusieurs complications que les données de supervision IT présentent rarement.

• Des vitesses de variation très différentes. Le niveau d'une cuve évolue lentement sur des heures ; un signal vibratoire change des milliers de fois par seconde. Une seule politique d'échantillonnage ne peut pas convenir aux deux.
• Des exigences de rétention longues. Les enquêtes qualité, les enregistrements réglementaires, l'entraînement de modèles ou le calibrage d'un jumeau numérique nécessitent souvent des années d'historique, pas des semaines.
• La qualité compte autant que la valeur. Une valeur issue d'un capteur déconnecté n'est pas une mesure. Les systèmes industriels associent un indicateur de qualité à chaque point.
• Des données d'événements et d'état mêlées aux données analogiques. États machines, alarmes, numéros de lot et changements de recette sont épars et doivent être croisés avec des signaux analogiques denses.
• Des horloges que vous ne maîtrisez pas. PLC, passerelles et serveurs ont chacun leur horloge, et elles dérivent.

Bien traiter ces points dès le départ coûte bien moins cher que de réparer plus tard des années d'historique incohérent.

Choisir les cadences d'échantillonnage

La cadence d'échantillonnage doit suivre la physique du signal et la question à laquelle vous voulez répondre, pas le temps de cycle par défaut du PLC.

Deux règles aident. D'abord, le principe de Nyquist : pour voir une variation, échantillonnez au moins deux fois plus vite qu'elle, et en pratique plusieurs fois plus vite. Ensuite, ne diffusez pas en continu des formes d'onde brutes à haute fréquence via MQTT. Calculez des indicateurs à l'edge et publiez-les ; n'envoyez des captures brutes qu'à la demande ou sur événement.

Pour les données discrètes, le report by exception est le modèle naturel. L'article sur le report by exception explique comment il réduit le trafic sans perte d'information.

Compression : garder le signal, éliminer le bruit

Dans les historians industriels, la compression désigne généralement un filtrage avec perte à l'ingestion, pas des fichiers zip. L'objectif est de stocker moins de points tout en pouvant reconstruire le signal avec une tolérance connue.

Filtrage par deadband (exception)

La méthode la plus simple : ne stocker un nouveau point que lorsque la valeur s'écarte de la dernière valeur stockée de plus d'un seuil. Une deadband de 0,2 degC sur une température signifie que les fluctuations inférieures à 0,2 degC ne sont pas stockées. La plupart des implémentations stockent aussi un point après un délai maximal même si rien n'a changé (un heartbeat), pour que les consommateurs puissent distinguer un signal stable d'un signal mort.

Compression swinging door

Le swinging door trending, utilisé par de nombreux historians commerciaux, ne conserve un point que lorsqu'une droite partant du dernier point stocké ne peut plus représenter les valeurs intermédiaires dans la tolérance. Il préserve bien mieux les pentes qu'une simple deadband : les rampes sont stockées avec quelques points au lieu de nombreux.

Choisir les tolérances

Fixez la tolérance par rapport à la précision du capteur, pas selon un pourcentage arbitraire. Filtrer sous le plancher de bruit du capteur revient à stocker du bruit ; filtrer bien au-dessus fait perdre de vrais changements de process. Documentez la tolérance par type de tag dans votre modèle de données IIoT pour qu'elle soit relue comme n'importe quel autre attribut du modèle.

Compression deadband sur des données MQTT en Python

Cet abonné lit de la télémétrie JSON depuis MQTT, applique une deadband avec heartbeat et écrit les points retenus en line protocol InfluxDB sur la sortie standard, que vous pouvez rediriger vers un fichier ou un processus d'écriture. Il utilise le serveur MQTT public gratuit, réservé aux tests et aux prototypes, et intègre un simulateur de démonstration.

import json
import math
import random
import threading
import time
import uuid

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PREFIX = f"connectmq/demo/{uuid.uuid4().hex[:8]}/plant1/line1"
DEADBAND = {"temperature": 0.2, "pressure": 0.05}
HEARTBEAT_S = 60

last_stored = {}  # topic -> (valeur, ts_secondes)


def keep(topic, metric, value, ts):
    prev = last_stored.get(topic)
    if prev is None:
        return True
    band = DEADBAND.get(metric, 0.0)
    return abs(value - prev[0]) > band or ts - prev[1] >= HEARTBEAT_S


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/+", qos=1)


def on_message(client, userdata, msg):
    asset, metric = msg.topic.split("/")[-2:]
    p = json.loads(msg.payload)
    ts = p["ts"] / 1000
    if keep(msg.topic, metric, p["value"], ts):
        last_stored[msg.topic] = (p["value"], ts)
        print(f'{metric},asset={asset},quality={p["quality"]} value={p["value"]} {int(p["ts"] * 1e6)}')


def simulate(pub):
    t0 = time.time()
    while True:
        t = time.time() - t0
        temp = 60 + 5 * math.sin(t / 120) + random.gauss(0, 0.05)
        pres = 2.5 + random.gauss(0, 0.01)
        ts = int(time.time() * 1000)
        for metric, val in [("temperature", temp), ("pressure", pres)]:
            body = {"value": round(val, 3), "ts": ts, "quality": "GOOD"}
            pub.publish(f"{PREFIX}/pump1/{metric}", json.dumps(body), qos=0)
        time.sleep(1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="tsdb-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
threading.Thread(target=simulate, args=(client,), daemon=True).start()
client.loop_forever()

Laissez-le tourner quelques minutes et comptez les lignes : la pression, qui ne présente que du bruit sous sa deadband, produit environ un point par minute (le heartbeat), tandis que la température produit des points au rythme de la sinusoïde. C'est exactement le comportement attendu d'un historian : dense là où le process bouge, épars là où il ne bouge pas.

Horodatage et ordre des données

Horodatez toujours à la source, au plus près de la mesure, en UTC avec les millisecondes. L'heure d'arrivée au broker n'est pas l'heure de mesure, en particulier lorsqu'une passerelle edge met des données en tampon pendant une coupure réseau et les transmet plus tard. Le modèle store and forward repose sur les horodatages source pour replacer les données tardives au bon endroit.

Synchronisez les horloges avec NTP ou PTP sur les passerelles et, si possible, sur les contrôleurs. Attendez-vous à des arrivées dans le désordre et assurez-vous que votre stockage accepte les écritures dans le passé. Certaines bases de séries temporelles gèrent efficacement les écritures tardives ; d'autres les pénalisent.

Options de stockage des séries temporelles industrielles

Beaucoup d'usines combinent un historian ou une TSDB pour les données récentes avec un lake pour l'analytique à long terme. L'article sur l'historian MQTT détaille la construction du volet historian directement à partir des topics MQTT.

Faire passer les données du broker au stockage

Un broker est un routeur, pas une base de données : les messages le traversent et les retained messages ne conservent que la dernière valeur par topic. Pour stocker l'historique, vous exécutez un processus d'écriture qui s'abonne et insère. Telegraf est un choix courant :

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["connectmq/plant1/#"]
  client_id = "YOUR_CLIENT_ID"
  username = "YOUR_USERNAME"
  password = "YOUR_PASSWORD"
  qos = 1
  data_format = "json"
  json_time_key = "ts"
  json_time_format = "unix_ms"
  tag_keys = ["quality"]

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/+/+/+/+"
    tags = "_/site/line/asset/_"
    measurement = "_/_/_/_/measurement"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "YOUR_INFLUX_TOKEN"
  organization = "plant"
  bucket = "telemetry"

Avec ConnectMQ comme broker MQTT managé, Telegraf n'est qu'un client authentifié de plus, en TLS sur le port 8883. Utilisez la QoS 1 pour l'abonnement du processus d'écriture afin qu'une brève déconnexion ne fasse pas disparaître des données en silence, et gardez sa session persistante si vous voulez que les messages soient mis en file pendant un redémarrage.

Checklist pour les séries temporelles industrielles

1. Classez chaque signal : analogique lent, analogique rapide, discret ou haute fréquence.
2. Définissez l'échantillonnage et la deadband par classe, en fonction de la précision des capteurs.
3. Ajoutez un heartbeat pour distinguer les signaux stables des signaux morts.
4. Horodatez à la source, en UTC avec les millisecondes.
5. Publiez la qualité avec chaque valeur.
6. Calculez des indicateurs à l'edge pour les signaux en kHz.
7. Choisissez le stockage chaud et froid, et définissez rétention et sous-échantillonnage.
8. Testez les écritures tardives et désordonnées avant la mise en service.

Pour l'architecture qui entoure tout cela, consultez le guide de l'IoT industriel avec MQTT.

Démarrez avec ConnectMQ

Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, puis connectez votre passerelle et un processus d'écriture Telegraf ou Python avec les identifiants générés. L'explorateur de topics en direct de ConnectMQ vous aide à vérifier les débits de publication et les payloads avant de vous engager sur une politique d'échantillonnage et de compression.

Questions fréquentes

Que sont les séries temporelles industrielles ?

Ce sont des données de mesures et d'événements horodatées produites par des équipements industriels, comme les températures, les pressions, les vitesses, les états et les compteurs. Elles sont généralement stockées avec un indicateur de qualité et conservées longtemps pour l'exploitation, la qualité et l'analytique.

Quelle est une bonne valeur de deadband ?

Partez d'une valeur proche de la précision ou du niveau de bruit du capteur pour la mesure concernée, puis vérifiez que les changements de process importants restent visibles dans les données stockées. Documentez la valeur par type de tag et revoyez-la quand les capteurs changent.

Un broker MQTT peut-il stocker l'historique des séries temporelles ?

Non. Un broker route les messages et conserve au plus la dernière valeur retenue par topic. Stockez l'historique avec un abonné comme Telegraf, un connecteur d'historian ou votre propre processus d'écriture.

Faut-il diffuser les données vibratoires brutes via MQTT ?

Généralement pas en continu. Calculez à l'edge des indicateurs comme le RMS, la crête et l'énergie par bande, publiez-les, puis n'envoyez des captures de forme d'onde brute qu'à la demande ou lorsqu'un événement les déclenche.