detectors/slsDetectorPackage
25
0
Fork 0
mirror of https://github.com/slsdetectorgroup/slsDetectorPackage.git synced 2025-06-07 18:40:42 +02:00
Code Issues Actions Packages Releases Activity

added REST and Standard implementations

Browse Source
This commit is contained in:
sala 2014-09-09 16:38:27 +02:00
parent 51b610e32a
commit c62594c7ab
5 changed files with 6263 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

295
slsReceiverSoftware/slsReceiver/RestHelper.h Normal file
View File

@ -0,0 +1,295 @@
/**
* @file RestHelper.h
* @author Leonardo Sala <leonardo.sala@psi.ch>
* @date Tue Mar 25 09:28:19 2014
*
* @brief
*
*
*/
#include <Poco/Net/HTTPClientSession.h>
#include <Poco/Net/HTTPRequest.h>
#include <Poco/Net/HTTPResponse.h>
#include <Poco/StreamCopier.h>
#include <Poco/Path.h>
#include <Poco/URI.h>
#include <Poco/Exception.h>
#include <Poco/Timespan.h>
#include "JsonBox/Value.h"
#include <iostream>
#include <sstream>
#include <string>
#include <exception>
#define EIGER_DEBUG
#ifdef EIGER_DEBUG
#define DEBUG(x) do { std::cerr << "[DEBUG] " << x << std::endl; } while (0)
#else
#define DEBUG(x)
#endif
using namespace Poco::Net;
using namespace Poco;
using namespace std;
class RestHelper {
public:
RestHelper(int timeout=10, int n_tries=3){
/**
*
*
* @param timeout default=10
* @param n_tries default=3
*/
http_timeout = timeout;
n_connection_tries = n_tries;
}
~RestHelper(){};
void set_connection_params(int timeout, int n_tries){
http_timeout = timeout;
n_connection_tries = n_tries;
}
void get_connection_params(int *timeout, int *n_tries){
*timeout = http_timeout;
*n_tries = n_connection_tries;
}
void init(string hostname, int port){
/** Initialize the RestHelper. Hostname and port parameters are not supposed to change.
*
*
* @param hostname FQDN of the host to connect to , e.g. www.iamfake.org, or sodoi.org
* @param port
*
* @return
*/
//Check for http:// string
string proto_str = "http://";
if( size_t found = hostname.find(proto_str) != string::npos ){
char c1[hostname.size()-found-1];
size_t length1 = hostname.copy(c1, hostname.size()-found-1, proto_str.size());
c1[length1]='\0';
hostname = c1;
}
full_hostname = "http://"+hostname;
session = new HTTPClientSession(hostname,port );
session->setKeepAliveTimeout( Timespan( http_timeout,0) );
};
void init(string hostname_port){
/** Initialize the RestHelper. Hostname_port parameters are not supposed to change.
*
*
* @param hostname FQDN and port of the host to connect to , e.g. www.iamfake.org:8080, or sodoi.org:1111. Default port is 8080
*
* @return
*/
//Check for http:// string
string proto_str = "http://";
if( size_t found = hostname_port.find(proto_str) != string::npos ){
char c1[hostname_port.size()-found-1];
size_t length1 = hostname_port.copy(c1, hostname_port.size()-found-1, proto_str.size());
c1[length1]='\0';
hostname_port = c1;
}
size_t found = hostname_port.rfind(":");
char c1[ found ], c2[hostname_port.size()-found-1];
string hostname;
size_t length1 = hostname_port.copy(c1, found);
c1[length1]='\0';
hostname = c1;
size_t length2 = hostname_port.copy(c2, found-1, found+1);
c2[length2]='\0';
int port = atoi(c2);
full_hostname = proto_str+hostname;
session = new HTTPClientSession(hostname,port );
session->setKeepAliveTimeout( Timespan( http_timeout,0) );
};
int get_json(string request, string* answer){
/** Retrieves a reply from the RESTful webservice.
*
*
* @param request Request without the hostname, e.g. if the full request would have been http://fake.org/fakemethod, request=fakemethod
* @param answer
*
* @return 0 if successful, -1 if failure happens.
*/
URI * uri = new URI(full_hostname+"/"+request);
string path(uri->getPathAndQuery());
if (path.empty()) path = "/";
// send request
HTTPRequest req(HTTPRequest::HTTP_GET, path, HTTPMessage::HTTP_1_1);
req.setContentType("application/json\r\n");
int code = send_request(session, req, answer);
delete uri;
return code;
};
int get_json(string request, JsonBox::Value* json_value){
/**
*
*
* @param request
* @param json_value
*
* @return
*/
URI *uri = new URI(full_hostname+"/"+request);
string path(uri->getPathAndQuery());
if (path.empty()) path = "/";
// send request
HTTPRequest req(HTTPRequest::HTTP_GET, path, HTTPMessage::HTTP_1_1);
req.setContentType("application/json\r\n");
string answer;
int code = send_request(session, req, &answer);
if(code == 0 ) {
DEBUG("ANSWER " << answer );
json_value->loadFromString(answer);
}
delete uri;
return code;
};
int post_json(string request, string *answer, string request_body=""){
/**
*
*
* @param request
* @param answer
* @param request_body Eventual arguments to the URL, e.g. action=login&name=mammamia
*
* @return
*/
//from: http://stackoverflow.com/questions/1499086/poco-c-net-ssl-how-to-post-https-request
URI *uri = new URI(full_hostname+"/"+request);
string path(uri->getPathAndQuery());
if (path.empty()) path = "/";
HTTPRequest req(HTTPRequest::HTTP_POST, path, HTTPMessage::HTTP_1_1 );
req.setContentType("application/json\r\n");
req.setContentLength( request.length() );
int code = send_request(session, req, answer, request_body);
delete uri;
return code;
}
int post_json(string request, JsonBox::Value* json_value, string request_body=""){
/**
*
*
* @param request
* @param json_value
* @param request_body Eventual arguments to the URL, e.g. action=login&name=mammamia
*
* @return
*/
URI *uri = new URI(full_hostname+"/"+request);
string path(uri->getPathAndQuery());
if (path.empty()) path = "/";
HTTPRequest req(HTTPRequest::HTTP_POST, path, HTTPMessage::HTTP_1_1 );
//this does not work
//req.setContentType("application/json\r\n");
//req.setContentLength( request.length() );
string answer;
int code = send_request(session, req, &answer, request_body);
if(code==0){
json_value->loadFromString(answer);
}
delete uri;
return code;
}
private:
HTTPClientSession *session;
string full_hostname;
/// HTTP timeout in seconds, default is 8
int http_timeout;
/// Number of connection tries
int n_connection_tries;
int send_request(HTTPClientSession *session, HTTPRequest &req, string *answer, string request_body=""){
/**
*
*
* @param session
* @param req
* @param answer
* @param request_body
*
* @return
*/
int n=0;
int code = -1;
while(n<n_connection_tries){
req.setContentType("application/json");
//without this you need to tell the lenght: http://pocoproject.org/forum/viewtopic.php?f=12&t=5741&p=10019&hilit=post+json#p10019
// request.setContentLength(my_string.length());
req.setChunkedTransferEncoding(true);
try {
//istringstream rs(request_body);
//req.read(rs);
//cout << " --- " << rs << endl;
if (request_body == "")
session->sendRequest( (req) );
else{
cout << request_body << endl;
ostream &os = session->sendRequest( req ) ;
os << request_body;
}
HTTPResponse res;
istream &is = session->receiveResponse(res);
StreamCopier::copyToString(is, *answer);
code = res.getStatus();
if (code != 200){
cout << "HTTP ERROR " << res.getStatus() << ": " << res.getReason() << endl;
code = -1;
}
else
code = 0;
return code;
}
catch (exception& e){
cout << "Exception connecting to "<< full_hostname << ": "<< e.what() << ", sleeping 5 seconds (" << n << "/"<<n_connection_tries << ")" << endl;
sleep(5);
}
n+=1;
}
return code;
}
};

