Moved whole threading code with UlDaq to C++
This commit is contained in:
parent
248fa8d8a4
commit
108e023026
|
|
@ -1,4 +1,7 @@
|
|||
include_directories(/usr/include/rtaudio)
|
||||
|
||||
add_library(cpp_daq lasp_cppuldaq.cpp)
|
||||
|
||||
set_source_files_properties(lasp_rtaudio.pyx PROPERTIES CYTHON_IS_CXX TRUE)
|
||||
set_source_files_properties(lasp_uldaq.pyx PROPERTIES CYTHON_IS_CXX TRUE)
|
||||
set_source_files_properties(lasp_rtaudio.cxx PROPERTIES COMPILE_FLAGS
|
||||
|
|
@ -8,8 +11,11 @@ cython_add_module(lasp_rtaudio lasp_rtaudio.pyx)
|
|||
cython_add_module(lasp_uldaq lasp_uldaq.pyx)
|
||||
|
||||
target_link_libraries(lasp_rtaudio pthread rtaudio)
|
||||
target_link_libraries(lasp_uldaq uldaq)
|
||||
target_link_libraries(lasp_uldaq cpp_daq uldaq pthread)
|
||||
if(win32)
|
||||
target_link_libraries(lasp_rtaudio python37)
|
||||
target_link_libraries(lasp_uldaq python37)
|
||||
endif(win32)
|
||||
|
||||
add_executable(test_uldaq test_uldaq.cpp)
|
||||
target_link_libraries(test_uldaq cpp_daq uldaq pthread)
|
||||
|
|
|
|||
18
lasp/device/lasp_cppdaq.h
Normal file
18
lasp/device/lasp_cppdaq.h
Normal file
|
|
@ -0,0 +1,18 @@
|
|||
#ifndef LAP_CPPDAQ_H
|
||||
#define LAP_CPPDAQ_H
|
||||
#include "lasp_cppqueue.h"
|
||||
|
||||
class Daq{
|
||||
|
||||
public:
|
||||
virtual void start(
|
||||
SafeQueue<double*> *inqueue,
|
||||
SafeQueue<double*> *outqueue) = 0;
|
||||
virtual void stop() = 0;
|
||||
virtual double samplerate() const = 0;
|
||||
|
||||
virtual ~Daq() {};
|
||||
|
||||
};
|
||||
|
||||
#endif // LAP_CPPDAQ_H
|
||||
0
lasp/device/lasp_cpprtaudiodaq.h
Normal file
0
lasp/device/lasp_cpprtaudiodaq.h
Normal file
547
lasp/device/lasp_cppuldaq.cpp
Normal file
547
lasp/device/lasp_cppuldaq.cpp
Normal file
|
|
@ -0,0 +1,547 @@
|
|||
#include "lasp_cppuldaq.h"
|
||||
#include <stdexcept>
|
||||
#include <uldaq.h>
|
||||
#include <cassert>
|
||||
#include <iostream>
|
||||
#include <cstring>
|
||||
#include <chrono>
|
||||
|
||||
using std::endl;
|
||||
using std::cerr;
|
||||
using std::runtime_error;
|
||||
typedef std::lock_guard<std::mutex> mutexlock;
|
||||
/* using std::this_thread; */
|
||||
|
||||
const us MAX_DEF_COUNT = 100;
|
||||
const us UL_ERR_MSG_LEN = 512;
|
||||
|
||||
inline void showErr(UlError err) {
|
||||
if(err != ERR_NO_ERROR) {
|
||||
char errmsg[UL_ERR_MSG_LEN];
|
||||
ulGetErrMsg(err, errmsg);
|
||||
std::cerr << "UlError: " << errmsg << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
DT9837A::DT9837A(
|
||||
us samplesPerBlock,
|
||||
boolvec& inChannels,
|
||||
boolvec& outChannels,
|
||||
double samplerate,
|
||||
bool monitorOutput,
|
||||
us deviceno
|
||||
):
|
||||
samplesPerBlock(samplesPerBlock),
|
||||
_samplerate(samplerate),
|
||||
inChannels(inChannels),
|
||||
outChannels(outChannels),
|
||||
monitorOutput(monitorOutput)
|
||||
{
|
||||
if(monitorOutput && !(nenoutchannels() > 0)) {
|
||||
throw runtime_error("Output monitoring only possible when output is enabled");
|
||||
}
|
||||
|
||||
stopThread = false;
|
||||
high_range = boolvec({false, false, false, false});
|
||||
|
||||
if(samplesPerBlock < 24 || samplesPerBlock > 8192) {
|
||||
throw runtime_error("Unsensible number of samples per block chosen");
|
||||
}
|
||||
|
||||
|
||||
// Some sanity checks
|
||||
if(inChannels.size() != 4) {
|
||||
throw runtime_error("Invalid length of enabled inChannels vector");
|
||||
}
|
||||
|
||||
// Some sanity checks
|
||||
if(outChannels.size() != 1) {
|
||||
throw runtime_error("Invalid length of enabled outChannels vector");
|
||||
}
|
||||
if(samplerate < 10000 || samplerate > 51000) {
|
||||
throw runtime_error("Invalid sample rate");
|
||||
}
|
||||
|
||||
DaqDeviceDescriptor devdescriptors[MAX_DEF_COUNT];
|
||||
DaqDeviceDescriptor descriptor;
|
||||
DaqDeviceInterface interfaceType = ANY_IFC;
|
||||
|
||||
UlError err;
|
||||
|
||||
us numdevs = MAX_DEF_COUNT;
|
||||
err = ulGetDaqDeviceInventory(interfaceType,
|
||||
devdescriptors,
|
||||
&numdevs);
|
||||
if(err != ERR_NO_ERROR){
|
||||
throw runtime_error("Device inventarization failed");
|
||||
}
|
||||
|
||||
if (deviceno >= numdevs) {
|
||||
throw runtime_error("Device number {deviceno} too high {err}. This could happen when the device is currently not connected");
|
||||
}
|
||||
|
||||
descriptor = devdescriptors[deviceno];
|
||||
// get a handle to the DAQ device associated with the first descriptor
|
||||
handle = ulCreateDaqDevice(descriptor);
|
||||
|
||||
if (handle == 0) {
|
||||
throw runtime_error("Unable to create a handle to the specified DAQ device. Is the device currently in use?");
|
||||
}
|
||||
|
||||
err = ulConnectDaqDevice(handle);
|
||||
if (err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
ulReleaseDaqDevice(handle);
|
||||
handle = 0;
|
||||
throw runtime_error("Unable to connect to device: {err}");
|
||||
}
|
||||
|
||||
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 runtime_error("Fatal: could normalize channel sensitivity");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
DT9837A::~DT9837A() {
|
||||
UlError err;
|
||||
if(isRunning()) {
|
||||
stop();
|
||||
}
|
||||
|
||||
if(handle) {
|
||||
err = ulDisconnectDaqDevice(handle);
|
||||
showErr(err);
|
||||
err = ulReleaseDaqDevice(handle);
|
||||
showErr(err);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
void DT9837A::setIEPEEnabled(boolvec& config) {
|
||||
if(isRunning()) {
|
||||
throw runtime_error("Cannot change config while sampling is running");
|
||||
}
|
||||
|
||||
// Some sanity checks
|
||||
if(config.size() != 4) {
|
||||
throw runtime_error("Invalid length of enabled inChannels vector");
|
||||
}
|
||||
|
||||
IepeMode iepe;
|
||||
UlError err;
|
||||
|
||||
for(us i=0; i< config.size(); i++) {
|
||||
|
||||
iepe = config[i] ? IEPE_ENABLED : IEPE_DISABLED;
|
||||
err = ulAISetConfig(handle, AI_CFG_CHAN_IEPE_MODE, i, iepe);
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
throw runtime_error("Fatal: could not set IEPE mode");
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
void DT9837A::setACCouplingMode(boolvec& coupling) {
|
||||
if(isRunning()) {
|
||||
throw runtime_error("Cannot change config while sampling is running");
|
||||
}
|
||||
|
||||
// Some sanity checks
|
||||
if(coupling.size() != 4) {
|
||||
throw runtime_error("Invalid length of enabled inChannels vector");
|
||||
}
|
||||
|
||||
CouplingMode cm;
|
||||
UlError err;
|
||||
|
||||
for(us i=0; i< coupling.size(); i++) {
|
||||
cm = coupling[i] ? CM_AC : CM_DC;
|
||||
|
||||
err = ulAISetConfig(handle, AI_CFG_CHAN_COUPLING_MODE, i, cm);
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
throw runtime_error("Fatal: could not set IEPE mode");
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
void DT9837A::setInputRange(boolvec& high_range) {
|
||||
if(isRunning()) {
|
||||
throw runtime_error("Cannot change config while sampling is running");
|
||||
}
|
||||
|
||||
// Some sanity checks
|
||||
if(high_range.size() != 4) {
|
||||
throw runtime_error("Invalid length of enabled high range vector");
|
||||
}
|
||||
|
||||
this->high_range = high_range;
|
||||
}
|
||||
|
||||
/* static inline void copysamples_transpose(const us samplesPerBlock,const us nchannels, */
|
||||
/* double* from, double* to, */
|
||||
/* const us from_startidx, */
|
||||
/* const us to_startidx, */
|
||||
/* const bool touldaqbuffer) { */
|
||||
|
||||
/* // Copy from "our type of buffer" to an uldaq buffer */
|
||||
/* if(touldaqbuffer) { */
|
||||
/* for(us sample=0;sample< samplesPerBlock;sample++) { */
|
||||
/* for(us channel=0;channel<nchannels;channel++) { */
|
||||
/* to[to_startidx+sample*nchannels + channel] = from[from_startidx + channel*samplesPerBlock+ sample]; */
|
||||
/* } */
|
||||
/* } */
|
||||
/* } */
|
||||
/* // Copy from an uldaq buffer to our type of buffer */
|
||||
/* else { */
|
||||
/* for(us sample=0;sample< samplesPerBlock;sample++) { */
|
||||
/* for(us channel=0;channel<nchannels;channel++) { */
|
||||
/* to[to_startidx + channel*samplesPerBlock+ sample] = from[from_startidx+sample*nchannels + channel]; */
|
||||
/* } */
|
||||
/* } */
|
||||
/* } */
|
||||
/* } */
|
||||
|
||||
|
||||
void threadfcn(DT9837A* td) {
|
||||
|
||||
std::cerr << "Starting DAQ Thread fcn" << endl;
|
||||
|
||||
const us nenoutchannels = td->nenoutchannels();
|
||||
us neninchannels = td->neninchannels();
|
||||
const us samplesPerBlock = td->samplesPerBlock;
|
||||
|
||||
const bool hasinput = neninchannels > 0;
|
||||
const bool hasoutput = nenoutchannels > 0;
|
||||
|
||||
bool monitorOutput = td->monitorOutput;
|
||||
|
||||
double* inbuffer = td->inbuffer;
|
||||
double* outbuffer = td->outbuffer;
|
||||
|
||||
SafeQueue<double*>* inqueue = td->inqueue;
|
||||
SafeQueue<double*>* outqueue = td->outqueue;
|
||||
|
||||
ScanStatus inscanstat;
|
||||
ScanStatus outscanstat;
|
||||
TransferStatus inxstat, outxstat;
|
||||
|
||||
double samplerate = td->samplerate();
|
||||
|
||||
const us buffer_mid_idx_in = neninchannels*samplesPerBlock;
|
||||
const us buffer_mid_idx_out = nenoutchannels*samplesPerBlock;
|
||||
|
||||
DaqDeviceHandle handle = td->handle;
|
||||
assert(handle);
|
||||
|
||||
DaqInScanFlag inscanflags = DAQINSCAN_FF_DEFAULT;
|
||||
AOutScanFlag outscanflags = AOUTSCAN_FF_DEFAULT;
|
||||
ScanOption scanoptions = SO_CONTINUOUS;
|
||||
UlError err = ERR_NO_ERROR;
|
||||
|
||||
DaqInChanDescriptor *indesc = NULL;
|
||||
|
||||
bool topinenqueued = false;
|
||||
// Start with true here, to not copy here the first time
|
||||
bool botinenqueued = true;
|
||||
|
||||
bool topoutenqueued = false;
|
||||
bool botoutenqueued = false;
|
||||
std::cerr << "SFSG" << endl;
|
||||
|
||||
// outitialize output, if any
|
||||
if(hasoutput) {
|
||||
assert(nenoutchannels == 1);
|
||||
assert(outqueue);
|
||||
|
||||
double *firstout;
|
||||
if(outqueue->empty()) {
|
||||
cerr << "******* WARNING: OUTPUTQUEUE UNDERFLOW, FILLING SIGNAL QUEUE WITH ZEROS ***********" << endl;
|
||||
firstout = static_cast<double*>(malloc(sizeof(double)*samplesPerBlock*nenoutchannels));
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
firstout[sample] = 0;
|
||||
}
|
||||
}
|
||||
else {
|
||||
firstout = outqueue->dequeue();
|
||||
}
|
||||
for(us sample=0;sample<2*samplesPerBlock;sample++) {
|
||||
outbuffer[sample] = firstout[sample];
|
||||
}
|
||||
memcpy(td->outbuffer, firstout, samplesPerBlock*nenoutchannels);
|
||||
/* free(firstout); */
|
||||
topoutenqueued = true;
|
||||
|
||||
err = ulAOutScan(handle,
|
||||
0,
|
||||
0,
|
||||
BIP10VOLTS,
|
||||
/* BIP60VOLTS, */
|
||||
2*td->samplesPerBlock, // Watch the 2 here!
|
||||
&samplerate,
|
||||
scanoptions,
|
||||
outscanflags,
|
||||
outbuffer);
|
||||
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
goto exit;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// Initialize input, if any
|
||||
if(hasinput) {
|
||||
|
||||
indesc = new DaqInChanDescriptor[neninchannels];
|
||||
us j = 0;
|
||||
for(us chin=0; chin < 4; chin++) {
|
||||
if(td->inChannels[chin] == true) {
|
||||
indesc[j].type = DAQI_ANALOG_SE;
|
||||
indesc[j].channel = chin;
|
||||
indesc[j].range = td->high_range[chin] ? BIP10VOLTS : BIP1VOLTS;
|
||||
j++;
|
||||
}
|
||||
|
||||
}
|
||||
// Overwrite last channel
|
||||
if(monitorOutput) {
|
||||
indesc[j].type = DAQI_DAC;
|
||||
indesc[j].channel = 0;
|
||||
indesc[j].range = BIP10VOLTS;
|
||||
j++;
|
||||
}
|
||||
assert(j==neninchannels);
|
||||
|
||||
err = ulDaqInScan(handle,
|
||||
indesc,
|
||||
neninchannels,
|
||||
2*td->samplesPerBlock, // Watch the 2 here!
|
||||
&samplerate,
|
||||
scanoptions,
|
||||
inscanflags,
|
||||
inbuffer);
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
goto exit;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
std::cerr << "Entering while loop" << endl;
|
||||
std::cerr << "hasinput: " << hasinput << endl;
|
||||
while(!td->stopThread && err == ERR_NO_ERROR) {
|
||||
/* std::cerr << "While..." << endl; */
|
||||
if(hasoutput) {
|
||||
err = ulAOutScanStatus(handle, &outscanstat, &outxstat);
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
goto exit;
|
||||
}
|
||||
assert(outscanstat == SS_RUNNING);
|
||||
/* std::cerr << "Samples scanned: " << outxstat.currentTotalCount << endl; */
|
||||
if(outxstat.currentIndex < buffer_mid_idx_out) {
|
||||
topoutenqueued = false;
|
||||
if(!botoutenqueued) {
|
||||
cerr << "Copying output buffer to bottom" << endl;
|
||||
double* bufcpy;
|
||||
if(!outqueue->empty()) {
|
||||
bufcpy = outqueue->dequeue();
|
||||
}
|
||||
else {
|
||||
cerr << "******* WARNING: OUTPUTQUEUE UNDERFLOW, FILLING SIGNAL QUEUE WITH ZEROS ***********" << endl;
|
||||
bufcpy = static_cast<double*>(malloc(sizeof(double)*samplesPerBlock*nenoutchannels));
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
bufcpy[sample] = 0;
|
||||
}
|
||||
}
|
||||
assert(nenoutchannels > 0);
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
outbuffer[buffer_mid_idx_out + sample] = bufcpy[sample];
|
||||
}
|
||||
free(bufcpy);
|
||||
botoutenqueued = true;
|
||||
}
|
||||
}
|
||||
else {
|
||||
botoutenqueued = false;
|
||||
if(!topoutenqueued) {
|
||||
topoutenqueued = true;
|
||||
cerr << "Copying output buffer to top" << endl;
|
||||
double* bufcpy;
|
||||
if(!outqueue->empty()) {
|
||||
bufcpy = outqueue->dequeue();
|
||||
}
|
||||
else {
|
||||
cerr << "******* WARNING: OUTPUTQUEUE UNDERFLOW, FILLING SIGNAL QUEUE WITH ZEROS ***********" << endl;
|
||||
bufcpy = static_cast<double*>(malloc(sizeof(double)*samplesPerBlock*nenoutchannels));
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
bufcpy[sample] = 0;
|
||||
}
|
||||
}
|
||||
assert(nenoutchannels > 0);
|
||||
for(us sample=0;sample<samplesPerBlock;sample++) {
|
||||
outbuffer[sample] = bufcpy[sample];
|
||||
}
|
||||
free(bufcpy);
|
||||
topoutenqueued = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if(hasinput) {
|
||||
err = ulDaqInScanStatus(handle, &inscanstat, &inxstat);
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
goto exit;
|
||||
}
|
||||
assert(inscanstat == SS_RUNNING);
|
||||
|
||||
if(inxstat.currentIndex < buffer_mid_idx_in) {
|
||||
topinenqueued = false;
|
||||
if(!botinenqueued) {
|
||||
cerr << "Copying in buffer bot" << endl;
|
||||
double* bufcpy = static_cast<double*>(malloc(sizeof(double)*samplesPerBlock*neninchannels));
|
||||
us monitoroffset = monitorOutput ? 1: 0;
|
||||
assert(neninchannels > 0);
|
||||
for(us channel=0;channel<(neninchannels-monitoroffset);channel++) {
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
bufcpy[(monitoroffset+channel)*samplesPerBlock+sample] = inbuffer[buffer_mid_idx_in + sample*neninchannels+channel];
|
||||
}
|
||||
}
|
||||
if(monitorOutput) {
|
||||
// Monitor output goes to first channel, that is
|
||||
// our convention
|
||||
us channel = neninchannels - 1;
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
bufcpy[sample] = inbuffer[buffer_mid_idx_in + sample*neninchannels+channel];
|
||||
}
|
||||
}
|
||||
inqueue->enqueue(bufcpy);
|
||||
botinenqueued = true;
|
||||
}
|
||||
}
|
||||
else {
|
||||
botinenqueued = false;
|
||||
if(!topinenqueued) {
|
||||
double* bufcpy = static_cast<double*>(malloc(sizeof(double)*samplesPerBlock*neninchannels));
|
||||
us monitoroffset = monitorOutput ? 1: 0;
|
||||
assert(neninchannels > 0);
|
||||
for(us channel=0;channel<(neninchannels-monitoroffset);channel++) {
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
bufcpy[(monitoroffset+channel)*samplesPerBlock+sample] = inbuffer[sample*neninchannels+channel];
|
||||
}
|
||||
}
|
||||
if(monitorOutput) {
|
||||
// Monitor output goes to first channel, that is
|
||||
// our convention
|
||||
us channel = neninchannels - 1;
|
||||
for(us sample=0;sample< samplesPerBlock;sample++) {
|
||||
bufcpy[sample] = inbuffer[sample*neninchannels+channel];
|
||||
}
|
||||
}
|
||||
|
||||
cerr << "Copying in buffer top" << endl;
|
||||
inqueue->enqueue(bufcpy);
|
||||
topinenqueued = true;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
|
||||
} // End of while loop
|
||||
std::cerr << "Exit of while loop" << endl;
|
||||
|
||||
|
||||
exit:
|
||||
|
||||
if(hasoutput) {
|
||||
ulAOutScanStop(handle);
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
}
|
||||
}
|
||||
|
||||
if(hasinput) {
|
||||
ulDaqInScanStop(handle);
|
||||
if(err != ERR_NO_ERROR) {
|
||||
showErr(err);
|
||||
}
|
||||
}
|
||||
|
||||
if(indesc) delete indesc;
|
||||
std::cerr << "Exit of thread fcn" << endl;
|
||||
|
||||
}
|
||||
|
||||
void DT9837A::start( SafeQueue<double*> *inqueue, SafeQueue<double*> *outqueue) {
|
||||
if(isRunning()) {
|
||||
throw runtime_error("Thread is already running");
|
||||
}
|
||||
|
||||
bool hasinput = neninchannels() > 0;
|
||||
bool hasoutput = nenoutchannels() > 0;
|
||||
|
||||
if(neninchannels() > 0 && !inqueue) {
|
||||
throw runtime_error("Inqueue not given, while input is enabled");
|
||||
}
|
||||
|
||||
if(nenoutchannels() > 0 && !outqueue) {
|
||||
throw runtime_error("outqueue not given, while output is enabled");
|
||||
}
|
||||
|
||||
if(hasinput) {
|
||||
assert(!inbuffer);
|
||||
inbuffer = new double[neninchannels()*samplesPerBlock*2];
|
||||
}
|
||||
if(hasoutput) {
|
||||
assert(!outbuffer);
|
||||
outbuffer = new double[nenoutchannels()*samplesPerBlock*2];
|
||||
}
|
||||
this->inqueue = inqueue;
|
||||
this->outqueue = outqueue;
|
||||
|
||||
stopThread = false;
|
||||
thread = new std::thread(threadfcn, this);
|
||||
|
||||
}
|
||||
|
||||
void DT9837A::stop() {
|
||||
if(!isRunning()) {
|
||||
throw runtime_error("No data acquisition running");
|
||||
}
|
||||
assert(thread);
|
||||
|
||||
stopThread = true;
|
||||
thread->join();
|
||||
delete thread;
|
||||
thread = NULL;
|
||||
|
||||
outqueue = NULL;
|
||||
inqueue = NULL;
|
||||
if(inbuffer) delete inbuffer;
|
||||
if(outbuffer) delete outbuffer;
|
||||
outbuffer = NULL;
|
||||
inbuffer = NULL;
|
||||
}
|
||||
|
||||
us DT9837A::neninchannels() const {
|
||||
mutexlock lock(mutex);
|
||||
us inch = std::count(inChannels.begin(), inChannels.end(), true);
|
||||
if(monitorOutput) inch++;
|
||||
return inch;
|
||||
}
|
||||
|
||||
us DT9837A::nenoutchannels() const {
|
||||
mutexlock lock(mutex);
|
||||
return std::count(outChannels.begin(), outChannels.end(), true);
|
||||
}
|
||||
73
lasp/device/lasp_cppuldaq.h
Normal file
73
lasp/device/lasp_cppuldaq.h
Normal file
|
|
@ -0,0 +1,73 @@
|
|||
#ifndef ULDAQ_THREAD_H
|
||||
#define ULDAQ_THREAD_H
|
||||
#include "lasp_cppqueue.h"
|
||||
#include "lasp_cppdaq.h"
|
||||
#include <uldaq.h>
|
||||
#include <algorithm>
|
||||
#include <vector>
|
||||
#include <mutex>
|
||||
#include <atomic>
|
||||
#include <thread>
|
||||
|
||||
using std::vector;
|
||||
using std::atomic;
|
||||
typedef vector<bool> boolvec;
|
||||
typedef unsigned int us;
|
||||
|
||||
class DT9837A: public Daq {
|
||||
us samplesPerBlock;
|
||||
double _samplerate;
|
||||
boolvec inChannels;
|
||||
boolvec high_range;
|
||||
boolvec outChannels;
|
||||
bool monitorOutput;
|
||||
atomic<bool> stopThread;
|
||||
DaqDeviceHandle handle = 0;
|
||||
|
||||
std::thread* thread = NULL;
|
||||
mutable std::mutex mutex;
|
||||
SafeQueue<double*> *inqueue = NULL;
|
||||
SafeQueue<double*> *outqueue = NULL;
|
||||
|
||||
double* inbuffer = NULL;
|
||||
double* outbuffer = NULL;
|
||||
|
||||
public:
|
||||
double samplerate() const {return this->_samplerate;}
|
||||
DT9837A(
|
||||
us samplesPerBlock,
|
||||
boolvec& inChannels,
|
||||
boolvec& outChannels,
|
||||
double samplerate,
|
||||
bool monitorOutput,
|
||||
us device_no = 0
|
||||
);
|
||||
DT9837A(const DT9837A&) = delete;
|
||||
|
||||
~DT9837A();
|
||||
|
||||
void setIEPEEnabled(boolvec& config);
|
||||
|
||||
// Coupling_ac: true, means AC coupling, false means DC coupling
|
||||
void setACCouplingMode(boolvec& coupling_ac);
|
||||
|
||||
void setInputRange(boolvec& high_range);
|
||||
|
||||
us neninchannels() const;
|
||||
us nenoutchannels() const;
|
||||
|
||||
bool isRunning() const { return bool(thread); }
|
||||
|
||||
virtual void start(
|
||||
SafeQueue<double*> *inqueue,
|
||||
SafeQueue<double*> *outqueue) final;
|
||||
|
||||
virtual void stop() final;
|
||||
|
||||
|
||||
friend void threadfcn(DT9837A*);
|
||||
};
|
||||
|
||||
#endif // ULDAQ_THREAD_H
|
||||
|
||||
|
||||
|
|
@ -1,6 +1,5 @@
|
|||
include "lasp_common_decls.pxd"
|
||||
|
||||
|
||||
cdef extern from "uldaq.h" nogil:
|
||||
|
||||
ctypedef enum DaqDeviceInterface:
|
||||
|
|
|
|||
|
|
@ -4,634 +4,37 @@ from .lasp_daqconfig import (DeviceInfo, InputMode, Range as pyRange,
|
|||
DAQChannel)
|
||||
from .lasp_avtype import AvType
|
||||
|
||||
__all__ = ['UlDT9837A', 'UlDaq']
|
||||
__all__ = ['UlDaq']
|
||||
|
||||
DEF MAX_DEF_COUNT = 100
|
||||
DEF UL_ERR_MSG_LEN = 512
|
||||
|
||||
cdef struct DaqThreadData:
|
||||
ctypedef unsigned us
|
||||
|
||||
unsigned int samplesPerBlock
|
||||
double samplerate
|
||||
cdef extern from "lasp_cppdaq.h":
|
||||
cdef cppclass Daq:
|
||||
void start(SafeQueue[double*] *inQueue,
|
||||
SafeQueue[double*] *outQueue)
|
||||
void stop()
|
||||
double samplerate()
|
||||
|
||||
DaqDeviceHandle handle
|
||||
|
||||
SafeQueue[void*] *inQueue
|
||||
SafeQueue[void*] *outQueue
|
||||
|
||||
DaqInChanDescriptor* inChanDescriptors
|
||||
unsigned ninChanDescriptors
|
||||
unsigned noutChanDescriptors
|
||||
|
||||
atomic[bool] stopThread
|
||||
atomic[bool] outThread_ready
|
||||
mutex* tdmutex
|
||||
|
||||
CPPThread[void*, void (*)(void*)] *inThread
|
||||
CPPThread[void*, void (*)(void*)] *outThread
|
||||
|
||||
double* inbuffer
|
||||
double* outbuffer
|
||||
|
||||
cdef void showErr(UlError err) nogil:
|
||||
cdef:
|
||||
char errmsg[UL_ERR_MSG_LEN]
|
||||
ulGetErrMsg(err, errmsg)
|
||||
fprintf(stderr, 'UlError: %s\n', errmsg)
|
||||
|
||||
cdef void inThreadFunction(void* threaddata_void) nogil:
|
||||
"""
|
||||
Stream input thread function
|
||||
"""
|
||||
cdef:
|
||||
DaqThreadData* td = <DaqThreadData*> threaddata_void
|
||||
|
||||
UlError err
|
||||
|
||||
ScanOption scanoptions
|
||||
DaqInScanFlag inscanflags
|
||||
TransferStatus xstat
|
||||
ScanStatus scanstat
|
||||
double samplerate
|
||||
bint top_enqueued, bottom_enqueued
|
||||
unsigned buffer_mid_idx, chan, sample
|
||||
|
||||
double sleep_time
|
||||
double *inbuffer_cpy
|
||||
unsigned sleep_time_mus # Thread sleep time in ms
|
||||
|
||||
samplerate = td.samplerate
|
||||
sleep_time = td.samplesPerBlock / td.samplerate / 2
|
||||
sleep_time_mus = <unsigned> (sleep_time * 1e6)
|
||||
|
||||
inscanflags = DAQINSCAN_FF_DEFAULT
|
||||
scanoptions = <ScanOption> SO_CONTINUOUS
|
||||
|
||||
buffer_mid_idx = td.samplesPerBlock*td.ninChanDescriptors
|
||||
top_enqueued = True
|
||||
bottom_enqueued = True
|
||||
|
||||
fprintf(stderr, 'Starting input thread\n')
|
||||
while not td.outThread_ready.load():
|
||||
CPPsleep_ms(50)
|
||||
|
||||
td.tdmutex.lock()
|
||||
err = ulDaqInScan(td.handle,
|
||||
td.inChanDescriptors,
|
||||
td.ninChanDescriptors,
|
||||
2*td.samplesPerBlock, # Watch the 2 here!
|
||||
&samplerate,
|
||||
scanoptions,
|
||||
inscanflags,
|
||||
td.inbuffer)
|
||||
td.tdmutex.unlock()
|
||||
|
||||
fprintf(stderr, 'Actual input sampling rate: %0.2f\n', samplerate)
|
||||
if err != ERR_NO_ERROR:
|
||||
fprintf(stderr, 'Error starting data in\n')
|
||||
showErr(err)
|
||||
return
|
||||
|
||||
|
||||
td.tdmutex.lock()
|
||||
err = ulDaqInScanStatus(td.handle, &scanstat, &xstat)
|
||||
td.tdmutex.unlock()
|
||||
if err != ERR_NO_ERROR:
|
||||
fprintf(stderr, 'Error obtaining input scan status\n')
|
||||
showErr(err)
|
||||
return
|
||||
|
||||
while td.stopThread.load() == False and err == ERR_NO_ERROR:
|
||||
td.tdmutex.lock()
|
||||
err = ulDaqInScanStatus(td.handle, &scanstat, &xstat)
|
||||
td.tdmutex.unlock()
|
||||
if err != ERR_NO_ERROR:
|
||||
showErr(err)
|
||||
break
|
||||
|
||||
if xstat.currentIndex < buffer_mid_idx:
|
||||
top_enqueued = False
|
||||
if not bottom_enqueued:
|
||||
# Copy the bottom of the buffer to the queue, while transposing
|
||||
# it.
|
||||
inbuffer_cpy = <double*> malloc(sizeof(double)*td.samplesPerBlock*td.ninChanDescriptors)
|
||||
for chan in range(td.ninChanDescriptors):
|
||||
for sample in range(td.samplesPerBlock):
|
||||
inbuffer_cpy[chan*td.samplesPerBlock+sample] = td.inbuffer[buffer_mid_idx+sample*td.ninChanDescriptors+chan]
|
||||
td.inQueue.enqueue(inbuffer_cpy)
|
||||
|
||||
bottom_enqueued = True
|
||||
else:
|
||||
# fprintf(stderr,'sleep...\n')
|
||||
CPPsleep_us(sleep_time_mus)
|
||||
# fprintf(stderr,'awake...\n')
|
||||
else:
|
||||
bottom_enqueued = False
|
||||
if not top_enqueued:
|
||||
inbuffer_cpy = <double*> malloc(sizeof(double)*td.samplesPerBlock*td.ninChanDescriptors)
|
||||
for chan in range(td.ninChanDescriptors):
|
||||
for sample in range(td.samplesPerBlock):
|
||||
inbuffer_cpy[chan*td.samplesPerBlock+sample] = td.inbuffer[sample*td.ninChanDescriptors+chan]
|
||||
td.inQueue.enqueue(inbuffer_cpy)
|
||||
|
||||
# Enqueue the top part of the queue
|
||||
top_enqueued = True
|
||||
else:
|
||||
# fprintf(stderr,'sleep...\n')
|
||||
CPPsleep_us(sleep_time_mus)
|
||||
# fprintf(stderr,'awake...\n')
|
||||
|
||||
fprintf(stderr, 'Exit while loop input thread\n')
|
||||
|
||||
td.tdmutex.lock()
|
||||
err = ulDaqInScanStop(td.handle)
|
||||
td.tdmutex.unlock()
|
||||
if err != ERR_NO_ERROR:
|
||||
fprintf(stderr, "Error stopping DAQ input thread\n")
|
||||
showErr(err)
|
||||
|
||||
return
|
||||
|
||||
cdef void outThreadFunction(void* threaddata_void) nogil:
|
||||
"""
|
||||
Stream output thread function
|
||||
"""
|
||||
cdef:
|
||||
DaqThreadData* td = <DaqThreadData*> threaddata_void
|
||||
|
||||
UlError err
|
||||
|
||||
ScanOption scanoptions
|
||||
AOutScanFlag outscanflags
|
||||
TransferStatus xstat
|
||||
ScanStatus scanstat
|
||||
bint top_enqueued, bottom_enqueued
|
||||
unsigned buffer_mid_idx, chan, sample
|
||||
|
||||
double sleep_time
|
||||
double *outbuffer_cpy
|
||||
unsigned sleep_time_mus # Thread sleep time in ms
|
||||
double samplerate
|
||||
|
||||
samplerate = td.samplerate
|
||||
sleep_time = td.samplesPerBlock / td.samplerate / 2
|
||||
sleep_time_mus = <unsigned> (sleep_time * 1e6)
|
||||
|
||||
# CAREFULL, a MEMCPY here is only allowed for a single channel!!
|
||||
if not td.noutChanDescriptors == 1:
|
||||
fprintf(stderr, 'Error: multiple output channels not implemented!\n')
|
||||
return
|
||||
|
||||
# Copy first queue blob to top of outbuffer
|
||||
outbuffer_cpy = <double*> td.outQueue.dequeue()
|
||||
memcpy(td.outbuffer, outbuffer_cpy, sizeof(double)*td.samplesPerBlock)
|
||||
free(outbuffer_cpy)
|
||||
|
||||
outscanflags = AOUTSCAN_FF_DEFAULT
|
||||
scanoptions = <ScanOption> SO_CONTINUOUS
|
||||
|
||||
buffer_mid_idx = td.samplesPerBlock
|
||||
|
||||
top_enqueued = True
|
||||
bottom_enqueued = False
|
||||
|
||||
fprintf(stderr, 'Starting output thread\n')
|
||||
|
||||
td.tdmutex.lock()
|
||||
fprintf(stderr, 'mutex locked\n')
|
||||
|
||||
err = ulAOutScan(td.handle,
|
||||
0,
|
||||
0,
|
||||
BIP10VOLTS,
|
||||
2*td.samplesPerBlock, # Watch the 2 here!
|
||||
&samplerate,
|
||||
scanoptions,
|
||||
outscanflags,
|
||||
td.outbuffer)
|
||||
fprintf(stderr, 'returned\n')
|
||||
td.tdmutex.unlock()
|
||||
fprintf(stderr, 'Actual output sampling rate: %0.2f\n', samplerate)
|
||||
|
||||
if err != ERR_NO_ERROR:
|
||||
fprintf(stderr, 'Error starting data out\n')
|
||||
showErr(err)
|
||||
return
|
||||
|
||||
|
||||
td.tdmutex.lock()
|
||||
err = ulAOutScanStatus(td.handle, &scanstat, &xstat)
|
||||
td.tdmutex.unlock()
|
||||
if err != ERR_NO_ERROR:
|
||||
showErr(err)
|
||||
return
|
||||
|
||||
td.outThread_ready.store(True)
|
||||
|
||||
while td.stopThread.load() == False:
|
||||
# printf('Running output thread in loop\n')
|
||||
td.tdmutex.lock()
|
||||
err = ulAOutScanStatus(td.handle, &scanstat, &xstat)
|
||||
td.tdmutex.unlock()
|
||||
if err != ERR_NO_ERROR:
|
||||
showErr(err)
|
||||
break
|
||||
|
||||
if xstat.currentIndex < buffer_mid_idx:
|
||||
top_enqueued = False
|
||||
# fprintf('xstat.currentIndex'
|
||||
if not bottom_enqueued:
|
||||
# Copy the bottom of the buffer to the queue, while transposing
|
||||
# it.
|
||||
if not td.outQueue.empty():
|
||||
outbuffer_cpy = <double*> td.outQueue.dequeue()
|
||||
else:
|
||||
# outbuffer_cpy = <double*> malloc(sizeof(double)*td.noutChanDescriptors*td.samplesPerBlock)
|
||||
outbuffer_cpy = <double*> malloc(sizeof(double)*td.samplesPerBlock)
|
||||
# for chan in range(td.noutChanDescriptors):
|
||||
# for sample in range(td.samplesPerBlock):
|
||||
# outbuffer_cpy[chan*td.samplesPerBlock+sample] = 0.0
|
||||
for sample in range(td.samplesPerBlock):
|
||||
outbuffer_cpy[sample] = 0.0
|
||||
fprintf(stderr, 'Output buffer queue empty! Filling with zeros\n')
|
||||
|
||||
for chan in range(td.noutChanDescriptors):
|
||||
for sample in range(td.samplesPerBlock):
|
||||
# td.outbuffer[buffer_mid_idx+sample*td.noutChanDescriptors+chan] = outbuffer_cpy[chan*td.samplesPerBlock+sample]
|
||||
# Simpler, valid for single chan
|
||||
td.outbuffer[buffer_mid_idx+sample] = outbuffer_cpy[sample]
|
||||
free(outbuffer_cpy)
|
||||
|
||||
bottom_enqueued = True
|
||||
else:
|
||||
CPPsleep_us(sleep_time_mus)
|
||||
else:
|
||||
bottom_enqueued = False
|
||||
if not top_enqueued:
|
||||
if not td.outQueue.empty():
|
||||
outbuffer_cpy = <double*> td.outQueue.dequeue()
|
||||
else:
|
||||
outbuffer_cpy = <double*> malloc(sizeof(double)*td.noutChanDescriptors*td.samplesPerBlock)
|
||||
# for chan in range(td.noutChanDescriptors):
|
||||
# for sample in range(td.samplesPerBlock):
|
||||
# outbuffer_cpy[chan*td.samplesPerBlock+sample] = 0.0
|
||||
for sample in range(td.samplesPerBlock):
|
||||
outbuffer_cpy[sample] = 0.0
|
||||
fprintf(stderr, 'Output buffer queue empty! Filling with zeros\n')
|
||||
|
||||
for chan in range(td.noutChanDescriptors):
|
||||
for sample in range(td.samplesPerBlock):
|
||||
# td.outbuffer[sample*td.noutChanDescriptors+chan] = outbuffer_cpy[chan*td.samplesPerBlock+sample]
|
||||
# Simpler, valid for single chan
|
||||
td.outbuffer[sample] = outbuffer_cpy[sample]
|
||||
|
||||
free(outbuffer_cpy)
|
||||
|
||||
# Enqueue the top part of the queue
|
||||
top_enqueued = True
|
||||
else:
|
||||
CPPsleep_us(sleep_time_mus)
|
||||
|
||||
fprintf(stderr, 'Exit while loop output thread\n')
|
||||
|
||||
|
||||
td.tdmutex.lock()
|
||||
err = ulAOutScanStop(td.handle)
|
||||
td.tdmutex.unlock()
|
||||
if err != ERR_NO_ERROR:
|
||||
fprintf(stderr, "Error stopping AOut output thread\n")
|
||||
showErr(err)
|
||||
|
||||
cdef class UlDT9837A:
|
||||
cdef:
|
||||
DaqDeviceHandle handle
|
||||
bint handle_connected
|
||||
|
||||
unsigned int ninchannels
|
||||
unsigned int noutchannels
|
||||
object input_range
|
||||
object enabled_inputs
|
||||
|
||||
bint out_enabled
|
||||
bint monitor_gen
|
||||
|
||||
DaqThreadData *td
|
||||
|
||||
def __cinit__(self, unsigned int deviceno):
|
||||
self.handle = 0
|
||||
self.handle_connected = False
|
||||
|
||||
self.ninchannels = 4
|
||||
self.noutchannels = 1
|
||||
self.input_range = 4*[False]
|
||||
self.enabled_inputs = 4*[False]
|
||||
|
||||
self.out_enabled = False
|
||||
self.monitor_gen = False
|
||||
|
||||
self.td = NULL
|
||||
|
||||
cdef:
|
||||
DaqDeviceDescriptor devdescriptors[MAX_DEF_COUNT]
|
||||
DaqDeviceDescriptor descriptor
|
||||
DaqDeviceInterface interfaceType = ANY_IFC
|
||||
|
||||
UlError err
|
||||
|
||||
unsigned int numdevs = MAX_DEF_COUNT
|
||||
|
||||
err = ulGetDaqDeviceInventory(interfaceType,
|
||||
devdescriptors,
|
||||
&numdevs)
|
||||
if(err != ERR_NO_ERROR):
|
||||
raise RuntimeError(f'Device inventarization failed: {err}')
|
||||
|
||||
if deviceno >= numdevs:
|
||||
raise ValueError(f'Device number {deviceno} too high {err}. This could happen when the device is currently not connected')
|
||||
|
||||
descriptor = devdescriptors[deviceno]
|
||||
# get a handle to the DAQ device associated with the first descriptor
|
||||
self.handle = ulCreateDaqDevice(descriptor);
|
||||
|
||||
if self.handle == 0:
|
||||
raise RuntimeError ("Unable to create a handle to the specified DAQ device. Is the device currently in use?");
|
||||
|
||||
err = ulConnectDaqDevice(self.handle)
|
||||
if err != ERR_NO_ERROR:
|
||||
raise RuntimeError(f'Unable to connect to device: {err}')
|
||||
|
||||
cdef void startScan(self,
|
||||
unsigned int samplesPerBlock,
|
||||
double samplerate,
|
||||
SafeQueue[void*] *inQueue,
|
||||
SafeQueue[void*] *outQueue):
|
||||
|
||||
cdef:
|
||||
int i, j, nnormalinchannels, neninchannels
|
||||
|
||||
# Sanity checks
|
||||
if inQueue and (not any(self.enabled_inputs) and (not self.monitor_gen)):
|
||||
raise ValueError('Input queue given, but no input channels enabled and monitor output disabled')
|
||||
if outQueue and not self.out_enabled:
|
||||
raise ValueError('Output queue given, but output channel is not enabled')
|
||||
# End sanity checks
|
||||
|
||||
cdef:
|
||||
DaqThreadData *td
|
||||
|
||||
td = <DaqThreadData*> malloc(sizeof(DaqThreadData))
|
||||
|
||||
if not td:
|
||||
raise MemoryError()
|
||||
|
||||
with nogil:
|
||||
td.samplesPerBlock = samplesPerBlock
|
||||
td.samplerate = <double> samplerate
|
||||
|
||||
td.handle = self.handle
|
||||
td.inQueue = inQueue
|
||||
td.outQueue = outQueue
|
||||
|
||||
td.inChanDescriptors = NULL
|
||||
td.ninChanDescriptors = 0
|
||||
td.noutChanDescriptors = 0
|
||||
|
||||
td.inThread = NULL
|
||||
td.outThread = NULL
|
||||
|
||||
td.stopThread.store(False)
|
||||
td.tdmutex = new mutex()
|
||||
|
||||
td.inbuffer = NULL
|
||||
td.outbuffer = NULL
|
||||
if not outQueue:
|
||||
td.outThread_ready.store(True)
|
||||
else:
|
||||
td.outThread_ready.store(False)
|
||||
|
||||
# Configure INPUTS
|
||||
py_nnormalinchannels = sum([1 if en_input else 0 for en_input in
|
||||
self.enabled_inputs])
|
||||
nnormalinchannels = py_nnormalinchannels
|
||||
|
||||
j = 0
|
||||
neninchannels = nnormalinchannels
|
||||
if self.monitor_gen:
|
||||
neninchannels += 1
|
||||
|
||||
fprintf(stderr, 'neninchannels: %u\n', neninchannels)
|
||||
|
||||
if neninchannels > 0:
|
||||
td.inChanDescriptors = <DaqInChanDescriptor*> malloc(neninchannels*
|
||||
sizeof(DaqInChanDescriptor))
|
||||
if not td.inChanDescriptors:
|
||||
self.cleanupThreadData(td)
|
||||
raise MemoryError()
|
||||
td.ninChanDescriptors = neninchannels
|
||||
td.inbuffer = <double*> malloc(2*sizeof(double)*td.samplesPerBlock*
|
||||
neninchannels)
|
||||
if not td.inbuffer:
|
||||
self.cleanupThreadData(td)
|
||||
raise MemoryError()
|
||||
|
||||
j = 0
|
||||
for i in range(4):
|
||||
if self.enabled_inputs[i]:
|
||||
td.inChanDescriptors[j].type = DAQI_ANALOG_SE
|
||||
td.inChanDescriptors[j].channel = i
|
||||
td.inChanDescriptors[j].range = BIP10VOLTS if self.input_range[i] else BIP1VOLTS
|
||||
j+=1
|
||||
|
||||
td.inbuffer = <double*> malloc(2*sizeof(double)*td.samplesPerBlock*neninchannels)
|
||||
if not td.inbuffer:
|
||||
self.cleanupThreadData(td)
|
||||
raise MemoryError()
|
||||
|
||||
# Create input channel descriptors
|
||||
if self.monitor_gen:
|
||||
td.inChanDescriptors[neninchannels-1].type = DAQI_DAC
|
||||
td.inChanDescriptors[neninchannels-1].channel = 0
|
||||
td.inChanDescriptors[neninchannels-1].range = BIP10VOLTS
|
||||
|
||||
fprintf(stderr, 'SFSG12\n')
|
||||
|
||||
# CONFIGURE OUTPUTS
|
||||
if self.out_enabled:
|
||||
fprintf(stderr, 'SFSG13\n')
|
||||
td.noutChanDescriptors = 1
|
||||
# WATCH THE TWO HERE!!!
|
||||
td.outbuffer = <double*> malloc(2*sizeof(double)*td.samplesPerBlock)
|
||||
if not td.outbuffer:
|
||||
self.cleanupThreadData(td)
|
||||
raise MemoryError()
|
||||
|
||||
# Create DAQ Threads
|
||||
with nogil:
|
||||
fprintf(stderr, 'SFSG14\n')
|
||||
if self.out_enabled:
|
||||
td.outThread = new CPPThread[void*, void (*)(void*)](
|
||||
outThreadFunction, <void*> td)
|
||||
|
||||
if not td.outThread:
|
||||
self.cleanupThreadData(td)
|
||||
raise RuntimeError('Error creating thread')
|
||||
|
||||
if neninchannels > 0:
|
||||
fprintf(stderr, 'SFSG15\n')
|
||||
td.inThread = new CPPThread[void*, void (*)(void*)](
|
||||
inThreadFunction, <void*> td)
|
||||
|
||||
if not td.inThread:
|
||||
self.cleanupThreadData(td)
|
||||
raise RuntimeError('Error creating thread')
|
||||
self.td = td
|
||||
|
||||
# def start(self):
|
||||
# cdef:
|
||||
# SafeQueue[void*] *inqueue
|
||||
# SafeQueue[void*] *outqueue
|
||||
# int i, sample, samples, samplesperbuf
|
||||
# double meas_seconds, samplerate
|
||||
# double* inbuf
|
||||
# double* outbuf
|
||||
|
||||
# inqueue = new SafeQueue[void*]()
|
||||
# outqueue = new SafeQueue[void*]()
|
||||
|
||||
# samplesperbuf = 512
|
||||
# samplerate = 10000
|
||||
# meas_seconds = 4
|
||||
|
||||
# for i in range(int(meas_seconds*samplerate/samplesperbuf)):
|
||||
# outbuf = <double*> malloc(sizeof(double)*samplesperbuf)
|
||||
# for sample in range(samplesperbuf):
|
||||
# outbuf[sample] = <double> 1
|
||||
# outqueue.enqueue(<void*> outbuf)
|
||||
|
||||
# self.startScan(
|
||||
# samplesperbuf,
|
||||
# samplerate, # Sample rate
|
||||
# inqueue,
|
||||
# outqueue)
|
||||
|
||||
# CPPsleep_ms(int(0.5*1000*meas_seconds))
|
||||
# self.stop()
|
||||
# while not inqueue.empty():
|
||||
# inbuf = <double*> inqueue.dequeue()
|
||||
# for sample in range(samplesperbuf):
|
||||
# print(f'Value monitor: {inbuf[sample]:1.2f}, value input: {inbuf[samplesperbuf+sample]:1.2f}')
|
||||
# pass
|
||||
# free(inbuf)
|
||||
|
||||
# while not outqueue.empty():
|
||||
# outbuf = <double*> outqueue.dequeue()
|
||||
# free(outbuf)
|
||||
# del inqueue
|
||||
# del outqueue
|
||||
|
||||
def stopScan(self):
|
||||
print(f'UlDT9837A: stopScan()')
|
||||
self.cleanupThreadData(self.td)
|
||||
self.td = NULL
|
||||
|
||||
def setInputChannelConfig(self, unsigned chnum, channelconfig: DAQChannel):
|
||||
if self.td:
|
||||
raise RuntimeError('Cannot change settings while sampling')
|
||||
cdef:
|
||||
int i
|
||||
UlError err
|
||||
IepeMode iepe
|
||||
CouplingMode cm
|
||||
|
||||
if chnum > 3:
|
||||
raise RuntimeError('Invalid input channel number')
|
||||
|
||||
# if chnum == 0:
|
||||
# fprintf(stderr, '====================== BIG WARNING ==============\n')
|
||||
# fprintf(stderr, 'We override IEPE to enabled on ch 0\n')
|
||||
# channelconfig.IEPE_enabled = True
|
||||
# fprintf(stderr, '====================== END BIG WARNING ==============\n')
|
||||
|
||||
|
||||
self.input_range[chnum] = True if channelconfig.range_ == pyRange.tenV else False
|
||||
self.enabled_inputs[chnum] = channelconfig.channel_enabled
|
||||
|
||||
iepe = IEPE_ENABLED if channelconfig.IEPE_enabled else IEPE_DISABLED
|
||||
cm = CM_AC if channelconfig.IEPE_enabled else CM_DC
|
||||
|
||||
err = ulAISetConfig(self.handle, AI_CFG_CHAN_IEPE_MODE, chnum,
|
||||
iepe)
|
||||
if err != ERR_NO_ERROR:
|
||||
raise RuntimeError('Fatal: could not set IEPE mode')
|
||||
|
||||
err = ulAISetConfig(self.handle, AI_CFG_CHAN_COUPLING_MODE, chnum, cm);
|
||||
if err != ERR_NO_ERROR:
|
||||
raise RuntimeError('Fatal: could not set coupling mode')
|
||||
|
||||
# err = ulAISetConfigDbl(self.handle, AI_CFG_CHAN_SENSOR_SENSITIVITY,
|
||||
# chnum, channelconfig.sensitivity)
|
||||
err = ulAISetConfigDbl(self.handle, AI_CFG_CHAN_SENSOR_SENSITIVITY,
|
||||
chnum, 1.0)
|
||||
# TODO: Fix this problem, of setting sensitivity twice, how do we do it
|
||||
# in the future?
|
||||
if err != ERR_NO_ERROR:
|
||||
raise RuntimeError('Fatal: could not set sensitivity')
|
||||
|
||||
def setOutputChannelConfig(self, unsigned chnum, channelconfig:
|
||||
DAQChannel, bint monitor_gen):
|
||||
if self.td:
|
||||
raise RuntimeError('Cannot change settings while sampling')
|
||||
if chnum > 0:
|
||||
raise RuntimeError('Invalid output channel number')
|
||||
|
||||
if monitor_gen and not channelconfig.channel_enabled:
|
||||
raise RuntimeError('Output channel should be enabled to enable channel monitoring')
|
||||
|
||||
self.out_enabled = channelconfig.channel_enabled
|
||||
self.monitor_gen = monitor_gen
|
||||
|
||||
def __dealloc__(self):
|
||||
fprintf(stderr, '__dealloc__\n')
|
||||
self.cleanupThreadData(self.td)
|
||||
self.td = NULL
|
||||
|
||||
if self.handle_connected:
|
||||
ulDisconnectDaqDevice(self.handle)
|
||||
ulReleaseDaqDevice(self.handle)
|
||||
|
||||
cdef void cleanupThreadData(self, DaqThreadData* td) nogil:
|
||||
fprintf(stderr, 'cleanupThreadData()\n')
|
||||
|
||||
if td is NULL:
|
||||
printf('TD is zero\n')
|
||||
return
|
||||
|
||||
td.stopThread.store(True)
|
||||
if td.inThread:
|
||||
td.inThread.join()
|
||||
del td.inThread
|
||||
fprintf(stderr, 'SFSG0\n')
|
||||
|
||||
if td.outThread:
|
||||
td.outThread.join()
|
||||
del td.outThread
|
||||
fprintf(stderr, 'SFSG1\n')
|
||||
|
||||
del td.tdmutex
|
||||
|
||||
if td.inChanDescriptors:
|
||||
free(td.inChanDescriptors)
|
||||
|
||||
fprintf(stderr, 'SFSG1\n')
|
||||
if td.inbuffer:
|
||||
free(td.inbuffer)
|
||||
fprintf(stderr, 'SFSG2\n')
|
||||
if td.outbuffer:
|
||||
free(td.outbuffer)
|
||||
fprintf(stderr, 'SFSG3\n')
|
||||
|
||||
free(td)
|
||||
fprintf(stderr, 'end cleanupThreadData()\n')
|
||||
cdef extern from "lasp_cppuldaq.h":
|
||||
cdef cppclass DT9837A(Daq):
|
||||
DT9837A(us samplesPerBlock,
|
||||
vector[bool] inChannels,
|
||||
vector[bool] outChannels,
|
||||
double samplerate,
|
||||
bint monitorOutput,
|
||||
us deviceno)
|
||||
void start(SafeQueue[double*] *inQueue,
|
||||
SafeQueue[double*] *outQueue)
|
||||
void stop()
|
||||
void setACCouplingMode(vector[bool] accoupling)
|
||||
void setInputRange(vector[bool] accoupling)
|
||||
void setIEPEEnabled(vector[bool] iepe)
|
||||
us neninchannels()
|
||||
us nenoutchannels()
|
||||
|
||||
ctypedef struct PyStreamData:
|
||||
PyObject* pyCallback
|
||||
|
|
@ -650,8 +53,8 @@ ctypedef struct PyStreamData:
|
|||
|
||||
# If these queue pointers are NULL, it means the stream does not have an
|
||||
# input, or output.
|
||||
SafeQueue[void*] *inQueue
|
||||
SafeQueue[void*] *outQueue
|
||||
SafeQueue[double*] *inQueue
|
||||
SafeQueue[double*] *outQueue
|
||||
CPPThread[void*, void (*)(void*)] *thread
|
||||
|
||||
cdef void audioCallbackPythonThreadFunction(void* voidsd) nogil:
|
||||
|
|
@ -659,8 +62,8 @@ cdef void audioCallbackPythonThreadFunction(void* voidsd) nogil:
|
|||
PyStreamData* sd
|
||||
cnp.NPY_TYPES npy_format
|
||||
|
||||
void* inbuffer = NULL
|
||||
void* outbuffer = NULL
|
||||
double* inbuffer = NULL
|
||||
double* outbuffer = NULL
|
||||
|
||||
unsigned noutchannels
|
||||
unsigned ninchannels
|
||||
|
|
@ -669,7 +72,6 @@ cdef void audioCallbackPythonThreadFunction(void* voidsd) nogil:
|
|||
|
||||
unsigned sw = sizeof(double)
|
||||
|
||||
|
||||
sd = <PyStreamData*> voidsd
|
||||
ninchannels = sd.ninchannels
|
||||
noutchannels = sd.noutchannels
|
||||
|
|
@ -677,7 +79,6 @@ cdef void audioCallbackPythonThreadFunction(void* voidsd) nogil:
|
|||
nBytesPerChan = sd.nBytesPerChan
|
||||
nFramesPerBlock = sd.nFramesPerBlock
|
||||
|
||||
|
||||
with gil:
|
||||
npy_format = cnp.NPY_FLOAT64
|
||||
callback = <object> sd.pyCallback
|
||||
|
|
@ -687,7 +88,7 @@ cdef void audioCallbackPythonThreadFunction(void* voidsd) nogil:
|
|||
while not sd.stopThread.load():
|
||||
|
||||
if sd.outQueue:
|
||||
outbuffer = malloc(nBytesPerChan*noutchannels)
|
||||
outbuffer = <double*> malloc(sizeof(double)*nBytesPerChan*noutchannels)
|
||||
|
||||
if sd.inQueue:
|
||||
if not sd.outQueue:
|
||||
|
|
@ -745,7 +146,7 @@ cdef void audioCallbackPythonThreadFunction(void* voidsd) nogil:
|
|||
return
|
||||
|
||||
if sd.outQueue:
|
||||
sd.outQueue.enqueue(outbuffer)
|
||||
sd.outQueue.enqueue(<double*> outbuffer)
|
||||
if not sd.inQueue:
|
||||
while sd.outQueue.size() > 10 and not sd.stopThread.load():
|
||||
# printf('Sleeping...\n')
|
||||
|
|
@ -762,7 +163,7 @@ cdef void audioCallbackPythonThreadFunction(void* voidsd) nogil:
|
|||
cdef class UlDaq:
|
||||
cdef:
|
||||
PyStreamData *sd
|
||||
UlDT9837A daq_device
|
||||
Daq* daq_device
|
||||
|
||||
def __cinit__(self):
|
||||
|
||||
|
|
@ -770,68 +171,9 @@ cdef class UlDaq:
|
|||
Acquires a daq handle, and opens the device
|
||||
|
||||
"""
|
||||
self.daq_device = None
|
||||
self.daq_device = NULL
|
||||
self.sd = NULL
|
||||
|
||||
def getDeviceInfo(self):
|
||||
"""
|
||||
Returns device information objects (DeviceInfo) for all available
|
||||
devices
|
||||
"""
|
||||
cdef:
|
||||
DaqDeviceDescriptor devdescriptors[MAX_DEF_COUNT]
|
||||
DaqDeviceDescriptor descriptor
|
||||
DaqDeviceInterface interfaceType = ANY_IFC
|
||||
DaqDeviceHandle handle
|
||||
|
||||
UlError err
|
||||
|
||||
unsigned int numdevs = MAX_DEF_COUNT
|
||||
unsigned deviceno
|
||||
|
||||
if self.sd is not NULL or self.daq_device is not None:
|
||||
assert self.daq_device is not None
|
||||
raise RuntimeError('Cannot acquire device info: stream is already opened.')
|
||||
|
||||
err = ulGetDaqDeviceInventory(interfaceType,
|
||||
devdescriptors,
|
||||
&numdevs)
|
||||
if(err != ERR_NO_ERROR):
|
||||
raise RuntimeError(f'Device inventarization failed: {err}')
|
||||
|
||||
|
||||
py_devinfo = []
|
||||
for deviceno in range(numdevs):
|
||||
descriptor = devdescriptors[deviceno]
|
||||
|
||||
if descriptor.productName == b'DT9837A':
|
||||
# Create proper interface name
|
||||
if descriptor.devInterface == DaqDeviceInterface.USB_IFC:
|
||||
name = 'USB - '
|
||||
elif descriptor.devInterface == DaqDeviceInterface.BLUETOOTH_IFC:
|
||||
name = 'Bluetooth - '
|
||||
elif descriptor.devInterface == DaqDeviceInterface.ETHERNET_IFC:
|
||||
name = 'Ethernet - '
|
||||
|
||||
name += descriptor.productName.decode('utf-8') + ', id ' + \
|
||||
descriptor.uniqueId.decode('utf-8')
|
||||
|
||||
d = DeviceInfo(
|
||||
api = -1,
|
||||
index = deviceno,
|
||||
probed = True,
|
||||
name = name,
|
||||
outputchannels = 1,
|
||||
inputchannels = 4,
|
||||
duplexchannels = 0,
|
||||
samplerates = [10000, 16000, 20000, 32000, 48000, 50000] ,
|
||||
sampleformats = ['64-bit floats'],
|
||||
prefsamplerate = 48000,
|
||||
hasInputIEPE = True)
|
||||
py_devinfo.append(d)
|
||||
return py_devinfo
|
||||
|
||||
|
||||
@cython.nonecheck(True)
|
||||
def start(self, avstream):
|
||||
"""
|
||||
|
|
@ -844,7 +186,7 @@ cdef class UlDaq:
|
|||
"""
|
||||
|
||||
if self.sd is not NULL:
|
||||
assert self.daq_device is not None
|
||||
assert self.daq_device is not NULL
|
||||
raise RuntimeError('Stream is already opened.')
|
||||
|
||||
daqconfig = avstream.daqconfig
|
||||
|
|
@ -862,6 +204,8 @@ cdef class UlDaq:
|
|||
bint in_stream=False
|
||||
bint out_stream=False
|
||||
|
||||
DT9837A* daq_device
|
||||
|
||||
if daqconfig.nFramesPerBlock > 8192 or daqconfig.nFramesPerBlock < 512:
|
||||
raise ValueError('Invalid number of nFramesPerBlock')
|
||||
|
||||
|
|
@ -911,47 +255,46 @@ cdef class UlDaq:
|
|||
self.sd.nBytesPerChan = daqconfig.nFramesPerBlock*sizeof(double)
|
||||
self.sd.nFramesPerBlock = daqconfig.nFramesPerBlock
|
||||
|
||||
if 'DT9837A' in device.name:
|
||||
self.daq_device = UlDT9837A(device.index)
|
||||
else:
|
||||
raise RuntimeError(f'Device {device.name} not found or not configured')
|
||||
|
||||
# Create channel maps for in channels, set in stream
|
||||
# parameters
|
||||
inch_enabled = 4*False
|
||||
if in_stream:
|
||||
print('Stream is input stream')
|
||||
for i, ch in enumerate(daqconfig.getInputChannels()):
|
||||
if ch.channel_enabled:
|
||||
self.sd.ninchannels += 1
|
||||
self.daq_device.setInputChannelConfig(i, ch)
|
||||
|
||||
self.sd.inQueue = new SafeQueue[void*]()
|
||||
inch_enabled = [True if ch.channel_enabled else False for ch in
|
||||
daqconfig.getInputChannels()]
|
||||
|
||||
self.sd.inQueue = new SafeQueue[double*]()
|
||||
|
||||
# Create channel maps for output channels
|
||||
outch_enabled = 4*False
|
||||
if out_stream:
|
||||
print('Stream is output stream')
|
||||
for i, ch in enumerate(daqconfig.getOutputChannels()):
|
||||
if ch.channel_enabled:
|
||||
self.sd.noutchannels += 1
|
||||
self.daq_device.setOutputChannelConfig(i, ch, monitorOutput)
|
||||
outch_enabled = [True if ch.channel_enabled else False for ch in
|
||||
daqconfig.getOutputChannels()]
|
||||
|
||||
self.sd.outQueue = new SafeQueue[void*]()
|
||||
self.sd.outQueue = new SafeQueue[double*]()
|
||||
|
||||
if monitorOutput and duplex_mode:
|
||||
self.sd.ninchannels += self.sd.noutchannels
|
||||
if 'DT9837A' in device.name:
|
||||
daq_device = new DT9837A(
|
||||
daqconfig.nFramesPerBlock,
|
||||
inch_enabled,
|
||||
outch_enabled,
|
||||
daqconfig.samplerate,
|
||||
monitorOutput,
|
||||
device.index)
|
||||
else:
|
||||
raise RuntimeError(f'Device {device.name} not found or not configured')
|
||||
|
||||
self.sd.pyCallback = <PyObject*> avstream._audioCallback
|
||||
self.sd.ninchannels = daq_device.neninchannels()
|
||||
self.sd.noutchannels = daq_device.nenoutchannels()
|
||||
|
||||
self.daq_device = daq_device
|
||||
|
||||
# Increase reference count to the callback
|
||||
Py_INCREF(<object> avstream._audioCallback)
|
||||
|
||||
|
||||
self.daq_device.startScan(
|
||||
daqconfig.nFramesPerBlock,
|
||||
samplerate,
|
||||
self.sd.inQueue,
|
||||
self.sd.outQueue)
|
||||
|
||||
with nogil:
|
||||
self.sd.thread = new CPPThread[void*, void (*)(void*)](audioCallbackPythonThreadFunction,
|
||||
<void*> self.sd)
|
||||
|
|
@ -959,15 +302,20 @@ cdef class UlDaq:
|
|||
# Allow it to start
|
||||
CPPsleep_ms(500)
|
||||
|
||||
self.daq_device.start(
|
||||
self.sd.inQueue,
|
||||
self.sd.outQueue)
|
||||
|
||||
return nFramesPerBlock, self.daq_device.td.samplerate
|
||||
return nFramesPerBlock, self.daq_device.samplerate()
|
||||
|
||||
def stop(self):
|
||||
if self.sd is NULL:
|
||||
raise RuntimeError('Stream is not opened')
|
||||
if self.daq_device is not NULL:
|
||||
self.daq_device.stop()
|
||||
del self.daq_device
|
||||
self.daq_device = NULL
|
||||
|
||||
self.daq_device.stopScan()
|
||||
self.daq_device = None
|
||||
self.cleanupStream(self.sd)
|
||||
self.sd = NULL
|
||||
|
||||
|
|
@ -1004,3 +352,62 @@ cdef class UlDaq:
|
|||
sd.pyCallback = NULL
|
||||
|
||||
free(sd)
|
||||
|
||||
def getDeviceInfo(self):
|
||||
"""
|
||||
Returns device information objects (DeviceInfo) for all available
|
||||
devices
|
||||
"""
|
||||
cdef:
|
||||
DaqDeviceDescriptor devdescriptors[MAX_DEF_COUNT]
|
||||
DaqDeviceDescriptor descriptor
|
||||
DaqDeviceInterface interfaceType = ANY_IFC
|
||||
DaqDeviceHandle handle
|
||||
|
||||
UlError err
|
||||
|
||||
unsigned int numdevs = MAX_DEF_COUNT
|
||||
unsigned deviceno
|
||||
|
||||
if self.sd is not NULL:
|
||||
assert self.daq_device is not NULL
|
||||
raise RuntimeError('Cannot acquire device info: stream is already opened.')
|
||||
|
||||
err = ulGetDaqDeviceInventory(interfaceType,
|
||||
devdescriptors,
|
||||
&numdevs)
|
||||
if(err != ERR_NO_ERROR):
|
||||
raise RuntimeError(f'Device inventarization failed: {err}')
|
||||
|
||||
|
||||
py_devinfo = []
|
||||
for deviceno in range(numdevs):
|
||||
descriptor = devdescriptors[deviceno]
|
||||
|
||||
if descriptor.productName == b'DT9837A':
|
||||
# Create proper interface name
|
||||
if descriptor.devInterface == DaqDeviceInterface.USB_IFC:
|
||||
name = 'USB - '
|
||||
elif descriptor.devInterface == DaqDeviceInterface.BLUETOOTH_IFC:
|
||||
name = 'Bluetooth - '
|
||||
elif descriptor.devInterface == DaqDeviceInterface.ETHERNET_IFC:
|
||||
name = 'Ethernet - '
|
||||
|
||||
name += descriptor.productName.decode('utf-8') + ', id ' + \
|
||||
descriptor.uniqueId.decode('utf-8')
|
||||
|
||||
d = DeviceInfo(
|
||||
api = -1,
|
||||
index = deviceno,
|
||||
probed = True,
|
||||
name = name,
|
||||
outputchannels = 1,
|
||||
inputchannels = 4,
|
||||
duplexchannels = 0,
|
||||
samplerates = [10000, 16000, 20000, 32000, 48000, 50000] ,
|
||||
sampleformats = ['64-bit floats'],
|
||||
prefsamplerate = 48000,
|
||||
hasInputIEPE = True)
|
||||
py_devinfo.append(d)
|
||||
return py_devinfo
|
||||
|
||||
|
|
|
|||
50
lasp/device/test_uldaq.cpp
Normal file
50
lasp/device/test_uldaq.cpp
Normal file
|
|
@ -0,0 +1,50 @@
|
|||
#include "lasp_cppuldaq.h"
|
||||
#include <chrono>
|
||||
#include <iostream>
|
||||
using std::cout;
|
||||
using std::endl;
|
||||
|
||||
|
||||
int main() {
|
||||
|
||||
boolvec inChannels = {true, false, false, false};
|
||||
/* boolvec inChannels = {false, false, false, false}; */
|
||||
boolvec outChannels = {true};
|
||||
double samplerate = 10000;
|
||||
const us samplesPerBlock = 2048;
|
||||
|
||||
DT9837A daq(
|
||||
samplesPerBlock,
|
||||
inChannels,
|
||||
outChannels,
|
||||
samplerate,
|
||||
false // monitor Output
|
||||
);
|
||||
|
||||
SafeQueue<double*> inqueue;
|
||||
SafeQueue<double*> outqueue;
|
||||
|
||||
daq.start(&inqueue, &outqueue);
|
||||
/* daq.start(NULL, &outqueue); */
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::seconds(5));
|
||||
/* std::string a; */
|
||||
/* std::cin >> a; */
|
||||
|
||||
daq.stop();
|
||||
|
||||
while(!inqueue.empty()) {
|
||||
double* buf = inqueue.dequeue();
|
||||
for(us i=0;i<samplesPerBlock;i++) {
|
||||
for(us ch=0;ch<daq.neninchannels();ch++) {
|
||||
cout << buf[ch*samplesPerBlock+i] << " ";
|
||||
}
|
||||
cout << endl;
|
||||
}
|
||||
free(buf);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user