RawDataSaverConsumerBase.cc

#include "otsdaq/DataManager/RawDataSaverConsumerBase.h"
#include "otsdaq/Macros/BinaryStringMacros.h"
#include "otsdaq/MessageFacility/MessageFacility.h"

#include <unistd.h>
#include <cassert>
#include <iostream>
//#include <string.h> //memcpy
#include <fstream>
#include <sstream>

using namespace ots;

//==============================================================================
RawDataSaverConsumerBase::RawDataSaverConsumerBase(std::string              supervisorApplicationUID,
                                                   std::string              bufferUID,
                                                   std::string              processorUID,
                                                   const ConfigurationTree& theXDAQContextConfigTree,
                                                   const std::string&       configurationPath)
    : WorkLoop(processorUID)
    , DataConsumer(supervisorApplicationUID, bufferUID, processorUID, HighConsumerPriority)
    , Configurable(theXDAQContextConfigTree, configurationPath)
    , filePath_(theXDAQContextConfigTree.getNode(configurationPath).getNode("FilePath").getValue<std::string>())
    , fileRadix_(theXDAQContextConfigTree.getNode(configurationPath).getNode("RadixFileName").getValue<std::string>())
    , maxFileSize_(theXDAQContextConfigTree.getNode(configurationPath).getNode("MaxFileSize").getValue<long>() * 1000000)  // Instead of 2^6=1048576
    , currentSubRunNumber_(0)

{
    // FILE *fp = fopen( "/home/otsdaq/tsave.txt","w");
    // if(fp)fclose(fp);
}

//==============================================================================
RawDataSaverConsumerBase::~RawDataSaverConsumerBase(void) {}

//==============================================================================
void RawDataSaverConsumerBase::startProcessingData(std::string runNumber)
{
    if(runNumber != "")  // If there is no number it means it was paused
    {
        currentSubRunNumber_ = 0;
        openFile(runNumber);
    }
    DataConsumer::startProcessingData(runNumber);
}

//==============================================================================
void RawDataSaverConsumerBase::stopProcessingData(void)
{
    DataConsumer::stopProcessingData();
    closeFile();
}

//==============================================================================
void RawDataSaverConsumerBase::openFile(std::string runNumber)
{
    currentRunNumber_ = runNumber;
    //  std::string fileName =   "Run" + runNumber + "_" + processorUID_ + "_Raw.dat";
    std::stringstream fileName;
    fileName << filePath_ << "/" << fileRadix_ << "_Run" << runNumber;
    // if split file is there then subrunnumber must be set!
    if(maxFileSize_ > 0)
        fileName << "_" << currentSubRunNumber_;
    fileName << "_Raw.dat";
    __CFG_COUT__ << "Saving file: " << fileName.str() << std::endl;
    outFile_.open(fileName.str().c_str(), std::ios::out | std::ios::binary);
    if(!outFile_.is_open())
    {
        __CFG_SS__ << "Can't open file " << fileName.str() << std::endl;
        __CFG_SS_THROW__;
    }

    writeHeader();  // write start of file header
}

//==============================================================================
void RawDataSaverConsumerBase::closeFile(void)
{
    if(outFile_.is_open())
    {
        writeFooter();  // write end of file footer
        outFile_.close();
    }
}

//==============================================================================
void RawDataSaverConsumerBase::save(const std::string& data)
{
    std::ofstream output;

    //  std::string outputPath = "/home/otsdaq/tsave.txt";
    //  if(0)
    //  {
    //      output.open(outputPath, std::ios::out | std::ios::app | std::ios::binary);
    //      output << data;
    //  }
    //  else
    //  {
    //      output.open(outputPath, std::ios::out | std::ios::app);
    //      output << data;
    //
    //      char str[5];
    //      for(unsigned int j=0;j<data.length();++j)
    //      {
    //          sprintf(str,"%2.2x",((unsigned int)data[j]) & ((unsigned int)(0x0FF)));
    //
    //          if(j%64 == 0) std::cout << "SAVE " << j << "\t: 0x\t";
    //          std::cout << str;
    //          if(j%8 == 7) std::cout << " ";
    //          if(j%64 == 63) std::cout << std::endl;
    //      }
    //      std::cout << std::endl;
    //      std::cout << std::endl;
    //  }
    //
    //  if(1)
    //  {
    //      char str[5];
    //      for(unsigned int j=0;j<data.length();++j)
    //      {
    //          sprintf(str,"%2.2x",((unsigned int)data[j]) & ((unsigned int)(0x0FF)));
    //
    //          if(j%64 == 0) std::cout << "SAVE " << j << "\t: 0x\t";
    //          std::cout << str;
    //          if(j%8 == 7) std::cout << " ";
    //          if(j%64 == 63) std::cout << std::endl;
    //      }
    //      std::cout << std::endl;
    //      std::cout << std::endl;
    //  }

    if(maxFileSize_ > 0)
    {
        long length = outFile_.tellp();
        if(length >= maxFileSize_ / 1000)
        {
            closeFile();
            ++currentSubRunNumber_;
            openFile(currentRunNumber_);
        }
    }

    writePacketHeader(data);  // write start of packet header
    outFile_.write((char*)&data[0], data.length());
    writePacketFooter(data);  // write start of packet footer
}

//==============================================================================
bool RawDataSaverConsumerBase::workLoopThread(toolbox::task::WorkLoop* /*workLoop*/)
{
    //__CFG_COUT__ << DataProcessor::processorUID_ << " running, because workloop: "
    //<< WorkLoop::continueWorkLoop_ << std::endl;
    fastRead();
    return WorkLoop::continueWorkLoop_;
}

//==============================================================================
void RawDataSaverConsumerBase::fastRead(void)
{
    //__CFG_COUT__ << processorUID_ << " running!" << std::endl;
    if(DataConsumer::read(dataP_, headerP_) < 0)
    {
        usleep(100);
        return;
    }
    //__CFG_COUTV__(dataP_->length());
    //std::string& buffer = *dataP_;

    //__CFG_COUT__ << "Reading from buffer length " << buffer.length() << " bytes!" <<
    //__E__;
    //__CFG_COUT__ << "Buffer Data: " <<
    // BinaryStringMacros::binaryNumberToHexString(buffer) << __E__;

    save(*dataP_);
    DataConsumer::setReadSubBuffer<std::string, std::map<std::string, std::string> >();
}

//==============================================================================
void RawDataSaverConsumerBase::slowRead(void)
{
    //__CFG_COUT__ << processorUID_ << " running!" << std::endl;
    // This is making a copy!!!
    if(DataConsumer::read(data_, header_) < 0)
    {
        usleep(1000);
        return;
    }
    save(data_);
}