2009
slsReceiverSoftware/slsReceiver/UDPRESTImplementation.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

819
slsReceiverSoftware/slsReceiver/UDPRESTImplementation.h Normal file
View File

@ -0,0 +1,819 @@
//#ifdef REST
#ifndef UDP_REST_IMPLEMENTATION_H
#define UDP_REST_IMPLEMENTATION_H
/********************************************//**
* @file UDPRESTImplementation.h
* @short does all the functions for a receiver, set/get parameters, start/stop etc.
***********************************************/
#include "sls_receiver_defs.h"
#include "receiver_defs.h"
#include "genericSocket.h"
#include "circularFifo.h"
#include "singlePhotonDetector.h"
#include "slsReceiverData.h"
#include "moenchCommonMode.h"
#include "UDPBaseImplementation.h"
#ifdef MYROOT1
#include <TTree.h>
#include <TFile.h>
#endif
#include "RestHelper.h"
#include <string.h>
#include <pthread.h>
#include <stdio.h>
#include <semaphore.h>
/**
* @short does all the functions for a receiver, set/get parameters, start/stop etc.
*/
class UDPRESTImplementation : private virtual slsReceiverDefs, public UDPBaseImplementation {
public:
/**
* Constructor
*/
UDPRESTImplementation();
/**
* Destructor
*/
virtual ~UDPRESTImplementation();
/**
* delete and free member parameters
*/
void deleteMembers();
/**
* initialize member parameters
*/
void initializeMembers();
/**
* Set receiver type
* @param det detector type
* Returns success or FAIL
*/
int setDetectorType(detectorType det);
//Frame indices and numbers caught
/**
* Returns current Frame Index Caught for an entire acquisition (including all scans)
*/
uint32_t getAcquisitionIndex();
/**
* Returns if acquisition started
*/
bool getAcquistionStarted();
/**
* Returns Frames Caught for each real time acquisition (eg. for each scan)
*/
int getFramesCaught();
/**
* Returns Total Frames Caught for an entire acquisition (including all scans)
*/
int getTotalFramesCaught();
/**
* Returns the frame index at start of each real time acquisition (eg. for each scan)
*/
uint32_t getStartFrameIndex();
/**
* Returns current Frame Index for each real time acquisition (eg. for each scan)
*/
uint32_t getFrameIndex();
/**
* Returns if measurement started
*/
bool getMeasurementStarted();
/**
* Resets the Total Frames Caught
* This is how the receiver differentiates between entire acquisitions
* Returns 0
*/
void resetTotalFramesCaught();
//file parameters
/**
* Returns File Path
*/
//char* getFilePath() const;
/**
* Set File Path
* @param c file path
*/
//char* setFilePath(const char c[]);
/**
* Returns File Name
*/
//char* getFileName() const;
/**
* Set File Name (without frame index, file index and extension)
* @param c file name
*/
//char* setFileName(const char c[]);
/**
* Returns File Index
*/
int getFileIndex();
/**
* Set File Index
* @param i file index
*/
int setFileIndex(int i);
/**
* Set Frame Index Needed
* @param i frame index needed
*/
int setFrameIndexNeeded(int i);
/**
* Set enable file write
* @param i file write enable
* Returns file write enable
*/
//int setEnableFileWrite(int i);
/**
* Enable/disable overwrite
* @param i enable
* Returns enable over write
*/
//int setEnableOverwrite(int i);
/**
* Returns file write enable
* 1: YES 0: NO
*/
//int getEnableFileWrite() const;
/**
* Returns file over write enable
* 1: YES 0: NO
*/
//int getEnableOverwrite() const;
//other parameters
/**
* abort acquisition with minimum damage: close open files, cleanup.
* does nothing if state already is 'idle'
*/
void abort() {};
/**
* Returns status of receiver: idle, running or error
*/
runStatus getStatus() const;
/**
* Set detector hostname
* @param c hostname
*/
void initialize(const char *detectorHostName);
/* Returns detector hostname
/returns hostname
* caller needs to deallocate the returned char array.
* if uninitialized, it must return NULL
*/
//char *getDetectorHostname() const;
/**
* Set Ethernet Interface or IP to listen to
*/
void setEthernetInterface(char* c);
/**
* Set UDP Port Number
*/
void setUDPPortNo(int p);
/*
* Returns number of frames to receive
* This is the number of frames to expect to receiver from the detector.
* The data receiver will change from running to idle when it got this number of frames
*/
//int getNumberOfFrames() const;
/**
* set frame number if a positive number
*/
//int32_t setNumberOfFrames(int32_t fnum);
/**
* Returns scan tag
*/
//int getScanTag() const;
/**
* set scan tag if its is a positive number
*/
//int32_t setScanTag(int32_t stag);
/**
* Returns the number of bits per pixel
*/
//int getDynamicRange() const;
/**
* set dynamic range if its is a positive number
*/
int32_t setDynamicRange(int32_t dr);
/**
* Set short frame
* @param i if shortframe i=1
*/
int setShortFrame(int i);
/**
* Set the variable to send every nth frame to gui
* or if 0,send frame only upon gui request
*/
int setNFrameToGui(int i);
/** set acquisition period if a positive number
*/
int64_t setAcquisitionPeriod(int64_t index);
/** get data compression, by saving only hits
*/
bool getDataCompression();
/** enabl data compression, by saving only hits
/returns if failed
*/
int enableDataCompression(bool enable);
/**
* enable 10Gbe
@param enable 1 for 10Gbe or 0 for 1 Gbe, -1 to read out
\returns enable for 10Gbe
*/
int enableTenGiga(int enable = -1);
//other functions
/**
* Returns the buffer-current frame read by receiver
* @param c pointer to current file name
* @param raw address of pointer, pointing to current frame to send to gui
* @param fnum frame number for eiger as it is not in the packet
*/
void readFrame(char* c,char** raw, uint32_t &fnum);
/**
* Closes all files
* @param ithr thread index
*/
void closeFile(int ithr = -1);
/**
* Starts Receiver - starts to listen for packets
* @param message is the error message if there is an error
* Returns success
*/
int startReceiver(char message[]);
/**
* Stops Receiver - stops listening for packets
* Returns success
*/
int stopReceiver();
/** set status to transmitting and
* when fifo is empty later, sets status to run_finished
*/
void startReadout();
/**
* shuts down the udp sockets
* \returns if success or fail
*/
int shutDownUDPSockets();
private:
/*
void not_implemented(string method_name){
std::cout << "[WARNING] Method " << method_name << " not implemented!" << std::endl;
};
*/
/**
* Deletes all the filter objects for single photon data
*/
void deleteFilter();
/**
* Constructs the filter for single photon data
*/
void setupFilter();
/**
* set up fifo according to the new numjobsperthread
*/
void setupFifoStructure ();
/**
* Copy frames to gui
* uses semaphore for nth frame mode
*/
void copyFrameToGui(char* startbuf[], uint32_t fnum=-1, char* buf=NULL);
/**
* creates udp sockets
* \returns if success or fail
*/
int createUDPSockets();
/**
* create listening thread
* @param destroy is true to kill all threads and start again
*/
int createListeningThreads(bool destroy = false);
/**
* create writer threads
* @param destroy is true to kill all threads and start again
*/
int createWriterThreads(bool destroy = false);
/**
* set thread priorities
*/
void setThreadPriorities();
/**
* initializes variables and creates the first file
* also does the startAcquisitionCallBack
* \returns FAIL or OK
*/
int setupWriter();
/**
* Creates new tree and file for compression
* @param ithr thread number
* @param iframe frame number
*\returns OK for succces or FAIL for failure
*/
int createCompressionFile(int ithr, int iframe);
/**
* Creates new file
*\returns OK for succces or FAIL for failure
*/
int createNewFile();
/**
* Static function - Thread started which listens to packets.
* Called by startReceiver()
* @param this_pointer pointer to this object
*/
static void* startListeningThread(void *this_pointer);
/**
* Static function - Thread started which writes packets to file.
* Called by startReceiver()
* @param this_pointer pointer to this object
*/
static void* startWritingThread(void *this_pointer);
/**
* Thread started which listens to packets.
* Called by startReceiver()
*
*/
int startListening();
/**
* Thread started which writes packets to file.
* Called by startReceiver()
*
*/
int startWriting();
/**
* Writing to file without compression
* @param buf is the address of buffer popped out of fifo
* @param numpackets is the number of packets
* @param framenum current frame number
*/
void writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum);
/**
* Its called for the first packet of a scan or acquistion
* Sets the startframeindices and the variables to know if acquisition started
* @param ithread listening thread number
*/
void startFrameIndices(int ithread);
/**
* This is called when udp socket is shut down
* It pops ffff instead of packet number into fifo
* to inform writers about the end of listening session
* @param ithread listening thread number
* @param rc number of bytes received
* @param pc packet count
* @param t total packets listened to
*/
void stopListening(int ithread, int rc, int &pc, int &t);
/**
* When acquisition is over, this is called
* @param ithread listening thread number
* @param wbuffer writer buffer
*/
void stopWriting(int ithread, char* wbuffer[]);
/**
* data compression for each fifo output
* @param ithread listening thread number
* @param wbuffer writer buffer
* @param npackets number of packets from the fifo
* @param data pointer to the next packet start
* @param xmax max pixels in x direction
* @param ymax max pixels in y direction
* @param nf nf
*/
void handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf);
/** structure of an eiger image header*/
typedef struct
{
unsigned char header_before[20];
unsigned char fnum[4];
unsigned char header_after[24];
} eiger_image_header;
/** structure of an eiger image header*/
typedef struct
{
unsigned char num1[4];
unsigned char num2[4];
} eiger_packet_header;
/** max number of listening threads */
const static int MAX_NUM_LISTENING_THREADS = EIGER_MAX_PORTS;
/** max number of writer threads */
const static int MAX_NUM_WRITER_THREADS = 15;
/** detector type */
detectorType myDetectorType;
/** detector hostname */
char detHostname[MAX_STR_LENGTH];
/** status of receiver */
runStatus status;
/** UDP Socket between Receiver and Detector */
genericSocket* udpSocket[MAX_NUM_LISTENING_THREADS];
/** Server UDP Port*/
int server_port[MAX_NUM_LISTENING_THREADS];
/** ethernet interface or IP to listen to */
char *eth;
/** max packets per file **/
int maxPacketsPerFile;
/** File write enable */
int enableFileWrite;
/** File over write enable */
int overwrite;
/** Complete File name */
char savefilename[MAX_STR_LENGTH];
/** File Name without frame index, file index and extension*/
char fileName[MAX_STR_LENGTH];
/** File Path */
char filePath[MAX_STR_LENGTH];
/** File Index */
int fileIndex;
/** scan tag */
int scanTag;
/** if frame index required in file name */
int frameIndexNeeded;
/* Acquisition started */
bool acqStarted;
/* Measurement started */
bool measurementStarted;
/** Frame index at start of each real time acquisition (eg. for each scan) */
uint32_t startFrameIndex;
/** Actual current frame index of each time acquisition (eg. for each scan) */
uint32_t frameIndex;
/** Frames Caught for each real time acquisition (eg. for each scan) */
int packetsCaught;
/** Total packets caught for an entire acquisition (including all scans) */
int totalPacketsCaught;
/** Pckets currently in current file, starts new file when it reaches max */
int packetsInFile;
/** Frame index at start of an entire acquisition (including all scans) */
uint32_t startAcquisitionIndex;
/** Actual current frame index of an entire acquisition (including all scans) */
uint32_t acquisitionIndex;
/** number of packets per frame*/
int packetsPerFrame;
/** frame index mask */
uint32_t frameIndexMask;
/** packet index mask */
uint32_t packetIndexMask;
/** frame index offset */
int frameIndexOffset;
/** acquisition period */
int64_t acquisitionPeriod;
/** frame number */
int32_t numberOfFrames;
/** dynamic range */
int dynamicRange;
/** short frames */
int shortFrame;
/** current frame number */
uint32_t currframenum;
/** Previous Frame number from buffer */
uint32_t prevframenum;
/** size of one frame */
int frameSize;
/** buffer size. different from framesize as we wait for one packet instead of frame for eiger */
int bufferSize;
/** oen buffer size */
int onePacketSize;
/** latest data */
char* latestData;
/** gui data ready */
int guiDataReady;
/** points to the data to send to gui */
char* guiData;
/** points to the filename to send to gui */
char* guiFileName;
/** temporary number for eiger frame number as its not included in the packet */
uint32_t guiFrameNumber;
/** send every nth frame to gui or only upon gui request*/
int nFrameToGui;
/** fifo size */
unsigned int fifosize;
/** number of jobs per thread for data compression */
int numJobsPerThread;
/** datacompression - save only hits */
bool dataCompression;
/** memory allocated for the buffer */
char *mem0[MAX_NUM_LISTENING_THREADS];
/** circular fifo to store addresses of data read */
CircularFifo<char>* fifo[MAX_NUM_LISTENING_THREADS];
/** circular fifo to store addresses of data already written and ready to be resued*/
CircularFifo<char>* fifoFree[MAX_NUM_LISTENING_THREADS];
/** Receiver buffer */
char *buffer[MAX_NUM_LISTENING_THREADS];
/** number of writer threads */
int numListeningThreads;
/** number of writer threads */
int numWriterThreads;
/** to know if listening and writer threads created properly */
int thread_started;
/** current listening thread index*/
int currentListeningThreadIndex;
/** current writer thread index*/
int currentWriterThreadIndex;
/** thread listening to packets */
pthread_t listening_thread[MAX_NUM_LISTENING_THREADS];
/** thread writing packets */
pthread_t writing_thread[MAX_NUM_WRITER_THREADS];
/** total frame count the listening thread has listened to */
int totalListeningFrameCount[MAX_NUM_LISTENING_THREADS];
/** mask showing which listening threads are running */
volatile uint32_t listeningthreads_mask;
/** mask showing which writer threads are running */
volatile uint32_t writerthreads_mask;
/** mask showing which threads have created files*/
volatile uint32_t createfile_mask;
/** OK if file created was successful */
int ret_createfile;
/** variable used to self terminate threads waiting for semaphores */
int killAllListeningThreads;
/** variable used to self terminate threads waiting for semaphores */
int killAllWritingThreads;
/** 10Gbe enable*/
int tengigaEnable;
//semaphores
/** semaphore to synchronize writer and guireader threads */
sem_t smp;
/** semaphore to synchronize listener threads */
sem_t listensmp[MAX_NUM_LISTENING_THREADS];
/** semaphore to synchronize writer threads */
sem_t writersmp[MAX_NUM_WRITER_THREADS];
//mutex
/** guiDataReady mutex */
pthread_mutex_t dataReadyMutex;
/** mutex for status */
pthread_mutex_t status_mutex;
/** mutex for progress variable currframenum */
pthread_mutex_t progress_mutex;
/** mutex for writing data to file */
pthread_mutex_t write_mutex;
/** File Descriptor */
FILE *sfilefd;
//filter
singlePhotonDetector<uint16_t> *singlePhotonDet[MAX_NUM_WRITER_THREADS];
slsReceiverData<uint16_t> *receiverdata[MAX_NUM_WRITER_THREADS];
moenchCommonMode *cmSub;
bool commonModeSubtractionEnable;
#ifdef MYROOT1
/** Tree where the hits are stored */
TTree *myTree[MAX_NUM_WRITER_THREADS];
/** File where the tree is saved */
TFile *myFile[MAX_NUM_WRITER_THREADS];
#endif
/**
callback arguments are
filepath
filename
fileindex
data size
return value is
0 callback takes care of open,close,write file
1 callback writes file, we have to open, close it
2 we open, close, write file, callback does not do anything
*/
int (*startAcquisitionCallBack)(char*, char*,int, int, void*);
void *pStartAcquisition;
/**
args to acquisition finished callback
total frames caught
*/
void (*acquisitionFinishedCallBack)(int, void*);
void *pAcquisitionFinished;
/**
args to raw data ready callback are
framenum
datapointer
datasize in bytes
file descriptor
guidatapointer (NULL, no data required)
*/
void (*rawDataReadyCallBack)(int, char*, int, FILE*, char*, void*);
void *pRawDataReady;
/** The action which decides what the user and default responsibilites to save data are
* 0 raw data ready callback takes care of open,close,write file
* 1 callback writes file, we have to open, close it
* 2 we open, close, write file, callback does not do anything */
int cbAction;
public:
/**
callback arguments are
filepath
filename
fileindex
datasize
return value is
0 callback takes care of open,close,wrie file
1 callback writes file, we have to open, close it
2 we open, close, write file, callback does not do anything
*/
void registerCallBackStartAcquisition(int (*func)(char*, char*,int, int, void*),void *arg){startAcquisitionCallBack=func; pStartAcquisition=arg;};
/**
callback argument is
toatal frames caught
*/
void registerCallBackAcquisitionFinished(void (*func)(int, void*),void *arg){acquisitionFinishedCallBack=func; pAcquisitionFinished=arg;};
/**
args to raw data ready callback are
framenum
datapointer
datasize in bytes
file descriptor
guidatapointer (NULL, no data required)
*/
void registerCallBackRawDataReady(void (*func)(int, char*, int, FILE*, char*, void*),void *arg){rawDataReadyCallBack=func; pRawDataReady=arg;};
//REST specific
bool isInitialized;
RestHelper * rest ;
};
#endif
//#endif /*REST*/

