La détection d'anomalies en temps réel en Python peut tenir en étonnamment peu de code : un abonné, une fenêtre glissante par capteur et un appel de publication pour les alertes. Dans ce tutoriel, vous allez construire un simulateur de capteurs et un détecteur en streaming qui évalue chaque message MQTT dès son arrivée et publie des événements d'anomalie anti-rebond, le tout sur le serveur MQTT public gratuit, utilisable dès maintenant pour les tests.
Ce que vous allez construire
L'installation comporte trois parties reliées par des topics :
- Un simulateur publie chaque seconde des mesures de température et de vibration pour quelques machines, avec des défauts injectés de temps en temps.
- Un détecteur s'abonne à toute la télémétrie, tient des statistiques glissantes par machine et par signal, et calcule un z-score robuste pour chaque mesure.
- Un topic d'alertes reçoit un événement JSON lorsqu'un signal reste anormal sur plusieurs mesures consécutives, puis un événement de rétablissement lorsqu'il revient à la normale.
connectmq/demo/<id>/plant/<machine>/telemetry simulator -> detector
connectmq/demo/<id>/plant/<machine>/score detector, every reading
connectmq/demo/<id>/plant/alerts detector, on state change
Publier un score pour chaque mesure et des alertes uniquement lors des changements garde des tableaux de bord informatifs sans submerger les équipes de notifications.
Prérequis
- Python 3.9 ou plus récent
- paho-mqtt 2.x
- Un accès réseau à broker.connectmq.com sur le port 1883 (ou 8883 pour TLS)
python -m venv venv
source venv/bin/activate # sous Windows : venv\Scripts\activate
pip install "paho-mqtt>=2.0"
Choisissez un identifiant aléatoire pour votre préfixe de topic, par exemple a71c2e. L'espace public de broker.connectmq.com est lisible par tous : n'utilisez donc que des données synthétiques. Le guide du serveur MQTT public gratuit en indique les limites, dont 10 messages par seconde et par client, que ce tutoriel respecte largement.
Étape 1 : le simulateur de capteurs
Enregistrez ce fichier sous simulator.py. Il produit un bruit réaliste, une dérive lente de type journalier et des épisodes de défaut aléatoires : un pic de température, un saut de vibration et un capteur bloqué.
import json
import math
import random
import time
import uuid
import paho.mqtt.client as mqtt
PREFIX = "connectmq/demo/a71c2e/plant"
MACHINES = ["press1", "press2", "pump1"]
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id=f"sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
state = {m: {"fault": None, "left": 0, "stuck": None} for m in MACHINES}
t = 0
while True:
for m in MACHINES:
s = state[m]
if s["left"] == 0 and random.random() < 0.01:
s["fault"] = random.choice(["spike", "vibration", "stuck"])
s["left"] = random.randint(5, 15)
temp = 60 + 3 * math.sin(t / 300) + random.gauss(0, 0.4)
vib = 2.0 + random.gauss(0, 0.08)
if s["left"] > 0:
if s["fault"] == "spike":
temp += 8
elif s["fault"] == "vibration":
vib += 0.9
elif s["fault"] == "stuck":
s["stuck"] = s["stuck"] or round(temp, 2)
temp = s["stuck"]
s["left"] -= 1
if s["left"] == 0:
s["fault"], s["stuck"] = None, None
payload = {"ts": time.time(), "tempC": round(temp, 2), "vibMm": round(vib, 3)}
client.publish(f"{PREFIX}/{m}/telemetry", json.dumps(payload), qos=0)
t += 1
time.sleep(1)
Lancez-le et laissez-le tourner. Vous pouvez observer les messages avec le client mosquitto :
mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-a71c2e -t "connectmq/demo/a71c2e/plant/#" -v
Étape 2 : choisir une méthode de score pour la détection d'anomalies en temps réel en Python
Un z-score classique utilise la moyenne et l'écart-type, qui sont eux-mêmes tirés par les valeurs aberrantes que vous cherchez à détecter. Un z-score robuste utilise à la place la médiane et l'écart absolu médian (MAD) :
robust_z = 0.6745 * (x - median) / MAD
La constante met le MAD à l'échelle pour que le score soit comparable à un z-score standard sur des données de distribution normale. Un seuil autour de 3,5 est un point de départ courant. Deux contrôles supplémentaires détectent les défaillances que le score manque :
- Détection de blocage. Si les N dernières mesures sont identiques, le capteur est probablement figé, même si la valeur semble normale.
- Préchauffage. N'évaluez rien tant que la fenêtre ne contient pas assez d'échantillons, sinon le bruit initial produit de fausses alertes.
Pour une comparaison du z-score, de l'IQR et de l'EWMA avec leurs compromis, consultez les méthodes de détection de valeurs aberrantes dans les données de capteurs.
Étape 3 : le détecteur en streaming
Enregistrez ce fichier sous detector.py. Il tient une fenêtre par machine et par signal, évalue chaque mesure, publie les scores et émet des alertes selon une règle de persistance simple : déclencher après 3 anomalies consécutives, lever après 5 mesures normales consécutives.
import json
import statistics
import time
import uuid
from collections import defaultdict, deque
import paho.mqtt.client as mqtt
PREFIX = "connectmq/demo/a71c2e/plant"
WINDOW = 120 # mesures conservées par signal
WARMUP = 30 # mesures avant le début de l'évaluation
THRESHOLD = 3.5 # limite du z-score robuste
RAISE_AFTER = 3
CLEAR_AFTER = 5
STUCK_AFTER = 8
windows = defaultdict(lambda: deque(maxlen=WINDOW))
recent = defaultdict(lambda: deque(maxlen=STUCK_AFTER))
streak = defaultdict(int)
calm = defaultdict(int)
active = set()
def robust_z(window, x):
med = statistics.median(window)
mad = statistics.median(abs(v - med) for v in window) or 1e-6
return 0.6745 * (x - med) / mad
def evaluate(key, x):
w = windows[key]
recent[key].append(x)
stuck = len(recent[key]) == STUCK_AFTER and len(set(recent[key])) == 1
if len(w) < WARMUP:
w.append(x)
return 0.0, False
z = robust_z(w, x)
anomalous = abs(z) > THRESHOLD or stuck
if not anomalous:
w.append(x) # n'apprendre qu'à partir des mesures normales
return z, anomalous
def on_connect(client, userdata, flags, reason_code, properties):
print("connected:", reason_code)
client.subscribe(f"{PREFIX}/+/telemetry", qos=0)
def on_message(client, userdata, msg):
machine = msg.topic.split("/")[-2]
try:
data = json.loads(msg.payload)
except json.JSONDecodeError:
return
scores = {}
for signal in ("tempC", "vibMm"):
if signal not in data:
continue
key = (machine, signal)
z, anomalous = evaluate(key, data[signal])
scores[signal] = round(z, 2)
if anomalous:
streak[key] += 1
calm[key] = 0
else:
calm[key] += 1
streak[key] = 0
if key not in active and streak[key] >= RAISE_AFTER:
active.add(key)
publish_alert(client, machine, signal, "raised", z, data[signal])
elif key in active and calm[key] >= CLEAR_AFTER:
active.discard(key)
publish_alert(client, machine, signal, "cleared", z, data[signal])
client.publish(f"{PREFIX}/{machine}/score", json.dumps(scores), qos=0)
def publish_alert(client, machine, signal, state, z, value):
event = {"ts": time.time(), "machine": machine, "signal": signal,
"state": state, "z": round(z, 2), "value": value,
"detector": "robust-z-v1", "threshold": THRESHOLD}
client.publish(f"{PREFIX}/alerts", json.dumps(event), qos=1)
print("ALERT", event)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id=f"detector-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=30)
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()
Après environ 30 secondes de préchauffage, vous devriez voir des lignes ALERT chaque fois que le simulateur injecte un défaut, suivies d'un événement cleared quelques secondes après la fin de celui-ci.
Étape 4 : suivre les alertes
Abonnez-vous au topic d'alertes depuis n'importe où :
mosquitto_sub -h broker.connectmq.com -p 1883 -i alerts-a71c2e -t "connectmq/demo/a71c2e/plant/alerts" -v
Les alertes sont publiées en QoS 1 pour survivre à une brève coupure réseau ; les scores utilisent la QoS 0, car le suivant arrive de toute façon une seconde plus tard. L'article les niveaux de QoS MQTT expliqués détaille ce compromis.
Étape 5 : ajouter un tableau de bord simple
Les alertes vous disent qu'il s'est passé quelque chose ; un graphique vous montre comment cela a évolué. Comme le détecteur publie un score pour chaque mesure, n'importe quel tableau de bord compatible MQTT peut tracer côte à côte les valeurs brutes et les scores sans toucher au détecteur. Quelques options qui fonctionnent bien :
- Node-RED avec les nœuds dashboard. Abonnez-vous à
+/telemetryet+/score, séparez par machine et alimentez des graphiques en courbes. Un nœud switch peut transformer le topic d'alertes en indicateur d'état coloré pour chaque machine. - Grafana avec un plugin de source de données MQTT. Pratique lorsque vous utilisez déjà Grafana pour d'autres métriques et voulez afficher les scores d'anomalie sur les mêmes écrans.
- Une page web utilisant MQTT over WebSocket. Connectez-vous avec MQTT.js à
wss://broker.connectmq.com:8084/mqttet dessinez les scores avec n'importe quelle bibliothèque de graphiques.
En regardant les graphiques, observez le comportement du score juste avant et juste après une alerte. Un détecteur propre montre un score calme qui bondit nettement pendant un défaut puis redevient calme. Un score bruité qui oscille la plupart du temps près du seuil signifie que le seuil est trop serré pour ce signal, ou que la fenêtre est trop courte pour estimer la référence de façon fiable.
Sur un broker ConnectMQ privé, l'explorateur de topics en direct vous donne le même aperçu rapide sans rien construire, ce qui est pratique pendant le réglage.
Pourquoi évaluer chaque mesure et alerter rarement
Il est tentant de ne publier que les alertes pour économiser de la bande passante. En pratique, c'est le flux de scores qui vous permet de régler le détecteur plus tard. Vous pouvez rejouer une journée de scores, vous demander combien d'alertes un autre seuil aurait produites et comparer équitablement des détecteurs. Les alertes, elles, sont destinées à des personnes et doivent être rares et pertinentes ; c'est pourquoi le détecteur applique une règle de persistance avant de les déclencher. Garder ces deux sorties sur des topics séparés permet à chaque consommateur de s'abonner exactement à ce dont il a besoin.
Liste de contrôle du réglage
| Paramètre | Effet si trop bas | Effet si trop haut | Point de départ |
|---|---|---|---|
| WINDOW | La référence saute avec le bruit | Lent à s'adapter aux changements légitimes | 2 à 5 minutes de données |
| THRESHOLD | Nombreuses fausses alertes | Manque les petits défauts | 3,5 |
| RAISE_AFTER | Alertes sur des parasites isolés | Détection retardée | 3 mesures |
| CLEAR_AFTER | Alertes qui oscillent | Alertes ouvertes trop longtemps | 5 mesures |
| STUCK_AFTER | Signale les signaux à variation lente | Les capteurs figés passent inaperçus | Dépend de la résolution |
Dépannage
- Aucun message n'arrive. Vérifiez que le simulateur et le détecteur utilisent le même préfixe et que chaque client dispose d'un client ID unique et non vide. Deux clients ayant le même ID se déconnectent mutuellement en boucle.
- Alertes permanentes sur un signal. Le signal est peut-être quantifié, de sorte que le MAD vaut zéro ; la petite valeur plancher fait alors exploser le score. Relevez le plancher à la résolution du capteur.
- Les alertes ne se lèvent jamais. La fenêtre a cessé d'apprendre, car chaque valeur est considérée comme anormale après un changement de niveau légitime. Ajoutez une réinitialisation périodique de la référence ou utilisez l'EWMA pour les signaux à variation lente.
- Le détecteur prend du retard. Python traite des milliers de messages par seconde avec ce schéma, mais des modèles lourds dans le callback bloquent la boucle. Déplacez l'évaluation vers une file de travail, ou exécutez plusieurs détecteurs avec les abonnements partagés de MQTT 5.
De la démo à la production
Passez à un broker privé avant de faire transiter la moindre donnée réelle. Remplacez la connexion par une connexion TLS avec identifiants :
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set() # utilise le magasin de CA du système ; ConnectMQ utilise un certificat Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)
Utilisez le client ID généré pour ces identifiants, car ConnectMQ lie les client IDs aux identifiants émis. Un broker MQTT managé signifie aussi que le détecteur n'est pas le seul élément que vous devez maintenir en fonctionnement. Lorsque les statistiques par signal ne suffisent plus, l'étape suivante est un modèle multivarié comme Isolation Forest sur des données de capteurs. Pour une vue d'ensemble de l'architecture, consultez les méthodes et l'architecture de détection d'anomalies IoT ainsi que le guide IA et IoT.
Démarrer avec ConnectMQ
Créez un broker avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire, copiez le nom d'utilisateur, le mot de passe et le client ID générés dans le détecteur, et suivez scores et alertes dans l'explorateur de topics en direct.
Questions fréquentes
Pourquoi utiliser un z-score robuste plutôt qu'un z-score classique ?
La médiane et le MAD sont à peine affectés par les valeurs aberrantes que vous cherchez à détecter, si bien que la référence reste stable pendant un défaut. Une moyenne et un écart-type classiques sont attirés vers l'anomalie et la masquent.
Combien de messages par seconde ce détecteur Python peut-il traiter ?
Les statistiques glissantes sont peu coûteuses : le débit est donc généralement limité par le réseau et l'analyse JSON plutôt que par le calcul des scores. Pour des volumes plus élevés, exécutez plusieurs instances du détecteur avec des abonnements partagés sur un broker qui les prend en charge.
Puis-je exécuter le détecteur sur un Raspberry Pi ?
Oui. Le script n'utilise que la bibliothèque standard et paho-mqtt, et tourne sans difficulté sur un Raspberry Pi faisant office de passerelle edge.
Les alertes doivent-elles être des retained messages ?
Pas les événements d'alerte eux-mêmes. Si vous voulez que les nouveaux tableaux de bord voient l'état actuel, publiez séparément un topic de statut retenu par machine contenant la liste des alertes actives.
