TinyML auf dem ESP32: trainieren, bereitstellen und Ergebnisse über MQTT veröffentlichen

Mit TinyML auf dem ESP32 klassifiziert ein günstiger WLAN-Mikrocontroller Sensordaten selbstständig und meldet nur die Antwort. In diesem Tutorial trainieren Sie einen kleinen Vibrationsklassifikator in Python, quantisieren ihn auf int8, führen ihn mit der Espressif-Portierung von TensorFlow Lite Micro auf einem ESP32 aus und veröffentlichen jede Klassifikation über MQTT an broker.connectmq.com. Alles läuft auf einem nackten ESP32-Board, denn die Firmware simuliert das Sensorsignal, bis Sie einen echten Beschleunigungssensor anschließen.

Was Sie bauen

Das Gerät tastet ein Vibrationsfenster mit 256 Punkten bei 1 kHz ab, berechnet drei Merkmale (RMS, Crest-Faktor und Nulldurchgangsrate), führt ein winziges neuronales Netz aus und klassifiziert die Maschine als idle, normal oder fault. Es veröffentlicht ein JSON-Ergebnis, sobald sich die Klasse ändert, dazu alle 30 Sekunden einen Heartbeat, und hält einen Retained-Status online/offline mit Last Will vor.

Warum ESP-IDF statt der Arduino IDE? Espressif pflegt esp-tflite-micro auf Basis des aktuellen Upstream-Stands von TensorFlow Lite Micro, mit optimierten Kernels für den ESP32-S3. Mehrere Arduino-Portierungen hinken der Upstream-API hinterher, was eine häufige Ursache für Kompilierfehler ist. Wenn TinyML für Sie neu ist, überfliegen Sie zuerst was TinyML ist.

Schritt 1: Das Modell in Python trainieren und quantisieren

Legen Sie für das ESP-IDF-Projekt einen Ordner tinyml_mqtt mit einem Unterordner main an und speichern Sie dieses Trainingsskript daneben. Es erzeugt gelabelte synthetische Vibrationsfenster, trainiert ein kleines Dense-Netz, wandelt es in ein vollständig int8-quantisiertes TensorFlow-Lite-Modell um und schreibt main/model_data.cc und main/model_data.h, einschließlich der Normalisierungskonstanten für die Merkmale, die die Firmware benötigt.

pip install tensorflow numpy

# train_model.py
import numpy as np
import tensorflow as tf

FS, N = 1000, 256
LABELS = ["idle", "normal", "fault"]
rng = np.random.default_rng(42)


def gen(cls):
    t = np.arange(N) / FS
    if cls == 0:                                   # idle: sensor noise only
        return 0.05 * rng.standard_normal(N)
    x = rng.uniform(0.8, 1.2) * np.sin(2 * np.pi * 50 * t + rng.uniform(0, 2 * np.pi))
    x += 0.1 * rng.standard_normal(N)
    if cls == 2:                                   # fault: periodic impacts
        idx = rng.choice(N, size=rng.integers(4, 10), replace=False)
        x[idx] += rng.uniform(2, 3, idx.size) * rng.choice([-1, 1], idx.size)
    return x


def features(x):
    rms = np.sqrt(np.mean(x ** 2))
    crest = np.max(np.abs(x)) / (rms + 1e-6)
    zcr = np.mean((x[1:] < 0) != (x[:-1] < 0))
    return np.array([rms, crest, zcr], dtype=np.float32)


y = rng.integers(0, 3, 6000)
X = np.stack([features(gen(c)) for c in y])
mean, std = X.mean(axis=0), X.std(axis=0)
Xn = ((X - mean) / std).astype(np.float32)

model = tf.keras.Sequential([
    tf.keras.Input(shape=(3,)),
    tf.keras.layers.Dense(16, activation="relu"),
    tf.keras.layers.Dense(3, activation="softmax"),
])
model.compile(optimizer="adam", loss="sparse_categorical_crossentropy", metrics=["accuracy"])
model.fit(Xn, y, epochs=30, batch_size=32, validation_split=0.2, verbose=2)


def representative_data():
    for i in range(300):
        yield [Xn[i:i + 1]]


converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.representative_dataset = representative_data
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type = tf.int8
converter.inference_output_type = tf.int8
tflite_model = converter.convert()
print("model size:", len(tflite_model), "bytes")

