OnMonProd_module.cc

Go to the documentation of this file.

///
/// \file    OnMonProd.cxx
/// \brief   Top level producer module for online monitoring
/// \version $Id: OnMonProd.cxx,v 1.25 2012-11-27 17:57:47 mbaird42 Exp $
/// \author  messier@indiana.edu, mbaird42@fnal.gov
///

#include <cstdlib>
#include <vector>
#include <algorithm>

#include "TSystem.h"
#include "TFile.h"
#include "TTree.h"
#include "art/Framework/Principal/Event.h"
#include "art/Framework/Principal/SubRun.h"
#include "art/Framework/Core/EDAnalyzer.h"

#include "NovaTimingUtilities/TimingUtilities.h"
#include "DAQDataFormats/RawEvent.h"
#include "DAQDataFormats/RawTrigger.h"
#include "RawData/FlatDAQData.h"

#include "OnlineMonitoring/util/HistoTable.h"
#include "OnlineMonitoring/util/HistoData.h"
#include "OnlineMonitoring/util/Ticker.h"
#include "OnlineMonitoring/util/Settings.h"
#include "OnlineMonitoring/util/IPC.h"
#include "OnlineMonitoring/producer/HistoSet.h"
#include "OnlineMonitoring/producer/FEBRatesByDiblock.h"
#include "OnlineMonitoring/producer/RawEventUnpacker.h"
#include "OnlineMonitoring/producer/HitCounts.h"
#include "OnlineMonitoring/producer/HitMaps.h"
#include "OnlineMonitoring/producer/TQPlots.h"
#include "OnlineMonitoring/producer/WLPlots.h"
#include "OnlineMonitoring/producer/TriggerPlots.h"
#include "OnlineMonitoring/producer/NanoErrors.h"
#include "OnlineMonitoring/producer/MicroErrors.h"
#include "OnlineMonitoring/producer/DataBlockErrors.h"
#include "OnlineMonitoring/producer/RawEventErrors.h"
#include "OnlineMonitoring/producer/WatchListManager.h"

/// Online Monitoring package header
namespace om {
  ///
  /// A class that knows how to serve monitoring histograms
  ///
  class OnMonProdIPC : public om::IPC 
  {
  public:
    OnMonProdIPC(int m, const char* hdl);
  private:
    TH1F* FindTH1F(const char* nm);
    TH2F* FindTH2F(const char* nm);
    void  HistoList(std::list<std::string>& hlist);
    void  AddToWatchList(int db, int dcm, int feb, int pix);
    void  RemoveFromWatchList(int db, int dcm, int feb, int pix);
    void  CurrentWatchList(std::vector<int>& db,
                           std::vector<int>& dcm,
                           std::vector<int>& feb,
                           std::vector<int>& pix);
  };
  
  ///
  /// Top level module for online monitoring
  ///
  class OnMonProd : public art::EDAnalyzer {
  public:
    explicit OnMonProd(fhicl::ParameterSet const &p);
    virtual ~OnMonProd();

    void beginRun(const art::Run& R);
    void beginJob();
    void endJob();
    void analyze(art::Event const &e);

    void reconfigure(fhicl::ParameterSet const &p);

    void NewRunNumber(unsigned int run, unsigned int subrun);
    void NewSubrunNumber(unsigned int run, unsigned int subrun);

    bool UnpackRawDaqData(art::Event const& evt,
                          daqdataformats::RawEvent& rawevt);

  private:
    OnMonProdIPC* fIPC;         ///< Communicates with viewer
    std::string   fSHMname;     ///< Shared memory for communication
    bool          fuseSHM;      ///< Use SHM to communicate with a viewer?
    unsigned int  fLastRun;     ///< Last run number analyzed
    unsigned int  fLastSubrun;  ///< Last subrun number analyzed
    unsigned int  fPartition;   ///< Partition of last event analyzed
    bool          fWriteToFile; ///< Output histogram file?
    bool          fTickerOn;    ///< Turned on in the control room

    TTree* fHeader;                   ///< Header object to be written to disk with the histograms
    int    fFirstEvent;               ///< Header: first event processed
    int    fLastEvent;                ///< Header: last event processed
    int    fNevents;                  ///< Header: number of events processed

