mirror of
https://github.com/slsdetectorgroup/slsDetectorPackage.git
synced 2025-06-21 09:08:00 +02:00
format receiver
This commit is contained in:
Erik Frojdh
317
slsReceiverSoftware/src/DataStreamer.cpp
Executable file → Normal file
317
slsReceiverSoftware/src/DataStreamer.cpp
Executable file → Normal file
@ -3,7 +3,6 @@
|
||||
* @short streams data from receiver via ZMQ
|
||||
***********************************************/
|
||||
|
||||
|
||||
#include "DataStreamer.h"
|
||||
#include "Fifo.h"
|
||||
#include "GeneralData.h"
|
||||
@ -15,211 +14,210 @@
|
||||
|
||||
const std::string DataStreamer::TypeName = "DataStreamer";
|
||||
|
||||
DataStreamer::DataStreamer(int ind, Fifo *f, uint32_t *dr, ROI *r, uint64_t *fi,
|
||||
int fd, int *nd, bool *qe, uint64_t *tot)
|
||||
: ThreadObject(ind, TypeName), fifo(f), dynamicRange(dr), roi(r),
|
||||
fileIndex(fi), flippedDataX(fd), quadEnable(qe), totalNumFrames(tot) {
|
||||
numDet[0] = nd[0];
|
||||
numDet[1] = nd[1];
|
||||
|
||||
DataStreamer::DataStreamer(int ind, Fifo* f, uint32_t* dr, ROI* r,
|
||||
uint64_t* fi, int fd, int* nd, bool* qe, uint64_t* tot) :
|
||||
ThreadObject(ind, TypeName),
|
||||
fifo(f),
|
||||
dynamicRange(dr),
|
||||
roi(r),
|
||||
fileIndex(fi),
|
||||
flippedDataX(fd),
|
||||
quadEnable(qe),
|
||||
totalNumFrames(tot)
|
||||
{
|
||||
numDet[0] = nd[0];
|
||||
numDet[1] = nd[1];
|
||||
|
||||
LOG(logDEBUG) << "DataStreamer " << ind << " created";
|
||||
}
|
||||
|
||||
|
||||
DataStreamer::~DataStreamer() {
|
||||
CloseZmqSocket();
|
||||
delete [] completeBuffer;
|
||||
CloseZmqSocket();
|
||||
delete[] completeBuffer;
|
||||
}
|
||||
|
||||
void DataStreamer::SetFifo(Fifo* f) {
|
||||
fifo = f;
|
||||
}
|
||||
void DataStreamer::SetFifo(Fifo *f) { fifo = f; }
|
||||
|
||||
void DataStreamer::ResetParametersforNewAcquisition(const std::string& fname){
|
||||
void DataStreamer::ResetParametersforNewAcquisition(const std::string &fname) {
|
||||
StopRunning();
|
||||
startedFlag = false;
|
||||
firstIndex = 0;
|
||||
startedFlag = false;
|
||||
firstIndex = 0;
|
||||
|
||||
fileNametoStream = fname;
|
||||
if (completeBuffer) {
|
||||
delete[] completeBuffer;
|
||||
completeBuffer = nullptr;
|
||||
}
|
||||
if (roi->xmin != -1) {
|
||||
if (generalData->myDetectorType == GOTTHARD) {
|
||||
adcConfigured = generalData->GetAdcConfigured(index, *roi);
|
||||
}
|
||||
completeBuffer = new char[generalData->imageSizeComplete];
|
||||
memset(completeBuffer, 0, generalData->imageSizeComplete);
|
||||
}
|
||||
delete[] completeBuffer;
|
||||
completeBuffer = nullptr;
|
||||
}
|
||||
if (roi->xmin != -1) {
|
||||
if (generalData->myDetectorType == GOTTHARD) {
|
||||
adcConfigured = generalData->GetAdcConfigured(index, *roi);
|
||||
}
|
||||
completeBuffer = new char[generalData->imageSizeComplete];
|
||||
memset(completeBuffer, 0, generalData->imageSizeComplete);
|
||||
}
|
||||
}
|
||||
|
||||
void DataStreamer::RecordFirstIndex(uint64_t fnum) {
|
||||
startedFlag = true;
|
||||
firstIndex = fnum;
|
||||
startedFlag = true;
|
||||
firstIndex = fnum;
|
||||
|
||||
LOG(logDEBUG1) << index << " First Index: " << firstIndex;
|
||||
LOG(logDEBUG1) << index << " First Index: " << firstIndex;
|
||||
}
|
||||
|
||||
void DataStreamer::SetGeneralData(GeneralData* g) {
|
||||
generalData = g;
|
||||
generalData->Print();
|
||||
void DataStreamer::SetGeneralData(GeneralData *g) {
|
||||
generalData = g;
|
||||
generalData->Print();
|
||||
}
|
||||
|
||||
void DataStreamer::SetNumberofDetectors(int* nd) {
|
||||
numDet[0] = nd[0];
|
||||
numDet[1] = nd[1];
|
||||
void DataStreamer::SetNumberofDetectors(int *nd) {
|
||||
numDet[0] = nd[0];
|
||||
numDet[1] = nd[1];
|
||||
}
|
||||
|
||||
void DataStreamer::SetFlippedDataX(int fd) {
|
||||
flippedDataX = fd;
|
||||
void DataStreamer::SetFlippedDataX(int fd) { flippedDataX = fd; }
|
||||
|
||||
void DataStreamer::SetAdditionalJsonHeader(
|
||||
const std::map<std::string, std::string> &json) {
|
||||
additionJsonHeader = json;
|
||||
}
|
||||
|
||||
void DataStreamer::SetAdditionalJsonHeader(const std::map<std::string, std::string> &json) {
|
||||
additionJsonHeader = json;
|
||||
void DataStreamer::CreateZmqSockets(int *nunits, uint32_t port,
|
||||
const sls::IpAddr ip) {
|
||||
uint32_t portnum = port + index;
|
||||
std::string sip = ip.str();
|
||||
try {
|
||||
zmqSocket = new ZmqSocket(portnum, (ip != 0 ? sip.c_str() : nullptr));
|
||||
} catch (...) {
|
||||
LOG(logERROR) << "Could not create Zmq socket on port " << portnum
|
||||
<< " for Streamer " << index;
|
||||
throw;
|
||||
}
|
||||
LOG(logINFO) << index << " Streamer: Zmq Server started at "
|
||||
<< zmqSocket->GetZmqServerAddress();
|
||||
}
|
||||
|
||||
void DataStreamer::CreateZmqSockets(int* nunits, uint32_t port, const sls::IpAddr ip) {
|
||||
uint32_t portnum = port + index;
|
||||
std::string sip = ip.str();
|
||||
try {
|
||||
zmqSocket = new ZmqSocket(portnum, (ip != 0? sip.c_str(): nullptr));
|
||||
} catch (...) {
|
||||
LOG(logERROR) << "Could not create Zmq socket on port " << portnum << " for Streamer " << index;
|
||||
throw;
|
||||
}
|
||||
LOG(logINFO) << index << " Streamer: Zmq Server started at " << zmqSocket->GetZmqServerAddress();
|
||||
}
|
||||
|
||||
|
||||
void DataStreamer::CloseZmqSocket() {
|
||||
if (zmqSocket) {
|
||||
delete zmqSocket;
|
||||
zmqSocket = nullptr;
|
||||
}
|
||||
if (zmqSocket) {
|
||||
delete zmqSocket;
|
||||
zmqSocket = nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
void DataStreamer::ThreadExecution() {
|
||||
char* buffer=nullptr;
|
||||
fifo->PopAddressToStream(buffer);
|
||||
LOG(logDEBUG5) << "DataStreamer " << index << ", "
|
||||
"pop 0x" << std::hex << (void*)(buffer) << std::dec << ":" << buffer;
|
||||
char *buffer = nullptr;
|
||||
fifo->PopAddressToStream(buffer);
|
||||
LOG(logDEBUG5) << "DataStreamer " << index
|
||||
<< ", "
|
||||
"pop 0x"
|
||||
<< std::hex << (void *)(buffer) << std::dec << ":" << buffer;
|
||||
|
||||
// check dummy
|
||||
uint32_t numBytes = (uint32_t)(*((uint32_t *)buffer));
|
||||
LOG(logDEBUG1) << "DataStreamer " << index << ", Numbytes:" << numBytes;
|
||||
if (numBytes == DUMMY_PACKET_VALUE) {
|
||||
StopProcessing(buffer);
|
||||
return;
|
||||
}
|
||||
|
||||
//check dummy
|
||||
uint32_t numBytes = (uint32_t)(*((uint32_t*)buffer));
|
||||
LOG(logDEBUG1) << "DataStreamer " << index << ", Numbytes:" << numBytes;
|
||||
if (numBytes == DUMMY_PACKET_VALUE) {
|
||||
StopProcessing(buffer);
|
||||
return;
|
||||
}
|
||||
|
||||
ProcessAnImage(buffer);
|
||||
|
||||
//free
|
||||
fifo->FreeAddress(buffer);
|
||||
ProcessAnImage(buffer);
|
||||
|
||||
// free
|
||||
fifo->FreeAddress(buffer);
|
||||
}
|
||||
|
||||
void DataStreamer::StopProcessing(char *buf) {
|
||||
LOG(logDEBUG1) << "DataStreamer " << index << ": Dummy";
|
||||
|
||||
sls_receiver_header *header = (sls_receiver_header *)(buf);
|
||||
// send dummy header and data
|
||||
if (!SendHeader(header, 0, 0, 0, true)) {
|
||||
LOG(logERROR) << "Could not send zmq dummy header for streamer "
|
||||
<< index;
|
||||
}
|
||||
|
||||
void DataStreamer::StopProcessing(char* buf) {
|
||||
LOG(logDEBUG1) << "DataStreamer " << index << ": Dummy";
|
||||
|
||||
sls_receiver_header* header = (sls_receiver_header*) (buf);
|
||||
//send dummy header and data
|
||||
if (!SendHeader(header, 0, 0, 0, true)) {
|
||||
LOG(logERROR) << "Could not send zmq dummy header for streamer " << index;
|
||||
}
|
||||
|
||||
fifo->FreeAddress(buf);
|
||||
StopRunning();
|
||||
LOG(logDEBUG1) << index << ": Streaming Completed";
|
||||
fifo->FreeAddress(buf);
|
||||
StopRunning();
|
||||
LOG(logDEBUG1) << index << ": Streaming Completed";
|
||||
}
|
||||
|
||||
/** buf includes only the standard header */
|
||||
void DataStreamer::ProcessAnImage(char* buf) {
|
||||
void DataStreamer::ProcessAnImage(char *buf) {
|
||||
|
||||
sls_receiver_header* header = (sls_receiver_header*) (buf + FIFO_HEADER_NUMBYTES);
|
||||
uint64_t fnum = header->detHeader.frameNumber;
|
||||
LOG(logDEBUG1) << "DataStreamer " << index << ": fnum:" << fnum;
|
||||
sls_receiver_header *header =
|
||||
(sls_receiver_header *)(buf + FIFO_HEADER_NUMBYTES);
|
||||
uint64_t fnum = header->detHeader.frameNumber;
|
||||
LOG(logDEBUG1) << "DataStreamer " << index << ": fnum:" << fnum;
|
||||
|
||||
if (!startedFlag) {
|
||||
RecordFirstIndex(fnum);
|
||||
}
|
||||
if (!startedFlag) {
|
||||
RecordFirstIndex(fnum);
|
||||
}
|
||||
|
||||
//shortframe gotthard
|
||||
if (completeBuffer) {
|
||||
// shortframe gotthard
|
||||
if (completeBuffer) {
|
||||
|
||||
//disregarding the size modified from callback (always using imageSizeComplete
|
||||
// instead of buf (32 bit) because gui needs imagesizecomplete and listener
|
||||
//write imagesize
|
||||
// disregarding the size modified from callback (always using
|
||||
// imageSizeComplete
|
||||
// instead of buf (32 bit) because gui needs imagesizecomplete and
|
||||
// listener
|
||||
// write imagesize
|
||||
|
||||
if (!SendHeader(header, generalData->imageSizeComplete,
|
||||
generalData->nPixelsXComplete, generalData->nPixelsYComplete, false)) {
|
||||
LOG(logERROR) << "Could not send zmq header for fnum " << fnum << " and streamer " << index;
|
||||
}
|
||||
memcpy(completeBuffer + ((generalData->imageSize) * adcConfigured),
|
||||
buf + FIFO_HEADER_NUMBYTES + sizeof(sls_receiver_header),
|
||||
(uint32_t)(*((uint32_t*)buf)) );
|
||||
if (!SendHeader(header, generalData->imageSizeComplete,
|
||||
generalData->nPixelsXComplete,
|
||||
generalData->nPixelsYComplete, false)) {
|
||||
LOG(logERROR) << "Could not send zmq header for fnum " << fnum
|
||||
<< " and streamer " << index;
|
||||
}
|
||||
memcpy(completeBuffer + ((generalData->imageSize) * adcConfigured),
|
||||
buf + FIFO_HEADER_NUMBYTES + sizeof(sls_receiver_header),
|
||||
(uint32_t)(*((uint32_t *)buf)));
|
||||
|
||||
if (!zmqSocket->SendData(completeBuffer, generalData->imageSizeComplete)) {
|
||||
LOG(logERROR) << "Could not send zmq data for fnum " << fnum << " and streamer " << index;
|
||||
}
|
||||
}
|
||||
if (!zmqSocket->SendData(completeBuffer,
|
||||
generalData->imageSizeComplete)) {
|
||||
LOG(logERROR) << "Could not send zmq data for fnum " << fnum
|
||||
<< " and streamer " << index;
|
||||
}
|
||||
}
|
||||
|
||||
// normal
|
||||
else {
|
||||
|
||||
//normal
|
||||
else {
|
||||
|
||||
if (!SendHeader(header, (uint32_t)(*((uint32_t*)buf)),
|
||||
generalData->nPixelsX, generalData->nPixelsY, false)) {// new size possibly from callback
|
||||
LOG(logERROR) << "Could not send zmq header for fnum " << fnum << " and streamer " << index;
|
||||
}
|
||||
if (!zmqSocket->SendData(buf + FIFO_HEADER_NUMBYTES + sizeof(sls_receiver_header),
|
||||
(uint32_t)(*((uint32_t*)buf)) )) {// new size possibly from callback
|
||||
LOG(logERROR) << "Could not send zmq data for fnum " << fnum << " and streamer " << index;
|
||||
}
|
||||
}
|
||||
if (!SendHeader(header, (uint32_t)(*((uint32_t *)buf)),
|
||||
generalData->nPixelsX, generalData->nPixelsY,
|
||||
false)) { // new size possibly from callback
|
||||
LOG(logERROR) << "Could not send zmq header for fnum " << fnum
|
||||
<< " and streamer " << index;
|
||||
}
|
||||
if (!zmqSocket->SendData(
|
||||
buf + FIFO_HEADER_NUMBYTES + sizeof(sls_receiver_header),
|
||||
(uint32_t)(
|
||||
*((uint32_t *)buf)))) { // new size possibly from callback
|
||||
LOG(logERROR) << "Could not send zmq data for fnum " << fnum
|
||||
<< " and streamer " << index;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int DataStreamer::SendHeader(sls_receiver_header *rheader, uint32_t size,
|
||||
uint32_t nx, uint32_t ny, bool dummy) {
|
||||
|
||||
zmqHeader zHeader;
|
||||
zHeader.data = !dummy;
|
||||
zHeader.jsonversion = SLS_DETECTOR_JSON_HEADER_VERSION;
|
||||
|
||||
int DataStreamer::SendHeader(sls_receiver_header* rheader, uint32_t size, uint32_t nx, uint32_t ny, bool dummy) {
|
||||
if (dummy) {
|
||||
return zmqSocket->SendHeader(index, zHeader);
|
||||
}
|
||||
|
||||
zmqHeader zHeader;
|
||||
zHeader.data = !dummy;
|
||||
zHeader.jsonversion = SLS_DETECTOR_JSON_HEADER_VERSION;
|
||||
sls_detector_header header = rheader->detHeader;
|
||||
|
||||
if (dummy) {
|
||||
return zmqSocket->SendHeader(index, zHeader);
|
||||
}
|
||||
uint64_t frameIndex = header.frameNumber - firstIndex;
|
||||
uint64_t acquisitionIndex = header.frameNumber;
|
||||
|
||||
sls_detector_header header = rheader->detHeader;
|
||||
|
||||
uint64_t frameIndex = header.frameNumber - firstIndex;
|
||||
uint64_t acquisitionIndex = header.frameNumber;
|
||||
|
||||
zHeader.dynamicRange = *dynamicRange;
|
||||
zHeader.fileIndex = *fileIndex;
|
||||
zHeader.ndetx = numDet[0];
|
||||
zHeader.ndety = numDet[1];
|
||||
zHeader.dynamicRange = *dynamicRange;
|
||||
zHeader.fileIndex = *fileIndex;
|
||||
zHeader.ndetx = numDet[0];
|
||||
zHeader.ndety = numDet[1];
|
||||
zHeader.npixelsx = nx;
|
||||
zHeader.npixelsy = ny;
|
||||
zHeader.imageSize = size;
|
||||
zHeader.acqIndex = acquisitionIndex;
|
||||
zHeader.frameIndex = frameIndex;
|
||||
zHeader.progress = 100 * ((double)(frameIndex + 1) / (double)(*totalNumFrames));
|
||||
zHeader.fname = fileNametoStream;
|
||||
zHeader.progress =
|
||||
100 * ((double)(frameIndex + 1) / (double)(*totalNumFrames));
|
||||
zHeader.fname = fileNametoStream;
|
||||
zHeader.frameNumber = header.frameNumber;
|
||||
zHeader.expLength = header.expLength;
|
||||
zHeader.packetNumber = header.packetNumber;
|
||||
@ -235,23 +233,22 @@ int DataStreamer::SendHeader(sls_receiver_header* rheader, uint32_t size, uint32
|
||||
zHeader.version = header.version;
|
||||
zHeader.flippedDataX = flippedDataX;
|
||||
zHeader.quad = *quadEnable;
|
||||
zHeader.completeImage = (header.packetNumber < generalData->packetsPerFrame ? false : true);
|
||||
zHeader.completeImage =
|
||||
(header.packetNumber < generalData->packetsPerFrame ? false : true);
|
||||
zHeader.addJsonHeader = additionJsonHeader;
|
||||
|
||||
return zmqSocket->SendHeader(index, zHeader);
|
||||
return zmqSocket->SendHeader(index, zHeader);
|
||||
}
|
||||
|
||||
|
||||
|
||||
void DataStreamer::RestreamStop() {
|
||||
//send dummy header
|
||||
zmqHeader zHeader;
|
||||
zHeader.data = false;
|
||||
zHeader.jsonversion = SLS_DETECTOR_JSON_HEADER_VERSION;
|
||||
int ret = zmqSocket->SendHeader(index, zHeader);
|
||||
if (!ret) {
|
||||
throw sls::RuntimeError("Could not restream Dummy Header via ZMQ for port " + std::to_string(zmqSocket->GetPortNumber()));
|
||||
}
|
||||
// send dummy header
|
||||
zmqHeader zHeader;
|
||||
zHeader.data = false;
|
||||
zHeader.jsonversion = SLS_DETECTOR_JSON_HEADER_VERSION;
|
||||
int ret = zmqSocket->SendHeader(index, zHeader);
|
||||
if (!ret) {
|
||||
throw sls::RuntimeError(
|
||||
"Could not restream Dummy Header via ZMQ for port " +
|
||||
std::to_string(zmqSocket->GetPortNumber()));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user