#include <fstream>
#include <H5Cpp.h>
#include <H5File.h>
#include <boost/filesystem.hpp>
#include "MicroDAQ.h"
namespace ChimeraTK {
/*********************************************************************************************************************/
namespace detail {
/** Callable class for use with boost::fusion::for_each: Attach the given
* accessor to the MicroDAQ with proper handling of the UserType. */
template<typename TRIGGERTYPE>
struct AccessorAttacher {
AccessorAttacher(VariableNetworkNode& feeder, MicroDAQ<TRIGGERTYPE>* owner, const std::string& name)
: _feeder(feeder), _owner(owner), _name(name) {}
template<typename PAIR>
void operator()(PAIR&) const {
// only continue if the call is for the right type
if(typeid(typename PAIR::first_type) != _feeder.getValueType()) return;
// register connection
_feeder >> _owner->template getAccessor<typename PAIR::first_type>(_name);
}
VariableNetworkNode& _feeder;
MicroDAQ<TRIGGERTYPE>* _owner;
const std::string& _name;
};
} // namespace detail
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
void MicroDAQ<TRIGGERTYPE>::addSource(const Module& source, const RegisterPath& namePrefix) {
// for simplification, first create a VirtualModule containing the correct
// hierarchy structure (obeying eliminate hierarchy etc.)
auto dynamicModel = source.findTag(".*"); /// @todo use virtualise() instead
// create variable group map for namePrefix if needed
if(groupMap.find(namePrefix) == groupMap.end()) {
// search for existing parent (if any)
auto parentPrefix = namePrefix;
while(groupMap.find(parentPrefix) == groupMap.end()) {
if(parentPrefix == "/") break; // no existing parent found
parentPrefix = std::string(parentPrefix).substr(0, std::string(parentPrefix).find_last_of("/"));
}
// create all not-yet-existing parents
while(parentPrefix != namePrefix) {
EntityOwner* owner = this;
if(parentPrefix != "/") owner = &groupMap[parentPrefix];
auto stop = std::string(namePrefix).find_first_of("/", parentPrefix.length() + 1);
if(stop == std::string::npos) stop = namePrefix.length();
RegisterPath name = std::string(namePrefix).substr(parentPrefix.length(), stop - parentPrefix.length());
parentPrefix /= name;
groupMap[parentPrefix] = VariableGroup(owner, std::string(name).substr(1), "");
}
}
// add all accessors on this hierarchy level
for(auto& acc : dynamicModel.getAccessorList()) {
boost::fusion::for_each(
accessorListMap.table, detail::AccessorAttacher<TRIGGERTYPE>(acc, this, namePrefix / acc.getName()));
}
// recurse into submodules
for(auto mod : dynamicModel.getSubmoduleList()) {
addSource(*mod, namePrefix / mod->getName());
}
}
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
void MicroDAQ<TRIGGERTYPE>::addSource(const DeviceModule& source, const RegisterPath& namePrefix) {
auto mod = source.virtualiseFromCatalog();
addSource(mod, namePrefix);
}
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
template<typename UserType>
VariableNetworkNode MicroDAQ<TRIGGERTYPE>::getAccessor(const std::string& variableName) {
// check if variable name already registered
for(auto& name : overallVariableList) {
if(name == variableName) {
throw ChimeraTK::logic_error("Cannot add '" + variableName +
"' to MicroDAQ since a variable with that "
"name is already registered.");
}
}
overallVariableList.push_back(variableName);
// add accessor and name to lists
auto& tmpAccessorList = boost::fusion::at_key<UserType>(accessorListMap.table);
auto& nameList = boost::fusion::at_key<UserType>(nameListMap.table);
auto dirName = variableName.substr(0, variableName.find_last_of("/"));
auto baseName = variableName.substr(variableName.find_last_of("/") + 1);
tmpAccessorList.emplace_back(&groupMap[dirName], baseName, "", 0, "");
nameList.push_back(variableName);
// return the accessor
return tmpAccessorList.back();
}
/*********************************************************************************************************************/
namespace detail {
template<typename TRIGGERTYPE>
struct H5storage {
H5storage(MicroDAQ<TRIGGERTYPE>* owner) : _owner(owner) {}
H5::H5File outFile;
std::string currentGroupName;
/** Unique list of groups, used to create the groups in the file */
std::list<std::string> groupList;
/** boost::fusion::map of UserTypes to std::lists containing the H5::DataSpace
* objects. */
template<typename UserType>
using dataSpaceList = std::list<H5::DataSpace>;
TemplateUserTypeMap<dataSpaceList> dataSpaceListMap;
/** boost::fusion::map of UserTypes to std::lists containing decimation
* factors. */
template<typename UserType>
using decimationFactorList = std::list<size_t>;
TemplateUserTypeMap<decimationFactorList> decimationFactorListMap;
uint32_t currentBuffer{0};
uint32_t nFillsInBuffer{0};
bool isOpened{false};
bool firstTrigger{true};
void processTrigger();
void writeData();
MicroDAQ<TRIGGERTYPE>* _owner;
};
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
struct DataSpaceCreator {
DataSpaceCreator(H5storage<TRIGGERTYPE>& storage) : _storage(storage) {}
template<typename PAIR>
void operator()(PAIR& pair) const {
typedef typename PAIR::first_type UserType;
// get the lists for the UserType
auto& accessorList = pair.second;
auto& decimationFactorList = boost::fusion::at_key<UserType>(_storage.decimationFactorListMap.table);
auto& dataSpaceList = boost::fusion::at_key<UserType>(_storage.dataSpaceListMap.table);
auto& nameList = boost::fusion::at_key<UserType>(_storage._owner->nameListMap.table);
// iterate through all accessors for this UserType
auto name = nameList.begin();
for(auto accessor = accessorList.begin(); accessor != accessorList.end(); ++accessor, ++name) {
// determine decimation factor
int factor = 1;
if(accessor->getNElements() > _storage._owner->decimationThreshold_) {
factor = _storage._owner->decimationFactor_;
}
decimationFactorList.push_back(factor);
// define data space
hsize_t dimsf[1]; // dataset dimensions
dimsf[0] = accessor->getNElements() / factor;
dataSpaceList.push_back(H5::DataSpace(1, dimsf));
// put all group names in list (each hierarchy level separately)
size_t idx = 0;
while((idx = name->find('/', idx + 1)) != std::string::npos) {
std::string groupName = name->substr(0, idx);
_storage.groupList.push_back(groupName);
}
}
}
H5storage<TRIGGERTYPE>& _storage;
};
} // namespace detail
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
void MicroDAQ<TRIGGERTYPE>::mainLoop() {
std::cout << "Initialising MicroDAQ system...";
// storage object
detail::H5storage<TRIGGERTYPE> storage(this);
// create the data spaces
boost::fusion::for_each(accessorListMap.table, detail::DataSpaceCreator<TRIGGERTYPE>(storage));
// sort group list and make unique to make sure lower levels get created first
storage.groupList.sort();
storage.groupList.unique();
// loop: process incoming triggers
while(true) {
trigger.read();
storage.processTrigger();
}
std::cout << " done." << std::endl;
}
/*********************************************************************************************************************/
namespace detail {
template<typename TRIGGERTYPE>
void H5storage<TRIGGERTYPE>::processTrigger() {
// update configuration variables
_owner->enable.readLatest();
_owner->nMaxFiles.readLatest();
_owner->nTriggersPerFile.readLatest();
// need to open or close file?
if(!isOpened && _owner->enable != 0) {
std::fstream bufferNumber;
// some things to be done only on first trigger
if(firstTrigger) {
// create sub-directory
boost::filesystem::create_directory("uDAQ");
// determine current buffer number
bufferNumber.open("uDAQ/currentBuffer", std::ofstream::in);
bufferNumber.seekg(0);
if(!bufferNumber.eof()) {
bufferNumber >> currentBuffer;
char filename[64];
std::sprintf(filename, "uDAQ/data%04d.h5", currentBuffer);
if(boost::filesystem::exists(filename) && boost::filesystem::file_size(filename) > 1000) currentBuffer++;
if(currentBuffer >= _owner->nMaxFiles) currentBuffer = 0;
}
else {
currentBuffer = 0;
}
bufferNumber.close();
}
// store current buffer number to disk
char filename[64];
std::sprintf(filename, "uDAQ/data%04d.h5", currentBuffer);
std::cout << "uDAQ: Starting with file: " << filename << std::endl;
bufferNumber.open("uDAQ/currentBuffer", std::ofstream::out);
bufferNumber << currentBuffer << std::endl;
bufferNumber.close();
// update file number process variables
_owner->currentFile = currentBuffer;
_owner->currentFile.write();
// open file
try {
outFile = H5::H5File(filename, H5F_ACC_TRUNC);
}
catch(H5::FileIException&) {
return;
}
isOpened = true;
}
else if(isOpened && _owner->enable == 0) {
outFile.close();
isOpened = false;
}
// if file is opened, this trigger should be included in the DAQ
if(isOpened) {
// write data
writeData();
// increment counter, after nTriggersPerFile triggers written to the same
// file, switch the file
nFillsInBuffer++;
if(nFillsInBuffer > _owner->nTriggersPerFile) {
// increment file number. use at most nMaxFiles files, overwrite old files
currentBuffer++;
if(currentBuffer >= _owner->nMaxFiles) currentBuffer = 0;
nFillsInBuffer = 0;
// just close the file here, will re-open on next trigger
outFile.close();
isOpened = false;
}
}
}
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
struct DataWriter {
DataWriter(detail::H5storage<TRIGGERTYPE>& storage) : _storage(storage) {}
template<typename PAIR>
void operator()(PAIR& pair) const {
typedef typename PAIR::first_type UserType;
// get the lists for the UserType
auto& accessorList = pair.second;
auto& decimationFactorList = boost::fusion::at_key<UserType>(_storage.decimationFactorListMap.table);
auto& dataSpaceList = boost::fusion::at_key<UserType>(_storage.dataSpaceListMap.table);
auto& nameList = boost::fusion::at_key<UserType>(_storage._owner->nameListMap.table);
// iterate through all accessors for this UserType
auto decimationFactor = decimationFactorList.begin();
auto dataSpace = dataSpaceList.begin();
auto name = nameList.begin();
for(auto accessor = accessorList.begin(); accessor != accessorList.end();
++accessor, ++decimationFactor, ++dataSpace, ++name) {
// form full path name of data set
std::string dataSetName = _storage.currentGroupName + "/" + *name;
// write to file (this is mainly a function call to allow template
// specialisations at this point)
try {
write2hdf<UserType>(*accessor, dataSetName, *decimationFactor, *dataSpace);
}
catch(H5::FileIException&) {
std::cout << "MicroDAQ: ERROR writing data set " << dataSetName << std::endl;
throw;
}
}
}
template<typename UserType>
void write2hdf(ArrayPollInput<UserType>& accessor, std::string& name, size_t decimationFactor,
H5::DataSpace& dataSpace) const;
H5storage<TRIGGERTYPE>& _storage;
};
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
template<typename UserType>
void DataWriter<TRIGGERTYPE>::write2hdf(ArrayPollInput<UserType>& accessor, std::string& dataSetName,
size_t decimationFactor, H5::DataSpace& dataSpace) const {
// prepare decimated buffer
size_t n = accessor.getNElements() / decimationFactor;
std::vector<float> buffer(n);
for(size_t i = 0; i < n; ++i) {
buffer[i] = userTypeToNumeric<float>(accessor[i * decimationFactor]);
}
// write data from internal buffer to data set in HDF5 file
H5::DataSet dataset = _storage.outFile.createDataSet(dataSetName, H5::PredType::NATIVE_FLOAT, dataSpace);
dataset.write(buffer.data(), H5::PredType::NATIVE_FLOAT);
}
/*********************************************************************************************************************/
template<typename TRIGGERTYPE>
void H5storage<TRIGGERTYPE>::writeData() {
// format current time
struct timeval tv;
gettimeofday(&tv, nullptr);
time_t t = tv.tv_sec;
if(t == 0) t = time(nullptr);
struct tm* tmp = localtime(&t);
char timeString[64];
std::sprintf(timeString, "%04d-%02d-%02d %02d:%02d:%02d.%03d", 1900 + tmp->tm_year, tmp->tm_mon + 1, tmp->tm_mday,
tmp->tm_hour, tmp->tm_min, tmp->tm_sec, static_cast<int>(tv.tv_usec / 1000));
// create groups
currentGroupName = std::string("/") + std::string(timeString);
try {
outFile.createGroup(currentGroupName);
for(auto& group : groupList) outFile.createGroup(currentGroupName + "/" + group);
}
catch(H5::FileIException&) {
outFile.close();
isOpened = false; // will re-open file on next trigger
return;
}
// read all input data
_owner->readAllLatest();
// write all data to file
boost::fusion::for_each(_owner->accessorListMap.table, DataWriter<TRIGGERTYPE>(*this));
}
/*********************************************************************************************************************/
} // namespace detail
INSTANTIATE_TEMPLATE_FOR_CHIMERATK_USER_TYPES(MicroDAQ);
} // namespace ChimeraTK