MQTT ve Python ile Gerçek Zamanlı Anomali Tespiti

Python ile gerçek zamanlı anomali tespiti şaşırtıcı derecede küçük olabilir: bir abone, her sensör için bir kayan pencere ve uyarılar için bir publish çağrısı. Bu eğitimde bir sensör simülatörü ve her MQTT mesajını geldiği anda puanlayıp sıçrama korumalı (debounced) anomali olayları yayınlayan bir akış dedektörü geliştireceksiniz; hepsi hemen şimdi kullanabileceğiniz genel bir test broker'ı üzerinde.

Ne Geliştireceksiniz

Kurulum, topic'lerle birbirine bağlanan üç parçadan oluşur:

Bir simülatör, birkaç makine için saniyede bir kez sıcaklık ve titreşim okumaları yayınlar ve ara sıra arıza enjekte eder.
Bir dedektör, tüm telemetriye abone olur, her makine ve her sinyal için kayan istatistikler tutar ve her okuma için sağlam (robust) bir z-skoru hesaplar.
Bir uyarı topic'i, bir sinyal ardışık birkaç okuma boyunca anormal kaldığında bir JSON olayı, normale döndüğünde ise bir toparlanma olayı alır.

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

Her okuma için bir skor, uyarıları ise yalnızca değişikliklerde yayınlamak, insanları bildirimlere boğmadan dashboard'ları bilgilendirici tutar.

Ön Koşullar

Python 3.9 veya üstü
paho-mqtt 2.x
broker.connectmq.com adresine 1883 portundan (veya TLS için 8883) ağ erişimi

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

Topic önekiniz için rastgele bir kimlik seçin, örneğin a71c2e. broker.connectmq.com'un genel alanı herkes tarafından okunabilir, bu yüzden yalnızca sentetik veri kullanın. Genel MQTT test broker'ı rehberi, client başına saniyede 10 mesaj dahil sınırlarını listeliyor; bu eğitim bu sınırın oldukça altında kalır.

Adım 1: Sensör Simülatörü

Bunu simulator.py olarak kaydedin. Gerçekçi gürültü, günlük döngüye benzer yavaş bir kayma ve rastgele arıza dönemleri üretir: bir sıcaklık sıçraması, bir titreşim basamağı ve takılı kalmış bir sensör.

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)

Çalıştırın ve açık bırakın. Mesajları mosquitto client'ı ile izleyebilirsiniz:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-a71c2e -t "connectmq/demo/a71c2e/plant/#" -v

Adım 2: Python ile Gerçek Zamanlı Anomali Tespiti için Puanlama Yöntemi Seçmek

Düz bir z-skoru ortalama ve standart sapmayı kullanır; ancak bunlar da yakalamak istediğiniz aykırı değerler tarafından çekiştirilir. Sağlam bir z-skoru bunların yerine medyanı ve medyan mutlak sapmayı (MAD) kullanır:

robust_z = 0.6745 * (x - median) / MAD

Sabit, MAD'i ölçekleyerek skorun normal dağılımlı veriler için standart bir z-skoruyla karşılaştırılabilir olmasını sağlar. 3,5 civarında bir eşik yaygın bir başlangıç noktasıdır. İki ek kontrol, skorun kaçırdığı arızaları yakalar:

Takılma tespiti. Son N okumanın hepsi aynıysa, değer normal görünse bile sensör muhtemelen donmuştur.
Isınma süresi. Pencerede yeterli örnek olana kadar puanlama yapmayın; aksi hâlde erken gürültü yanlış uyarılar üretir.

Z-skoru, IQR ve EWMA'nın ödünleşimleriyle birlikte karşılaştırması için sensör verisinde aykırı değer tespiti yöntemleri yazısına bakın.

Adım 3: Akış Dedektörü

Bunu detector.py olarak kaydedin. Her makine ve sinyal için bir pencere tutar, her okumayı puanlar, skorları yayınlar ve basit bir süreklilik kuralıyla uyarı üretir: ardışık 3 anomaliden sonra uyarı ver, ardışık 5 normal okumadan sonra temizle.

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          # sinyal başına tutulan okuma sayısı
WARMUP = 30           # puanlama başlamadan önceki okuma sayısı
THRESHOLD = 3.5       # sağlam z-skoru sınırı
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)  # yalnızca normal okumalardan öğren
    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()

