lasp/src/lasp/device/lasp_uldaq_impl.cpp

/* #define DEBUGTRACE_ENABLED */
#include "debugtrace.hpp"
#include "lasp_config.h"

#if LASP_HAS_ULDAQ == 1
#include "lasp_daqconfig.h"
#include "lasp_uldaq.h"
#include "lasp_uldaq_impl.h"

using namespace std::literals::chrono_literals;

/**
 * @brief Reserve some space for an error message from UlDaq
 */
const us UL_ERR_MSG_LEN = 512;

/**
 * @brief Return a string corresponding to the UlDaq API error
 *
 * @param err error code
 *
 * @return Error string
 */
string getErrMsg(UlError err) {
  string errstr;
  errstr.reserve(UL_ERR_MSG_LEN);
  char errmsg[UL_ERR_MSG_LEN];
  errstr = "UlDaq API Error: ";
  ulGetErrMsg(err, errmsg);
  errstr += errmsg;
  return errstr;
}
inline void showErr(string errstr) {
  std::cerr << "\b\n**************** UlDAQ backend error **********\n";
  std::cerr << errstr << std::endl;
  std::cerr << "***********************************************\n\n";
}
inline void showErr(UlError err) { showErr(getErrMsg(err)); }

DT9837A::~DT9837A() {
  UlError err;
  if (isRunning()) {
    stop();
  }

  if (_handle) {
    err = ulDisconnectDaqDevice(_handle);
    showErr(err);
    err = ulReleaseDaqDevice(_handle);
    showErr(err);
  }
}
DT9837A::DT9837A(const DeviceInfo &devinfo, const DaqConfiguration &config)
    : Daq(devinfo, config),
      _nFramesPerBlock(availableFramesPerBlock.at(framesPerBlockIndex)) {

  // Some sanity checks
  if (inchannel_config.size() != 4) {
    throw rte("Invalid length of enabled inChannels vector");
  }

  if (outchannel_config.size() != 1) {
    throw rte("Invalid length of enabled outChannels vector");
  }

  if (_nFramesPerBlock < 24 || _nFramesPerBlock > 8192) {
    throw rte("Unsensible number of samples per block chosen");
  }

  if (samplerate() < 10000 || samplerate() > 51000) {
    throw rte("Invalid sample rate");
  }

  std::vector<DeviceInfo> devs;
  DaqDeviceDescriptor devdescriptors[MAX_ULDAQ_DEV_COUNT_PER_API];
  fillUlDaqDeviceInfo(devs, static_cast<void *>(devdescriptors));

  if (devs.size() == 0) {
    throw rte("Unable to find any UlDaq devices");
  }

  if (devinfo.api_specific_devindex < 0 ||
      devinfo.api_specific_devindex >= (int) MAX_ULDAQ_DEV_COUNT_PER_API) {
    throw rte("Invalid device index");
  }

  // get a handle to the DAQ device associated with the first descriptor
  _handle = ulCreateDaqDevice(devdescriptors[devinfo.api_specific_devindex]);

  if (_handle == 0) {
    throw rte("Unable to create a handle to the specified DAQ "
              "device. Is the device currently in use? Please make sure to set "
              "the DAQ configuration in duplex mode if simultaneous input and "
              "output is required.");
  }

  UlError err = ulConnectDaqDevice(_handle);

  if (err != ERR_NO_ERROR) {
    ulReleaseDaqDevice(_handle);
    _handle = 0;
    throw rte("Unable to connect to device: " + getErrMsg(err));
  }

  /// Loop over input channels, set parameters
  for (us ch = 0; ch < 4; ch++) {

    err = ulAISetConfigDbl(_handle, AI_CFG_CHAN_SENSOR_SENSITIVITY, ch, 1.0);
    showErr(err);
    if (err != ERR_NO_ERROR) {
      throw rte("Fatal: could normalize channel sensitivity");
    }

    CouplingMode cm = inchannel_config.at(ch).ACCouplingMode ? CM_AC : CM_DC;
    err = ulAISetConfig(_handle, AI_CFG_CHAN_COUPLING_MODE, ch, cm);
    if (err != ERR_NO_ERROR) {
      showErr(err);
      throw rte("Fatal: could not set AC/DC coupling mode");
    }

    IepeMode iepe =
        inchannel_config.at(ch).IEPEEnabled ? IEPE_ENABLED : IEPE_DISABLED;
    err = ulAISetConfig(_handle, AI_CFG_CHAN_IEPE_MODE, ch, iepe);
    if (err != ERR_NO_ERROR) {
      showErr(err);
      throw rte("Fatal: could not set IEPE mode");
    }
  }
}

