Commencer
IoT et IA

D'Edge Impulse à MQTT : publier les prédictions d'un modèle depuis un ESP32 et Linux

Sortez un modèle Edge Impulse entraîné du Studio pour l'intégrer à un vrai système : exportez-le en bibliothèque Arduino ou en fichier .eim pour Linux, exécutez l'inférence sur l'appareil et publiez chaque prédiction via MQTT.

Par L'équipe ConnectMQ8 min de lecture
Illustration Edge Impulse MQTT avec un impulse entraîné exporté vers un ESP32 et un Raspberry Pi qui publient des messages de prédiction vers un broker cloud
Essayez-le sur votre propre broker MQTTObtenez un broker MQTT privé et managé avec l’offre Free et partez d’exemples de code prêts à l’emploi.
Essayer gratuitement Exemples de code Aucune carte bancaire requise

L'intégration Edge Impulse MQTT est la dernière étape qui transforme un modèle entraîné dans Edge Impulse Studio en quelque chose d'exploitable par le reste de votre système. Dans ce tutoriel, vous allez exporter un impulse sous forme de bibliothèque Arduino, l'exécuter sur un ESP32 et publier chaque prédiction en JSON vers broker.connectmq.com, puis faire de même sur un Raspberry Pi avec le runner Linux et Python. À la fin, les tableaux de bord et les alertes pourront s'abonner à la sortie de votre modèle comme à n'importe quel autre capteur.

Comment les éléments s'articulent

Edge Impulse gère la collecte de données, les blocs de traitement du signal, l'entraînement et l'optimisation. Il produit un artefact déployable : une bibliothèque C++ (également empaquetée en bibliothèque Arduino) pour les microcontrôleurs, ou un exécutable .eim pour les cibles Linux. Aucun des deux ne connaît votre réseau. Vous ajoutez un peu de code autour du classificateur pour publier les résultats vers un broker MQTT, et à partir de là, autant de consommateurs que nécessaire peuvent réagir.

Cible Option d'export Appel d'inférence Client MQTT
ESP32 / ESP32-S3 Bibliothèque Arduino (.zip) run_classifier() en C++ PubSubClient
Raspberry Pi / Linux .eim via le runner Linux ImpulseRunner.classify() en Python paho-mqtt 2.x

Étape 1 : exporter le modèle depuis Edge Impulse Studio

  1. Ouvrez votre projet et vérifiez que l'impulse est entraîné et que les résultats de Model testing sont acceptables.
  2. Allez dans Deployment, choisissez Arduino library et sélectionnez le modèle quantifié (int8) si les options EON Compiler et quantifiée sont proposées. Les modèles quantifiés utilisent moins de RAM et s'exécutent plus vite sur les microcontrôleurs.
  3. Cliquez sur Build et téléchargez le .zip. Le nom de la bibliothèque reprend celui de votre projet, par exemple vibration_monitor_inferencing.
  4. Dans l'Arduino IDE, utilisez Sketch > Include Library > Add .ZIP Library pour l'installer.
  5. Installez aussi PubSubClient de Nick O'Leary depuis le gestionnaire de bibliothèques, et vérifiez que le paquet de cartes ESP32 d'Espressif est installé.

Pour tester sans câbler de capteur, ouvrez Live classification ou un échantillon de test dans Studio et cliquez sur Copy to clipboard à côté des caractéristiques brutes. Vous collerez ce tableau dans le sketch.

Étape 2 : un sketch ESP32 qui publie les prédictions

Le sketch ci-dessous exécute le classificateur sur un tampon de caractéristiques statique toutes les deux secondes et publie le résultat. Remplacez l'include par le nom d'en-tête de votre bibliothèque et collez vos caractéristiques brutes. Une version alimentée par un capteur est présentée ensuite.

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

Flashez-le, ouvrez le moniteur série à 115200 bauds : la classification de l'échantillon collé apparaît toutes les deux secondes. Ses scores doivent correspondre à ce que Studio affichait pour le même échantillon. Si ce n'est pas le cas, les caractéristiques ont été copiées depuis un autre échantillon ou depuis le mauvais bloc de traitement.

Utiliser des données de capteur en direct

Pour un vrai déploiement, remplissez un tampon depuis le capteur à l'intervalle d'échantillonnage utilisé pour entraîner l'impulse, puis encapsulez-le dans un signal. EI_CLASSIFIER_INTERVAL_MS et EI_CLASSIFIER_RAW_SAMPLES_PER_FRAME proviennent de la bibliothèque exportée :

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

Appelez mqtt.loop() entre les fenêtres. Avec des fenêtres longues, une boucle d'échantillonnage en attente active peut priver le keepalive MQTT de temps processeur : gardez donc des fenêtres courtes ou échantillonnez depuis une interruption de timer ou une tâche FreeRTOS.

Étape 3 : Edge Impulse sur Raspberry Pi avec Python

Sur les cibles Linux, Edge Impulse fournit un runner qui télécharge un modèle .eim optimisé pour la carte. Installez la CLI en suivant la documentation Edge Impulse Linux pour votre appareil, puis :

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

Ce script charge le modèle, classe un vecteur de caractéristiques et publie le résultat. Remplacez get_features() par votre propre source de données, par exemple des valeurs lues sur un capteur en I2C.

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

Pour les projets caméra et microphone, le même SDK fournit des runners image et audio qui capturent et classent en une seule étape ; publiez leurs résultats de la même manière.

