Edge Impulse zu MQTT: Modellvorhersagen von ESP32 und Linux veröffentlichen

Die Edge Impulse MQTT-Integration ist der letzte Schritt, der ein in Edge Impulse Studio trainiertes Modell in etwas verwandelt, das der Rest Ihres Systems nutzen kann. In diesem Tutorial exportieren Sie einen Impulse als Arduino-Bibliothek, führen ihn auf einem ESP32 aus und veröffentlichen jede Vorhersage als JSON an broker.connectmq.com. Anschließend machen Sie dasselbe auf einem Raspberry Pi mit dem Linux-Runner und Python. Am Ende können Dashboards und Alarmierung die Ausgabe Ihres Modells wie jeden anderen Sensor abonnieren.

Wie die Teile zusammenpassen

Edge Impulse übernimmt Datenerfassung, Signalverarbeitungsblöcke, Training und Optimierung. Heraus kommt ein bereitstellbares Artefakt: eine C++-Bibliothek (auch als Arduino-Bibliothek verpackt) für Mikrocontroller oder eine ausführbare .eim-Datei für Linux-Ziele. Keines davon weiß etwas über Ihr Netzwerk. Sie ergänzen um den Klassifikator herum ein wenig Code, der Ergebnisse an einen MQTT-Broker veröffentlicht, und von dort können beliebig viele Konsumenten reagieren.

Schritt 1: Das Modell aus Edge Impulse Studio exportieren

1. Öffnen Sie Ihr Projekt und stellen Sie sicher, dass der Impulse trainiert ist und die Ergebnisse unter Model testing akzeptabel sind.
2. Gehen Sie zu Deployment, wählen Sie Arduino library und das quantisierte (int8) Modell, sofern der EON Compiler und quantisierte Optionen angeboten werden. Quantisierte Modelle brauchen weniger RAM und laufen auf Mikrocontrollern schneller.
3. Klicken Sie auf Build und laden Sie die .zip-Datei herunter. Der Bibliotheksname folgt Ihrem Projektnamen, zum Beispiel vibration_monitor_inferencing.
4. Installieren Sie sie in der Arduino IDE über Sketch > Include Library > Add .ZIP Library.
5. Installieren Sie außerdem PubSubClient von Nick O'Leary über den Bibliotheksverwalter und stellen Sie sicher, dass das ESP32-Boardpaket von Espressif installiert ist.

Um ohne angeschlossenen Sensor zu testen, öffnen Sie in Studio Live classification oder ein Testsample und klicken neben den Rohmerkmalen auf Copy to clipboard. Dieses Array fügen Sie in den Sketch ein.

Schritt 2: ESP32-Sketch, der Vorhersagen veröffentlicht

Der folgende Sketch führt den Klassifikator alle zwei Sekunden auf einem statischen Merkmalspuffer aus und veröffentlicht das Ergebnis. Ersetzen Sie das Include durch den Header-Namen Ihrer Bibliothek und fügen Sie Ihre Rohmerkmale ein. Eine sensorgesteuerte Version folgt danach.

#include <WiFi.h>
#include <PubSubClient.h>
#include <vibration_monitor_inferencing.h>   // your exported library

const char *WIFI_SSID = "your-ssid";
const char *WIFI_PASS = "your-password";
const char *MQTT_HOST = "broker.connectmq.com";
const uint16_t MQTT_PORT = 1883;
const char *CLIENT_ID = "ei-esp32-4b7e19";                       // make unique
const char *TOPIC_BASE = "connectmq/demo/4b7e19/line1/fan3";    // make unique

// Paste the raw features copied from Edge Impulse Studio
static const float features[] = {
  0.0
};

WiFiClient net;
PubSubClient mqtt(net);

int raw_feature_get_data(size_t offset, size_t length, float *out_ptr) {
  memcpy(out_ptr, features + offset, length * sizeof(float));
  return 0;
}

void connectWiFi() {
  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASS);
  while (WiFi.status() != WL_CONNECTED) { delay(500); Serial.print("."); }
  Serial.printf("\nWi-Fi OK, IP %s\n", WiFi.localIP().toString().c_str());
}

void connectMqtt() {
  String statusTopic = String(TOPIC_BASE) + "/status";
  while (!mqtt.connected()) {
    Serial.print("MQTT connecting... ");
    if (mqtt.connect(CLIENT_ID, statusTopic.c_str(), 1, true, "offline")) {
      mqtt.publish(statusTopic.c_str(), "online", true);
      Serial.println("connected");
    } else {
      Serial.printf("failed rc=%d, retry in 3 s\n", mqtt.state());
      delay(3000);
    }
  }
}

