JCLB.cc

Go to the documentation of this file.

#include <string>
#include <iostream>
#include <iomanip>
#include <limits>
#include <vector>
#include <map>
#include <algorithm>
 
#include "DataQueue/DataFormats/clb_common_header.hh"
 
#include "km3net-dataformat/online/JDAQHit.hh"
#include "km3net-dataformat/online/JDAQSuperFrame.hh"
#include "km3net-dataformat/online/JDAQTimeslice.hh"
 
#include "JDAQ/JDAQTimesliceIO.hh"
 
#include "JSupport/JFileRecorder.hh"
#include "JSupport/JSupport.hh"
#include "JSupport/JMeta.hh"
 
#include "Jeep/JParser.hh"
#include "Jeep/JMessage.hh"
 
 
namespace {
 
  const uint32_t TDC  =  1414808643;           //!< TDC data type
  
  using KM3NETDAQ::JDAQTimeslice;
 
  typedef std::vector<char>                    buffer_type;        //!< UDP packet
  typedef std::vector<buffer_type>             collection_type;    //!< collection of UDP packets
  typedef std::map<int, collection_type>       timeslice_type;     //!< module identifier -> collection of UDP packets
 
 
  /**
   * Auxiliary data structure to organise CLB data.
   */
  struct map_type :
    public std::map<uint64_t, timeslice_type>                      // UTC time -> time slice data
  {
    /**
     * Constructor.
     *
     * \param  detector           detector identifier
     */
    map_type(int detector) :
      detector(detector),
      frame_index(0)
    {}
 
 
    /**
     * Convert first element in container and remove it.
     *
     * \return                    time slice
     */
    const JDAQTimeslice& pop()
    {
      using namespace std;
      using namespace KM3NETDAQ;
 
      frame_index += 1;
 
      timeslice.clear();
 
      for (timeslice_type::iterator collection = this->begin()->second.begin(); collection != this->begin()->second.end(); ++collection) {
 
        sort(collection->second.begin(), collection->second.end(), compare);
      
        const CLBCommonHeader* header = (const CLBCommonHeader*) collection->second.rbegin()->data();
 
        const JDAQChronometer chronometer(detector, 
                                          header->runNumber(), 
                                          frame_index, 
                                          JDAQUTCExtended(header->timeStamp().sec(), header->timeStamp().tics()));
 
        if (timeslice.empty()) {
          timeslice = JDAQTimeslice(chronometer);
        }
 
        const uint32_t number_of_packets  =  collection->second.size();
        const uint32_t sequence_number    =  header->udpSequenceNumber();
 
        const JDAQFrameStatus status(DAQ_UDP_RECEIVED_PACKETS.write(number_of_packets) |                 //
                                     DAQ_UDP_SEQUENCE_NUMBER .write(sequence_number),                    // DAQ  status
                                     header->domStatus(1),                                               // TDC  status
                                     header->domStatus(2) |                                              // 
                                     DAQ_UDP_TRAILER.write(sequence_number + 1 == number_of_packets));   // FIFO status
 
        JDAQSuperFrame frame(JDAQSuperFrameHeader(chronometer, collection->first, status));
 
        for (collection_type::const_iterator i = collection->second.begin(); i != collection->second.end(); ++i) {
 
          const size_t   size = (i->size() - sizeof(CLBCommonHeader)) / sizeof(JDAQHit);
          const JDAQHit* data = (const JDAQHit*) (i->data() + sizeof(CLBCommonHeader));
          
          frame.add(size, data);
        }
 
        timeslice.push_back(frame);
      }
 
      this->erase(this->begin());
 
      return timeslice;
    }
 
 
    /**
     * Compare two frames by UDP sequence number.
     *
     * \param  first              first  UDP packet
     * \param  second             second UDP packet
     * \return                    true if first packet before second; else false
     */
    static inline bool compare(const buffer_type& first,const buffer_type& second) 
    {
      return (((const CLBCommonHeader*) first .data())->udpSequenceNumber() <
              ((const CLBCommonHeader*) second.data())->udpSequenceNumber());
    }
 
  private:
    int           detector;
    int           frame_index;
    JDAQTimeslice timeslice;
  };
}
 
 
/**
 * \file
 *
 * Auxiliary program to convert raw CLB data to KM3NETDAQ::JDAQTimeslice data.
 * \author mdejong
 */
int main(int argc, char **argv)
{
  using namespace std;
  using namespace JPP;
  using namespace KM3NETDAQ;
 
 
  string             inputFile;
  int                numberOfEvents;
  JFileRecorder<JTYPELIST<JDAQTimeslice, JMeta>::typelist> outputFile;
  int                detector;
  size_t             queue;
  int                debug;
 
  try {
 
    JParser<> zap("Auxiliary program to convert raw CLB data to KM3NETDAQ::JDAQTimeslice data.");
 
    zap['f'] = make_field(inputFile,         "input file).");
    zap['o'] = make_field(outputFile,        "output file.")            = "clb.root";
    zap['n'] = make_field(numberOfEvents)                               = numeric_limits<int>::max();
    zap['Q'] = make_field(queue,             "queue depth")             = 1000;
    zap['D'] = make_field(detector,          "detector identifier")     = 1;
    zap['d'] = make_field(debug,             "debug.")                  = 1;
 
    zap(argc, argv);
  }
  catch(const exception& error) {
    FATAL(error.what() << endl);
  }
 
 
  outputFile.open();
 
  if (!outputFile.is_open()) {
    FATAL("Error opening file " << outputFile << endl);
  }
 
  outputFile.put(JMeta(argc, argv));
 
  map_type data(detector);
 
  uint32_t run  = numeric_limits<uint32_t>::max();
  uint32_t size;
 
  ifstream in(inputFile.c_str(), ios::binary);
 
  for (int count = 0; count != numberOfEvents && in.read((char*) &size, sizeof(uint32_t)); ++count) {
 
    STATUS(setw(8) << count << '\r'); DEBUG(endl);
 
    buffer_type buffer(size);
 
    if (in.read(buffer.data(), size)) {
 
      if (size >= sizeof(CLBCommonHeader)) {
 
        const CLBCommonHeader* header = (const CLBCommonHeader*) buffer.data();
 
        if (run == numeric_limits<uint32_t>::max()) {
          run = header->runNumber();
        }
 
        DEBUG("data: "
              << setw(8) << header->runNumber()         << ' '
              << setw(9) << header->domIdentifier()     << ' ' 
              << setw(2) << header->udpSequenceNumber() << endl);
 
        if (header->dataType() == TDC) {
      
          if (header->runNumber() == run) {
 
            data[header->timeStamp().inMilliSeconds()][header->domIdentifier()].push_back(buffer);
 
            while (data.size() > queue) {
              outputFile.put(data.pop());
            }
 
          } else {
 
            WARNING("Run numbers differ " << run << ' ' << header->runNumber() << " -> skip data." << endl);
          }
        }
      }
    }
  }
  STATUS(endl);
 
  in.close();
 
  while (!data.empty()) {
    outputFile.put(data.pop());
  }
 
  outputFile.close();
}