daqman.cc

Go to the documentation of this file.

#include "V172X_Daq.hh"
#include "V172X_Event.hh"
#include "ConfigHandler.hh"
#include "CommandSwitchFunctions.hh"
#include "EventHandler.hh"
#include "AsyncEventHandler.hh"
#include "RawWriter.hh"
#include "ProcessedPlotter.hh"
#include "Reader.hh"

#include "GenericAnalysis.hh"
#include "BaselineFinder.hh"
#include "PulseFinder.hh"
#include "Integrator.hh"
#include "EvalRois.hh"
#include "SpectrumMaker.hh"
#include "ConvertData.hh"
#include "SumChannels.hh"
#include "TriggerHistory.hh"

#include "runinfo.hh"

#include "kbhit.h"
#include <exception>
#include <string>
#include "Message.hh"
#include <time.h>
#include <fstream>
#include <algorithm>
#include <numeric>
#include "boost/thread/thread.hpp"
bool stop_run = false;

//utility class to print run statistics
struct PrintStats{
  time_t last_print_time;
  unsigned long events_processed;
  unsigned long long bytes_processed;
  void Clear() { 
    last_print_time = time(0); 
    events_processed = 0;
    bytes_processed = 0;
  }
  PrintStats() { Clear(); }
};
std::ostream& operator<<(std::ostream& out, const PrintStats& stats){
  time_t now = time(0);
  out<<"In last "<<now-stats.last_print_time<<" seconds, processed "
     <<stats.events_processed<<" events at "
     <<stats.bytes_processed/(now-stats.last_print_time)/1024<<" kiB/s";
  return out;
}

//utility class to add real-time spectra
class SpectrumAdder{
  std::vector<SpectrumMaker*>& _spectra;
public:
  SpectrumAdder(std::vector<SpectrumMaker*>& v) : _spectra(v) {}
  std::istream& operator()(std::istream& in){
    //first parameter is the name of the spectrum module
    std::string name;
    in>>name;
    _spectra.push_back(new SpectrumMaker(name));
    //expect to read parameters immediately after name
    in>>*_spectra.back();
    return in;
  }
};
  
class InsertComment{
  runinfo* _info;
public:
  InsertComment(runinfo* info) : _info(info) {}
  int operator()(const char* val){ 
    _info->SetMetadata("comment",std::string(val)) ; return 0; 
  }
};