Étape 4 : s'abonner et vérifier

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

Vous devriez voir le statut en retained et un flux de messages d'inférence. Le serveur MQTT public gratuit est ouvert à tous : utilisez un préfixe aléatoire et n'envoyez jamais de données privées. En production, créez un broker privé et passez à TLS sur le port 8883 avec WiFiClientSecure sur l'ESP32 ou client.tls_set() en Python, en utilisant le nom d'utilisateur, le mot de passe et le client ID fournis par le portail.

Transformer les prédictions en actions

Une fois les prédictions arrivées sur un broker, le travail de l'appareil est terminé et le reste relève de l'intégration MQTT classique. Quelques schémas couvrent la plupart des projets.

Tableaux de bord. Un tableau de bord dans le navigateur peut s'abonner en WebSocket sécurisé sur le port 8084 avec le chemin /mqtt et tracer l'objet scores de chaque appareil. Comme le topic de statut est en retained, le tableau de bord sait immédiatement au chargement quels appareils sont en ligne.

Alertes. Si votre impulse inclut un bloc d'anomalie, le score d'anomalie est le champ le plus utile pour les alertes, car il augmente pour des entrées que le modèle n'a jamais vues, y compris des modes de défaillance que vous n'avez pas enregistrés. Les règles d'alerte à seuil de ConnectMQ peuvent surveiller une valeur numérique et enregistrer les alertes dans le journal des alertes, ce qui suffit souvent pendant un pilote. Pour des notifications sur téléphone ou dans des outils de messagerie, exécutez un flux Node-RED ou un petit abonné Python qui écoute les topics d'inférence et appelle le service de notification que vous utilisez déjà.

Historians. Telegraf, Node-RED ou un abonné Python peuvent écrire chaque prédiction dans une base de données de séries temporelles, ce qui vous donne un historique des états des machines à comparer avec les rapports de maintenance. Cet historique est aussi la meilleure source de candidats pour le réentraînement.

Boucle fermée. Si une prédiction doit modifier quelque chose de physique, comme arrêter un ventilateur ou ouvrir une vanne, gardez cette logique sur l'appareil ou sur un contrôleur local et publiez l'action effectuée dans un message distinct. Le broker enregistre alors ce qui s'est passé sans se trouver sur le chemin critique.

Bonnes pratiques Edge Impulse MQTT

  1. Publiez sur changement ou sur seuil, et non à chaque inférence, une fois la phase de test passée. Ajoutez un heartbeat pour qu'un silence ne soit jamais ambigu.
  2. Incluez la version du modèle. Les projets Edge Impulse sont souvent réentraînés ; ajoutez une chaîne de version de déploiement à chaque message.
  3. Utilisez le score d'anomalie si votre impulse comporte un bloc d'anomalie. Il détecte les entrées qui ne ressemblent à rien de l'entraînement, ce que la classification seule ne peut pas faire.
  4. Laissez le seuil de confiance côté abonné lorsque plusieurs systèmes consomment le même topic avec des tolérances différentes aux fausses alarmes.
  5. Respectez les limites des messages. Publier tous les scores de classes ne pose pas de problème ; publier des fenêtres brutes, si, sauf à échantillonner rarement pour collecter des données de réentraînement.

Pour un regard au niveau du framework sur ce qui tourne sous le capot, lisez la prise en main de TensorFlow Lite Micro. Pour construire le même pipeline sans plateforme, suivez TinyML sur ESP32. Le guide IA et IoT remet les deux en perspective.

Envoyez vos prédictions Edge Impulse vers ConnectMQ

Quand le modèle fonctionne, donnez-lui une destination de niveau production. ConnectMQ est un broker MQTT managé avec TLS, MQTT 5, des règles d'alerte à seuil et un explorateur de topics en direct, pour suivre les prédictions et alerter sur les scores d'anomalie sans exploiter vous-même un broker. Commencez avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire.

Questions fréquentes

Edge Impulse prend-il directement en charge MQTT ?

La bibliothèque de modèle exportée effectue uniquement l'inférence. Vous ajoutez la publication MQTT avec un client standard comme PubSubClient sur les microcontrôleurs ou paho-mqtt sous Linux, comme le montre ce tutoriel.

Pourquoi mon ESP32 échoue-t-il à publier les résultats Edge Impulse ?

La cause la plus fréquente est le tampon par défaut de 256 octets de PubSubClient. Appelez setBufferSize() avec une valeur plus grande, ou publiez moins de champs.

Puis-je exécuter des modèles Edge Impulse sur un ESP32 sans l'Arduino IDE ?

Oui. Edge Impulse exporte aussi une bibliothèque C++ générique que vous pouvez compiler avec ESP-IDF, et vous pouvez publier avec le client MQTT d'ESP-IDF au lieu de PubSubClient.

Comment collecter de nouvelles données d'entraînement depuis des appareils déployés ?

Publiez occasionnellement des fenêtres brutes ou des échantillons à faible confiance sur un topic séparé, stockez-les avec un abonné et téléversez-les dans votre projet Edge Impulse pour étiquetage et réentraînement.

#Edge Impulse#TinyML#ESP32#Raspberry Pi#MQTT

Articles liés

Tous les articles →

Votre broker MQTT privé, à quelques clics

Créez un compte, récupérez les identifiants de votre broker et connectez votre premier appareil — aucun serveur à installer, mettre à jour ou dimensionner.

Commencer Voir les tarifs