bool DT9837A::isRunning() const {
  DEBUGTRACE_ENTER;
  return _thread.joinable();
}
void DT9837A::stop() {
  DEBUGTRACE_ENTER;
  StreamStatus status = _streamStatus;
  if (!isRunning()) {
    throw rte("No data acquisition running");
  }

  _stopThread = true;
  if (_thread.joinable()) {
    _thread.join();
  }
  _stopThread = false;
  status.isRunning = false;
  _streamStatus = status;
}

/**
 * @brief Throws an exception in case it happens. Does nothing in case of no
 * error.
 *
 * @param e
 */
inline void throwUlException(UlError err) {
  if (err == ERR_NO_ERROR) {
    return;
  }
  string errstr = getErrMsg(err);
  showErr(errstr);
  Daq::StreamStatus::StreamError serr;
  if ((int)err < 5) {
    serr = Daq::StreamStatus::StreamError::logicError;
  } else if ((int)err < 9) {
    serr = Daq::StreamStatus::StreamError::systemError;
  } else if ((int)err < 18) {
    serr = Daq::StreamStatus::StreamError::logicError;
  } else if ((int)err == 18) {
    serr = Daq::StreamStatus::StreamError::inputXRun;
  } else if ((int)err == 19) {
    serr = Daq::StreamStatus::StreamError::outputXRun;
  } else {
    serr = Daq::StreamStatus::StreamError::driverError;
  }

  throw Daq::StreamException(serr, errstr);
}

void DT9837A::start(InDaqCallback inCallback, OutDaqCallback outCallback) {
  DEBUGTRACE_ENTER;
  if (isRunning()) {
    throw rte("DAQ is already running");
  }
  if (neninchannels() > 0) {
    if (!inCallback)
      throw rte("DAQ requires a callback for input data");
  }
  if (nenoutchannels() > 0) {
    if (!outCallback)
      throw rte("DAQ requires a callback for output data");
  }
  assert(neninchannels() + nenoutchannels() > 0);
  _thread = std::thread(&DT9837A::threadFcn, this, inCallback, outCallback);
}

InBufHandler::InBufHandler(DT9837A &daq, InDaqCallback cb)
    : BufHandler(daq, daq.neninchannels()), cb(cb)