int main(int argc, char** argv)
{
  
  //set up the config handler
  ConfigHandler* config = ConfigHandler::GetInstance();
  
  //register all processing modules
  EventHandler* modules = EventHandler::GetInstance();
  //only RawWriter and reading headers go synchronously 
  RawWriter* writer = modules->AddModule<RawWriter>();
  ConvertData* read_headers = modules->AddModule<ConvertData>("ReadHeaders");
  read_headers->SetHeadersOnly(true);
  
  std::vector<AsyncEventHandler*> async_threads;
  
  //All analyzing modules run on a single asynchronous thread
  AsyncEventHandler thread1;
  modules->AddAsyncReceiver(&thread1);
  thread1.AddModule(new ConvertData);
  thread1.AddModule(new SumChannels);
  thread1.AddModule(new BaselineFinder);
  thread1.AddModule(new Integrator);
  thread1.AddModule(new EvalRois);
  async_threads.push_back(&thread1);
  // any more advanced functions?
  
  //ProcessedPlotter and RootGraphix share a thread and refresh speed
  AsyncEventHandler thread2;
  //add RootGraphix first so dependencies pass, but call afterward...
  RootGraphix* rootgraphix = new RootGraphix;
  modules->AddModule(rootgraphix, false, true);
  thread1.AddReceiver(&thread2);
  ProcessedPlotter* plotter = new ProcessedPlotter;
  thread2.AddModule(plotter);
  thread2.AddModule(new TriggerHistory);
  thread2.AddModule(rootgraphix, false);
  async_threads.push_back(&thread2);
  
  //allow as many spectra as we want
  std::vector<SpectrumMaker*> spectra;
  modules->RegisterParameter("spectra",spectra, 
                             "List of live analysis spectra to dispay");
  
  
  
  //keep this for backward compatibility
  modules->RegisterReadFunction("add_spectrum", SpectrumAdder(spectra),
                                "Add a new real-time spectrum to display");

    
  //initialize some options for command switches
  long stop_events = -1, stop_time = -1; 
  long long stop_size = -1;
  int stattime=0;
  bool write_only = false;
  runinfo* info = modules->GetRunInfo();
  std::string testmode_file="";
  int testmode_dt = 0;
  int graphics_refresh = 1;
  config->AddCommandSwitch('i', "info", "Set run database info to <info>",
                           CommandSwitch::DefaultRead<runinfo>(*info),
                           "info");
  config->AddCommandSwitch('m',"message","Set database comment to <message>",
                           InsertComment(info),"message");
                           
  config->AddCommandSwitch('e',"stop_events","Stop after <n> events",
                           CommandSwitch::DefaultRead<long>(stop_events),
                           "n");
  config->AddCommandSwitch('t',"stop_time","Stop after <n> seconds",
                           CommandSwitch::DefaultRead<long>(stop_time),
                           "n");
  config->AddCommandSwitch('s',"stop_size","Stop after saving <size> MB",
                           CommandSwitch::DefaultRead<long long>(stop_size),
                           "size");
  config->AddCommandSwitch(' ',"write-only",
                           "Disable all modules except for the RawWriter",
                           CommandSwitch::SetValue<bool>(write_only, true));
  config->AddCommandSwitch(' ',"testmode","Fake DAQ from input file <file>",
                           CommandSwitch::DefaultRead<std::string>
                           (testmode_file) ,"file");
  config->AddCommandSwitch(' ',"testmode-dt","Sleep <N> ms between each event",
                           CommandSwitch::DefaultRead<int>(testmode_dt),"N");
  config->AddCommandSwitch(' ',"stat-time","Print stats every <secs> seconds",
                           CommandSwitch::DefaultRead<int>(stattime),"secs");
  config->AddCommandSwitch(' ',"refresh","Time in s between graphics update",
                           CommandSwitch::DefaultRead<int>(graphics_refresh),
                           "secs");
  config->RegisterParameter("stop_size",stop_size,
                            "Maximum file size in bytes before abort the run");
  config->RegisterParameter("stop_time",stop_time,
                            "Maximum time in seconds before we abort the run");
  config->RegisterParameter("stop_events",stop_events,
                            "Maximum number of events before we abort the run");
  config->RegisterParameter("stat-time",stattime,
                            "Time between printing of event/data rates");
  config->RegisterReadFunction("require_comment",DeprecatedParameter<bool>());
  V172X_Daq daq;
    
  config->SetProgramUsageString("daqman [options]");
  config->SetDefaultCfgFile("daqman.cfg");
  
  if(config->ProcessCommandLine(argc, argv))
    return 1;
  
  if(config->GetNCommandArgs()){
    Message(ERROR)<<"Too many arguments specified."<<std::endl;
    config->PrintSwitches(true);
  }
  
  AsyncEventHandler thread3;
  if(spectra.size() > 0){
    //for right now, put the spectra all on one thread
    thread1.AddReceiver(&thread3);
    for(size_t i=0; i<spectra.size(); ++i)
      thread3.AddModule((spectra[i]), true, false);
    async_threads.push_back(&thread3);
  }
  
  if(writer->enabled){
    modules->SetRunIDFromFilename(writer->GetFilename());
  }
  
  //see if we want to disable everything
  if(write_only){
    std::vector<BaseModule*>* mods = modules->GetListOfModules();
    for(size_t i=0; i < mods->size(); i++){
      if(mods->at(i) != writer) mods->at(i)->enabled = false;
    }
  }
  
  //set the graphics refresh time
  thread2.SetSleepMillisec(graphics_refresh*1000);
  
  //see if we're in test mode
  Reader* reader = 0;
  if(testmode_file!=""){
    reader = new Reader(testmode_file);
    if(!reader->IsOk()){
      Message(ERROR)<<"Testmode specified, but can't load file "
                    <<testmode_file<<".\n";
      return 1;
    }
  }
  
  //set the terminal into unbuffered mode
  keyboard board;
  //initialize all modules
  Message(INFO)<<"Initializing modules...\n";
  if(modules->Initialize()){
    Message(CRITICAL)<<"Unable to initialize all modules.\n";
    return 1;
  }
  //start up the asynchronous event handlers
  for(size_t i=0; i<async_threads.size(); ++i)
    async_threads[i]->StartRunning();
  try
    {
      if(!reader){
        Message(INFO)<<"Initializing DAQ...\n";
        if(daq.Initialize() == 0){
          Message(INFO)<<"Starting Run...\n";
          daq.StartRun();
        }
        else{
          Message(CRITICAL)<<"Initialization Error!"<<std::endl;
          modules->Finalize();
          return -1;
        }
      }
      //start the thread which takes commands
      time_t start_time = time(0);
      long events_downloaded = 0;
      long long data_downloaded = 0;
      PrintStats stats;
      while(!stop_run){
        //see if we should end the run
        if( (stop_time > 0 && time(0)-start_time >= stop_time) ||
            (stop_events > 0 && events_downloaded >= stop_events) ||
            (stop_size > 0 && writer->GetBytesWritten()/1000000 >= stop_size) ){
          stop_run = true;
        }
        //process user input
        if(board.kbhit()){
          //process user entered command
          char c = board.getch();
          switch(c){
          case 'q':
          case 'Q':
            stop_run = true;
            break;
          case 'p':
          case 'P':
            plotter->TogglePause();
            break;
          default:
            Message(ERROR)<<"Unknown control character '"<<c<<"'\n";
          };
        }
        if(stop_run) break;
        //get the next event
        RawEventPtr evt;
        if(reader){
          evt = reader->GetNextEvent();
          if(testmode_dt>0)
            boost::this_thread::sleep(boost::posix_time::millisec(testmode_dt));
        }
        else
          evt = daq.GetNextEvent(500000);
        if(!evt){
          //see if there's an error
          if(reader){
            Message(INFO)<<"Reached end of file, exiting testmode.\n";
            stop_run = true;
            break;
          }
          else if(daq.GetStatus() != BaseDaq::NORMAL){
            Message(ERROR)<<"An error occurred while getting next event.\n";
            Message(ERROR)<<"Attempting to abort run...\n";
            stop_run = true;
            break;
          }
          else{
            Message(DEBUG)<<"Waiting for new event ready in memory...\n";
            continue;
          }
        }
        if(modules->Process(evt)){
          Message(ERROR)<<"Problem encountered processing event.\n";
          //stop_run = true;
          continue;
        }

        data_downloaded += evt->GetDataSize();
        events_downloaded++;
        stats.events_processed++;
        stats.bytes_processed += evt->GetDataSize();
        if(stattime>0 && time(0)-stats.last_print_time >= stattime){
          Message(INFO)<<stats<<std::endl;
          stats.Clear();
        }
      }
      Message(INFO)<<"Ending Run....\n";
      if(!reader){
        Message(DEBUG)<<"Processing remaining events in queue...\n";
        daq.EndRun();
        RawEventPtr evt;
        while((stop_events<=0 || events_downloaded < stop_events) &&
              daq.GetStatus() == BaseDaq::NORMAL &&
              (evt = daq.GetNextEvent()) && 
              !modules->Process(evt)){
          data_downloaded += evt->GetDataSize();
          events_downloaded++;
          stats.events_processed++;
          stats.bytes_processed += evt->GetDataSize();
          
        }
      }
      time_t delta_time = time(0)-start_time;
      //end the asyncronous threads
      for(size_t i=0; i<async_threads.size(); ++i)
        async_threads[i]->StopRunning();
      
      modules->Finalize();
      //print out some statistics
      Message(INFO)<<events_downloaded<<" events processed.\n";
      Message(INFO)<<delta_time<<" seconds elapsed.\n";
      if(delta_time>0){
        Message(INFO)<<data_downloaded<<" total bytes processed at "
                     <<data_downloaded/(delta_time)
                     <<" bytes/s\n";
      }
    }
  catch(std::exception &e)
    {
      std::cerr<<"Caught Exception: "<<e.what()<<"\n";
    }
  
  return daq.GetStatus();
}