esp32-Serial-Logger/serial_task.cpp

#include "serial_task.h"
#include <time.h>
#include <sys/time.h>

QueueHandle_t queueSD  = NULL;
QueueHandle_t queueWeb = NULL;
QueueHandle_t queueTX  = NULL;

volatile uint32_t serialBaud     = DEFAULT_BAUD_RATE;
volatile uint8_t  serialDataBits = 8;
volatile char     serialParity   = 'N';
volatile uint8_t  serialStopBits = 1;

static SemaphoreHandle_t serialMutex = NULL;

void getTimestamp(char *buf, size_t len) {
  struct timeval tv;
  gettimeofday(&tv, NULL);
  struct tm timeinfo;
  localtime_r(&tv.tv_sec, &timeinfo);
  int ms = tv.tv_usec / 1000;
  snprintf(buf, len, "%04d-%02d-%02d %02d:%02d:%02d.%03d",
    timeinfo.tm_year + 1900, timeinfo.tm_mon + 1, timeinfo.tm_mday,
    timeinfo.tm_hour, timeinfo.tm_min, timeinfo.tm_sec, ms);
}

static uint32_t getSerialConfig(uint8_t dataBits, char parity, uint8_t stopBits) {
  // Use standard Arduino serial config constants
  if (dataBits == 8 && parity == 'N' && stopBits == 1) return SERIAL_8N1;
  if (dataBits == 8 && parity == 'E' && stopBits == 1) return SERIAL_8E1;
  if (dataBits == 8 && parity == 'O' && stopBits == 1) return SERIAL_8O1;
  if (dataBits == 8 && parity == 'N' && stopBits == 2) return SERIAL_8N2;
  if (dataBits == 8 && parity == 'E' && stopBits == 2) return SERIAL_8E2;
  if (dataBits == 8 && parity == 'O' && stopBits == 2) return SERIAL_8O2;
  if (dataBits == 7 && parity == 'N' && stopBits == 1) return SERIAL_7N1;
  if (dataBits == 7 && parity == 'E' && stopBits == 1) return SERIAL_7E1;
  if (dataBits == 7 && parity == 'O' && stopBits == 1) return SERIAL_7O1;
  if (dataBits == 6 && parity == 'N' && stopBits == 1) return SERIAL_6N1;
  if (dataBits == 6 && parity == 'E' && stopBits == 1) return SERIAL_6E1;
  if (dataBits == 6 && parity == 'O' && stopBits == 1) return SERIAL_6O1;
  if (dataBits == 5 && parity == 'N' && stopBits == 1) return SERIAL_5N1;
  if (dataBits == 5 && parity == 'E' && stopBits == 1) return SERIAL_5E1;
  if (dataBits == 5 && parity == 'O' && stopBits == 1) return SERIAL_5O1;
  return SERIAL_8N1;  // default
}

void reconfigureSerial(uint32_t baud, uint8_t dataBits, char parity, uint8_t stopBits) {
  if (serialMutex) xSemaphoreTake(serialMutex, portMAX_DELAY);

  Serial2.flush();
  Serial2.end();
  delay(50);

  uint32_t config = getSerialConfig(dataBits, parity, stopBits);
  Serial2.begin(baud, config, SERIAL2_RX_PIN, SERIAL2_TX_PIN);
  Serial2.setRxBufferSize(DEFAULT_RX_BUFFER);

  serialBaud = baud;
  serialDataBits = dataBits;
  serialParity = parity;
  serialStopBits = stopBits;

  Serial.printf("[Serial] Reconfigured: %lu %d%c%d\n", baud, dataBits, parity, stopBits);

  if (serialMutex) xSemaphoreGive(serialMutex);
}

void serialTaskInit() {
  queueSD  = xQueueCreate(QUEUE_SD_SIZE,  sizeof(LogEntry*));
  queueWeb = xQueueCreate(QUEUE_WEB_SIZE, sizeof(LogEntry*));
  queueTX  = xQueueCreate(QUEUE_TX_SIZE,  sizeof(LogEntry*));

  serialMutex = xSemaphoreCreateMutex();

  uint32_t config = getSerialConfig(serialDataBits, serialParity, serialStopBits);
  Serial2.begin(serialBaud, config, SERIAL2_RX_PIN, SERIAL2_TX_PIN);
  Serial2.setRxBufferSize(DEFAULT_RX_BUFFER);

  Serial.printf("[Serial] Init: %lu %d%c%d on TX=%d RX=%d\n",
    serialBaud, serialDataBits, (char)serialParity, serialStopBits,
    SERIAL2_TX_PIN, SERIAL2_RX_PIN);

  xTaskCreatePinnedToCore(serialRxTask, "SerialRX", TASK_STACK_SERIAL,
    NULL, TASK_PRIORITY_SERIAL, NULL, 1);

  xTaskCreatePinnedToCore(serialTxTask, "SerialTX", TASK_STACK_SERIAL,
    NULL, TASK_PRIORITY_SERIAL - 1, NULL, 1);
}