Yaklaşık 30 saniyelik ısınma süresinden sonra, simülatör her arıza enjekte ettiğinde ALERT satırları, arıza bittikten birkaç saniye sonra da bir cleared olayı görmelisiniz.

Adım 4: Uyarıları İzlemek

Uyarı topic'ine istediğiniz yerden abone olun:

mosquitto_sub -h broker.connectmq.com -p 1883 -i alerts-a71c2e -t "connectmq/demo/a71c2e/plant/alerts" -v

Uyarılar kısa bir ağ kesintisinden etkilenmesin diye QoS 1 ile yayınlanır; skorlar ise QoS 0 kullanır, çünkü bir sonraki zaten bir saniye sonra gelir. MQTT QoS seviyeleri yazısı bu ödünleşimi derinlemesine ele alıyor.

Adım 5: Basit Bir Dashboard Eklemek

Uyarılar bir şey olduğunu söyler; bir grafik ise nasıl geliştiğini gösterir. Dedektör her okuma için bir skor yayınladığından, MQTT destekleyen herhangi bir dashboard dedektöre dokunmadan ham değerleri ve skorları yan yana çizebilir. İyi çalışan birkaç seçenek:

Dashboard düğümleriyle Node-RED. +/telemetry ve +/score topic'lerine abone olun, makineye göre ayırın ve çizgi grafiklere besleyin. Bir switch düğümü, uyarı topic'ini her makine için renkli bir durum göstergesine dönüştürebilir.
MQTT veri kaynağı eklentisiyle Grafana. Diğer metrikler için zaten Grafana kullanıyorsanız ve anomali skorlarını aynı ekranlarda görmek istiyorsanız faydalıdır.
MQTT over WebSocket kullanan bir tarayıcı sayfası. MQTT.js ile wss://broker.connectmq.com:8084/mqtt adresine bağlanın ve skorları herhangi bir grafik kütüphanesiyle çizin.

Grafiklere bakarken skorun bir uyarıdan hemen önce ve sonra nasıl davrandığına dikkat edin. Temiz bir dedektör, arıza sırasında keskin şekilde sıçrayan ve sonrasında yeniden sakinleşen sakin bir skor gösterir. Çoğu zaman eşiğin yakınında gezinen gürültülü bir skor, eşiğin o sinyal için fazla dar olduğu ya da pencerenin referansı güvenilir şekilde tahmin etmek için fazla kısa olduğu anlamına gelir.

Özel bir ConnectMQ broker'ında canlı topic gezgini, hiçbir şey geliştirmeden aynı hızlı görünümü sunar; bu da ayar yaparken kullanışlıdır.

Neden Her Okumayı Puanlayıp Nadiren Uyarı Vermeli

Bant genişliğinden tasarruf etmek için yalnızca uyarıları yayınlamak cazip gelebilir. Pratikte dedektörü sonradan ayarlamanızı sağlayan şey skor akışıdır. Bir günlük skoru yeniden oynatabilir, farklı bir eşiğin kaç uyarı üreteceğini sorgulayabilir ve dedektörleri adil şekilde karşılaştırabilirsiniz. Uyarılar ise insanlara gider ve nadir ve anlamlı olmalıdır; bu yüzden dedektör onları üretmeden önce süreklilik kuralı uygular. Bu iki çıktıyı ayrı topic'lerde tutmak, her tüketicinin tam olarak ihtiyaç duyduğu şeye abone olmasını sağlar.

Ayar Kontrol Listesi

Parametre	Çok düşükse etkisi	Çok yüksekse etkisi	Başlangıç noktası
WINDOW	Referans gürültüyle sıçrar	Meşru değişikliklere geç uyum sağlar	2 ila 5 dakikalık veri
THRESHOLD	Çok sayıda yanlış uyarı	Küçük arızaları kaçırır	3,5
RAISE_AFTER	Tekil aksaklıklarda uyarı verir	Gecikmeli tespit	3 okuma
CLEAR_AFTER	Uyarı gidip gelir	Uyarılar çok uzun açık kalır	5 okuma
STUCK_AFTER	Yavaş değişen sinyalleri işaretler	Donmuş sensörler fark edilmez	Çözünürlüğe bağlı

