RxDriverESP32.h

 
#pragma once
#include <driver/rmt_rx.h>
#include <driver/rmt_tx.h>
#include <freertos/FreeRTOS.h>
#include <freertos/queue.h>
#include <freertos/task.h>
 
#include <stddef.h>
#include <stdint.h>
 
#include "Config.h"
#include "pulse/Vector.h"
#include "pulse/Logger.h"
#include "pulse/RxDriver.h"
#include "pulse/TxDriver.h"
#include "pulse/Codec.h"
#include "pulse/RingBuffer.h"
 
namespace pulsewire {
 
class RxDriverESP32 : public RxDriver {
public:
  RxDriverESP32(ManchesterCodec& codec, uint8_t pin,
                  uint32_t freqHz = DEFAULT_BIT_FREQ_HZ, bool useChecksum = false, uint32_t timeoutUs = 5000)
      : _codec(codec), _rxPin(pin), _freqHz(freqHz), _useChecksum(useChecksum), _timeoutUs(timeoutUs) {
    _lastSymbolTime = 0;
  }
 
  void setFrameSize(uint8_t size) override { _frameSize = size; }
 
  void setRxBufferSize(size_t size) override { _rxBufferSize = size; }
 
  bool begin(uint8_t frameSize) override {
    setFrameSize(frameSize);
    return begin();
  }
 
  bool begin() override {
    if (!_codec.begin()){
      return false;
    }
    manchesterBits.reserve(_codec.getBitCount());
    // Enable DMA for large frames
    uint8_t dma = (_frameSize > 64) ? 1 : 0;
    rmt_rx_channel_config_t rx_config = {.gpio_num = (gpio_num_t)_rxPin,
                                         .clk_src = RMT_CLK_SRC_DEFAULT,
                                         .resolution_hz = 1000000,
                                         .mem_block_symbols = 64,
                                         .flags = {
                                             .invert_in = 0,
                                             .with_dma = dma,
                                         }};
 
    // Setup channel
    rmt_channel_handle_t rx_channel = nullptr;
    esp_err_t rmt_result = rmt_new_rx_channel(&rx_config, &rx_channel);
    if (rmt_result != ESP_OK || rx_channel == nullptr) {
      Logger::error("RMT RX channel initialization failed: %d", rmt_result);
      return false;
    }
    _rxChannel = rx_channel;
 
    // Setup FreeRTOS queue for received frames
    int count = _rxBufferSize / _frameSize;
    if (_rxBufferSize % _frameSize != 0) count++;
 
    // Create Queue
    _frameQueue = xQueueCreate(count, _frameSize);  // n frames max in queue
    if (_frameQueue == nullptr) {
      Logger::error("Failed to create frame queue");
      return false;
    }
    rxOverflow.resize(_rxBufferSize);
    _stopTask = false;
 
    // Create task
    BaseType_t taskResult = xTaskCreatePinnedToCore(rxTask, "rmt_rx", 4096,
                                                    this, 1, &_taskHandle, 1);
    if (taskResult != pdPASS || _taskHandle == nullptr) {
      Logger::error("Failed to create RX task: %d", taskResult);
      return false;
    }
    return true;
  }
 
  void end() override {
    _stopTask = true;
    if (_taskHandle) {
      // Wait for task to exit
      while (eTaskGetState(_taskHandle) != eDeleted) {
        delay(1);
      }
      _taskHandle = nullptr;
    }
    if (_rxChannel) {
      rmt_del_channel(_rxChannel);
      _rxChannel = nullptr;
    }
  }
 
  size_t readBytes(uint8_t* buffer, size_t length) override {
    size_t totalRead = 0;
    // First, serve from rxOverflow if any
    while (length > 0 && rxOverflow.available() > 0) {
      int b = rxOverflow.read();
      if (b < 0) break;
      *buffer++ = (uint8_t)b;
      --length;
      ++totalRead;
    }
    // Then, read full frames from the queue as needed
    uint8_t frame[_frameSize]{};
    while (length > 0 && _frameQueue &&
           uxQueueMessagesWaiting(_frameQueue) > 0) {
      if (xQueueReceive(_frameQueue, frame, 0) == pdTRUE) {
        size_t toCopy = (length < _frameSize) ? length : _frameSize;
        memcpy(buffer, frame, toCopy);
        totalRead += toCopy;
        buffer += toCopy;
        length -= toCopy;
        // If we didn't consume the whole frame, store the rest in rxOverflow
        if (toCopy < _frameSize) {
          rxOverflow.writeArray(frame + toCopy, _frameSize - toCopy);
        }
      }
    }
    return totalRead;
  }
 