void publishResult(const ei_impulse_result_t &result) {
  // Find the top class and build a JSON object with all scores
  size_t best = 0;
  String scores = "{";
  for (size_t i = 0; i < EI_CLASSIFIER_LABEL_COUNT; i++) {
    if (result.classification[i].value > result.classification[best].value) best = i;
    scores += "\"" + String(result.classification[i].label) + "\":" +
              String(result.classification[i].value, 3);
    if (i < EI_CLASSIFIER_LABEL_COUNT - 1) scores += ",";
  }
  scores += "}";

  String payload = "{\"label\":\"" + String(result.classification[best].label) + "\"" +
                   ",\"p\":" + String(result.classification[best].value, 3) +
                   ",\"scores\":" + scores +
                   ",\"dsp_ms\":" + String(result.timing.dsp) +
                   ",\"nn_ms\":" + String(result.timing.classification);
#if EI_CLASSIFIER_HAS_ANOMALY == 1
  payload += ",\"anomaly\":" + String(result.anomaly, 3);
#endif
  payload += "}";

  String topic = String(TOPIC_BASE) + "/inference";
  bool ok = mqtt.publish(topic.c_str(), payload.c_str());
  Serial.printf("%s %s\n", ok ? "sent" : "FAILED", payload.c_str());
}

void setup() {
  Serial.begin(115200);
  delay(500);
  if (sizeof(features) / sizeof(float) != EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE) {
    Serial.printf("Feature count mismatch: got %u, expected %u\n",
                  (unsigned)(sizeof(features) / sizeof(float)),
                  (unsigned)EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE);
  }
  connectWiFi();
  mqtt.setServer(MQTT_HOST, MQTT_PORT);
  mqtt.setBufferSize(512);   // default 256 bytes is too small for all scores
  connectMqtt();
}

void loop() {
  if (!mqtt.connected()) connectMqtt();
  mqtt.loop();

  signal_t signal;
  signal.total_length = EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE;
  signal.get_data = &raw_feature_get_data;

  ei_impulse_result_t result = { 0 };
  EI_IMPULSE_ERROR err = run_classifier(&signal, &result, false);
  if (err != EI_IMPULSE_OK) {
    Serial.printf("run_classifier failed (%d)\n", err);
  } else {
    publishResult(result);
  }
  delay(2000);
}

Flashen Sie ihn, öffnen Sie den seriellen Monitor mit 115200 Baud, und die Klassifikation für Ihr eingefügtes Sample erscheint alle zwei Sekunden. Die Werte sollten mit dem übereinstimmen, was Studio für dasselbe Sample angezeigt hat. Ist das nicht der Fall, wurden die Merkmale aus einem anderen Sample oder dem falschen Verarbeitungsblock kopiert.

Live-Sensordaten verwenden

Für ein echtes Deployment füllen Sie einen Puffer vom Sensor im Abtastintervall, mit dem der Impulse trainiert wurde, und verpacken ihn in ein Signal. EI_CLASSIFIER_INTERVAL_MS und EI_CLASSIFIER_RAW_SAMPLES_PER_FRAME stammen aus der exportierten Bibliothek:

static float buffer[EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE];

void readAccel(float &x, float &y, float &z);   // implement for your sensor

bool classifyLive(ei_impulse_result_t &result) {
  for (size_t i = 0; i < EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE; i += EI_CLASSIFIER_RAW_SAMPLES_PER_FRAME) {
    uint64_t next = micros() + (uint64_t)(EI_CLASSIFIER_INTERVAL_MS * 1000);
    readAccel(buffer[i], buffer[i + 1], buffer[i + 2]);   // assumes a 3-axis impulse
    while (micros() < next) { }
  }
  signal_t signal;
  if (numpy::signal_from_buffer(buffer, EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE, &signal) != 0) return false;
  return run_classifier(&signal, &result, false) == EI_IMPULSE_OK;
}

Rufen Sie zwischen den Fenstern mqtt.loop() auf. Bei langen Fenstern kann eine Abtastschleife mit aktivem Warten das MQTT-Keepalive aushungern. Halten Sie die Fenster daher kurz oder tasten Sie über einen Timer-Interrupt oder einen FreeRTOS-Task ab.

Schritt 3: Edge Impulse auf dem Raspberry Pi mit Python