hex_bytes = ", ".join(f"0x{b:02x}" for b in tflite_model)
with open("tinyml_mqtt/main/model_data.cc", "w") as f:
    f.write('#include "model_data.h"\n\n')
    f.write(f"alignas(16) const unsigned char g_model[] = {{{hex_bytes}}};\n")
    f.write(f"const unsigned int g_model_len = {len(tflite_model)};\n")

with open("tinyml_mqtt/main/model_data.h", "w") as f:
    f.write("#pragma once\n\n")
    f.write("extern const unsigned char g_model[];\nextern const unsigned int g_model_len;\n\n")
    f.write("constexpr float kFeatMean[3] = {%s};\n" % ", ".join(f"{v:.6f}f" for v in mean))
    f.write("constexpr float kFeatStd[3] = {%s};\n" % ", ".join(f"{v:.6f}f" for v in std))
    f.write('constexpr const char *kLabels[3] = {"idle", "normal", "fault"};\n')
print("wrote main/model_data.cc and main/model_data.h")

Das Modell besteht aus zwei FULLY_CONNECTED-Schichten und einem SOFTMAX, und die int8-Version ist nur wenige Kilobyte groß. Die Validierungsgenauigkeit sollte nahezu perfekt sein, weil die synthetischen Klassen gut getrennt sind. Bei echten Daten verbringen Sie den Großteil Ihrer Zeit voraussichtlich mit Datenerfassung und Merkmalen, nicht mit dem Netz.

Warum vollständige int8-Quantisierung

Die obigen Konvertereinstellungen erzeugen ein Modell, dessen Gewichte, Aktivierungen, Eingabe und Ausgabe allesamt 8-Bit-Ganzzahlen sind. Das ist auf dem ESP32 aus drei Gründen wichtig. Das Modell ist etwa ein Viertel so groß wie die Gleitkommaversion, die optimierten Kernels in esp-tflite-micro sind für int8 geschrieben, und auch die Arena für Zwischentensoren schrumpft. Der Preis: Sie müssen Eingaben selbst in den Ganzzahlbereich umrechnen, mit Skalierung und Nullpunkt aus dem Eingabetensor, und Ausgaben zurückrechnen. Genau das erledigt die Firmware weiter unten.

Mit dem repräsentativen Datensatz wählt der Konverter diese Skalierungen. Deckt er nicht den vollen Bereich realer Eingaben ab, werden Werte außerhalb des Bereichs abgeschnitten, und die Genauigkeit auf dem Gerät sinkt, obwohl sie in Python perfekt aussah. Wenn Sie zu echten Sensordaten wechseln, speisen Sie echte normalisierte Merkmalsvektoren aus jeder Klasse in representative_data() ein.

Schritt 2: Das ESP-IDF-Projekt anlegen

Projektstruktur:

tinyml_mqtt/
  CMakeLists.txt
  sdkconfig.defaults
  main/
    CMakeLists.txt
    main.cc
    model_data.cc   (generated)
    model_data.h    (generated)

Die oberste CMakeLists.txt bindet den Beispiel-WLAN-Helfer von Espressif ein, damit das Tutorial kurz bleibt:

cmake_minimum_required(VERSION 3.16)
set(EXTRA_COMPONENT_DIRS $ENV{IDF_PATH}/examples/common_components/protocol_examples_common)
include($ENV{IDF_PATH}/tools/cmake/project.cmake)
project(tinyml_mqtt)

main/CMakeLists.txt:

idf_component_register(SRCS "main.cc" "model_data.cc" INCLUDE_DIRS ".")

sdkconfig.defaults mit Ihrem WLAN:

CONFIG_EXAMPLE_CONNECT_WIFI=y
CONFIG_EXAMPLE_WIFI_SSID="your-ssid"
CONFIG_EXAMPLE_WIFI_PASSWORD="your-password"

Fügen Sie dann im Projektordner die Inferenz-Runtime aus der ESP Component Registry hinzu:

idf.py set-target esp32s3      # or esp32
idf.py add-dependency "espressif/esp-tflite-micro"

Schritt 3: Firmware mit Inferenz und MQTT

main/main.cc richtet WLAN und MQTT ein, lädt das Modell in eine statische Tensor-Arena und führt die Schleife aus Abtastung, Merkmalen, Inferenz und Veröffentlichung aus. Ändern Sie TOPIC_BASE und die Client ID vor dem Flashen in etwas Eindeutiges.

#include <algorithm>
#include <cmath>
#include <cstdio>

