Edge Impulse MQTT: publica predicciones del modelo

La integración de Edge Impulse con MQTT es el último paso que convierte un modelo entrenado en Edge Impulse Studio en algo que el resto de tu sistema puede usar. En este tutorial exportarás un impulse como librería de Arduino, lo ejecutarás en un ESP32 y publicarás cada predicción como JSON en broker.connectmq.com, y después harás lo mismo en una Raspberry Pi con el runner de Linux y Python. Al terminar, los paneles y las alertas podrán suscribirse a la salida de tu modelo como a la de cualquier otro sensor.

Cómo encajan las piezas

Edge Impulse se encarga de la recogida de datos, los bloques de procesamiento de señal, el entrenamiento y la optimización. Lo que produce es un artefacto desplegable: una librería C++ (también empaquetada como librería de Arduino) para microcontroladores o un ejecutable .eim para destinos Linux. Ninguno de los dos sabe nada de tu red. Tú añades un poco de código alrededor del clasificador que publica los resultados en un broker MQTT, y a partir de ahí cualquier número de consumidores puede reaccionar.

Destino	Opción de exportación	Llamada de inferencia	Cliente MQTT
ESP32 / ESP32-S3	Librería de Arduino (.zip)	`run_classifier()` en C++	PubSubClient
Raspberry Pi / Linux	`.eim` mediante el runner de Linux	`ImpulseRunner.classify()` en Python	paho-mqtt 2.x

Paso 1: exportar el modelo desde Edge Impulse Studio

Abre tu proyecto y asegúrate de que el impulse está entrenado y de que los resultados de Model testing son aceptables.
Ve a Deployment, elige Arduino library y selecciona el modelo cuantizado (int8) si se ofrecen el EON Compiler y las opciones cuantizadas. Los modelos cuantizados usan menos RAM y se ejecutan más rápido en microcontroladores.
Haz clic en Build y descarga el .zip. El nombre de la librería sigue el nombre de tu proyecto, por ejemplo vibration_monitor_inferencing.
En el Arduino IDE, usa Sketch > Include Library > Add .ZIP Library para instalarla.
Instala también PubSubClient de Nick O'Leary desde el Gestor de librerías y asegúrate de tener instalado el paquete de placas ESP32 de Espressif.

Para probar sin cablear un sensor, abre Live classification o una muestra de prueba en Studio y haz clic en Copy to clipboard junto a las características en bruto. Pegarás ese array en el sketch.

Paso 2: sketch de ESP32 que publica predicciones

El sketch siguiente ejecuta el clasificador sobre un búfer estático de características cada dos segundos y publica el resultado. Sustituye el include por el nombre de la cabecera de tu librería y pega tus características en bruto. Después se muestra una versión que lee de un sensor.

#include <WiFi.h>
#include <PubSubClient.h>
#include <vibration_monitor_inferencing.h>   // tu librería exportada

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";                       // hazlo único
const char *TOPIC_BASE = "connectmq/demo/4b7e19/line1/fan3";    // hazlo único

// Pega las características en bruto copiadas de 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) {
  // Busca la clase principal y construye un objeto JSON con todas las puntuaciones
  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);   // los 256 bytes por defecto no bastan para todas las puntuaciones
  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éalo, abre el Monitor Serie a 115200 baudios y la clasificación de la muestra pegada aparecerá cada dos segundos. Sus puntuaciones deberían coincidir con las que mostraba Studio para esa misma muestra. Si no coinciden, las características se copiaron de otra muestra o del bloque de procesamiento equivocado.

Usar datos de sensor en directo

Para un despliegue real, llena un búfer con lecturas del sensor al intervalo de muestreo con el que se entrenó el impulse y envuélvelo en una señal. EI_CLASSIFIER_INTERVAL_MS y EI_CLASSIFIER_RAW_SAMPLES_PER_FRAME proceden de la librería exportada:

static float buffer[EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE];

void readAccel(float &x, float &y, float &z);   // impleméntala para tu 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]);   // supone un impulse de 3 ejes
    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;
}

Llama a mqtt.loop() entre ventanas. Con ventanas largas, un bucle de muestreo con espera activa puede dejar sin atender el keepalive de MQTT, así que mantén las ventanas cortas o muestrea desde una interrupción de temporizador o una tarea de FreeRTOS.

Paso 3: Edge Impulse en Raspberry Pi con Python

En destinos Linux, Edge Impulse ofrece un runner que descarga un modelo .eim optimizado para la placa. Instala la CLI siguiendo la documentación de Edge Impulse para Linux correspondiente a tu dispositivo y, después:

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

Este script carga el modelo, clasifica un vector de características y publica el resultado. Sustituye get_features() por tu propia fuente de datos, por ejemplo valores leídos de un sensor por 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):
    """Marcador de posición: devuelve `count` floats en el mismo formato que las características en bruto de Studio."""
    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()

Para proyectos con cámara y micrófono, el mismo SDK ofrece runners de imagen y audio que capturan y clasifican en un solo paso; publica sus resultados de la misma forma.

