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();
}