TinyML en ESP32: entrena, despliega y publica resultados por MQTT

TinyML en ESP32 permite que un microcontrolador Wi-Fi barato clasifique por sí mismo los datos de un sensor y comunique solo la respuesta. En este tutorial entrenarás un pequeño clasificador de vibraciones en Python, lo cuantizarás a int8, lo ejecutarás en un ESP32 con la adaptación de TensorFlow Lite Micro de Espressif y publicarás cada clasificación por MQTT en broker.connectmq.com, el servidor MQTT público gratuito de ConnectMQ para pruebas y prototipos. Todo funciona con una placa ESP32 sin nada más, porque el firmware simula la señal del sensor hasta que conectes un acelerómetro real.

Qué vas a construir

El dispositivo muestrea una ventana de vibración de 256 puntos a 1 kHz, calcula tres características (RMS, factor de cresta y tasa de cruces por cero), ejecuta una red neuronal diminuta y clasifica la máquina como idle, normal o fault. Publica un resultado JSON cada vez que cambia la clase, más un latido cada 30 segundos, y mantiene un estado online/offline retenido con Last Will.

¿Por qué ESP-IDF en lugar del Arduino IDE? Espressif mantiene esp-tflite-micro al día con la versión actual de TensorFlow Lite Micro, con kernels optimizados para el ESP32-S3. Varias adaptaciones para Arduino van por detrás de la API original, lo que es una fuente habitual de errores de compilación. Si TinyML es nuevo para ti, echa antes un vistazo a qué es TinyML.

Paso 1: entrenar y cuantizar el modelo en Python

Crea una carpeta tinyml_mqtt para el proyecto ESP-IDF con una subcarpeta main y coloca este script de entrenamiento a su lado. Genera ventanas de vibración sintéticas etiquetadas, entrena una pequeña red densa, la convierte en un modelo TensorFlow Lite totalmente int8 y escribe main/model_data.cc y main/model_data.h, incluidas las constantes de normalización de características que necesita el firmware.

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: solo ruido del sensor
        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: impactos periódicos
        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")

El modelo tiene dos capas FULLY_CONNECTED y una SOFTMAX, y la versión int8 ocupa solo unos pocos kilobytes. La precisión de validación debería ser casi perfecta, porque las clases sintéticas están bien separadas. Con datos reales, cuenta con dedicar la mayor parte del tiempo a la recogida de datos y a las características, no a la red.

Por qué cuantización int8 completa

Los ajustes del conversor anteriores producen un modelo cuyos pesos, activaciones, entrada y salida son todos enteros de 8 bits. Esto importa en el ESP32 por tres motivos. El modelo ocupa aproximadamente una cuarta parte de la versión en coma flotante, los kernels optimizados de esp-tflite-micro están escritos para int8 y la arena necesaria para los tensores intermedios también se reduce. El precio es que tienes que convertir tú mismo las entradas al dominio entero, usando la escala y el punto cero almacenados en el tensor de entrada, y volver a convertir las salidas. El firmware siguiente hace exactamente eso.

El conjunto de datos representativo es lo que usa el conversor para elegir esas escalas. Si no cubre todo el rango de entradas reales, los valores fuera de rango se recortan y la precisión cae en el dispositivo aunque en Python pareciera perfecta. Cuando pases a datos de sensor reales, alimenta representative_data() con vectores de características normalizados reales de todas las clases.

Paso 2: crear el proyecto ESP-IDF

Estructura del proyecto:

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

CMakeLists.txt de nivel superior. Incorpora el asistente Wi-Fi de ejemplo de Espressif para que el tutorial sea breve:

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 con tu red Wi-Fi:

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

Después añade el runtime de inferencia desde el ESP Component Registry, dentro de la carpeta del proyecto:

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

Paso 3: firmware con inferencia y MQTT

main/main.cc configura el Wi-Fi y MQTT, carga el modelo en una tensor arena estática y ejecuta el bucle de muestreo, características, inferencia y publicación. Cambia TOPIC_BASE y el client ID por algo único antes de flashear.

#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"   // hazlo único
#define CLIENT_ID  "esp32-tinyml-7c41a9"                  // hazlo único

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

// ---- sensor simulado: sustitúyelo por muestras reales de un acelerómetro ----
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: cambia el estado de la máquina cada 10 ventanas
    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));
  }
}

Compila, flashea y abre el monitor:

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

Paso 4: ver cómo llegan los resultados

Desde cualquier ordenador, suscríbete a tu prefijo:

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