// Flush assembled line to SD and Web queues
static void flushLineToQueues(char *lineBuf, int &linePos) {
  if (linePos <= 0) return;
  lineBuf[linePos] = '\0';

  LogEntry *entry = (LogEntry*)pvPortMalloc(sizeof(LogEntry));
  if (entry) {
    getTimestamp(entry->timestamp, sizeof(entry->timestamp));
    entry->direction = 'R';
    memcpy(entry->data, lineBuf, linePos + 1);
    entry->dataLen = linePos;

    if (xQueueSend(queueSD, &entry, 0) != pdTRUE) {
      LogEntry *old;
      if (xQueueReceive(queueSD, &old, 0) == pdTRUE) vPortFree(old);
      xQueueSend(queueSD, &entry, 0);
    }

    LogEntry *webEntry = (LogEntry*)pvPortMalloc(sizeof(LogEntry));
    if (webEntry) {
      memcpy(webEntry, entry, sizeof(LogEntry));
      if (xQueueSend(queueWeb, &webEntry, 0) != pdTRUE) vPortFree(webEntry);
    }
  }
  linePos = 0;
}

void serialRxTask(void *param) {
  static char lineBuf[LOG_LINE_MAX_LEN];
  int linePos = 0;
  TickType_t lastActivity = xTaskGetTickCount();

  Serial.println("[Task] SerialRX started on core " + String(xPortGetCoreID()));

  while (true) {
    int available = Serial2.available();

    if (available > 0) {
      int space = (int)(LOG_LINE_MAX_LEN - linePos - 1);
      if (space <= 0) { flushLineToQueues(lineBuf, linePos); continue; }

      int toRead = (available < space) ? available : space;
      for (int i = 0; i < toRead; i++) {
        char c = Serial2.read();
        if (c == '\n' || c == '\r') {
          if (linePos > 0) flushLineToQueues(lineBuf, linePos);
          continue;
        }
        // Only accept printable ASCII (0x20~0x7E) and TAB (0x09)
        // Skip null bytes and noise (0x00, 0xFF, etc.)
        if ((c >= 0x20 && c <= 0x7E) || c == '\t') {
          lineBuf[linePos++] = c;
        }
      }
      lastActivity = xTaskGetTickCount();
    } else {
      if (linePos > 0 && (xTaskGetTickCount() - lastActivity) > pdMS_TO_TICKS(100)) {
        flushLineToQueues(lineBuf, linePos);
      }
      vTaskDelay(pdMS_TO_TICKS(1));
    }
  }
}

void serialTxTask(void *param) {
  Serial.println("[Task] SerialTX started on core " + String(xPortGetCoreID()));

  while (true) {
    LogEntry *entry;
    if (xQueueReceive(queueTX, &entry, pdMS_TO_TICKS(50)) == pdTRUE) {
      if (serialMutex) xSemaphoreTake(serialMutex, portMAX_DELAY);
      Serial2.write((uint8_t*)entry->data, entry->dataLen);
      Serial2.flush();
      if (serialMutex) xSemaphoreGive(serialMutex);

      LogEntry *sdEntry = (LogEntry*)pvPortMalloc(sizeof(LogEntry));
      if (sdEntry) {
        memcpy(sdEntry, entry, sizeof(LogEntry));
        sdEntry->direction = 'T';
        if (xQueueSend(queueSD, &sdEntry, pdMS_TO_TICKS(10)) != pdTRUE) vPortFree(sdEntry);
      }

      LogEntry *webEntry = (LogEntry*)pvPortMalloc(sizeof(LogEntry));
      if (webEntry) {
        memcpy(webEntry, entry, sizeof(LogEntry));
        webEntry->direction = 'T';
        if (xQueueSend(queueWeb, &webEntry, 0) != pdTRUE) vPortFree(webEntry);
      }

      vPortFree(entry);
    }
  }
}