2330
slsReceiverSoftware/slsReceiver/UDPStandardImplementation.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

810
slsReceiverSoftware/slsReceiver/UDPStandardImplementation.h Normal file
View File

@ -0,0 +1,810 @@
//#ifdef UDP_BASE_IMPLEMENTATION
#ifndef UDP_STANDARD_IMPLEMENTATION_H
#define UDP_STANDARD_IMPLEMENTATION_H
/********************************************//**
* @file UDPBaseImplementation.h
* @short does all the functions for a receiver, set/get parameters, start/stop etc.
***********************************************/
#include "sls_receiver_defs.h"
#include "receiver_defs.h"
#include "genericSocket.h"
#include "circularFifo.h"
#include "singlePhotonDetector.h"
#include "slsReceiverData.h"
#include "moenchCommonMode.h"
//#include "UDPInterface.h"
#include "UDPBaseImplementation.h"
#ifdef MYROOT1
#include <TTree.h>
#include <TFile.h>
#endif
#include <string.h>
#include <pthread.h>
#include <stdio.h>
#include <semaphore.h>
/**
* @short does all the functions for a receiver, set/get parameters, start/stop etc.
*/
class UDPStandardImplementation: private virtual slsReceiverDefs, public UDPBaseImplementation {
public:
/**
* Constructor
*/
UDPStandardImplementation();
/**
* Destructor
*/
virtual ~UDPStandardImplementation();
/**
* delete and free member parameters
*/
void deleteMembers();
/**
* initialize member parameters
*/
void initializeMembers();
/**
* Set receiver type
* @param det detector type
* Returns success or FAIL
*/
int setDetectorType(detectorType det);
//Frame indices and numbers caught
/**
* Returns current Frame Index Caught for an entire acquisition (including all scans)
*/
//uint32_t getAcquisitionIndex();
/**
* Returns if acquisition started
*/
//bool getAcquistionStarted();
/**
* Returns Frames Caught for each real time acquisition (eg. for each scan)
*/
//int getFramesCaught();
/**
* Returns Total Frames Caught for an entire acquisition (including all scans)
*/
//int getTotalFramesCaught();
/**
* Returns the frame index at start of each real time acquisition (eg. for each scan)
*/
//uint32_t getStartFrameIndex();
/**
* Returns current Frame Index for each real time acquisition (eg. for each scan)
*/
//uint32_t getFrameIndex();
/**
* Returns if measurement started
*/
//bool getMeasurementStarted();
/**
* Resets the Total Frames Caught
* This is how the receiver differentiates between entire acquisitions
* Returns 0
*/
//void resetTotalFramesCaught();
//file parameters
/**
* Returns File Path
*/
//char* getFilePath() const;
/**
* Set File Path
* @param c file path
*/
//char* setFilePath(const char c[]);
/**
* Returns File Name
*/
//char* getFileName() const;
/**
* Set File Name (without frame index, file index and extension)
* @param c file name
*/
//char* setFileName(const char c[]);
/**
* Returns File Index
*/
//int getFileIndex();
/**
* Set File Index
* @param i file index
*/
//int setFileIndex(int i);
/**
* Set Frame Index Needed
* @param i frame index needed
*/
//int setFrameIndexNeeded(int i);
/**
* Set enable file write
* @param i file write enable
* Returns file write enable
*/
//int setEnableFileWrite(int i);
/**
* Enable/disable overwrite
* @param i enable
* Returns enable over write
*/
//int setEnableOverwrite(int i);
/**
* Returns file write enable
* 1: YES 0: NO
*/
//int getEnableFileWrite() const;
/**
* Returns file over write enable
* 1: YES 0: NO
*/
//int getEnableOverwrite() const;
//other parameters
/**
* abort acquisition with minimum damage: close open files, cleanup.
* does nothing if state already is 'idle'
*/
void abort() {};
/**
* Returns status of receiver: idle, running or error
*/
runStatus getStatus() const;
/**
* Set detector hostname
* @param c hostname
*/
void initialize(const char *detectorHostName);
/* Returns detector hostname
/returns hostname
* caller needs to deallocate the returned char array.
* if uninitialized, it must return NULL
*/
char *getDetectorHostname() const;
/**
* Set Ethernet Interface or IP to listen to
*/
void setEthernetInterface(char* c);
/**
* Set UDP Port Number
*/
void setUDPPortNo(int p);
/*
* Returns number of frames to receive
* This is the number of frames to expect to receiver from the detector.
* The data receiver will change from running to idle when it got this number of frames
*/
int getNumberOfFrames() const;
/**
* set frame number if a positive number
*/
int32_t setNumberOfFrames(int32_t fnum);
/**
* Returns scan tag
*/
int getScanTag() const;
/**
* set scan tag if its is a positive number
*/
int32_t setScanTag(int32_t stag);
/**
* Returns the number of bits per pixel
*/
int getDynamicRange() const;
/**
* set dynamic range if its is a positive number
*/
int32_t setDynamicRange(int32_t dr);
/**
* Set short frame
* @param i if shortframe i=1
*/
int setShortFrame(int i);
/**
* Set the variable to send every nth frame to gui
* or if 0,send frame only upon gui request
*/
int setNFrameToGui(int i);
/** set acquisition period if a positive number
*/
int64_t setAcquisitionPeriod(int64_t index);
/** get data compression, by saving only hits
*/
bool getDataCompression();
/** enabl data compression, by saving only hits
/returns if failed
*/
int enableDataCompression(bool enable);
/**
* enable 10Gbe
@param enable 1 for 10Gbe or 0 for 1 Gbe, -1 to read out
\returns enable for 10Gbe
*/
int enableTenGiga(int enable = -1);
//other functions
/**
* Returns the buffer-current frame read by receiver
* @param c pointer to current file name
* @param raw address of pointer, pointing to current frame to send to gui
* @param fnum frame number for eiger as it is not in the packet
*/
void readFrame(char* c,char** raw, uint32_t &fnum);
/**
* Closes all files
* @param ithr thread index
*/
void closeFile(int ithr = -1);
/**
* Starts Receiver - starts to listen for packets
* @param message is the error message if there is an error
* Returns success
*/
int startReceiver(char message[]);
/**
* Stops Receiver - stops listening for packets
* Returns success
*/
int stopReceiver();
/** set status to transmitting and
* when fifo is empty later, sets status to run_finished
*/
void startReadout();
/**
* shuts down the udp sockets
* \returns if success or fail
*/
int shutDownUDPSockets();
private:
/*
void not_implemented(string method_name){
std::cout << "[WARNING] Method " << method_name << " not implemented!" << std::endl;
};
*/
/**
* Deletes all the filter objects for single photon data
*/
void deleteFilter();
/**
* Constructs the filter for single photon data
*/
void setupFilter();
/**
* set up fifo according to the new numjobsperthread
*/
void setupFifoStructure ();
/**
* Copy frames to gui
* uses semaphore for nth frame mode
*/
void copyFrameToGui(char* startbuf[], uint32_t fnum=-1, char* buf=NULL);
/**
* creates udp sockets
* \returns if success or fail
*/
int createUDPSockets();
/**
* create listening thread
* @param destroy is true to kill all threads and start again
*/
int createListeningThreads(bool destroy = false);
/**
* create writer threads
* @param destroy is true to kill all threads and start again
*/
int createWriterThreads(bool destroy = false);
/**
* set thread priorities
*/
void setThreadPriorities();
/**
* initializes variables and creates the first file
* also does the startAcquisitionCallBack
* \returns FAIL or OK
*/
int setupWriter();
/**
* Creates new tree and file for compression
* @param ithr thread number
* @param iframe frame number
*\returns OK for succces or FAIL for failure
*/
int createCompressionFile(int ithr, int iframe);
/**
* Creates new file
*\returns OK for succces or FAIL for failure
*/
int createNewFile();
/**
* Static function - Thread started which listens to packets.
* Called by startReceiver()
* @param this_pointer pointer to this object
*/
static void* startListeningThread(void *this_pointer);
/**
* Static function - Thread started which writes packets to file.
* Called by startReceiver()
* @param this_pointer pointer to this object
*/
static void* startWritingThread(void *this_pointer);
/**
* Thread started which listens to packets.
* Called by startReceiver()
*
*/
int startListening();
/**
* Thread started which writes packets to file.
* Called by startReceiver()
*
*/
int startWriting();
/**
* Writing to file without compression
* @param buf is the address of buffer popped out of fifo
* @param numpackets is the number of packets
* @param framenum current frame number
*/
void writeToFile_withoutCompression(char* buf,int numpackets, uint32_t framenum);
/**
* Its called for the first packet of a scan or acquistion
* Sets the startframeindices and the variables to know if acquisition started
* @param ithread listening thread number
*/
void startFrameIndices(int ithread);
/**
* This is called when udp socket is shut down
* It pops ffff instead of packet number into fifo
* to inform writers about the end of listening session
* @param ithread listening thread number
* @param rc number of bytes received
* @param pc packet count
* @param t total packets listened to
*/
void stopListening(int ithread, int rc, int &pc, int &t);
/**
* When acquisition is over, this is called
* @param ithread listening thread number
* @param wbuffer writer buffer
*/
void stopWriting(int ithread, char* wbuffer[]);
/**
* data compression for each fifo output
* @param ithread listening thread number
* @param wbuffer writer buffer
* @param npackets number of packets from the fifo
* @param data pointer to the next packet start
* @param xmax max pixels in x direction
* @param ymax max pixels in y direction
* @param nf nf
*/
void handleDataCompression(int ithread, char* wbuffer[], int &npackets, char* data, int xmax, int ymax, int &nf);
/** structure of an eiger image header*/
typedef struct
{
unsigned char header_before[20];
unsigned char fnum[4];
unsigned char header_after[24];
} eiger_image_header;
/** structure of an eiger image header*/
typedef struct
{
unsigned char num1[4];
unsigned char num2[4];
} eiger_packet_header;
/** max number of listening threads */
const static int MAX_NUM_LISTENING_THREADS = EIGER_MAX_PORTS;
/** max number of writer threads */
const static int MAX_NUM_WRITER_THREADS = 15;
/** detector type */
detectorType myDetectorType;
/** detector hostname */
char detHostname[MAX_STR_LENGTH];
/** status of receiver */
runStatus status;
/** UDP Socket between Receiver and Detector */
genericSocket* udpSocket[MAX_NUM_LISTENING_THREADS];
/** Server UDP Port*/
int server_port[MAX_NUM_LISTENING_THREADS];
/** ethernet interface or IP to listen to */
char *eth;
/** max packets per file **/
int maxPacketsPerFile;
/** File write enable */
int enableFileWrite;
/** File over write enable */
int overwrite;
/** Complete File name */
char savefilename[MAX_STR_LENGTH];
/** File Name without frame index, file index and extension*/
char fileName[MAX_STR_LENGTH];
/** File Path */
char filePath[MAX_STR_LENGTH];
/** File Index */
int fileIndex;
/** scan tag */
int scanTag;
/** if frame index required in file name */
int frameIndexNeeded;
/* Acquisition started */
bool acqStarted;
/* Measurement started */
bool measurementStarted;
/** Frame index at start of each real time acquisition (eg. for each scan) */
uint32_t startFrameIndex;
/** Actual current frame index of each time acquisition (eg. for each scan) */
uint32_t frameIndex;
/** Frames Caught for each real time acquisition (eg. for each scan) */
int packetsCaught;
/** Total packets caught for an entire acquisition (including all scans) */
int totalPacketsCaught;
/** Pckets currently in current file, starts new file when it reaches max */
int packetsInFile;
/** Frame index at start of an entire acquisition (including all scans) */
uint32_t startAcquisitionIndex;
/** Actual current frame index of an entire acquisition (including all scans) */
uint32_t acquisitionIndex;
/** number of packets per frame*/
int packetsPerFrame;
/** frame index mask */
uint32_t frameIndexMask;
/** packet index mask */
uint32_t packetIndexMask;
/** frame index offset */
int frameIndexOffset;
/** acquisition period */
int64_t acquisitionPeriod;
/** frame number */
int32_t numberOfFrames;
/** dynamic range */
int dynamicRange;
/** short frames */
int shortFrame;
/** current frame number */
uint32_t currframenum;
/** Previous Frame number from buffer */
uint32_t prevframenum;
/** size of one frame */
int frameSize;
/** buffer size. different from framesize as we wait for one packet instead of frame for eiger */
int bufferSize;
/** oen buffer size */
int onePacketSize;
/** latest data */
char* latestData;
/** gui data ready */
int guiDataReady;
/** points to the data to send to gui */
char* guiData;
/** points to the filename to send to gui */
char* guiFileName;
/** temporary number for eiger frame number as its not included in the packet */
uint32_t guiFrameNumber;
/** send every nth frame to gui or only upon gui request*/
int nFrameToGui;
/** fifo size */
unsigned int fifosize;
/** number of jobs per thread for data compression */
int numJobsPerThread;
/** datacompression - save only hits */
bool dataCompression;
/** memory allocated for the buffer */
char *mem0[MAX_NUM_LISTENING_THREADS];
/** circular fifo to store addresses of data read */
CircularFifo<char>* fifo[MAX_NUM_LISTENING_THREADS];
/** circular fifo to store addresses of data already written and ready to be resued*/
CircularFifo<char>* fifoFree[MAX_NUM_LISTENING_THREADS];
/** Receiver buffer */
char *buffer[MAX_NUM_LISTENING_THREADS];
/** number of writer threads */
int numListeningThreads;
/** number of writer threads */
int numWriterThreads;
/** to know if listening and writer threads created properly */
int thread_started;
/** current listening thread index*/
int currentListeningThreadIndex;
/** current writer thread index*/
int currentWriterThreadIndex;
/** thread listening to packets */
pthread_t listening_thread[MAX_NUM_LISTENING_THREADS];
/** thread writing packets */
pthread_t writing_thread[MAX_NUM_WRITER_THREADS];
/** total frame count the listening thread has listened to */
int totalListeningFrameCount[MAX_NUM_LISTENING_THREADS];
/** mask showing which listening threads are running */
volatile uint32_t listeningthreads_mask;
/** mask showing which writer threads are running */
volatile uint32_t writerthreads_mask;
/** mask showing which threads have created files*/
volatile uint32_t createfile_mask;
/** OK if file created was successful */
int ret_createfile;
/** variable used to self terminate threads waiting for semaphores */
int killAllListeningThreads;
/** variable used to self terminate threads waiting for semaphores */
int killAllWritingThreads;
/** 10Gbe enable*/
int tengigaEnable;
//semaphores
/** semaphore to synchronize writer and guireader threads */
sem_t smp;
/** semaphore to synchronize listener threads */
sem_t listensmp[MAX_NUM_LISTENING_THREADS];
/** semaphore to synchronize writer threads */
sem_t writersmp[MAX_NUM_WRITER_THREADS];
//mutex
/** guiDataReady mutex */
pthread_mutex_t dataReadyMutex;
/** mutex for status */
pthread_mutex_t status_mutex;
/** mutex for progress variable currframenum */
pthread_mutex_t progress_mutex;
/** mutex for writing data to file */
pthread_mutex_t write_mutex;
/** File Descriptor */
FILE *sfilefd;
//filter
singlePhotonDetector<uint16_t> *singlePhotonDet[MAX_NUM_WRITER_THREADS];
slsReceiverData<uint16_t> *receiverdata[MAX_NUM_WRITER_THREADS];
moenchCommonMode *cmSub;
bool commonModeSubtractionEnable;
#ifdef MYROOT1
/** Tree where the hits are stored */
TTree *myTree[MAX_NUM_WRITER_THREADS];
/** File where the tree is saved */
TFile *myFile[MAX_NUM_WRITER_THREADS];
#endif
/**
callback arguments are
filepath
filename
fileindex
data size
return value is
0 callback takes care of open,close,write file
1 callback writes file, we have to open, close it
2 we open, close, write file, callback does not do anything
*/
int (*startAcquisitionCallBack)(char*, char*,int, int, void*);
void *pStartAcquisition;
/**
args to acquisition finished callback
total frames caught
*/
void (*acquisitionFinishedCallBack)(int, void*);
void *pAcquisitionFinished;
/**
args to raw data ready callback are
framenum
datapointer
datasize in bytes
file descriptor
guidatapointer (NULL, no data required)
*/
void (*rawDataReadyCallBack)(int, char*, int, FILE*, char*, void*);
void *pRawDataReady;
/** The action which decides what the user and default responsibilites to save data are
* 0 raw data ready callback takes care of open,close,write file
* 1 callback writes file, we have to open, close it
* 2 we open, close, write file, callback does not do anything */
int cbAction;
public:
/**
callback arguments are
filepath
filename
fileindex
datasize
return value is
0 callback takes care of open,close,wrie file
1 callback writes file, we have to open, close it
2 we open, close, write file, callback does not do anything
*/
void registerCallBackStartAcquisition(int (*func)(char*, char*,int, int, void*),void *arg){startAcquisitionCallBack=func; pStartAcquisition=arg;};
/**
callback argument is
toatal frames caught
*/
void registerCallBackAcquisitionFinished(void (*func)(int, void*),void *arg){acquisitionFinishedCallBack=func; pAcquisitionFinished=arg;};
/**
args to raw data ready callback are
framenum
datapointer
datasize in bytes
file descriptor
guidatapointer (NULL, no data required)
*/
void registerCallBackRawDataReady(void (*func)(int, char*, int, FILE*, char*, void*),void *arg){rawDataReadyCallBack=func; pRawDataReady=arg;};
};
#endif
//#endif