Sorun Giderme

Hiç mesaj gelmiyor. Simülatörün ve dedektörün aynı öneki kullandığını ve her client'ın benzersiz, boş olmayan bir client ID'ye sahip olduğunu kontrol edin. Aynı ID'ye sahip iki client birbirinin bağlantısını sürekli koparır.
Bir sinyalde sürekli uyarı. Sinyal kuantize olabilir ve MAD sıfır çıkar; bu durumda küçük taban değer skoru patlatır. Tabanı sensörün çözünürlüğüne yükseltin.
Uyarılar hiç temizlenmiyor. Meşru bir seviye kaymasından sonra her değer anormal sayıldığı için pencere öğrenmeyi bırakmıştır. Periyodik bir referans yenileme ekleyin ya da yavaş değişen sinyaller için EWMA kullanın.
Dedektör geride kalıyor. Python bu kalıpta saniyede binlerce mesajı işleyebilir, ancak callback içindeki ağır modeller döngüyü bloke eder. Puanlamayı bir iş kuyruğuna taşıyın veya MQTT 5 shared subscription'larıyla birkaç dedektör çalıştırın.

Demodan Üretime

Herhangi bir gerçek veri akmadan önce özel bir broker'a geçin. Bağlantıyı TLS ve kimlik bilgileriyle değiştirin:

client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # sistem CA deposunu kullanır; ConnectMQ bir Let's Encrypt sertifikası kullanır
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)

Bu kimlik bilgileri için oluşturulan client ID'yi kullanın, çünkü ConnectMQ client ID'leri verilen kimlik bilgilerine bağlar. Yönetilen bir MQTT broker, çalışır durumda tutmanız gereken tek şeyin dedektör olmaması anlamına da gelir. Tek sinyalli istatistikler yetersiz kalmaya başladığında sonraki adım, sensör verisinde Isolation Forest gibi çok değişkenli bir modeldir. Daha geniş mimari resim için IoT anomali tespiti yöntemleri ve mimarisi yazısına ve yapay zeka ve IoT rehberine bakın.

ConnectMQ ile Başlayın

app.connectmq.com adresinde ücretsiz Serverless planında kredi kartı gerektirmeden bir broker oluşturun, oluşturulan kullanıcı adını, parolayı ve client ID'yi dedektöre kopyalayın ve skorları ile uyarıları canlı topic gezgininde izleyin.

Sık Sorulan Sorular

Neden normal z-skoru yerine sağlam z-skoru kullanılmalı?

Medyan ve MAD, tespit etmeye çalıştığınız aykırı değerlerden neredeyse hiç etkilenmez; bu yüzden referans bir arıza sırasında kararlı kalır. Normal ortalama ve standart sapma ise anomaliye doğru çekilir ve onu gizler.

Bu Python dedektörü saniyede kaç mesaj işleyebilir?

Kayan istatistikler ucuzdur; bu yüzden işlem hacmi genellikle puanlamadan değil, ağ ve JSON ayrıştırmadan sınırlanır. Daha yüksek hacimler için, shared subscription destekleyen bir broker'da birkaç dedektör örneğini shared subscription'larla çalıştırın.

Dedektörü bir Raspberry Pi üzerinde çalıştırabilir miyim?

Evet. Betik yalnızca standart kütüphaneyi ve paho-mqtt'yi kullanır ve uç gateway olarak çalışan bir Raspberry Pi üzerinde rahatlıkla çalışır.

Uyarılar retained mesaj olmalı mı?

Uyarı olaylarının kendisi değil. Yeni dashboard'ların mevcut durumu görmesini istiyorsanız, her makine için aktif uyarıların listesini içeren ayrı bir retained status topic'i yayınlayın.

MQTT Sensör Akışlarında Python ile Gerçek Zamanlı Anomali Tespiti