#include "esp_event.h"
#include "esp_log.h"
#include "esp_netif.h"
#include "esp_random.h"
#include "esp_timer.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "mqtt_client.h"
#include "nvs_flash.h"
#include "protocol_examples_common.h"
#include "tensorflow/lite/micro/micro_interpreter.h"
#include "tensorflow/lite/micro/micro_mutable_op_resolver.h"
#include "tensorflow/lite/schema/schema_generated.h"

#include "model_data.h"

#define TOPIC_BASE "connectmq/demo/esp32-7c41a9/motor1"   // make this unique
#define CLIENT_ID  "esp32-tinyml-7c41a9"                  // make this unique

static const char *TAG = "tinyml";
static constexpr int kN = 256;
static constexpr float kFs = 1000.0f;
static constexpr int kArenaSize = 8 * 1024;
alignas(16) static uint8_t tensor_arena[kArenaSize];

static esp_mqtt_client_handle_t mqtt = nullptr;
static volatile bool mqtt_ready = false;

static void mqtt_event(void *arg, esp_event_base_t base, int32_t id, void *data) {
  if (id == MQTT_EVENT_CONNECTED) {
    mqtt_ready = true;
    esp_mqtt_client_publish(mqtt, TOPIC_BASE "/status", "online", 0, 1, 1);
    ESP_LOGI(TAG, "MQTT connected");
  } else if (id == MQTT_EVENT_DISCONNECTED) {
    mqtt_ready = false;
    ESP_LOGW(TAG, "MQTT disconnected");
  }
}

static void start_mqtt() {
  esp_mqtt_client_config_t cfg = {};
  cfg.broker.address.uri = "mqtt://broker.connectmq.com:1883";
  cfg.credentials.client_id = CLIENT_ID;
  cfg.session.last_will.topic = TOPIC_BASE "/status";
  cfg.session.last_will.msg = "offline";
  cfg.session.last_will.qos = 1;
  cfg.session.last_will.retain = 1;
  mqtt = esp_mqtt_client_init(&cfg);
  esp_mqtt_client_register_event(mqtt, MQTT_EVENT_ANY, mqtt_event, nullptr);
  esp_mqtt_client_start(mqtt);
}

// ---- simulated sensor: replace with real accelerometer samples ----
static float uniform(float a, float b) { return a + (b - a) * (esp_random() / 4294967296.0f); }
static float randn() {
  float u1 = (esp_random() + 1.0f) / 4294967297.0f;
  float u2 = esp_random() / 4294967296.0f;
  return sqrtf(-2.0f * logf(u1)) * cosf(2.0f * M_PI * u2);
}
static void read_window(int scenario, float *x) {
  float amp = uniform(0.8f, 1.2f), phase = uniform(0.0f, 2.0f * M_PI);
  for (int i = 0; i < kN; i++) {
    x[i] = scenario == 0 ? 0.05f * randn()
                         : amp * sinf(2.0f * M_PI * 50.0f * i / kFs + phase) + 0.1f * randn();
  }
  if (scenario == 2) {
    int impacts = 4 + esp_random() % 6;
    for (int k = 0; k < impacts; k++) {
      int idx = esp_random() % kN;
      x[idx] += uniform(2.0f, 3.0f) * ((esp_random() & 1) ? 1.0f : -1.0f);
    }
  }
}

static void compute_features(const float *x, float *f) {
  float sumsq = 0, peak = 0;
  int crossings = 0;
  for (int i = 0; i < kN; i++) {
    sumsq += x[i] * x[i];
    peak = std::max(peak, fabsf(x[i]));
    if (i > 0 && ((x[i] < 0) != (x[i - 1] < 0))) crossings++;
  }
  float rms = sqrtf(sumsq / kN);
  f[0] = rms;
  f[1] = peak / (rms + 1e-6f);
  f[2] = (float)crossings / (kN - 1);
}

