source: Sophya/trunk/AddOn/TAcq/brbaseproc.cc@ 4016

//----------------------------------------------------------------
// Projet BAORadio - (C) LAL/IRFU  2008-2010
// Classes de base pour les threads de traitememt donnees BAORadio 
//----------------------------------------------------------------

#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fstream>
#include "brbaseproc.h"


using namespace SOPHYA;
//---------------------------------------------------------------------
// Classe BRBaseProcessor : classe de base pour l'analyse des donnees 
// BAORadio - Les classes de traitement heritent de cette classe et 
// redefinissent la mathode Process() qui est appele paquet par paquet
//---------------------------------------------------------------------
/*!
   \class BRBaseProcessor
   \ingroup TAcq

   \brief Base class for BAORadio processing thread classes.

   Thread classes for BAORadio data processing and analysis can inherit from 
   this class and redefine/implement the Process() method which is called 
   for each BRpaquet. 
*/ 

/* --Methode-- */
BRBaseProcessor::BRBaseProcessor(RAcqMemZoneMgr& memgr, MemZaction mmact)
  : memgr_(memgr), fbuff_(memgr.NbFibres()), fprocbuff_(memgr.NbFibres()),
    vpaq_(memgr.NbFibres()), vprocpaq_(memgr.NbFibres()),
    vpchk_(memgr.NbFibres()), vfgok_(memgr.NbFibres()), 
    curfc_(memgr.NbFibres()), fcfirst_(memgr.NbFibres()), ttfirst_(memgr.NbFibres())
{
  SetMemZAction(mmact);
  stop_ = false;
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
    fbuff_[fib]=NULL; 
    fprocbuff_[fib]=NULL; 
    vpaq_[fib]=BRPaquet(NULL,memgr_.PaqSize());
    vprocpaq_[fib]=NULL; 
    vpchk_[fib]=BRPaqChecker(true,0); 
    vfgok_[fib]=true;
    curfc_[fib]=0;
    fcfirst_[fib]=0;
    ttfirst_[fib]=0;
  }
  fgokallfibers_=true;
  totprocnpaq_=0;
  startdate_=0;
  setNameId("baseproc",0); 
  ClearProcTimeWindow();
  SetPrintLevel();
}

/* --Methode-- */
MemZStatus BRBaseProcessor::SetMemZAction(MemZaction mmact)
{
  mmact_=mmact;
  mmsta_=MemZS_ProcA;
  switch (mmact) {
  case MemZA_ProcA:
    mmsta_=MemZS_ProcA;    
    break;
  case MemZA_ProcB:
    mmsta_=MemZS_ProcB;    
    break;
  case MemZA_ProcC:
    mmsta_=MemZS_ProcC;    
    break;
  case MemZA_ProcD:
    mmsta_=MemZS_ProcD;    
    break;
  case MemZA_ProcE:
    mmsta_=MemZS_ProcE;    
    break;
  case MemZA_ProcF:
    mmsta_=MemZS_ProcF;    
    break;
  case MemZA_ProcG:
    mmsta_=MemZS_ProcG;    
    break;
  case MemZA_ProcH:
    mmsta_=MemZS_ProcH;    
    break;
  case MemZA_ProcI:
    mmsta_=MemZS_ProcI;    
    break;
  case MemZA_ProcJ:
    mmsta_=MemZS_ProcI;    
    break;
  default:
    mmact_=MemZA_ProcA;
    mmsta_=MemZS_ProcA;
    break;
  }
  return mmsta_;
}