Verás el estado online retenido y, a continuación, un mensaje de inferencia cada vez que la máquina simulada cambie de estado, más un latido cada 30 segundos. El firmware evalúa dos ventanas por segundo, pero publica solo un puñado de mensajes por minuto, que es precisamente la idea de TinyML: calcular en local e informar con moderación. El servidor público, pensado para pruebas y no para datos de producción, admite hasta 10 mensajes por segundo por cliente, así que este diseño deja margen de sobra.

Diseño de los topics y del payload

El tutorial usa dos topics bajo un mismo prefijo de dispositivo: status, retenido y respaldado por un Last Will, e inference, que transporta los resultados. Esa separación permite que un panel muestre al instante la salud del dispositivo mientras los resultados fluyen como eventos. A medida que crece la flota, añade más contexto a la jerarquía, por ejemplo plant1/utilities/compressors/comp-04/inference, para que un suscriptor pueda seleccionar una máquina, una línea o toda la planta con comodines.

Mantén el payload corto pero autodescriptivo. El campo model te dice qué firmware produjo un resultado tras una actualización OTA, p permite a los suscriptores aplicar su propio umbral de confianza y us revela cuándo aumenta el tiempo de inferencia tras un cambio de modelo. Si necesitas ajustar el comportamiento a distancia, suscribe el dispositivo a un topic cmd para ajustes como el intervalo de latido y valida cada valor antes de aplicarlo. En un broker privado de ConnectMQ puedes ver ambos topics en el explorador de topics en directo y definir una regla de alerta por umbral sobre los campos numéricos mientras haces pruebas.

Pasar a datos de sensor reales

Sustituye read_window() por muestras de un acelerómetro como un MPU6050 o un ADXL345, tomadas a una frecuencia fija con un temporizador hardware o con la FIFO del sensor. Después vuelve a entrenar con ventanas grabadas: registra ventanas reales por MQTT o por puerto serie en cada estado de la máquina, etiquétalas y pásalas por el mismo código de características y entrenamiento. Mantén idéntico el cálculo de características en Python y en C++, incluida la definición de cruce por cero, o la precisión en el dispositivo no coincidirá con la del entrenamiento.

Cuando pases a producción, cambia a TLS y credenciales: usa la URI mqtts://broker.connectmq.com:8883 con el host de tu broker de ConnectMQ, configura cfg.credentials.username, cfg.credentials.authentication.password y el client ID emitido, y proporciona el certificado de la CA (el paquete de certificados de ESP-IDF cubre Let's Encrypt). El tutorial de ESP32 con MQTT explica en detalle la configuración de TLS.

Solución de problemas de TinyML en ESP32

Próximos pasos

Compara este pipeline construido a mano con el flujo de trabajo de Edge Impulse con MQTT, que automatiza la recogida de datos y el diseño de características, o profundiza en el propio runtime en primeros pasos con TensorFlow Lite Micro. La guía de IA e IoT muestra dónde encajan dispositivos como este en un sistema AIoT más amplio.

Empieza con ConnectMQ

Tu nodo TinyML solo es tan útil como los sistemas que reciben sus resultados. ConnectMQ es un broker MQTT gestionado con TLS, credenciales generadas y un explorador de topics en directo para ver los mensajes de inferencia en tiempo real. Crea un broker con el plan Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y apunta a él el firmware.

Preguntas frecuentes

¿Qué ESP32 es el mejor para TinyML?

El ESP32-S3 es la opción habitual porque sus instrucciones vectoriales aceleran los kernels de redes neuronales en esp-tflite-micro y muchas placas incluyen PSRAM. El ESP32 original funciona bien con modelos pequeños como el de este tutorial.

¿Puedo hacer TinyML en ESP32 con el Arduino IDE?

Sí, mediante librerías de Arduino que incluyen TensorFlow Lite Micro o exportaciones de Edge Impulse. Comprueba que la versión de la librería coincida con la API actual de TFLM, porque las adaptaciones antiguas usan constructores y cabeceras diferentes.

¿Qué tamaño puede tener un modelo TinyML en ESP32?

Depende de la flash para los pesos y de la RAM para la tensor arena. Los modelos densos y convolucionales pequeños caben sin problema en la memoria interna; los modelos de visión más grandes suelen necesitar un ESP32-S3 con PSRAM.

¿Por qué publicar solo cuando hay cambios?

La inferencia se ejecuta muchas veces por segundo, pero a los sistemas posteriores solo les interesa cuando cambia el estado. Publicar al cambiar más un latido ahorra energía de radio y ancho de banda y te mantiene dentro de los límites de frecuencia del broker.