TinyML no ESP32: Treine, Implante e Publique Resultados via MQTT

O TinyML no ESP32 permite que um microcontrolador Wi-Fi barato classifique dados de sensores sozinho e informe apenas a resposta. Neste tutorial você vai treinar um pequeno classificador de vibração em Python, quantizá-lo para int8, executá-lo em um ESP32 com o port do TensorFlow Lite Micro da Espressif e publicar cada classificação no broker.connectmq.com via MQTT. Tudo roda em uma placa ESP32 sem nenhum periférico, porque o firmware simula o sinal do sensor até você conectar um acelerômetro real.

O que você vai construir

O dispositivo amostra uma janela de vibração de 256 pontos a 1 kHz, calcula três features (RMS, fator de crista e taxa de cruzamento por zero), executa uma pequena rede neural e classifica a máquina como idle, normal ou fault. Ele publica um resultado em JSON sempre que a classe muda, além de um heartbeat a cada 30 segundos, e mantém um status online/offline retido com Last Will.

Por que ESP-IDF em vez da Arduino IDE? A Espressif mantém o esp-tflite-micro alinhado ao TensorFlow Lite Micro upstream atual, com kernels otimizados para o ESP32-S3. Vários ports para Arduino estão atrasados em relação à API upstream, o que é uma fonte comum de erros de compilação. Se TinyML é novidade para você, dê uma olhada antes em o que é TinyML.

Passo 1: treine e quantize o modelo em Python

Crie uma pasta tinyml_mqtt para o projeto ESP-IDF, com uma subpasta main, e coloque este script de treinamento ao lado dela. Ele gera janelas de vibração sintéticas rotuladas, treina uma pequena rede densa, converte-a em um modelo TensorFlow Lite totalmente int8 e grava main/model_data.cc e main/model_data.h, incluindo as constantes de normalização de features de que o firmware precisa.

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: apenas ruído do 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")

O modelo tem duas camadas FULLY_CONNECTED e uma SOFTMAX, e a versão int8 tem apenas alguns kilobytes. A precisão de validação deve ficar próxima da perfeição, porque as classes sintéticas são bem separadas. Com dados reais, espere passar a maior parte do tempo na coleta de dados e nas features, e não na rede.

Por que quantização int8 completa

As configurações do conversor acima produzem um modelo cujos pesos, ativações, entrada e saída são todos inteiros de 8 bits. Isso importa no ESP32 por três motivos. O modelo fica com cerca de um quarto do tamanho da versão em float, os kernels otimizados do esp-tflite-micro são escritos para int8 e a arena necessária para os tensores intermediários também diminui. O preço é que você precisa converter as entradas para o domínio inteiro por conta própria, usando a escala e o zero point armazenados no tensor de entrada, e converter as saídas de volta. O firmware abaixo faz exatamente isso.

O conjunto de dados representativo é o que o conversor usa para escolher essas escalas. Se ele não cobrir toda a faixa de entradas reais, os valores fora da faixa são cortados e a precisão cai no dispositivo, mesmo que tenha parecido perfeita em Python. Quando passar para dados reais de sensores, alimente representative_data() com vetores de features normalizados reais de todas as classes.

Passo 2: crie o projeto ESP-IDF

Estrutura do projeto:

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

CMakeLists.txt de nível superior. Ele inclui o helper de Wi-Fi dos exemplos da Espressif para manter o tutorial curto:

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 com a sua rede Wi-Fi:

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

Em seguida, adicione o runtime de inferência a partir do ESP Component Registry, de dentro da pasta do projeto:

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

Passo 3: firmware com inferência e MQTT

O main/main.cc configura o Wi-Fi e o MQTT, carrega o modelo em uma tensor arena estática e executa o loop de amostragem, features, inferência e publicação. Altere TOPIC_BASE e o client ID para algo único antes de gravar o firmware.

#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"   // torne isto único
#define CLIENT_ID  "esp32-tinyml-7c41a9"                  // torne isto ú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: substitua por amostras reais de 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: muda o estado da máquina a cada 10 janelas
    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));
  }
}

Compile, grave o firmware e abra o monitor:

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

Passo 4: acompanhe a chegada dos resultados

De qualquer computador, assine o seu prefixo:

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