  int available() override {
    if (!_frameQueue) return 0;
    return uxQueueMessagesWaiting(_frameQueue) * _frameSize;
  }
 
 protected:
  size_t _rxBufferSize = 256;
  uint8_t _rxPin;
  uint16_t _frameSize = DEFAULT_FRAME_SIZE;
  uint32_t _freqHz;
  rmt_channel_handle_t _rxChannel = nullptr;
  TaskHandle_t _taskHandle = nullptr;
  volatile bool _stopTask = false;
  uint16_t _preambleBits = 0;
  bool _inFrame = false;
  QueueHandle_t _frameQueue = nullptr;
  RingBuffer<uint8_t> rxOverflow;
  ManchesterCodec& _codec;
  Vector<uint8_t> manchesterBits;
  bool _useChecksum = false;
  uint32_t _timeoutUs = 5000;
  uint32_t _lastSymbolTime = 0;
 
  static void rxTask(void* arg) {
    auto* self = static_cast<RxDriverESP32*>(arg);
    size_t symbolCount = self->_frameSize * self->_codec.getBitCount();
    Vector<rmt_symbol_word_t> symbols(symbolCount);
    uint8_t data[self->_frameSize]{};
    self->_lastSymbolTime = micros();
    while (!self->_stopTask) {
      size_t rx_size = 0;
      rmt_receive_config_t rx_cfg = {};
      esp_err_t err =
          rmt_receive(self->_rxChannel, symbols.data(),
                      symbols.size() * sizeof(rmt_symbol_word_t), &rx_cfg);
      bool gotSymbol = false;
      if (err == ESP_OK && rx_size > 0) {
        size_t num_bits = rx_size / sizeof(rmt_symbol_word_t);
        // Use decodeEdge for each symbol
        uint32_t bitPeriodUs = 1000000UL / self->_freqHz;
        uint32_t minUs = bitPeriodUs / 2;
        uint32_t maxUs = bitPeriodUs * 2;
        size_t frameLen = 0;
        for (size_t i = 0; i < num_bits; ++i) {
          uint32_t duration = symbols[i].duration0 + symbols[i].duration1;
          bool isOne = (symbols[i].level0 == 1 && symbols[i].level1 == 0);
          bool isZero = (symbols[i].level0 == 0 && symbols[i].level1 == 1);
          if (isOne || isZero) {
            bool level = isOne ? 1 : 0;
            self->_lastSymbolTime = micros();
            gotSymbol = true;
            if (self->_codec.decodeEdge(duration, level, minUs, maxUs, data,
                                        self->_frameSize, frameLen)) {
              if (self->_useChecksum) {
                // Validate checksum
                if (frameLen < 2) continue;
                uint8_t sum = 0;
                for (size_t j = 0; j < frameLen - 1; ++j) sum += data[j];
                if (sum == data[frameLen - 1]) {
                  xQueueSend(self->_frameQueue, data, 0);
                }
              } else {
                xQueueSend(self->_frameQueue, data, 0);
              }
            }
          }
        }
      }
      // Timeout logic: if no symbol received for timeoutUs, flush buffer
      uint32_t now = micros();
      if (!gotSymbol && (now - self->_lastSymbolTime > self->_timeoutUs)) {
        // Try to decode with a zero-duration edge to flush the buffer
        size_t frameLen = 0;
        if (self->_codec.decodeEdge(0, 0, 0, 0, data, self->_frameSize, frameLen)) {
          if (self->_useChecksum) {
            if (frameLen >= 2) {
              uint8_t sum = 0;
              for (size_t j = 0; j < frameLen - 1; ++j) sum += data[j];
              if (sum == data[frameLen - 1]) {
                xQueueSend(self->_frameQueue, data, 0);
              }
            }
          } else {
            xQueueSend(self->_frameQueue, data, 0);
          }
        }
        self->_lastSymbolTime = now;
      }
      delay(1);
    }
    vTaskDelete(nullptr);
  }
 
  // processSymbols is no longer needed; decoding is handled by decodeEdge
};
 
}  // namespace pulsewire