extern "C" void app_main(void) {
  ESP_ERROR_CHECK(nvs_flash_init());
  ESP_ERROR_CHECK(esp_netif_init());
  ESP_ERROR_CHECK(esp_event_loop_create_default());
  ESP_ERROR_CHECK(example_connect());
  start_mqtt();

  const tflite::Model *model = tflite::GetModel(g_model);
  if (model->version() != TFLITE_SCHEMA_VERSION) {
    ESP_LOGE(TAG, "schema mismatch");
    return;
  }
  static tflite::MicroMutableOpResolver<4> resolver;
  resolver.AddFullyConnected();
  resolver.AddSoftmax();
  resolver.AddQuantize();
  resolver.AddDequantize();

  static tflite::MicroInterpreter interpreter(model, resolver, tensor_arena, kArenaSize);
  if (interpreter.AllocateTensors() != kTfLiteOk) {
    ESP_LOGE(TAG, "AllocateTensors failed: increase kArenaSize");
    return;
  }
  ESP_LOGI(TAG, "arena used: %u bytes", (unsigned)interpreter.arena_used_bytes());
  TfLiteTensor *in = interpreter.input(0);
  TfLiteTensor *out = interpreter.output(0);

  static float window[kN];
  int last_label = -1;
  int64_t last_sent_us = 0;

  for (int n = 0;; n++) {
    int scenario = (n / 10) % 3;          // demo: change machine state every 10 windows
    read_window(scenario, window);

    float feat[3];
    compute_features(window, feat);
    for (int i = 0; i < 3; i++) {
      float norm = (feat[i] - kFeatMean[i]) / kFeatStd[i];
      int q = (int)lroundf(norm / in->params.scale) + in->params.zero_point;
      in->data.int8[i] = (int8_t)std::min(127, std::max(-128, q));
    }

    int64_t t0 = esp_timer_get_time();
    if (interpreter.Invoke() != kTfLiteOk) {
      ESP_LOGE(TAG, "Invoke failed");
      continue;
    }
    int infer_us = (int)(esp_timer_get_time() - t0);

    int best = 0;
    float best_p = -1;
    for (int i = 0; i < 3; i++) {
      float p = (out->data.int8[i] - out->params.zero_point) * out->params.scale;
      if (p > best_p) { best_p = p; best = i; }
    }
    ESP_LOGI(TAG, "%s p=%.2f (%d us)", kLabels[best], best_p, infer_us);

    int64_t now = esp_timer_get_time();
    if (mqtt_ready && (best != last_label || now - last_sent_us > 30000000)) {
      char payload[128];
      snprintf(payload, sizeof(payload),
               "{\"label\":\"%s\",\"p\":%.2f,\"rms\":%.3f,\"us\":%d,\"model\":\"vib-mlp-v1\"}",
               kLabels[best], best_p, feat[0], infer_us);
      esp_mqtt_client_publish(mqtt, TOPIC_BASE "/inference", payload, 0, 1, 0);
      last_label = best;
      last_sent_us = now;
    }
    vTaskDelay(pdMS_TO_TICKS(500));
  }
}

Bauen, flashen und den Monitor öffnen:

python train_model.py
cd tinyml_mqtt
idf.py build flash monitor

Schritt 4: Die Ergebnisse eintreffen sehen

Abonnieren Sie Ihr Präfix von einem beliebigen Computer aus:

mosquitto_sub -h broker.connectmq.com -p 1883 -i tinyml-viewer-31 \
  -t 'connectmq/demo/esp32-7c41a9/#' -v

Sie sehen den Retained-Status online, dann jedes Mal eine Inferenznachricht, wenn die simulierte Maschine ihren Zustand ändert, plus alle 30 Sekunden einen Heartbeat. Die Firmware wertet zwei Fenster pro Sekunde aus, veröffentlicht aber nur eine Handvoll Nachrichten pro Minute, und genau darum geht es bei TinyML: lokal rechnen, sparsam melden. Der kostenlose öffentliche MQTT-Broker erlaubt bis zu 10 Nachrichten pro Sekunde und Client, sodass dieses Design reichlich Spielraum lässt.

Topics und Payload gestalten

Das Tutorial verwendet zwei Topics unter einem Gerätepräfix: status, als Retained Message und mit Last Will abgesichert, und inference für die Ergebnisse. Durch diese Aufteilung kann ein Dashboard den Gerätezustand sofort anzeigen, während die Ergebnisse als Events fließen. Wächst die Flotte, packen Sie mehr Kontext in die Hierarchie, etwa plant1/utilities/compressors/comp-04/inference, damit ein Abonnent mit Wildcards eine Maschine, eine Linie oder den ganzen Standort auswählen kann. In einer Fabrik ist das der Unified Namespace (UNS) nach ISA-95, zum Beispiel acme/werk-muenchen/druckluft/kompressoren/comp-04/inference: So landen die Ergebnisse Ihrer Predictive-Maintenance-Knoten direkt neben SPS- und Energiedaten.