Für Linux-Ziele stellt Edge Impulse einen Runner bereit, der ein für das Board optimiertes .eim-Modell herunterlädt. Installieren Sie die CLI gemäß der Edge-Impulse-Linux-Dokumentation für Ihr Gerät und führen Sie dann Folgendes aus:

edge-impulse-linux-runner --download model.eim
pip install edge_impulse_linux "paho-mqtt>=2.0"

Dieses Skript lädt das Modell, klassifiziert einen Merkmalsvektor und veröffentlicht das Ergebnis. Ersetzen Sie get_features() durch Ihre eigene Datenquelle, zum Beispiel Werte, die über I2C von einem Sensor gelesen werden.

import json
import time
import uuid

import paho.mqtt.client as mqtt
from edge_impulse_linux.runner import ImpulseRunner

MODEL = "model.eim"
RUN = uuid.uuid4().hex[:8]
BASE = f"connectmq/demo/{RUN}/line1/fan3"


def get_features(count):
    """Placeholder: return `count` floats in the same format as Studio's raw features."""
    return [0.0] * count


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"ei-pi-{RUN}")
client.will_set(f"{BASE}/status", "offline", qos=1, retain=True)
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
client.publish(f"{BASE}/status", "online", qos=1, retain=True)

runner = ImpulseRunner(MODEL)
try:
    info = runner.init()
    params = info["model_parameters"]
    print("loaded", info["project"]["name"], "expects", params["input_features_count"], "features")
    while True:
        res = runner.classify(get_features(params["input_features_count"]))
        scores = res["result"]["classification"]
        label = max(scores, key=scores.get)
        msg = {"label": label, "p": round(scores[label], 3),
               "scores": {k: round(v, 3) for k, v in scores.items()},
               "ms": res.get("timing", {}).get("classification")}
        if "anomaly" in res["result"]:
            msg["anomaly"] = round(res["result"]["anomaly"], 3)
        client.publish(f"{BASE}/inference", json.dumps(msg), qos=1)
        print(msg)
        time.sleep(2)
finally:
    runner.stop()
    client.loop_stop()

Für Kamera- und Mikrofonprojekte bietet dasselbe SDK Bild- und Audio-Runner, die in einem Schritt erfassen und klassifizieren; deren Ergebnisse veröffentlichen Sie auf dieselbe Weise.

Schritt 4: Abonnieren und überprüfen

mosquitto_sub -h broker.connectmq.com -p 1883 -i ei-viewer-77 -t 'connectmq/demo/+/line1/fan3/#' -v

Sie sollten den Retained-Status und einen Strom von Inferenznachrichten sehen. Der öffentliche MQTT-Broker ist für Tests und Prototypen gedacht und für alle offen. Verwenden Sie daher ein zufälliges Präfix und senden Sie niemals private Daten. Erstellen Sie für die Produktion einen privaten Broker und wechseln Sie zu TLS auf Port 8883, mit WiFiClientSecure auf dem ESP32 oder client.tls_set() in Python, und verwenden Sie Benutzername, Passwort und Client ID aus dem Portal.

Vorhersagen in Aktionen verwandeln

Sobald Vorhersagen auf einem Broker eintreffen, ist die Aufgabe des Geräts erledigt, und der Rest ist gewöhnliche MQTT-Integration. In der Fabrik hängen Sie die Vorhersagen am besten an den Knoten der Maschine in Ihrem Unified Namespace (UNS), dessen Topic-Struktur der ISA-95-Hierarchie folgt, etwa acme/werk-stuttgart/montage/linie-1/luefter-3/inference; dann stehen sie MES und Instandhaltung direkt neben den übrigen Maschinendaten zur Verfügung. Einige Muster decken die meisten Projekte ab.

Dashboards. Ein Browser-Dashboard kann über sicheres WebSocket auf Port 8084 mit dem Pfad /mqtt abonnieren und das scores-Objekt für jedes Gerät grafisch darstellen. Da das Status-Topic eine Retained Message ist, weiß das Dashboard beim Laden sofort, welche Geräte online sind.

Alarme. Wenn Ihr Impulse einen Anomalieblock enthält, ist der Anomaliewert das nützlichste Feld für Alarme, weil er bei Eingaben steigt, die das Modell nie gesehen hat, einschließlich Fehlerarten, die Sie nie aufgezeichnet haben. Schwellenwert-Alarmregeln in ConnectMQ können einen numerischen Wert überwachen und Alarme im Alarmprotokoll festhalten, was während eines Pilotprojekts oft ausreicht. Für Benachrichtigungen an Smartphones oder Chat-Tools betreiben Sie einen Node-RED-Flow oder einen kleinen Python-Abonnenten, der auf den Inferenz-Topics lauscht und den Benachrichtigungsdienst aufruft, den Sie bereits nutzen.