    unsigned long int fStartNOvATime; ///< Header: time of first processed event (in NOvA time - 64 MHz clock ticks) - initalized to -1 to make it a HUGE positive number
    unsigned long int fEndNOvATime;   ///< Header: time of last processed event (in NOvA time - 64 MHz clock ticks)

    uint32_t fStartYear;  ///< Header: year of first processed event (actual years since 0 A.D.)
    uint16_t fStartMonth; ///< Header: month of first processed event (actual month 1-12)
    uint16_t fStartDay;   ///< Header: day of first processed event (day of the month 1-31)
    double   fStartHour;  ///< Header: hour of first processed event (since midnight 0.0 - 23.99999)

    uint32_t fEndYear;    ///< Header: year of last processed event (actual years since 0 A.D.)
    uint16_t fEndMonth;   ///< Header: month of last processed event (actual month 1-12)
    uint16_t fEndDay;     ///< Header: day of last processed event (day of the month 1-31)
    double   fEndHour;    ///< Header: hour of last processed event (since midnight 0.0 - 23.99999)

    std::vector<int> fIncompleteEvents;   ///< Header: a list of incomplete events
  };
}

using namespace om;

OnMonProdIPC::OnMonProdIPC(int m, const char* hdl) : om::IPC(m, hdl) { }
TH1F* OnMonProdIPC::FindTH1F(const char* nm) {
  return HistoSet::Instance().FindTH1F(nm);
}
TH2F* OnMonProdIPC::FindTH2F(const char* nm) {
  return HistoSet::Instance().FindTH2F(nm);
}
void OnMonProdIPC::HistoList(std::list<std::string>& hlist) {
  HistoSet::Instance().GetNames(hlist);
}
void OnMonProdIPC::AddToWatchList(int db, int dcm, int feb, int pix) {
  WatchListManager::Instance().Add(db, dcm, feb, pix);
}
void OnMonProdIPC::RemoveFromWatchList(int db, int dcm, int feb, int pix) {
  WatchListManager::Instance().Remove(db, dcm, feb, pix);
}
void OnMonProdIPC::CurrentWatchList(std::vector<int>& db,
                                    std::vector<int>& dcm,
                                    std::vector<int>& feb,
                                    std::vector<int>& pix) {
  WatchListManager::Instance().GetList(db, dcm, feb, pix);
}

//......................................................................

OnMonProd::OnMonProd(fhicl::ParameterSet const &p) : 
  EDAnalyzer(p),
  fIPC(0), 
  fSHMname(""),
  fuseSHM(false),
  fLastRun(0), 
  fLastSubrun(0),
  fPartition(4),
  fWriteToFile(false),
  fTickerOn(false),
  fHeader(0),
  fFirstEvent(1e9),
  fLastEvent(-1),
  fNevents(0),
  fStartNOvATime(-1),
  fEndNOvATime(0)
{
  this->reconfigure(p);

  HistoTable::Instance(Settings::Instance().fCSVFile.c_str(),
                       Settings::Instance().fDet);

  // Histogram fillers
  static FEBRatesByDiblock fbdb;
  static HitCounts         hc;
  static HitMaps           hm;
  static TQPlots           tqp;
  static WLPlots           wlp;
  static TriggerPlots      trig;
  static NanoErrors        nerr;
  static MicroErrors       mcerr;
  static DataBlockErrors   dberr;
  static RawEventErrors    reerr;
}

//......................................................................

OnMonProd::~OnMonProd() { }

//......................................................................

//
// Unpack the raw DAQ data from the ART event
//
bool OnMonProd::UnpackRawDaqData(art::Event const& evt,
                                 daqdataformats::RawEvent& rawevt)
{
  try {
    art::Handle<std::vector<rawdata::FlatDAQData> > flatdaq_handle;
    evt.getByLabel("daq",flatdaq_handle);
    if (flatdaq_handle->size()!=1) {
      std::cerr << "Required to have one and only one flat daq data object" 
                << std::endl;
      return false;
    }
    rawevt.readData((*flatdaq_handle)[0].getRawBufferPointer());
    return true;
  }
  catch (...) {
    std::cerr << "arf! Bad unpack of flatdaq!" << std::endl;
    return false;
  }
  return false;
}

//......................................................................