Você verá o status online retido e, em seguida, uma mensagem de inferência cada vez que a máquina simulada mudar de estado, além de um heartbeat a cada 30 segundos. O firmware avalia duas janelas por segundo, mas publica apenas algumas mensagens por minuto, e essa é a essência do TinyML: processar localmente, informar de forma esparsa. O servidor MQTT público (gratuito), pensado para testes e protótipos, permite até 10 mensagens por segundo por cliente, então este projeto deixa folga de sobra.

Projetando os topics e o payload

O tutorial usa dois topics sob um mesmo prefixo de dispositivo: status, que é retido e respaldado por um Last Will, e inference, que transporta os resultados. Essa divisão permite que um dashboard mostre a saúde do dispositivo instantaneamente enquanto os resultados fluem como eventos. À medida que a frota cresce, por exemplo nos compressores de uma planta em Joinville, coloque mais contexto na hierarquia, por exemplo plant1/utilities/compressors/comp-04/inference, para que um assinante possa selecionar uma máquina, uma linha ou a planta inteira com wildcards.

Mantenha o payload curto, mas autodescritivo. O campo model diz qual firmware produziu um resultado após uma atualização over-the-air, p permite que os assinantes apliquem seu próprio limiar de confiança e us revela quando o tempo de inferência aumenta após uma troca de modelo. Se precisar ajustar o comportamento remotamente, faça o dispositivo assinar um topic cmd para configurações como o intervalo do heartbeat e valide cada valor antes de aplicá-lo. Em um broker ConnectMQ privado, você pode acompanhar os dois topics no explorador de topics em tempo real e definir uma regra de alerta por limiar em campos numéricos enquanto testa.

Passando para dados reais de sensores

Substitua read_window() por amostras de um acelerômetro, como um MPU6050 ou ADXL345, amostradas a uma taxa fixa com um timer de hardware ou o FIFO do sensor. Depois, retreine com janelas gravadas: registre janelas reais via MQTT ou serial para cada estado da máquina, rotule-as e alimente-as no mesmo código de features e treinamento. Mantenha o cálculo de features idêntico em Python e em C++, incluindo a definição de cruzamento por zero, ou a precisão no dispositivo não vai corresponder à do treinamento.

Quando for para produção, mude para TLS e credenciais: use a URI mqtts://broker.connectmq.com:8883 com o host do seu broker ConnectMQ, defina cfg.credentials.username, cfg.credentials.authentication.password e o client ID emitido e forneça o certificado da CA (o bundle de certificados do ESP-IDF cobre o Let's Encrypt). O tutorial de ESP32 com MQTT detalha a configuração do TLS.

Solução de problemas de TinyML no ESP32

Próximos passos

Compare este pipeline construído à mão com o fluxo de trabalho de Edge Impulse com MQTT, que automatiza a coleta de dados e o projeto de features, ou aprofunde-se no próprio runtime em primeiros passos com TensorFlow Lite Micro. O guia de IA e IoT mostra onde dispositivos como este se encaixam em um sistema AIoT maior.

Comece a usar o ConnectMQ

O seu nó TinyML é tão útil quanto os sistemas que recebem os seus resultados. O ConnectMQ é um broker MQTT gerenciado com TLS, credenciais geradas e um explorador de topics em tempo real para acompanhar as mensagens de inferência. Crie um broker no plano gratuito Serverless em app.connectmq.com, sem cartão de crédito, e aponte o firmware para ele.

Perguntas frequentes

Qual ESP32 é o melhor para TinyML?

O ESP32-S3 é a escolha usual, porque suas instruções vetoriais aceleram os kernels de redes neurais no esp-tflite-micro e muitas placas incluem PSRAM. O ESP32 original funciona bem para modelos pequenos como o deste tutorial.

Posso fazer TinyML no ESP32 com a Arduino IDE?

Sim, por meio de bibliotecas Arduino que empacotam o TensorFlow Lite Micro ou exportações do Edge Impulse. Verifique se a versão da biblioteca corresponde à API atual do TFLM, porque ports mais antigos usam construtores e headers diferentes.

Qual o tamanho máximo de um modelo TinyML no ESP32?

Depende da flash para os pesos e da RAM para a tensor arena. Modelos densos e convolucionais pequenos cabem facilmente na memória interna; modelos de visão maiores geralmente precisam de um ESP32-S3 com PSRAM.

Por que publicar apenas na mudança?

A inferência roda muitas vezes por segundo, mas os sistemas a jusante só se importam quando o estado muda. Publicar na mudança mais um heartbeat economiza energia de rádio e largura de banda e mantém você dentro dos limites de taxa do broker.