JSupernovaMonitor.cc

Go to the documentation of this file.

#include <iostream>
#include <iomanip>
#include <queue>
#include <string>
 
#include "JDAQ/JDAQTimesliceIO.hh"
#include "JDAQ/JDAQEventIO.hh"
#include "JDAQ/JDAQSummarysliceIO.hh"
#include "JDAQ/JDAQTags.hh"
 
#include "JDetector/JDetector.hh"
#include "JDetector/JDetectorToolkit.hh"
 
#include "JNet/JControlHostObjectIterator.hh"
 
#include "Jeep/JParser.hh"
#include "Jeep/JMessage.hh"
 
#include "JROOT/JROOTClassSelector.hh"
 
#include "JSupport/JSupport.hh"
 
#include "JSupernova.hh"
 
/**
 * \file
 *
 * Online supernova monitor
 * \author mlincett
 */
int main(int argc, char* argv[]) {
  using namespace std;
  using namespace JPP;
  using namespace KM3NETDAQ;
  
  string    controlhost;
  string    ligier;
  JTimeval  timeout_us;
  int       numberOfTimeouts;
  int       debug;
  int       queueLength;
  int       windowLength;
  int       statPrintInterval_s;
 
  string detectorFile;
  JROOTClassSelector selector;
  string summaryFile;
 
  int       TMax_ns;
  int       preTriggerThreshold;
  JRange<int> M;
  double TVeto_ns;
 
  try {
 
    JParser<> zap("Example program to test receiving of objects from ControlHost server.");
 
    zap['H'] = make_field(controlhost, "CH server (input)")         = "localhost";
    zap['L'] = make_field(ligier, "Ligier server (output)")         = "";
    zap['t'] = make_field(timeout_us)       = 100000;
    zap['n'] = make_field(numberOfTimeouts) = 100;
    zap['a'] = make_field(detectorFile);
    zap['C'] = make_field(selector)         = getROOTClassSelection<JDAQTimesliceTypes_t>();
    zap['Q'] = make_field(queueLength,  "number of timeslices of trigger queue")          = 100;
    zap['W'] = make_field(windowLength, "number of timeslices of trigger sliding window") = 5;
    zap['T'] = make_field(TMax_ns, "coincidence time window [ns]") = 10;
    zap['M'] = make_field(M, "multiplicity range for SN coincidences") = JRange<int>(6,10);
    zap['S'] = make_field(preTriggerThreshold, "muon veto multiplicity threshold") = 4;
    zap['V'] = make_field(TVeto_ns, "muon veto time interval") = 1000;
    zap['s'] = make_field(summaryFile, "summary output file");
    zap['P'] = make_field(statPrintInterval_s, "statistics & file print interval [s]") = 30;
    zap['d'] = make_field(debug)            = 1;
 
    zap(argc, argv);
 
  } 
 
  catch(const exception &error) {
    FATAL(error.what() << endl);
  }
 
  if (queueLength < windowLength) {
    FATAL("Length of the trigger window must be smaller than the queue.");
  }
 
 
  setDAQLongprint(debug >= JEEP::debug_t);
 
  using namespace JSUPERNOVA;
 
  // -------------------------------
  // load detector and build routers
  // -------------------------------
 
  JDetector detector;
 
  try {
    load(detectorFile, detector);
  }
  catch(const JException& error) {
    FATAL(error);
  }
 
  const JModuleRouter moduleRouter(detector);
 
  const int DETID = detector.getID();
 
  const int detectorSize = detector.size();
 
  // -------------------------------
  // initialize processing queue
  // -------------------------------
 
  typedef JTriggerSN trigger_type; 
 
  typedef priority_queue<trigger_type, vector<trigger_type>, greater<trigger_type> > queue_type;
 
  typedef deque<trigger_type> window_type; 
 
  typedef map<int, map<int, double> > rates_type;  
 
  queue_type  trgQueue;
  window_type trgWindow;
  rates_type  moduleRates;
 
  JTriggerSNStats stats(detectorSize);
 
  long int counter_live_ts = 0;
  long int counter_lost_ts = 0;
 
  double frameTime_s = getFrameTime() / 1.0e9;
 
  // -------------------------------
  // main processing
  // -------------------------------
  
  try {
    
    // setup input
 
    typedef JDAQTimesliceSN  data_type;
    typedef JDAQSummaryslice summary_type;
 
    JControlHostObjectIterator<data_type>    in(controlhost, timeout_us, true);
    JControlHostObjectIterator<summary_type> sm(controlhost, 1000.0, true);
 
    // setup output
    JControlHost* out = NULL;
 
    if (ligier != "") {
      out = new JControlHost(ligier);
      out->MyId(argv[0]); // pid
    }
 
    const string outputTag = "SNT";
 
    // setup state
 
    int RUN = 0;
 
    for (int i = 0; i != numberOfTimeouts; ) {
 
      if (in.hasNext()) {
 
        data_type* timeslice = in.next();
 
        DEBUG(timeslice->getDAQHeader() << endl);
        
        // -----------------
        // run change reset
        // -----------------
 
        const int r = timeslice->getRunNumber();
 
        if (r != RUN) {
 
          if (RUN != 0) {
 
            // a run change has been detected
 
            NOTICE("RUN CHANGE" << endl);
 
            while (trgQueue.size() > 0) { trgQueue.pop(); }
 
            trgWindow.clear();  
 
            moduleRates.clear();
 
          }
 
          RUN = r;
          
        }
 
        while ( sm.hasNext() ) {
 
          JDAQSummaryslice* summary = sm.next();
 
          int index = summary->getFrameIndex();
 
          for (JDAQSummaryslice::const_iterator summary_frame = summary->begin(); summary_frame != summary->end(); ++summary_frame) {
 
            int DOMID = summary_frame->getModuleID();
 
            for (int ipmt = 0 ; ipmt < NUMBER_OF_PMTS ; ipmt++) {
              moduleRates[index][DOMID] += summary_frame->getRate(ipmt, 1.0/1000);
            }
          }
        }
 
        // -----------------
        // process timeslice
        // -----------------
 
        JDataSN preTrigger(TMax_ns, preTriggerThreshold);
 
        preTrigger(timeslice, moduleRouter);
 
        JTriggerSN trigger(M, TVeto_ns);
 
        trigger(preTrigger);
 
        trgQueue.push(trigger);
 
        //----------------
        // compute trigger
        //----------------
 
        if (trgQueue.size() >= (unsigned) queueLength) {
          
          while (trgWindow.size() <= (unsigned) windowLength) {
 
            trigger_type pending = trgQueue.top();
              
            if ( trgWindow.size() == 0 || pending > trgWindow.back() ) {
              
              trgWindow.push_back( pending );
 
              counter_live_ts++;
 
            } else {
              // latecoming (out of order) timeslice 
              counter_lost_ts++;
            }
 
            trgQueue.pop();
          
          }
 
          // build triggered modules
          
          set<int> triggeredModules;
 
          for (int its = 0; its < windowLength; its++) {
            
            set<int> current = trgWindow[its].getModules();
 
            triggeredModules.insert(current.begin(), current.end());
 
          }
 
          // trigger window slide of one element
 
          int currentFrame    = trgWindow[0].frameIndex;
          JDAQUTCExtended currentTime  = trgWindow[0].timeUTC;
 
          trgWindow.pop_front();
 
          // calculate trigger
 
          int trigger = triggeredModules.size();
          
          ++stats[trigger];
 
          DEBUG("TRIGGER => TS = " << currentFrame << "; LEVEL = " << trigger << endl);
 
          // calculate active modules
          int activeModules = -1;
 
          if (!moduleRates.empty()          &&
                moduleRates.count(currentFrame)) {
              
            activeModules = 0;
 
            for (map<int, double>::const_iterator p = moduleRates.at(currentFrame).begin(); 
                 p != moduleRates.at(currentFrame).end(); p++ ) {
              if (p->second > 0) {
                activeModules++;
              }
            }
          }
 
 
          if (out != NULL) {
 
 
            JSummarySN summary(DETID, activeModules, RUN, currentFrame, currentTime, trigger);
 
            stringstream message;
            message << summary;
            
            out->PutFullString(outputTag, message.str());
          }
          
 
          // print stats
 
          if ( (counter_live_ts % ((int)(statPrintInterval_s / frameTime_s)) == 0 ) ) {
            
            double livetime = counter_live_ts * frameTime_s;
            
            stats.setLiveTime(livetime);
 
            NOTICE(endl);
            NOTICE(stats.toString());
            NOTICE("=> discarded out-of-order timeslices = " << counter_lost_ts << endl);          
 
            if (summaryFile != "") {
              ofstream of(summaryFile.c_str());
              of << stats.toSummaryFile();
              of.close();
            }
 
          }
 
        } else { 
          NOTICE("Filling trigger queue: " << trgQueue.size() << "/" << queueLength << '\r'); 
        }
 
      }  else {
 
        NOTICE("timeout " << setw(3) << i << endl);
 
        ++i;
      }
    }
  }
  
  catch(const JSocketException& error) {
    ERROR(error.what() << endl);
  }
 
}