Edge Impulse MQTT: Publique Previsões de Modelos

A integração Edge Impulse MQTT é o último passo para transformar um modelo treinado no Edge Impulse Studio em algo que o resto do seu sistema possa usar. Neste tutorial você vai exportar um impulse como biblioteca Arduino, executá-lo em um ESP32 e publicar cada previsão como JSON no servidor MQTT público broker.connectmq.com; depois fará o mesmo em um Raspberry Pi com o runner para Linux e Python. Ao final, dashboards e alertas poderão assinar a saída do seu modelo como a de qualquer outro sensor.

Como as peças se encaixam

O Edge Impulse cuida da coleta de dados, dos blocos de processamento de sinais, do treinamento e da otimização. O que ele produz é um artefato implantável: uma biblioteca C++ (também empacotada como biblioteca Arduino) para microcontroladores ou um executável .eim para alvos Linux. Nenhum dos dois sabe nada sobre a sua rede. Você adiciona um pouco de código em torno do classificador para publicar os resultados em um broker MQTT, e a partir daí qualquer número de consumidores pode reagir.

Alvo	Opção de exportação	Chamada de inferência	Cliente MQTT
ESP32 / ESP32-S3	Biblioteca Arduino (.zip)	`run_classifier()` em C++	PubSubClient
Raspberry Pi / Linux	`.eim` via runner para Linux	`ImpulseRunner.classify()` em Python	paho-mqtt 2.x

Passo 1: exporte o modelo do Edge Impulse Studio

Abra o seu projeto e verifique se o impulse está treinado e se os resultados de Model testing são aceitáveis.
Vá em Deployment, escolha Arduino library e selecione o modelo quantizado (int8) se as opções EON Compiler e quantizado estiverem disponíveis. Modelos quantizados usam menos RAM e rodam mais rápido em microcontroladores.
Clique em Build e baixe o .zip. O nome da biblioteca segue o nome do seu projeto, por exemplo vibration_monitor_inferencing.
Na Arduino IDE, use Sketch > Include Library > Add .ZIP Library para instalá-la.
Instale também a PubSubClient de Nick O'Leary pelo Gerenciador de Bibliotecas e verifique se o pacote de placas ESP32 da Espressif está instalado.

Para testar sem conectar um sensor, abra Live classification ou uma amostra de teste no Studio e clique em Copy to clipboard ao lado das features brutas. Você vai colar esse array no sketch.

Passo 2: sketch para ESP32 que publica previsões

O sketch abaixo executa o classificador sobre um buffer estático de features a cada dois segundos e publica o resultado. Substitua o include pelo nome do header da sua biblioteca e cole as suas features brutas. Uma versão alimentada por sensor aparece em seguida.

#include <WiFi.h>
#include <PubSubClient.h>
#include <vibration_monitor_inferencing.h>   // sua biblioteca 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";                       // torne único
const char *TOPIC_BASE = "connectmq/demo/4b7e19/line1/fan3";    // torne único

// Cole as features brutas copiadas do 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) {
  // Encontra a classe principal e monta um objeto JSON com todos os 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);   // o padrão de 256 bytes é pequeno demais para todos os 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);
}

Grave o firmware, abra o Monitor Serial a 115200 baud e a classificação da amostra colada aparecerá a cada dois segundos. Os scores devem corresponder ao que o Studio mostrou para a mesma amostra. Se não corresponderem, as features foram copiadas de outra amostra ou do bloco de processamento errado.

Usando dados de sensor em tempo real

Para uma implantação real, preencha um buffer a partir do sensor no intervalo de amostragem com que o impulse foi treinado e depois envolva-o em um signal. EI_CLASSIFIER_INTERVAL_MS e EI_CLASSIFIER_RAW_SAMPLES_PER_FRAME vêm da biblioteca exportada:

static float buffer[EI_CLASSIFIER_DSP_INPUT_FRAME_SIZE];

void readAccel(float &x, float &y, float &z);   // implemente para o seu 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]);   // supõe um impulse de 3 eixos
    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;
}

Chame mqtt.loop() entre as janelas. Com janelas longas, um loop de amostragem com espera ativa pode impedir o keepalive do MQTT; portanto, mantenha as janelas curtas ou faça a amostragem a partir de uma interrupção de timer ou de uma task do FreeRTOS.

Passo 3: Edge Impulse no Raspberry Pi com Python

Em alvos Linux, o Edge Impulse fornece um runner que baixa um modelo .eim otimizado para a placa. Instale a CLI seguindo a documentação do Edge Impulse para Linux referente ao seu dispositivo e depois execute:

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

Este script carrega o modelo, classifica um vetor de features e publica o resultado. Substitua get_features() pela sua própria fonte de dados, por exemplo valores lidos de um sensor via 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: retorna `count` floats no mesmo formato das features brutas do 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 projetos com câmera e microfone, o mesmo SDK oferece runners de imagem e áudio que capturam e classificam em uma única etapa; publique os resultados deles da mesma forma.