void OnMonProd::NewRunNumber(unsigned int run, unsigned int subrun) 
{
  if (run==0 && subrun==0) return;

  if(fWriteToFile) {

    char buff[1024];
    
    if(fuseSHM)
      sprintf(buff, "%s_CR_r%.8d_s%.3d.onmon.root", fSHMname.c_str(), run, subrun);
    else {
      if(Settings::Instance().fDet == kFARDET)
        sprintf(buff, "fardet_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
      if(Settings::Instance().fDet == kNEARDET)
        sprintf(buff, "neardet_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
      if(Settings::Instance().fDet == kNDOS)
        sprintf(buff, "ndos_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
      if(Settings::Instance().fDet == kTESTBEAM)
        sprintf(buff, "testbeam_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
    }

    std::cout << "Writing end-of-run histograms to root file " 
              << buff << "...";

    // Convert times from NOvA times to unix times.
    // using unsigned long long rather than uint64_t to ensure consistent behavior between OSX and LINUX
    uint16_t           hour = 0;
    uint16_t           min  = 0;
    uint16_t           sec  = 0;

#ifndef DARWINBUILD
    uint64_t           RoundedTime = 0.;  // We need this to have > 1 second precision.
#else
    unsigned long long RoundedTime = 0;   // We need this to have > 1 second precision.
#endif

    novadaq::timeutils::convertNovaTimeToUnixTime(fStartNOvATime, sec, min, hour,
                                                  fStartDay, fStartMonth, fStartYear);
    novadaq::timeutils::convertUnixTimeToNovaTime(sec, min, hour, fStartDay,
                                                  fStartMonth, fStartYear, RoundedTime);

    fStartHour = (double)hour + (double)min/60.0 + (double)sec/3600.0
                 + (long double)(fStartNOvATime-RoundedTime)/64.0e6/3600.0;
    
    hour = 0;
    min  = 0;
    sec  = 0;
    RoundedTime = 0;
    novadaq::timeutils::convertNovaTimeToUnixTime(fEndNOvATime, sec, min, hour,
                                                  fEndDay, fEndMonth, fEndYear);
    novadaq::timeutils::convertUnixTimeToNovaTime(sec, min, hour, fEndDay,
                                                  fEndMonth, fEndYear, RoundedTime);
    
    fEndHour = (double)hour + (double)min/60.0 + (double)sec/3600.0
               + (long double)(fEndNOvATime-RoundedTime)/64.0e6/3600.0;

    // Convert to "normal" time units
    fStartYear  += 1900;
    fStartMonth += 1;
    fEndYear    += 1900;
    fEndMonth   += 1;

    // sort incomplete event numbers
    std::sort(fIncompleteEvents.begin(),fIncompleteEvents.end());

    // Write histos and header to disk.
    TFile* f = new TFile(buff,"RECREATE");
    HistoSet::Instance().WriteToRootFile(f);
    fHeader->Fill();
    fHeader->Write("Header",TObject::kOverwrite,0);
    f->Close();
    delete f; f=0;

    // reset header varaibles
    fHeader->Reset();
    fFirstEvent    = 1e9;
    fLastEvent     = -1;
    fNevents       = 0;
    fStartNOvATime = -1;
    fEndNOvATime   = 0;
    fIncompleteEvents.clear();
    
    std::cout << "done." << std::endl;
    
  }

}

//......................................................................

void OnMonProd::NewSubrunNumber(unsigned int run, unsigned int subrun) 
{
  if (run==0 && subrun==0) return;
  
  if(fWriteToFile) {
    
    char buff[1024];

    if(fuseSHM)
      sprintf(buff, "%s_CR_r%.8d_s%.3d.onmon.root", fSHMname.c_str(), run, subrun);
    else {
      if(Settings::Instance().fDet == kFARDET)
        sprintf(buff, "fardet_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
      if(Settings::Instance().fDet == kNEARDET)
        sprintf(buff, "neardet_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
      if(Settings::Instance().fDet == kNDOS)
        sprintf(buff, "ndos_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
      if(Settings::Instance().fDet == kTESTBEAM)
        sprintf(buff, "testbeam_nearline_r%.8d_s%.3d.onmon.root", run, subrun);
    }
    
    std::cout << "Writing end-of-subrun histograms to root file " 
              << buff << "...";
    
    // Convert times from NOvA times to unix times.
    // using unsigned long long rather than uint64_t to ensure consistent behavior on OSX and LINUX
    uint16_t           hour = 0;
    uint16_t           min  = 0;
    uint16_t           sec  = 0;

#ifndef DARWINBUILD
    uint64_t           RoundedTime = 0.;  // We need this to have > 1 second precision.
#else
    unsigned long long RoundedTime = 0;   // We need this to have > 1 second precision.
#endif

    novadaq::timeutils::convertNovaTimeToUnixTime(fStartNOvATime, sec, min, hour,
                                                  fStartDay, fStartMonth, fStartYear);
    novadaq::timeutils::convertUnixTimeToNovaTime(sec, min, hour, fStartDay,
                                                  fStartMonth, fStartYear, RoundedTime);

    fStartHour = (double)hour + (double)min/60.0 + (double)sec/3600.0
                 + (long double)(fStartNOvATime-RoundedTime)/64.0e6/3600.0;
    
    hour = 0;
    min  = 0;
    sec  = 0;
    RoundedTime = 0;
    novadaq::timeutils::convertNovaTimeToUnixTime(fEndNOvATime, sec, min, hour,
                                                  fEndDay, fEndMonth, fEndYear);
    novadaq::timeutils::convertUnixTimeToNovaTime(sec, min, hour, fEndDay,
                                                  fEndMonth, fEndYear, RoundedTime);
    
    fEndHour = (double)hour + (double)min/60.0 + (double)sec/3600.0
               + (long double)(fEndNOvATime-RoundedTime)/64.0e6/3600.0;

    // Convert to "normal" time units
    fStartYear  += 1900;
    fStartMonth += 1;
    fEndYear    += 1900;
    fEndMonth   += 1;

    // sort incomplete event numbers
    std::sort(fIncompleteEvents.begin(),fIncompleteEvents.end());

    // Write histos and header to disk.
    TFile* f = new TFile(buff,"RECREATE");
    HistoSet::Instance().WriteToRootFile(f);
    fHeader->Fill();
    fHeader->Write("Header",TObject::kOverwrite,0);
    f->Close();
    delete f; f=0;

    // reset header varaibles
    fHeader->Reset();
    fFirstEvent    = 1e9;
    fLastEvent     = -1;
    fNevents       = 0;
    fStartNOvATime = -1;
    fEndNOvATime   = 0;
    fIncompleteEvents.clear();
    
    std::cout << "done." << std::endl;
    
  }

  std::cout << "Removing inactive WatchList items: ";
  WatchListManager::Instance().EndSubrunCleanUp();
  std::cout << "done." << std::endl;

}

//......................................................................

void OnMonProd::reconfigure(const fhicl::ParameterSet& p) 
{
  //
  // Define the values in the Settings object
  //

  // try to find the correct path to the .csv file.
  std::string filename = p.get< std::string > ("CSVFile");
  std::string csvpath;
  for(int i = 0; i < 4; ++i) {
    switch(i) {
    case 0: csvpath = "./";      break;
    case 1: csvpath = "./util/"; break;
    case 2:
      csvpath =  getenv("SRT_PRIVATE_CONTEXT");
      csvpath += "/OnlineMonitoring/util/";
      break;
    case 3:
      csvpath =  getenv("SRT_PUBLIC_CONTEXT");
      csvpath += "/OnlineMonitoring/util/";
      break;
    }
    csvpath += filename;
    if(access(csvpath.c_str(), F_OK) != -1) {
      Settings::Instance().fCSVFile = csvpath;
      break;
    }
  } // end for i

  std::string d = p.get< std::string > ("Detector");
  if      (d == "NDOS") Settings::Instance().fDet = kNDOS;
  else if (d == "ND")   Settings::Instance().fDet = kNEARDET;
  else if (d == "FD")   Settings::Instance().fDet = kFARDET;
  else if (d == "TB")   Settings::Instance().fDet = kTESTBEAM;
  else abort();
  
  fSHMname = p.get<std::string>("SHMHandle");
  fuseSHM  = p.get<bool       >("useSHM");

  if (fIPC) delete fIPC;
  if (fuseSHM) fIPC = new OnMonProdIPC(kIPC_SERVER, fSHMname.c_str());

  bool debug = p.get< bool > ("Debug");
  RawEventUnpacker::Instance().SetDebug(debug);

  fWriteToFile = p.get<bool>("WriteToFile");
  fTickerOn    = p.get<bool>("TickerOn");
}


//......................................................................

void OnMonProd::beginRun(const art::Run& R) { }

//......................................................................

void OnMonProd::beginJob()
{
  if(fHeader) { delete fHeader; fHeader = 0; }
  fHeader = new TTree("Header","Header Information");

  fHeader->Branch("Run",&fLastRun);
  fHeader->Branch("Subrun",&fLastSubrun);
  fHeader->Branch("Partition",&fPartition);
  fHeader->Branch("FirstEvent",&fFirstEvent);
  fHeader->Branch("LastEvent",&fLastEvent);
  fHeader->Branch("Nevents",&fNevents);
  fHeader->Branch("StartYear",&fStartYear);
  fHeader->Branch("StartMonth",&fStartMonth);
  fHeader->Branch("StartDay",&fStartDay);
  fHeader->Branch("StartHour",&fStartHour);
  fHeader->Branch("EndYear",&fEndYear);
  fHeader->Branch("EndMonth",&fEndMonth);
  fHeader->Branch("EndDay",&fEndDay);
  fHeader->Branch("EndHour",&fEndHour);
  fHeader->Branch("IncompleteEvents",&fIncompleteEvents);

  // I would like to add these two variables to the header object but
  // root complains that an unsigned long int is an unacceptable type.
  //
  // fHeader->Branch("StartNovaTime", &fStartNOvATime);
  // fHeader->Branch("EndNovaTime", &fEndNOvATime);
}

//......................................................................

void OnMonProd::endJob() 
{
  if(Settings::Instance().fDet != kFARDET)
    this->NewSubrunNumber(fLastRun, fLastSubrun);
  if(Settings::Instance().fDet == kFARDET)
    this->NewRunNumber(fLastRun, fLastSubrun);
  if(fIPC)    { delete fIPC; fIPC = 0; }
  if(fHeader) { delete fHeader; fHeader = 0; }
}

//......................................................................

void OnMonProd::analyze(art::Event const& evt)
{
  bool aok;

  //
  // Handle any pending requests for histograms
  //
  if (fuseSHM) fIPC->HandleRequests();
  
  static daqdataformats::RawEvent rawevt;
  aok = this->UnpackRawDaqData(evt, rawevt);
  if (aok==false) return;
  
  unsigned int run       = rawevt.getRunNumber();
  unsigned int subrun    = rawevt.getSubrun();
  unsigned int event     = rawevt.getEventNumber();
  unsigned int partition = rawevt.getPartition();
  unsigned long int time = rawevt.getTrigger()->getTriggerStart();



  //
  // Check if the run/subrun number has changed
  //
  if ( (subrun != fLastSubrun) && (Settings::Instance().fDet != kFARDET) ) {
    this->NewSubrunNumber(fLastRun, fLastSubrun);
  }
  if ( (run != fLastRun) && (Settings::Instance().fDet == kFARDET) ) {
    this->NewRunNumber(fLastRun, fLastSubrun);
  }

  //
  // update Header variables.
  //
  fNevents++;
  fLastRun    = run;
  fLastSubrun = subrun;
  fPartition  = partition;
  
  // first/last event are not necessarily coorelated with earliest/latest time
  if(time < fStartNOvATime)
    fStartNOvATime = time;
  if((int)event < fFirstEvent)
    fFirstEvent = event;
  if(time > fEndNOvATime)
    fEndNOvATime = time;
  if((int)event > fLastEvent)
    fLastEvent = event;
  
  if( rawevt.getHeader()->isEventIncomplete() ) {
    fIncompleteEvents.push_back(event);
  }
  



  static unsigned int count = 0;
  if (++count%10==0) {
    if (fuseSHM) fIPC->PostResources(run, subrun, event);
    std::cout << "onmon_prod: run/sub/evt="
              << run << "/" << subrun << "/" << event << " "
              << " type:size="
              << rawevt.getEventType() << ":"
              << rawevt.getEventSize()
              << std::endl;
  }
  
  //
  // Update the ticker so it can notify its subscribers.
  // Do this BEFORE unpacking so that there are no overlaps in plots
  // reset every 24 hours.
  //
  if (fTickerOn) Ticker::Instance().Update(run, subrun);

  //
  // Unpack the raw event and publish results to plot-making
  // subscribers
  //
  RawEventUnpacker::Instance().UnpackRawEvent(rawevt);  
}

////////////////////////////////////////////////////////////////////////


#include "art/Framework/Core/ModuleMacros.h"
DEFINE_ART_MODULE(om::OnMonProd)