TinyML sur ESP32 : entraîner, déployer et publier les résultats via MQTT

Le TinyML sur ESP32 permet à un microcontrôleur Wi-Fi bon marché de classer lui-même des données de capteurs et de ne transmettre que la réponse. Dans ce tutoriel, vous allez entraîner un petit classificateur de vibrations en Python, le quantifier en int8, l'exécuter sur un ESP32 avec le portage de TensorFlow Lite Micro d'Espressif et publier chaque classification via MQTT vers broker.connectmq.com, le serveur MQTT public gratuit réservé aux tests. Tout fonctionne sur une simple carte ESP32, car le firmware simule le signal du capteur jusqu'à ce que vous branchiez un vrai accéléromètre.

Ce que vous allez construire

L'appareil échantillonne une fenêtre de vibrations de 256 points à 1 kHz, calcule trois caractéristiques (valeur efficace, facteur de crête et taux de passage par zéro), exécute un minuscule réseau de neurones et classe la machine en idle, normal ou fault. Il publie un résultat JSON à chaque changement de classe, plus un heartbeat toutes les 30 secondes, et maintient un statut online/offline en retained avec un Last Will.

Pourquoi ESP-IDF plutôt que l'Arduino IDE ? Espressif maintient esp-tflite-micro en phase avec la version amont actuelle de TensorFlow Lite Micro, avec des noyaux optimisés pour l'ESP32-S3. Plusieurs portages Arduino sont en retard sur l'API amont, ce qui est une source fréquente d'erreurs de compilation. Si le TinyML est nouveau pour vous, parcourez d'abord ce qu'est le TinyML.

Étape 1 : entraîner et quantifier le modèle en Python

Créez un dossier tinyml_mqtt pour le projet ESP-IDF avec un sous-dossier main, et placez ce script d'entraînement à côté. Il génère des fenêtres de vibrations synthétiques étiquetées, entraîne un petit réseau dense, le convertit en modèle TensorFlow Lite entièrement int8 et écrit main/model_data.cc et main/model_data.h, y compris les constantes de normalisation des caractéristiques dont le firmware a besoin.

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")

Le modèle comporte deux couches FULLY_CONNECTED et un SOFTMAX, et la version int8 ne pèse que quelques kilo-octets. La précision de validation devrait être quasi parfaite, car les classes synthétiques sont bien séparées. Avec des données réelles, attendez-vous à passer l'essentiel de votre temps sur la collecte de données et les caractéristiques, pas sur le réseau.

Pourquoi une quantification entièrement int8

Les paramètres du convertisseur ci-dessus produisent un modèle dont les poids, les activations, l'entrée et la sortie sont tous des entiers 8 bits. C'est important sur l'ESP32 pour trois raisons : le modèle fait environ un quart de la taille de la version flottante, les noyaux optimisés d'esp-tflite-micro sont écrits pour l'int8, et l'arène nécessaire aux tenseurs intermédiaires diminue aussi. En contrepartie, vous devez vous-même convertir les entrées dans le domaine entier, à l'aide de l'échelle et du point zéro stockés dans le tenseur d'entrée, puis reconvertir les sorties. C'est exactement ce que fait le firmware ci-dessous.

Le jeu de données représentatif est ce que le convertisseur utilise pour choisir ces échelles. S'il ne couvre pas toute la plage des entrées réelles, les valeurs hors plage sont écrêtées et la précision chute sur l'appareil, même si elle semblait parfaite en Python. Lorsque vous passerez à des données de capteurs réelles, fournissez à representative_data() de vrais vecteurs de caractéristiques normalisés issus de chaque classe.

Étape 2 : créer le projet ESP-IDF

Structure du projet :

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

Le CMakeLists.txt de premier niveau. Il intègre l'assistant Wi-Fi des exemples d'Espressif pour garder le tutoriel concis :

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 avec votre réseau Wi-Fi :

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

Ajoutez ensuite le runtime d'inférence depuis l'ESP Component Registry, à partir du dossier du projet :

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

Étape 3 : le firmware avec inférence et MQTT

main/main.cc configure le Wi-Fi et MQTT, charge le modèle dans une arène de tenseurs statique et exécute la boucle échantillonnage, caractéristiques, inférence et publication. Remplacez TOPIC_BASE et le client ID par des valeurs uniques avant de flasher.

#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));
  }
}

Compilez, flashez et ouvrez le moniteur :

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

Étape 4 : regarder arriver les résultats

Depuis n'importe quel ordinateur, abonnez-vous à votre préfixe :

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