Passo 4: assine e verifique

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

Você deverá ver o status retido e um fluxo de mensagens de inferência. O servidor MQTT público é aberto a todos e serve para testes e protótipos, então use um prefixo aleatório e nunca envie dados privados ou de produção. Para produção, crie um broker privado e mude para TLS na porta 8883 com WiFiClientSecure no ESP32 ou client.tls_set() em Python, usando o nome de usuário, a senha e o client ID do portal.

Transformando previsões em ações

Assim que as previsões chegam a um broker, o trabalho do dispositivo termina e o resto é integração MQTT comum. Alguns padrões cobrem a maioria dos projetos.

Dashboards. Um dashboard no navegador pode assinar via WebSocket seguro na porta 8084 com o caminho /mqtt e plotar o objeto scores de cada dispositivo. Como o topic de status é retido, o dashboard sabe imediatamente, ao carregar, quais dispositivos estão online.

Alertas. Se o seu impulse inclui um bloco de anomalia, o índice de anomalia é o campo mais útil para gerar alertas, porque ele sobe para entradas que o modelo nunca viu, incluindo modos de falha que você não registrou. As regras de alerta por limiar do ConnectMQ podem monitorar um valor numérico e registrar alertas no log de alertas, o que muitas vezes basta durante um piloto. Para notificações em celulares ou ferramentas de chat, rode um fluxo Node-RED ou um pequeno assinante Python que escute os topics de inferência e chame o serviço de notificação que você já usa.

Historiadores. Telegraf, Node-RED ou um assinante Python podem gravar cada previsão em um banco de dados de séries temporais, dando a você um histórico dos estados das máquinas (por exemplo, dos exaustores de uma planta em Joinville) para comparar com os registros de manutenção. Esse histórico também é a melhor fonte de candidatos para retreinamento.

Malha fechada. Se uma previsão deve alterar algo físico, como parar um ventilador ou abrir uma válvula, mantenha essa lógica no dispositivo ou em um controlador local e publique a ação executada como uma mensagem própria. Assim, o broker registra o que aconteceu em vez de ficar no caminho crítico.

Boas práticas de Edge Impulse com MQTT

Publique na mudança ou ao cruzar um limiar, e não a cada inferência, depois que sair da fase de testes. Adicione um heartbeat para que o silêncio nunca seja ambíguo.
Inclua a versão do modelo. Projetos do Edge Impulse são retreinados com frequência; adicione uma string de versão da implantação a cada mensagem.
Use o índice de anomalia se o seu impulse tiver um bloco de anomalia. Ele captura entradas diferentes de tudo o que havia no treinamento, algo que a classificação sozinha não consegue.
Mantenha o limiar de confiança do lado do assinante quando vários sistemas consumirem o mesmo topic com tolerâncias diferentes a alarmes falsos.
Respeite os limites de mensagem. Publicar os scores de todas as classes é aceitável; publicar janelas brutas não é, a menos que você faça amostragens raras para coletar dados de retreinamento.

Para uma visão no nível do framework sobre o que roda por baixo, leia primeiros passos com TensorFlow Lite Micro. Para construir o mesmo pipeline sem uma plataforma, siga TinyML no ESP32. O guia de IA e IoT coloca ambos em contexto.

Envie as previsões do Edge Impulse para o ConnectMQ

Quando o modelo funcionar, dê a ele um destino de nível de produção. O ConnectMQ é um broker MQTT gerenciado com TLS, MQTT 5, regras de alerta por limiar e um explorador de topics em tempo real, para que você possa acompanhar as previsões e gerar alertas com base nos índices de anomalia sem operar um broker por conta própria. Comece no plano gratuito Serverless em app.connectmq.com, sem cartão de crédito.

Perguntas frequentes

O Edge Impulse oferece suporte direto a MQTT?

A biblioteca do modelo exportado executa apenas a inferência. Você adiciona a publicação MQTT com um cliente padrão, como PubSubClient em microcontroladores ou paho-mqtt no Linux, como mostrado neste tutorial.

Por que a publicação do meu ESP32 falha com os resultados do Edge Impulse?

A causa mais comum é o buffer padrão de 256 bytes da PubSubClient. Chame setBufferSize() com um valor maior ou publique menos campos.

Posso rodar modelos do Edge Impulse em um ESP32 sem a Arduino IDE?

Sim. O Edge Impulse também exporta uma biblioteca C++ genérica que você pode compilar com o ESP-IDF, e é possível publicar com o cliente MQTT do ESP-IDF em vez da PubSubClient.

Como coletar novos dados de treinamento de dispositivos implantados?

Publique ocasionalmente janelas brutas ou amostras de baixa confiança em um topic separado, armazene-as com um assinante e envie-as ao seu projeto no Edge Impulse para rotulagem e retreinamento.

Do Edge Impulse ao MQTT: Publique Previsões de Modelos a partir do ESP32 e do Linux