{
  DEBUGTRACE_ENTER;
  assert(daq.getHandle() != 0);

  monitorOutput = daq.monitorOutput;

  DaqInScanFlag inscanflags = DAQINSCAN_FF_DEFAULT;
  ScanOption scanoptions = SO_CONTINUOUS;
  UlError err = ERR_NO_ERROR;

  std::vector<DaqInChanDescriptor> indescs;
  boolvec eninchannels_without_mon = daq.eninchannels(false);

  // Set ranges for each input. Below asks only channels that are not a
  // monitor channel (hence the false flag).
  dvec ranges = daq.inputRangeForEnabledChannels(false);
  for (us chin = 0; chin < 4; chin++) {
    if (eninchannels_without_mon[chin] == true) {
      DaqInChanDescriptor indesc;
      indesc.type = DAQI_ANALOG_SE;
      indesc.channel = chin;

      double rangeval = ranges.at(chin);
      Range rangenum;
      if (fabs(rangeval - 1.0) < 1e-8) {
        rangenum = BIP1VOLTS;
      } else if (fabs(rangeval - 10.0) < 1e-8) {
        rangenum = BIP10VOLTS;
      } else {
        throw Daq::StreamException(Daq::StreamStatus::StreamError::logicError);
        std::cerr << "Fatal: input range value is invalid" << endl;
        return;
      }
      indesc.range = rangenum;
      indescs.push_back(indesc);
    }
  }

  // Add possibly last channel as monitor
  if (monitorOutput) {
    DaqInChanDescriptor indesc;
    indesc.type = DAQI_DAC;
    indesc.channel = 0;
    /// The output only has a range of 10V, therefore the monitor of the
    /// output also has to be set to this value.
    indesc.range = BIP10VOLTS;
    indescs.push_back(indesc);
  }
  assert(indescs.size() == nchannels);

  DEBUGTRACE_MESSAGE("Starting input scan");

  err = ulDaqInScan(daq.getHandle(), indescs.data(), nchannels,
                    2 * nFramesPerBlock, // Watch the 2 here!
                    &samplerate, scanoptions, inscanflags, buf.data());
  throwUlException(err);
}
void InBufHandler::start() {

  ScanStatus status;
  TransferStatus transferStatus;
  UlError err = ulDaqInScanStatus(daq.getHandle(), &status, &transferStatus);
  throwUlException(err);

  totalFramesCount = transferStatus.currentTotalCount;
  topenqueued = true;
  botenqueued = true;
}

bool InBufHandler::operator()() {

  /* DEBUGTRACE_ENTER; */

  bool ret = true;

  auto runCallback = ([&](us totalOffset) {
    /* DEBUGTRACE_ENTER; */

    DaqData data(nFramesPerBlock, nchannels,
                 DataTypeDescriptor::DataType::dtype_fl64);

    us monitorOffset = monitorOutput ? 1 : 0;
    /* /// Put the output monitor in front */
    if (monitorOutput) {
      for (us frame = 0; frame < nFramesPerBlock; frame++) {
        data.value<double>(frame, 0) =
            buf[totalOffset + (frame * nchannels) + (nchannels - 1)];
      }
    }

    for (us channel = 0; channel < nchannels - monitorOffset; channel++) {
      /* DEBUGTRACE_PRINT(channel); */
      for (us frame = 0; frame < nFramesPerBlock; frame++) {
        data.value<double>(frame, channel + monitorOffset) =
            buf[totalOffset + (frame * nchannels) + channel];
      }
    }
    return cb(data);
  });

  ScanStatus status;
  TransferStatus transferStatus;

  UlError err = ulDaqInScanStatus(daq.getHandle(), &status, &transferStatus);
  throwUlException(err);

  increment = transferStatus.currentTotalCount - totalFramesCount;
  totalFramesCount += increment;

  if (increment > nFramesPerBlock) {
    throw Daq::StreamException(Daq::StreamStatus::StreamError::inputXRun);
  }
  assert(status == SS_RUNNING);

  if (transferStatus.currentIndex < (long long)buffer_mid_idx) {
    topenqueued = false;
    if (!botenqueued) {
      ret = runCallback(nchannels * nFramesPerBlock);
      botenqueued = true;
    }
  } else {
    botenqueued = false;
    if (!topenqueued) {
      ret = runCallback(0);
      topenqueued = true;
    }
  }
  return ret;
}
InBufHandler::~InBufHandler() {
  // At exit of the function, stop scanning.
  DEBUGTRACE_ENTER;
  UlError err = ulDaqInScanStop(daq.getHandle());
  if (err != ERR_NO_ERROR) {
    showErr(err);
  }
}