Paso 4: suscribirse y verificar

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

Deberías ver el estado retenido y un flujo de mensajes de inferencia. El servidor MQTT público gratuito es para pruebas y prototipos y está abierto a todo el mundo, así que usa un prefijo aleatorio y no envíes nunca datos privados. Para producción, crea un broker privado y cambia a TLS en el puerto 8883 con WiFiClientSecure en el ESP32 o client.tls_set() en Python, usando el usuario, la contraseña y el client ID del portal.

Convertir las predicciones en acciones

Una vez que las predicciones llegan a un broker, el trabajo del dispositivo ha terminado y el resto es integración MQTT corriente. Unos cuantos patrones cubren la mayoría de los proyectos.

Paneles. Un panel en el navegador puede suscribirse por WebSocket seguro en el puerto 8084 con la ruta /mqtt y representar el objeto scores de cada dispositivo. Como el topic de estado es retenido, el panel sabe al cargar qué dispositivos están en línea.

Alertas. Si tu impulse incluye un bloque de anomalías, la puntuación de anomalía es el campo más útil para alertar, porque aumenta con entradas que el modelo nunca ha visto, incluidos modos de fallo que no registraste. Las reglas de alerta por umbral de ConnectMQ pueden vigilar un valor numérico y registrar las alertas en el registro de alertas, lo que suele bastar durante un piloto. Para notificaciones al móvil o a herramientas de chat, ejecuta un flujo de Node-RED o un pequeño suscriptor en Python que escuche los topics de inferencia y llame al servicio de notificaciones que ya utilices.

Historiadores. Telegraf, Node-RED o un suscriptor en Python pueden escribir cada predicción en una base de datos de series temporales, lo que te da un histórico de estados de la máquina para compararlo con los registros de mantenimiento. Ese histórico es también la mejor fuente de candidatos para reentrenar.

Bucle cerrado. Si una predicción debe cambiar algo físico, como detener un ventilador o abrir una válvula, mantén esa lógica en el dispositivo o en un controlador local y publica la acción realizada como un mensaje propio. Así el broker registra lo ocurrido en lugar de estar en la ruta crítica.

Buenas prácticas de Edge Impulse con MQTT

Publica cuando haya cambios o se supere un umbral, no en cada inferencia, una vez superada la fase de pruebas. Añade un latido para que el silencio nunca sea ambiguo.
Incluye la versión del modelo. Los proyectos de Edge Impulse se reentrenan a menudo; añade una cadena con la versión del despliegue a cada mensaje.
Usa la puntuación de anomalía si tu impulse tiene un bloque de anomalías. Detecta entradas distintas a todo lo visto en el entrenamiento, algo que la clasificación por sí sola no puede hacer.
Aplica el umbral de confianza en el lado del suscriptor cuando varios sistemas consumen el mismo topic con distinta tolerancia a las falsas alarmas.
Respeta los límites de los mensajes. Publicar todas las puntuaciones de clase está bien; publicar ventanas en bruto no, salvo que muestrees esporádicamente para recoger datos de reentrenamiento.

Para ver a nivel de framework lo que se ejecuta por debajo, lee primeros pasos con TensorFlow Lite Micro. Para construir el mismo pipeline sin una plataforma, sigue TinyML en ESP32. La guía de IA e IoT pone ambos en contexto.

Envía tus predicciones de Edge Impulse a ConnectMQ

Cuando el modelo funcione, dale un destino de nivel de producción. ConnectMQ es un broker MQTT gestionado con TLS, MQTT 5, reglas de alerta por umbral y un explorador de topics en directo, para que puedas ver las predicciones y alertar sobre las puntuaciones de anomalía sin operar tú mismo un broker. Empieza con el plan Serverless gratuito en app.connectmq.com, sin tarjeta de crédito.

Preguntas frecuentes

¿Edge Impulse es compatible con MQTT directamente?

La librería del modelo exportado solo realiza la inferencia. Tú añades la publicación MQTT con un cliente estándar, como PubSubClient en microcontroladores o paho-mqtt en Linux, tal como se muestra en este tutorial.

¿Por qué falla la publicación de los resultados de Edge Impulse en mi ESP32?

La causa más habitual es el búfer por defecto de 256 bytes de PubSubClient. Llama a setBufferSize() con un valor mayor o publica menos campos.

¿Puedo ejecutar modelos de Edge Impulse en un ESP32 sin el Arduino IDE?

Sí. Edge Impulse también exporta una librería C++ genérica que puedes compilar con ESP-IDF, y puedes publicar con el cliente MQTT de ESP-IDF en lugar de PubSubClient.

¿Cómo recojo nuevos datos de entrenamiento de los dispositivos desplegados?

Publica de vez en cuando ventanas en bruto o muestras de baja confianza en un topic separado, guárdalas con un suscriptor y súbelas a tu proyecto de Edge Impulse para etiquetarlas y reentrenar.

De Edge Impulse a MQTT: publica predicciones del modelo desde ESP32 y Linux