Files
slsDetectorPackage/slsReceiverSoftware/src/DataProcessor.h
T
maliakal_d c3ad3d2e73
Build and Deploy on local RHEL9 / build (push) Successful in 2m13s
Build on RHEL9 docker image / build (push) Successful in 3m53s
Build and Deploy on local RHEL8 / build (push) Successful in 4m58s
Build on RHEL8 docker image / build (push) Successful in 5m47s
Run Simulator Tests on local RHEL9 / build (push) Successful in 18m55s
Run Simulator Tests on local RHEL8 / build (push) Successful in 22m26s
Dev/rx disable datastream port (#1448)
- changed 
   - command from `datastream` to `udp_datastream` to be more specific
   - Detector API:
      - setNumberofUDPInterfaces(pos) =>setNumberofUDPInterfaces() # no position anymore
      - getDataStream(portPosition, pos) =>getUDPDataStream(portPosition, pos)
      - setDataStream(portPosition, pos) => setUDPDataStream(portPosition) # no position anymore
      
   - modified the dataprocessing thread: 
      - if gui or call back, start zmq processing (connecting sockets) before starting receiver (disabled ports sends dummy that the client zmq sockets are not ready for)
      - if only progress, starting progress processing thread after startReceiver
      -


- added
   - Detector API:
      - getRxDisabledUDPPortIndices()
      - getPortPositionList()  
   - allow disabling/enabling udp interface in receiver for Jungfrua/Moench when both udp interfaces is enabled  (This is to go fast, where only one half of the module matters)
   - write ports type and disabled ports to master file (tests for it as well)
   - tests for python commands as well
   - datastream command to deprecated commands
   - refactored tests to make the new parameters fit

* first draft of disabling data port at the receiver side. Todo: master file and stream for gui to use

* updated help in command

* formatting

* md5 unchanged

* fix test, check port position for detector type

* wip, udp port enable metadata in client, to:receiver, zmq streaming

* wip, master attributes

* wip, imple=> save only disabled ports with port index, 1 interface, is empty

* works in file. still needs refactoring

* updated writer versions

* udp ports type pass, disabled with api yet

* fixed disabled ports meta data as well

* fixed in developer via other PR

* renamed.more intuitive

* fix in gui that wont wait for disabled ports

* added tests for master file

* extra tests

* move the master fiel tests into its own tests with the disable marker for github tests

* refactoring

* python test

* more intuitive and doc

* release notes

* example for python

* back to before

* minor documentation fix and sending nports unnecessarily for getRxUDPPortDisableMetadata

* doc

* review fixes, 1. using std::array instead of vector, 2. moving udp_datastream command testing with different values into master attributes testing file because we test the master file

* minor

* review: update help

* progress printed before assigned

* try catch inside hdf5

* auto generated code for CLI and formatting

* missing packets should not be checked at github workflow level

* formatting

* improved python doc for udp_datastream help

* indent doc python done
2026-07-13 14:58:25 +02:00

221 lines
7.5 KiB
C++

// SPDX-License-Identifier: LGPL-3.0-or-other
// Copyright (C) 2021 Contributors to the SLS Detector Package
#pragma once
/************************************************
* @file DataProcessor.h
* @short creates data processor thread that
* pulls pointers to memory addresses from fifos
* and processes data stored in them & writes them to file
***********************************************/
/**
*@short creates & manages a data processor thread each
*/
#include "ThreadObject.h"
#include "receiver_defs.h"
#include <atomic>
#include <filesystem>
#include <mutex>
#include <vector>
namespace sls {
class GeneralData;
class Fifo;
class File;
class DataStreamer;
class MasterAttributes;
class DataProcessor : private virtual slsDetectorDefs, public ThreadObject {
public:
DataProcessor(int index);
~DataProcessor() override;
bool GetStartedFlag() const;
void SetFifo(Fifo *f);
void SetGeneralData(GeneralData *generalData);
void SetUdpPortNumber(const uint16_t portNumber);
void SetActivate(bool enable);
void SetPortROI(const ROI arg);
void setMultiROIMetadata(const std::vector<slsDetectorDefs::ROI> &args);
void SetDataStreamEnable(bool enable);
void SetStreamingFrequency(uint32_t value);
void SetStreamingTimerInMs(uint32_t value);
void SetStreamingStartFnum(uint32_t value);
void SetFramePadding(bool enable);
void SetQuadEnable(bool value);
void SetFlipRows(bool fd);
void SetNumberofTotalFrames(uint64_t value);
void
SetAdditionalJsonHeader(const std::map<std::string, std::string> &json);
void ResetParametersforNewAcquisition();
void CloseFiles();
void DeleteFiles();
void SetupFileWriter(const bool filewriteEnable,
const fileFormat fileFormatType,
std::mutex *hdf5LibMutex);
void CreateFirstFiles(const std::filesystem::path &filePath,
const std::string &fileNamePrefix,
const uint64_t fileIndex, const bool overWriteEnable,
const bool silentMode, const bool udpDataStream);
#ifdef HDF5C
uint32_t GetFilesInAcquisition() const;
std::string CreateVirtualFile(const std::filesystem::path &filePath,
const std::string &fileNamePrefix,
const uint64_t fileIndex,
const bool overWriteEnable,
const bool silentMode, const int modulePos,
const int numModX, const int numModY,
std::mutex *hdf5LibMutex, bool gotthard25um);
void LinkFileInMaster(const std::string &masterFileName,
const std::string &virtualFileName,
const bool silentMode, std::mutex *hdf5LibMutex);
#endif
std::string CreateMasterFile(const std::filesystem::path &filePath,
const std::string &fileNamePrefix,
const uint64_t fileIndex,
const bool overWriteEnable, bool silentMode,
const fileFormat fileFormatType,
MasterAttributes *attr,
std::mutex *hdf5LibMutex);
/** params: sls_receiver_header, pointer to data, image size */
void registerCallBackRawDataReady(void (*func)(sls_receiver_header &,
dataCallbackHeader, char *,
size_t &, void *),
void *arg);
protected:
/**
* Align corresponding digital bits together (CTB only if ctbDbitlist is not
* empty)
* set variable reorder to true if data should be rearranged such that
* it groups each signal (0-63) from all the different samples together
*/
void ArrangeDbitData(size_t &size, char *data);
/**
* remove trailing bits in digital data stream
*/
void RemoveTrailingBits(size_t &size, char *data);
private:
void RecordFirstIndex(uint64_t fnum);
/**
* Thread Exeution for DataProcessor Class
* Pop bound addresses, process them,
* write to file if needed & free the address
*/
void ThreadExecution() override;
/**
* Frees dummy buffer,
* reset running mask by calling StopRunning()
*/
void StopProcessing(char *buf);
/**
* Process an image popped from fifo,
* write to file if fw enabled & update parameters
*/
void ProcessAnImage(sls_receiver_header &header, size_t &size,
size_t &firstImageIndex, char *data);
/**
* Calls CheckTimer and CheckCount for streaming frequency and timer
* and determines if the current image should be sent to streamer
* @returns true if it should to streamer, else false
*/
bool SendToStreamer();
/**
* This function should be called only in random frequency mode
* Checks if timer is done and ready to send to stream
* @returns true if ready to send to stream, else false
*/
bool CheckTimer();
/**
* This function should be called only in non random frequency mode
* Checks if count is done and ready to send to stream
* @returns true if ready to send to stream, else false
*/
bool CheckCount();
void PadMissingPackets(sls_receiver_header header, char *data);
void CropImage(size_t &size, char *data);
static const std::string typeName;
GeneralData *generalData{nullptr};
Fifo *fifo;
uint16_t udpPortNumber{0};
bool dataStreamEnable;
bool activated{false};
ROI portRoi{};
bool isPartiallyInRoi{false};
bool isOutsideRoi{false};
std::vector<ROI> multiRoiMetadata{};
std::unique_ptr<char[]> completeImageToStreamBeforeCropping;
/** if 0, sending random images with a timer */
uint32_t streamingFrequency;
uint32_t streamingTimerInMs;
uint32_t streamingStartFnum;
uint32_t currentFreqCount{0};
struct timespec timerbegin {};
bool framePadding;
std::atomic<bool> startedFlag{false};
std::atomic<uint64_t> firstIndex{0};
bool quadEnable{false};
bool flipRows{false};
uint64_t nTotalFrames{0};
std::map<std::string, std::string> additionalJsonHeader;
/** Used by streamer thread to update local copy (reduce number of locks
* during streaming) */
std::atomic<bool> isAdditionalJsonUpdated{false};
/** mutex to update json and to read and update local copy */
mutable std::mutex additionalJsonMutex;
/** local copy of additional json header (it can be update on the fly) */
std::map<std::string, std::string> localAdditionalJsonHeader;
// for statistics
uint64_t numFramesCaught{0};
/** Frame Number of latest processed frame number */
std::atomic<uint64_t> currentFrameIndex{0};
/** first streamer frame to add frame index in fifo header */
bool firstStreamerFrame{false};
bool streamCurrentFrame{false};
File *dataFile{nullptr};
// call back
/**
* Call back for raw data
* args to raw data ready callback are
* sls_receiver_header frame metadata
* dataPointer is the pointer to the data
* dataSize in bytes is the size of the data in bytes.
*/
void (*rawDataReadyCallBack)(sls_receiver_header &, dataCallbackHeader,
char *, size_t &, void *) = nullptr;
void *pRawDataReady{nullptr};
};
} // namespace sls