OutBufHandler::OutBufHandler(DT9837A &daq, OutDaqCallback cb)
    : BufHandler(daq, daq.nenoutchannels()), cb(cb) {

  DEBUGTRACE_MESSAGE("Starting output scan");
  DEBUGTRACE_PRINT(nchannels);
  AOutScanFlag outscanflags = AOUTSCAN_FF_DEFAULT;
  ScanOption scanoptions = SO_CONTINUOUS;
  UlError err = ulAOutScan(daq.getHandle(), 0, 0, BIP10VOLTS,
                           2 * nFramesPerBlock, // Watch the 2 here!
                           &samplerate, scanoptions, outscanflags, buf.data());

  throwUlException(err);
}
void OutBufHandler::start() {

  ScanStatus status;
  TransferStatus transferStatus;

  UlError err = ulAOutScanStatus(daq.getHandle(), &status, &transferStatus);
  if (err != ERR_NO_ERROR) {
    showErr(err);
    throw rte("Unable to start output on DAQ");
  }
  if (status != SS_RUNNING) {
    throw rte("Unable to start output on DAQ");
  }
  totalFramesCount = transferStatus.currentTotalCount;
  topenqueued = true;
  botenqueued = true;
}
bool OutBufHandler::operator()() {

  /* DEBUGTRACE_ENTER; */
  bool res = true;
  assert(daq.getHandle() != 0);

  UlError err = ERR_NO_ERROR;

  ScanStatus status;
  TransferStatus transferStatus;

  err = ulAOutScanStatus(daq.getHandle(), &status, &transferStatus);
  throwUlException(err);
  if (status != SS_RUNNING) {
    return false;
  }
  increment = transferStatus.currentTotalCount - totalFramesCount;
  totalFramesCount += increment;

  if (increment > nFramesPerBlock) {
    throw Daq::StreamException(Daq::StreamStatus::StreamError::outputXRun);
  }

  if (transferStatus.currentIndex < buffer_mid_idx) {
    topenqueued = false;
    if (!botenqueued) {
      DaqData d(nFramesPerBlock, 1, DataTypeDescriptor::DataType::dtype_fl64);
      res = cb(d);
      d.copyToRaw(0, reinterpret_cast<byte_t *>(&(buf[buffer_mid_idx])));

      botenqueued = true;
    }
  } else {
    botenqueued = false;
    if (!topenqueued) {
      DaqData d(nFramesPerBlock, 1, DataTypeDescriptor::DataType::dtype_fl64);
      res = cb(d);
      d.copyToRaw(0, reinterpret_cast<byte_t *>(&(buf[0])));

      topenqueued = true;
    }
  }
  return res;
}

OutBufHandler::~OutBufHandler() {
  DEBUGTRACE_ENTER;
  UlError err = ulAOutScanStop(daq.getHandle());
  if (err != ERR_NO_ERROR) {
    showErr(err);
  }
}

void DT9837A::threadFcn(InDaqCallback inCallback, OutDaqCallback outCallback) {

  DEBUGTRACE_ENTER;

  try {

    std::unique_ptr<OutBufHandler> obh;
    std::unique_ptr<InBufHandler> ibh;

    StreamStatus status = _streamStatus;
    status.isRunning = true;
    _streamStatus = status;

    if (nenoutchannels() > 0) {
      assert(outCallback);
      obh = std::make_unique<OutBufHandler>(*this, outCallback);
    }
    if (neninchannels() > 0) {
      assert(inCallback);
      ibh = std::make_unique<InBufHandler>(*this, inCallback);
    }
    if (obh)
      obh->start();
    if (ibh)
      ibh->start();

    const double sleeptime_s =
        static_cast<double>(_nFramesPerBlock) / (16 * samplerate());
    const us sleeptime_us = static_cast<us>(sleeptime_s * 1e6);

    while (!_stopThread) {
      if (ibh) {
        if (!(*ibh)()) {
          _stopThread = true;
        }
      }
      if (obh) {
        if (!(*obh)()) {
          _stopThread = true;
        }
      } else {
        std::this_thread::sleep_for(std::chrono::microseconds(sleeptime_us));
      }
    }

    /// Update stream status that we are not running anymore
    status.isRunning = false;
    _streamStatus = status;
    _stopThread = false;

  } catch (StreamException &e) {

    StreamStatus status = _streamStatus;
    // Copy over error type
    status.errorType = e.e;
    _streamStatus = status;

    /*
    cerr << "\n******************\n";
    cerr << "Catched error in UlDAQ thread: " << e.what() << endl;
    cerr << "\n******************\n";
    */
  }
}

#endif // LASP_HAS_ULDAQ