Edge Impulse to MQTT: Publish Model Predictions from ESP32 and Linux

Edge Impulse MQTT integration is the last step that turns a model you trained in Edge Impulse Studio into something the rest of your system can use. In this tutorial you will export an impulse as an Arduino library, run it on an ESP32 and publish each prediction as JSON to broker.connectmq.com, then do the same on a Raspberry Pi with the Linux runner and Python. By the end, dashboards and alerting can subscribe to your model's output like any other sensor.

How the Pieces Fit

Edge Impulse handles data collection, signal processing blocks, training and optimization. What it produces is a deployable artifact: a C++ library (also packaged as an Arduino library) for microcontrollers, or an .eim executable for Linux targets. Neither of these knows anything about your network. You add a small amount of code around the classifier that publishes results to an MQTT broker, and from there any number of consumers can react.

Step 1: Export the Model from Edge Impulse Studio

1. Open your project and make sure the impulse is trained and the Model testing results are acceptable.
2. Go to Deployment, choose Arduino library, and select the quantized (int8) model if the EON Compiler and quantized options are offered. Quantized models use less RAM and run faster on microcontrollers.
3. Click Build and download the .zip. The library name follows your project name, for example vibration_monitor_inferencing.
4. In the Arduino IDE, use Sketch > Include Library > Add .ZIP Library to install it.
5. Also install PubSubClient by Nick O'Leary from the Library Manager, and make sure the ESP32 board package from Espressif is installed.

To test without wiring a sensor, open Live classification or a test sample in Studio and click Copy to clipboard next to the raw features. You will paste that array into the sketch.

Step 2: ESP32 Sketch that Publishes Predictions

The sketch below runs the classifier on a static feature buffer every two seconds and publishes the result. Replace the include with your library's header name and paste your raw features. A sensor-driven version is shown after it.

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

Flash it, open the Serial Monitor at 115200 baud, and the classification for your pasted sample appears every two seconds. Its scores should match what Studio showed for the same sample. If they do not, the features were copied from a different sample or the wrong processing block.

Using Live Sensor Data

For a real deployment, fill a buffer from the sensor at the sampling interval the impulse was trained with, then wrap it in a signal. EI_CLASSIFIER_INTERVAL_MS and EI_CLASSIFIER_RAW_SAMPLES_PER_FRAME come from the exported library:

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

Call mqtt.loop() between windows. With long windows, a busy-wait sampling loop can starve the MQTT keepalive, so either keep windows short or sample from a timer interrupt or FreeRTOS task.

Step 3: Edge Impulse on Raspberry Pi with Python

On Linux targets, Edge Impulse provides a runner that downloads an optimized .eim model for the board. Install the CLI by following the Edge Impulse Linux documentation for your device, then:

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

This script loads the model, classifies a feature vector and publishes the result. Replace get_features() with your own data source, for example values read from a sensor over 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()

For camera and microphone projects, the same SDK provides image and audio runners that capture and classify in one step; publish their results the same way.

Step 4: Subscribe and Verify

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

You should see the retained status and a stream of inference messages. The public MQTT test broker is open to everyone, so use a random prefix and never send private data. For production, create a private broker and switch to TLS on port 8883 with WiFiClientSecure on the ESP32 or client.tls_set() in Python, using the username, password and client ID from the portal.

Turning Predictions into Actions

Once predictions arrive on a broker, the device's job is done and the rest is ordinary MQTT integration. A few patterns cover most projects.

Dashboards. A browser dashboard can subscribe over secure WebSocket on port 8084 with the /mqtt path and plot the scores object for each device. Because the status topic is retained, the dashboard knows immediately which devices are online when it loads.

Alerts. If your impulse includes an anomaly block, the anomaly score is the most useful field to alert on, because it rises for inputs the model has never seen, including failure modes you did not record. ConnectMQ threshold alert rules can watch a numeric value and record alerts in the alerts log, which is often enough during a pilot. For notifications to phones or chat tools, run a Node-RED flow or a small Python subscriber that listens on the inference topics and calls the notification service you already use.

Historians. Telegraf, Node-RED or a Python subscriber can write each prediction to a time-series database, giving you a history of machine states to compare against maintenance records. That history is also the best source of candidates for retraining.

Closed loop. If a prediction should change something physical, such as stopping a fan or opening a valve, keep that logic on the device or a local controller and publish the action taken as its own message. The broker then records what happened rather than being in the critical path.

Edge Impulse MQTT Best Practices

1. Publish on change or on threshold, not every inference, once you move past testing. Add a heartbeat so silence is never ambiguous.
2. Include the model version. Edge Impulse projects are retrained often; add a deployment version string to each message.
3. Use the anomaly score if your impulse has an anomaly block. It catches inputs unlike anything in training, which classification alone cannot.
4. Keep the confidence threshold on the subscriber side when several systems consume the same topic with different tolerance for false alarms.
5. Respect message limits. Publishing all class scores is fine; publishing raw windows is not, unless you sample rarely to collect retraining data.

For a framework-level look at what runs under the hood, read TensorFlow Lite Micro getting started. To build the same pipeline without a platform, follow TinyML on ESP32. The AI and IoT guide puts both in context.

Send Your Edge Impulse Predictions to ConnectMQ

When the model works, give it a production-grade destination. ConnectMQ is a hosted MQTT broker with TLS, MQTT 5, threshold alert rules and a live topic explorer, so you can watch predictions and alert on anomaly scores without running a broker yourself. Start on the free Serverless plan at app.connectmq.com, no credit card required.

Frequently Asked Questions

Does Edge Impulse support MQTT directly?

The exported model library performs inference only. You add MQTT publishing with a standard client such as PubSubClient on microcontrollers or paho-mqtt on Linux, as shown in this tutorial.

Why does my ESP32 publish fail with Edge Impulse results?

The most common cause is PubSubClient's default 256-byte buffer. Call setBufferSize() with a larger value, or publish fewer fields.

Can I run Edge Impulse models on an ESP32 without the Arduino IDE?

Yes. Edge Impulse also exports a generic C++ library that you can build with ESP-IDF, and you can publish with the ESP-IDF MQTT client instead of PubSubClient.

How do I collect new training data from deployed devices?

Publish occasional raw windows or low-confidence samples to a separate topic, store them with a subscriber, and upload them to your Edge Impulse project for labelling and retraining.