Halten Sie den Payload kurz, aber selbstbeschreibend. Das Feld model zeigt Ihnen nach einem Over-the-Air-Update, welche Firmware ein Ergebnis erzeugt hat, p erlaubt Abonnenten eigene Konfidenzschwellen, und us verrät, wenn die Inferenzzeit nach einer Modelländerung ansteigt. Wenn Sie das Verhalten aus der Ferne anpassen müssen, lassen Sie das Gerät ein cmd-Topic für Einstellungen wie das Heartbeat-Intervall abonnieren und validieren Sie jeden Wert vor der Übernahme. Auf einem privaten ConnectMQ-Broker können Sie beide Topics im Live-Topic-Explorer beobachten und während des Testens eine Schwellenwert-Alarmregel für numerische Felder festlegen.

Umstieg auf echte Sensordaten

Ersetzen Sie read_window() durch Messwerte eines Beschleunigungssensors wie MPU6050 oder ADXL345, die mit fester Rate über einen Hardware-Timer oder den FIFO des Sensors abgetastet werden. Trainieren Sie dann mit aufgezeichneten Fenstern neu: Protokollieren Sie echte Fenster aus jedem Maschinenzustand über MQTT oder die serielle Schnittstelle, labeln Sie sie und speisen Sie sie in denselben Merkmals- und Trainingscode ein. Halten Sie die Merkmalsberechnung in Python und C++ identisch, einschließlich der Definition des Nulldurchgangs, sonst entspricht die Genauigkeit auf dem Gerät nicht der beim Training.

Wechseln Sie für die Produktion zu TLS und Zugangsdaten: Verwenden Sie die URI mqtts://broker.connectmq.com:8883 mit dem Host Ihres ConnectMQ-Brokers, setzen Sie cfg.credentials.username, cfg.credentials.authentication.password und die ausgegebene Client ID und hinterlegen Sie das CA-Zertifikat (das Zertifikatsbündel von ESP-IDF deckt Let's Encrypt ab). Das ESP32-MQTT-Tutorial behandelt die TLS-Einrichtung im Detail.

Fehlerbehebung bei TinyML auf dem ESP32

Nächste Schritte

Vergleichen Sie diese handgebaute Pipeline mit dem Edge-Impulse-MQTT-Workflow, der Datenerfassung und Merkmalsdesign automatisiert, oder tauchen Sie mit dem Einstieg in TensorFlow Lite Micro tiefer in die Runtime selbst ein. Der Leitfaden zu KI und IoT zeigt, wo solche Geräte in ein größeres AIoT-System passen.

Starten Sie mit ConnectMQ

Ihr TinyML-Knoten ist nur so nützlich wie die Systeme, die seine Ergebnisse empfangen. ConnectMQ ist ein verwalteter MQTT-Broker mit TLS, generierten Zugangsdaten und einem Live-Topic-Explorer, in dem Sie Inferenznachrichten in Echtzeit verfolgen. Erstellen Sie einen Broker im kostenlosen Serverless-Plan unter app.connectmq.com, ohne Kreditkarte, und richten Sie die Firmware darauf aus.

Häufig gestellte Fragen

Welcher ESP32 eignet sich am besten für TinyML?

Der ESP32-S3 ist die übliche Wahl, weil seine Vektorbefehle die Kernels neuronaler Netze in esp-tflite-micro beschleunigen und viele Boards PSRAM mitbringen. Der ursprüngliche ESP32 funktioniert für kleine Modelle wie das in diesem Tutorial problemlos.

Kann ich TinyML auf dem ESP32 mit der Arduino IDE umsetzen?

Ja, über Arduino-Bibliotheken, die TensorFlow Lite Micro oder Edge-Impulse-Exporte bündeln. Prüfen Sie, ob die Bibliotheksversion zur aktuellen TFLM-API passt, denn ältere Portierungen verwenden andere Konstruktoren und Header.

Wie groß darf ein TinyML-Modell auf dem ESP32 sein?

Das hängt vom Flash für die Gewichte und vom RAM für die Tensor-Arena ab. Kleine Dense- und Faltungsmodelle passen problemlos in den internen Speicher; größere Bildverarbeitungsmodelle brauchen meist einen ESP32-S3 mit PSRAM.

Warum nur bei Änderungen veröffentlichen?

Die Inferenz läuft viele Male pro Sekunde, aber nachgelagerte Systeme interessiert nur, wann sich der Zustand ändert. Veröffentlichen bei Änderungen plus Heartbeat spart Funkenergie und Bandbreite und hält Sie innerhalb der Ratenlimits des Brokers.