Vous verrez le statut online en retained, puis un message d'inférence à chaque changement d'état de la machine simulée, plus un heartbeat toutes les 30 secondes. Le firmware évalue deux fenêtres par seconde mais ne publie qu'une poignée de messages par minute, et c'est tout l'intérêt du TinyML : calculer localement, communiquer avec parcimonie. Le serveur MQTT public gratuit autorise jusqu'à 10 messages par seconde et par client, ce qui laisse une large marge à cette conception.

Concevoir les topics et le payload

Le tutoriel utilise deux topics sous un même préfixe d'appareil : status, en retained et adossé à un Last Will, et inference, qui transporte les résultats. Cette séparation permet à un tableau de bord d'afficher instantanément l'état de l'appareil tandis que les résultats circulent sous forme d'événements. À mesure que le parc grandit, enrichissez la hiérarchie de contexte, par exemple plant1/utilities/compressors/comp-04/inference, pour qu'un abonné puisse sélectionner une machine, une ligne ou le site entier avec des wildcards.

Gardez le payload court mais autodescriptif. Le champ model vous indique quel firmware a produit un résultat après une mise à jour à distance, p permet aux abonnés d'appliquer leur propre seuil de confiance, et us révèle une hausse du temps d'inférence après un changement de modèle. Si vous devez ajuster le comportement à distance, abonnez l'appareil à un topic cmd pour des réglages comme l'intervalle de heartbeat et validez chaque valeur avant de l'appliquer. Sur un broker ConnectMQ privé, vous pouvez suivre les deux topics dans l'explorateur de topics en direct et définir une règle d'alerte à seuil sur des champs numériques pendant vos tests.

Passer à des données de capteurs réelles

Remplacez read_window() par des échantillons provenant d'un accéléromètre comme un MPU6050 ou un ADXL345, échantillonnés à fréquence fixe avec un timer matériel ou la FIFO du capteur. Réentraînez ensuite sur des fenêtres enregistrées : journalisez de vraies fenêtres via MQTT ou la liaison série pour chaque état de la machine, étiquetez-les et injectez-les dans le même code de caractéristiques et d'entraînement. Gardez un calcul des caractéristiques strictement identique en Python et en C++, y compris la définition du passage par zéro, sinon la précision sur l'appareil ne correspondra pas à celle de l'entraînement.

Pour la production, passez à TLS et aux identifiants : utilisez l'URI mqtts://broker.connectmq.com:8883 avec l'hôte de votre broker ConnectMQ, définissez cfg.credentials.username, cfg.credentials.authentication.password et le client ID attribué, et fournissez le certificat de l'autorité de certification (le bundle de certificats d'ESP-IDF gère Let's Encrypt). Le tutoriel ESP32 MQTT détaille la configuration TLS.

Dépannage du TinyML sur ESP32

Étapes suivantes

Comparez ce pipeline construit à la main avec le flux de travail Edge Impulse MQTT, qui automatise la collecte de données et la conception des caractéristiques, ou explorez le runtime lui-même avec la prise en main de TensorFlow Lite Micro. Le guide IA et IoT montre la place d'appareils comme celui-ci dans un système AIoT plus vaste.

Démarrer avec ConnectMQ

Votre nœud TinyML n'est utile que si des systèmes reçoivent ses résultats. ConnectMQ est un broker MQTT managé avec TLS, des identifiants générés et un explorateur de topics en direct pour suivre les messages d'inférence en temps réel. Créez un broker avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire, et dirigez-y le firmware.

Questions fréquentes

Quel ESP32 choisir pour le TinyML ?

L'ESP32-S3 est le choix habituel, car ses instructions vectorielles accélèrent les noyaux de réseaux de neurones d'esp-tflite-micro et de nombreuses cartes intègrent de la PSRAM. L'ESP32 d'origine convient parfaitement aux petits modèles comme celui de ce tutoriel.

Peut-on faire du TinyML sur ESP32 avec l'Arduino IDE ?

Oui, via des bibliothèques Arduino qui embarquent TensorFlow Lite Micro ou des exports Edge Impulse. Vérifiez que la version de la bibliothèque correspond à l'API TFLM actuelle, car les anciens portages utilisent des constructeurs et des en-têtes différents.

Quelle taille peut atteindre un modèle TinyML sur ESP32 ?

Cela dépend de la flash pour les poids et de la RAM pour l'arène de tenseurs. Les petits modèles denses et convolutifs tiennent facilement en mémoire interne ; les modèles de vision plus grands nécessitent généralement un ESP32-S3 avec PSRAM.

Pourquoi ne publier que sur changement ?

L'inférence s'exécute plusieurs fois par seconde, mais les systèmes en aval ne s'intéressent qu'aux changements d'état. Publier sur changement avec un heartbeat économise l'énergie radio et la bande passante, et vous maintient dans les limites de débit du broker.