/* --Methode-- */
void BRBaseProcessor::run()
{
  setRC(1);     
  int rc=0;
  try {
    cout << "BRBaseProcessor[" << bpnom_ << "," << bpid_ << "]::run() - Starting " << " NFibers=" 
         << memgr_.NbFibres() << endl;  
    if (fgproctimewin_) cout << " BRBaseProcessor...TimeWindow Start=" << tmstart_ << " End=" << tmend_ << endl;

    size_t paqsz=memgr_.PaqSize();
    size_t procpaqsz=memgr_.ProcPaqSize();

    double tstartdays = tmstart_.ToDays(); 
    double tenddays = tmend_.ToDays(); 

    bool fgrun=true;
    while (fgrun) {
      if (stop_) break;
      if (memgr_.GetRunState() == MemZR_Stopped) break;
      mid_ = memgr_.FindMemZoneId(mmact_);  // (MemZA_ProcA);
      //      Byte* buffg = memgr_.GetMemZone(mid_);
      //      if (buffg == NULL) {
      if (mid_ < 0) {
        cout << "BRBaseProcessor[" << bpnom_ << "," << bpid_ << "]::run()/ERROR FindMemZoneId(" 
             << (int)mmact_ << ") ->" << mid_ << ") -> NULL" << endl;
        setRC(7);      fgrun=false;             
        break;  
      }
      cts_=memgr_.GetAuxData(mid_)->FillTime();   // get associated date/time (DATEOBS)
      if (startdate_==0)  startdate_=cts_.DaysPart();

      for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
        fbuff_[fib] = memgr_.GetMemZone(mid_,fib);
        if (fbuff_[fib] == NULL) { // cela ne devrait pas arriver 
          cout << "BRBaseProcessor[" << bpnom_ << "," << bpid_ << "]::run()/ERROR memgr.GetMemZone(" 
               << mid_ << "," << fib << ") -> NULL" << endl;
          setRC(9);       fgrun=false;          
          break;
        }
        if ((procpaqsz>0)&&((fprocbuff_[fib]=memgr_.GetProcMemZone(mid_,fib))==NULL)) {   // cela ne devrait pas arriver non plus
          cout << "BRBaseProcessor[" << bpnom_ << "," << bpid_ << "]::run()/ERROR memgr.GetProcMemZone(" 
               << mid_ << "," << fib << ") -> NULL" << endl;
          setRC(9);       fgrun=false;          
          break;
        }
      }

      // On ne traite les paquets que si pas de fenetre en temps ou tmstart_ <= FillTime() <= tmend_
      double ftmdays = memgr_.GetAuxData(mid_)->FillTime().ToDays();
      if ((!fgproctimewin_) || ( (ftmdays<=tenddays)&&(ftmdays>=tstartdays) ) ) {

        for(size_t jp=0; jp<memgr_.NbPaquets(); jp++) {   // boucle sur les paquets d'une zone  
          fgokallfibers_=true;
          for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
            vpaq_[fib].Set(fbuff_[fib]+jp*paqsz);
            vfgok_[fib]=vpchk_[fib].Check(vpaq_[fib],curfc_[fib]);
            if (!vfgok_[fib])  fgokallfibers_=false;
            if (procpaqsz>0)    vprocpaq_[fib] = fprocbuff_[fib]+jp*procpaqsz;
          }
          //--- Traitement 
          int rcp=0;
          if (fgokallfibers_)  {
            if (totprocnpaq_==0) {
              for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
                fcfirst_[fib]=curfc_[fib];
                ttfirst_[fib]=vpaq_[fib].TimeTag();
              }
            }
            rcp=Process();
            totprocnpaq_++;
          }
          //    if (rcp != 0)   { fgrun=false; break; }
        }  // Fin de boucle sur les paquets
      }  // Fin de test pour fenetre en temps de traitement des paquets 
 
      memgr_.FreeMemZone(mid_, mmsta_);   // (MemZS_ProcA);
    } // Fin de boucle sur les zones a traiter 
    //------------------------------------
    cout << " ---- END BRBaseProcessor[" << bpnom_ << "," << bpid_ << "]::run() , TotNbProcPaq=" 
         << totprocnpaq_ << endl;
    /*
    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  vpchk_[fib].Print();
    cout << " ------------------------------------ " << endl;
    */
  }
  catch (std::exception& exc) {
    cout << "BRBaseProcessor[" << bpnom_ << "," << bpid_ << "]::run()/catched std::exception " << exc.what() << endl;
    setRC(98);  
    return; 
  }
  catch(...) {
    cout << "BRBaseProcessor[" << bpnom_ << "," << bpid_ << "]::run()/catched unknown ... exception " << endl;
    setRC(99);  
    return; 
  }

}


/* --Methode-- */
int BRBaseProcessor::Process()
{
  // Cette methode est appele pour chaque "trigger', cad la serie de paquets pour toutes les fibres
  // la methode par defaut ne fait rien 
  // if (fgokallfibers_) { faire le traitement }
  //  return 0;
  /*    Exemple de code test  */
  if ((prtlev_>0)&&(totprocnpaq_ %prtmodulo_==0))  {
    cout << " BRBaseProcessor::Process(totnpaq=" << totprocnpaq_ << ")  ObsTime=" << getObsTime() 
         << " TimeTag=" << getCurTimeTagSeconds() << " s. FrameCounter="  << getCurFrameCounter() << endl;
    if (prtlev_<2) return 0;
    cout << " ... FrameCnt[0..." << memgr_.NbFibres()-1 << "]=" ;
    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)   
      cout << getCurFrameCounter(fib) << "," ;
    cout << endl;
    cout << " ... TimeTag[0..." << memgr_.NbFibres()-1 << "]=" ;
    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)   
      cout << getCurTimeTag(fib) << "," ;
    cout << endl;
    if (prtlev_<3) return 0;
    cout << " ... TimeTagSeconds[0..." << memgr_.NbFibres()-1 << "]=" ;
    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)   
      cout << getCurTimeTagSeconds(fib) << "," ;
    cout << endl;
    if (prtlev_<4) return 0;
    cout << " ... FileSequenceNumber[0..." << memgr_.NbFibres()-1 << "]=" ;
    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)   
      cout << memgr_.GetAuxData(mid_)->FileSequenceNumber(fib) << ",";
    cout << endl;
  }
  return 0;
}