← Datalogging MQTT offline-first en un PLC ESP32 con SD

Datalogging MQTT offline-first en un PLC ESP32 con SD — ejemplo completo

Monitorización que no pierde datos: publicación MQTT por eventos con búfer en tarjeta SD, RTC sincronizado por NTP y contador de pulsos en un PLC ESP32.

Programa completo y ejecutable para el ESP32 PLC (mqtt-events-sd-buffering.ino): incluye cabecera de conexionado, requisitos y notas de integración.

Vista de solo lectura.

/*
 * COMPLETE EXAMPLE — Machine monitoring: event-driven MQTT + SD buffering
 *
 * Hardware: ESP32 PLC (Industrial Shields, any model with WiFi + SD + RTC)
 * Based on: textile monitoring project, TERROT_BDE.ino + modules
 *
 * Wiring:
 *   I0_0        Production pulse sensor (counts machine cycles)
 *   I0_1..I0_5  Machine status signals (running, stop, alarm...)
 *
 * Offline-first architecture:
 *   1. Change on an input -> JSON event published instantly over MQTT
 *   2. Every 20 s -> full snapshot with the pulse counter
 *   3. EVERYTHING is also written to the SD card (one file per day): if
 *      the network goes down nothing is lost; when it comes back the
 *      history can be recovered
 *   4. Real time from an NTP-synchronized RTC, and a unique messageID
 *      per message
 */

#include 
#include 
#include 
#include           // integrated RTC of the PLC (Industrial Shields library)
#include 

const char *WIFI_SSID = "WIFI_SSID";
const char *WIFI_PASS = "WIFI_PASS";
const char *MQTT_HOST = "BROKER_IP";
const int   MQTT_PORT = 1883;
const int   PLC_ID    = 1;

const char *TOPIC_DATA   = "planta/plc1/data";
const char *TOPIC_EVENTS = "planta/plc1/events";

WiFiClient net;
PubSubClient mqtt(net);

const int pins[] = {I0_0, I0_1, I0_2, I0_3, I0_4, I0_5};
const int NPINS = 6;
int lastState[NPINS];
volatile uint32_t pulses = 0;
uint32_t tSend = 0, tWifi = 0;

// ------------------------------------------------- Identity and persistence
String messageId() {
  return String(PLC_ID) + "_" + String(RTC.getTime()) + "_" + String(millis());
}

String dailyFileName() {
  if (!RTC.read()) return "/error_date.json";
  char n[20];
  sprintf(n, "/%04d-%02d-%02d.json", RTC.getYear(), RTC.getMonth(), RTC.getMonthDay());
  return String(n);
}

void saveToSD(const String &json) {
  File f = SD.open(dailyFileName(), FILE_APPEND);
  if (f) { f.println(json); f.close(); }
}

// Publishes if there is a connection; in any case the message stays on the SD
void publishMessage(const char *topic, const String &json) {
  saveToSD(json);
  if (mqtt.connected()) mqtt.publish(topic, json.c_str());
}

// ------------------------------------------------- Non-blocking connectivity
void checkWiFi() {
  if (WiFi.status() == WL_CONNECTED) return;
  if (millis() - tWifi < 10000) return;        // retry every 10 s, without blocking
  tWifi = millis();
  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASS);
}

void checkMQTT() {
  if (WiFi.status() != WL_CONNECTED || mqtt.connected()) return;
  mqtt.connect(("plc" + String(PLC_ID)).c_str());
}

void syncRTC() {                               // NTP -> physical RTC chip
  configTime(3600, 3600, "pool.ntp.org");
  struct tm ti;
  if (getLocalTime(&ti, 10000)) {
    RTC.setTime((uint32_t)mktime(&ti));
    RTC.write();
  }
}

void setup() {
  Serial.begin(115200);
  for (int i = 0; i < NPINS; i++) { pinMode(pins[i], INPUT); lastState[i] = digitalRead(pins[i]); }
  attachInterrupt(I0_0, [] { pulses++; }, FALLING);   // production counter

  SD.begin();
  RTC.begin();
  checkWiFi();
  mqtt.setServer(MQTT_HOST, MQTT_PORT);

  uint32_t t0 = millis();                      // initial WiFi wait (max 15 s)
  while (WiFi.status() != WL_CONNECTED && millis() - t0 < 15000) delay(500);
  if (WiFi.status() == WL_CONNECTED) syncRTC();
}

void loop() {
  checkWiFi();
  checkMQTT();
  mqtt.loop();

  // A. Spontaneous events: publish every input change instantly
  for (int i = 1; i < NPINS; i++) {
    int s = digitalRead(pins[i]);
    if (s != lastState[i]) {
      String evt = "{\"id\":" + String(PLC_ID) +
                   ",\"id_msg\":\"" + messageId() +
                   "\",\"input\":\"I0_" + String(i) +
                   "\",\"value\":" + String(s) + "}";
      publishMessage(TOPIC_EVENTS, evt);
      lastState[i] = s;
    }
  }

  // B. Periodic snapshot every 20 s with the pulse counter
  if (millis() - tSend > 20000) {
    tSend = millis();
    uint32_t n = pulses; pulses = 0;           // read and reset
    String json = "{\"id\":" + String(PLC_ID) +
                  ",\"id_msg\":\"" + messageId() +
                  "\",\"pulses\":" + String(n) + ",\"inputs\":{";
    for (int i = 0; i < NPINS; i++)
      json += "\"I0_" + String(i) + "\":" + String(digitalRead(pins[i])) +
              (i < NPINS - 1 ? "," : "");
    json += "}}";
    publishMessage(TOPIC_DATA, json);
  }

  delay(20);
}