Historians. Telegraf, Node-RED oder ein Python-Abonnent können jede Vorhersage in eine Zeitreihendatenbank schreiben. So erhalten Sie eine Historie der Maschinenzustände, die Sie mit Wartungsprotokollen vergleichen können. Diese Historie ist zugleich die beste Quelle für Kandidaten für das erneute Training.

Geschlossener Regelkreis. Soll eine Vorhersage etwas Physisches verändern, etwa einen Lüfter stoppen oder ein Ventil öffnen, behalten Sie diese Logik auf dem Gerät oder einer lokalen Steuerung und veröffentlichen die ausgeführte Aktion als eigene Nachricht. Der Broker protokolliert dann, was passiert ist, statt im kritischen Pfad zu liegen.

Best Practices für Edge Impulse MQTT

1. Bei Änderungen oder Schwellenwerten veröffentlichen, nicht bei jeder Inferenz, sobald die Testphase vorbei ist. Ergänzen Sie einen Heartbeat, damit Stille nie mehrdeutig ist.
2. Die Modellversion mitsenden. Edge-Impulse-Projekte werden häufig neu trainiert; fügen Sie jeder Nachricht eine Deployment-Versionskennung hinzu.
3. Den Anomaliewert nutzen, wenn Ihr Impulse einen Anomalieblock hat. Er erfasst Eingaben, die nichts im Training ähneln, was die Klassifikation allein nicht kann.
4. Den Konfidenz-Schwellenwert auf der Abonnentenseite halten, wenn mehrere Systeme dasselbe Topic mit unterschiedlicher Toleranz gegenüber Fehlalarmen konsumieren.
5. Nachrichtenlimits beachten. Alle Klassenwerte zu veröffentlichen ist in Ordnung; Rohfenster zu veröffentlichen nicht, es sei denn, Sie ziehen selten Stichproben, um Daten für das erneute Training zu sammeln.

Einen Blick auf die Framework-Ebene unter der Haube bietet der Einstieg in TensorFlow Lite Micro. Um dieselbe Pipeline ohne Plattform aufzubauen, folgen Sie TinyML auf dem ESP32. Der Leitfaden zu KI und IoT ordnet beides ein.

Senden Sie Ihre Edge-Impulse-Vorhersagen an ConnectMQ

Wenn das Modell funktioniert, geben Sie ihm ein produktionsreifes Ziel. ConnectMQ ist ein verwalteter MQTT-Broker mit TLS, MQTT 5, Schwellenwert-Alarmregeln und einem Live-Topic-Explorer, sodass Sie Vorhersagen beobachten und bei Anomaliewerten alarmieren können, ohne selbst einen Broker zu betreiben. Starten Sie mit dem kostenlosen Serverless-Plan unter app.connectmq.com, ohne Kreditkarte.

Häufig gestellte Fragen

Unterstützt Edge Impulse MQTT direkt?

Die exportierte Modellbibliothek führt nur die Inferenz aus. Die MQTT-Veröffentlichung ergänzen Sie mit einem Standard-Client wie PubSubClient auf Mikrocontrollern oder paho-mqtt unter Linux, wie in diesem Tutorial gezeigt.

Warum schlägt das Veröffentlichen von Edge-Impulse-Ergebnissen auf meinem ESP32 fehl?

Die häufigste Ursache ist der Standardpuffer von PubSubClient mit 256 Bytes. Rufen Sie setBufferSize() mit einem größeren Wert auf oder veröffentlichen Sie weniger Felder.

Kann ich Edge-Impulse-Modelle ohne Arduino IDE auf einem ESP32 ausführen?

Ja. Edge Impulse exportiert auch eine generische C++-Bibliothek, die Sie mit ESP-IDF bauen können, und Sie können mit dem MQTT-Client von ESP-IDF statt mit PubSubClient veröffentlichen.

Wie sammle ich neue Trainingsdaten von bereitgestellten Geräten?

Veröffentlichen Sie gelegentlich Rohfenster oder Samples mit geringer Konfidenz auf einem separaten Topic, speichern Sie sie mit einem Abonnenten und laden Sie sie zum Labeln und erneuten Trainieren in Ihr Edge-Impulse-Projekt hoch.