source: Sophya/trunk/AddOn/TAcq/brfitsrd.cc@ 3980

//----------------------------------------------------------------
// Projet BAORadio - (C) LAL/IRFU  2008-2011
// Classes de threads pour lecture fichiers fits BAORadio
//----------------------------------------------------------------


#include "brfitsrd.h"

#include <stdlib.h>
#include <unistd.h>
#include <fstream>
#include <exception>

#include "pexceptions.h"
#include "timestamp.h"
#include "ctimer.h"

#include "brpaqu.h"

#include "resusage.h" // Pour mesure temps elapsed/CPU ...
#include <sys/time.h>  // pour gettimeofday

using namespace SOPHYA;

//---------------------------------------------------------------------
// Classe thread de lecture de Multi-fibres de fichiers FITS BAORadio  
//---------------------------------------------------------------------

/* --Methode-- */
BRMultiFitsReader::BRMultiFitsReader(RAcqMemZoneMgr& mem, vector<string>& dirs, bool rdsamefc, 
                                     uint_4 imin, uint_4 imax, uint_4 istep)
  :  memgr_(mem), dirs_(dirs), stop_(false), rdsamefc_(rdsamefc), imin_(imin), imax_(imax), istep_(istep) 
{
  SetPrintLevel();
  totnbytesrd_ = 0;
  totsamefc_ = 0;
  if (memgr_.NbFibres() > MAXANAFIB) 
    throw BAORadioException("BRMultiFitsReader::BRMultiFitsReader/ NbFibres>MAXANAFIB "); 
  if (dirs_.size() != memgr_.NbFibres()) 
    throw BAORadioException("BRMultiFitsReader::BRMultiFitsReader/ NbFibres != Nb Data Directories"); 

  packsize_=memgr_.PaqSize();
  mid_=-2; 
  mmbuf_=NULL;
  max_targ_npaq = memgr_.NbPaquets();
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) mmbufib_[fib]=NULL;

  cpaqdeltatime_=0.;

  char flnm[1024];
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
    sprintf(flnm,"%s/signal%d.fits",dirs_[fib].c_str(),imin_);
    mff_[fib].Open(flnm, MF_Read);
    cout << " BRMultiFitsReader::BRMultiFitsReader() opening " << flnm << endl;
    if (mff_[fib].NAxis1() != memgr_.PaqSize())  {
      cout << " BRMultiFitsReader::BRMultiFitsReader/ fib=" << fib << " File=" << flnm <<
        " NAxis1()= " << mff_[fib].NAxis1() << " <> PaqSize()=" << memgr_.PaqSize() << endl;
      throw BAORadioException("BRMultiFitsReader::BRMultiFitsReader/ mff.NAxis1() != memgr_.PaqSize() "); 
    }
    // Extraction de qques parametres utiles depuis les fichiers FITS 
    string fkvs;
    if (fib==0) {
      fkvs=mff_[fib].GetKeyValue("DATEOBS");
      if (fkvs.length()>0)  cdateobs_.Set(fkvs);
      fkvs=mff_[fib].GetKeyValue("TMSTART");
      if (fkvs.length()>0) {
        ctmstart_.Set(fkvs);
        fkvs=mff_[fib].GetKeyValue("TMEND");
        SOPHYA::TimeStamp tmend_=ctmstart_;
        if (fkvs.length()>0) tmend_.Set(fkvs);
        cpaqdeltatime_=((double)(tmend_.DaysPart()-cdateobs_.DaysPart())*86400.+
                        (tmend_.SecondsPart()-cdateobs_.SecondsPart()))/(double)mff_[fib].NAxis2();
        cout << " BRMultiFitsReader/First file (for fiber 0) TMSTART=" << fkvs << " TMEND-START=" 
             << cpaqdeltatime_*(double)mff_[fib].NAxis2() << endl;
      }
    }
    fkvs=mff_[fib].GetKeyValue("FIBERNUM");
    memgr_.FiberId(fib) = atoi( fkvs.c_str() );

    vfilenum_.push_back(imin_);
    vfpos_.push_back(0);
    vpaq_.push_back(BRPaquet(NULL,memgr_.PaqSize()));
    vpchk_.push_back(BRPaqChecker(true,0)); 
    curfc_.push_back(0);
    totnpqrd_.push_back(0);
    totnpqok_.push_back(0);    
  }
}


/* --Methode-- */
void BRMultiFitsReader::run()
{
  setRC(1); 
  try {
    TimeStamp ts; 
    Timer tm("BRMultiFitsReader", false);
    cout << " BRMultiFitsReader::run() - Starting " << ts << " NbFibres()=" << memgr_.NbFibres()
         << "  PaqSize() = " << memgr_.PaqSize() << endl;
    cout << " ...ReadMode: " << ((rdsamefc_)?"Paquets With SameFrameCounter":"All OK paquets") 
         << " signalII.fits IMin=" << imin_ << " IMax=" << imax_ << " IStep=" << istep_ << endl;

    uint_8  prtcnt=0;
    Byte* nextpaq=NULL;
    bool fgok=true;
    while (fgok) {
      if (stop_) break;
      if ( MoveToNextTarget() ) {
        cout << "BRMultiFitsReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
        setRC(7);  fgok=false;  break;
      }
      for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
        nextpaq=GetPaquetTarget(fib);
        if (nextpaq == NULL) { // Cela ne devrait pas arriver 
          cout << "BRMultiFitsReader::run()/Error-A2- GetPaquetTarget(fib)  returned NULL ->STOP 9" << endl;
          setRC(9);  fgok=false;  break;
        }
        vpaq_[fib].Set(nextpaq);
      }
      if (ReadNextAllFibers()) { fgok=false;  break; }
      prtcnt++;
      if ((prtlev_>0)&&(prtcnt%prtmodulo_==0))  {
          cout << "BRMultiFitsReader: NbPaqMFRead=" << prtcnt << " NSameFC=" 
               << totsamefc_ << " / NPaqFib0Read=" << totnpqrd_[0] 
               << " FracSameFC=" << 100*totsamefc_/totnpqrd_[0] << " %" <<  endl;
      }
    }

    MoveToNextTarget();  // Pour faire traiter le dernier paquet si plein 
    MZoneManage(true);   // Nettoyage final 

    cout << " ------------------  BRMultiFitsReader::run() END ----------------- " << endl;
    ts.SetNow();
    tm.SplitQ();
    cout << "BRMultiFitsReader::run():  END reading : " << ts << endl;
    cout << "... NbPaqMFRead=" << prtcnt << " NSameFC=" 
         << totsamefc_ << " / NPaqFib0Read=" << totnpqrd_[0] 
         << " FracSameFC=" << 100*totsamefc_/totnpqrd_[0] << " %" << endl;

    for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
      int perc=0;
      if (totnpqrd_[fib]>0) perc=100*totsamefc_/totnpqrd_[fib];
      cout << " Fiber" << fib << " TotNPaqRd=" << totnpqrd_[fib] << " TotNPaqOK=" << totnpqok_[fib] 
           << " FracSameFC=" << perc << " %" << endl;
      if (prtlev_ > 0)  vpchk_[fib].Print(cout);
    }
    cout << "  TotalDiskRead= " << totnbytesrd_/(1024*1024) << " MBytes Disk-Read rate= " 
         << (double)(totnbytesrd_)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;    
    cout << " BRMultiFitsReader::run()/Timing: \n"; 
    tm.Print();
    cout << " ---------------------------------------------------------- " << endl;

    usleep(250000);       // Attente de traitement du dernier paquet
    memgr_.Stop();        // Arret 

  }  // Fin du bloc try 
  catch (std::exception& exc) {
    cout << " BRMultiFitsReader::run()/catched execption msg=  " << exc.what() << endl;
    setRC(3);   
    return; 
  }
  catch(...) {
    cout << " BRMultiFitsReader::run()/catched unknown ... exception " << endl;
    setRC(4);   
    return; 
  }
  setRC(0);
  return;
}

/* --Methode-- */
bool BRMultiFitsReader::ReadNextAllFibers()
{
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)  {
    if (ReadNext(fib)) return true;  // probleme 
  }
  if (!rdsamefc_ || (memgr_.NbFibres()<2))  {
    uint_8 cfc=curfc_[0];
    bool fgsamefc=true;
    for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
      if (curfc_[fib]!=cfc)   fgsamefc=false;
    }
    if (fgsamefc)  totsamefc_++;  
    return false;  // c'est OK 
  }
  //  On va essayer de lire jusqu'a avoir same_frame_counter 
  bool echec=true;
  while (echec) {
    uint_8 cfc=curfc_[0];
    bool fgsamefc=true;
    for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
      if (curfc_[fib]!=cfc) {
        fgsamefc=false;
        if (curfc_[fib] > cfc)  cfc=curfc_[fib]; 
      }
    }
    if (fgsamefc) { 
      totsamefc_++;  echec=false; return false;  // c'est OK  , same framecounter
    }
    else {  // else !fgsame
      for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) {
        while (curfc_[fib]<cfc) {
          if (ReadNext(fib)) return true;  // probleme 
        }
      }
    }   // fin de  else !fgsame 
  }  // fin de while(echec): lecture jusqu'a same_frame_counter 

  return true; // probleme 
}

/* --Methode-- */
bool BRMultiFitsReader::ReadNext(int fib)
{
  if (!mff_[fib].IsOpen())  return true;
  bool fggood=false;
  while(!fggood) { 
    if (vfpos_[fib] >= mff_[fib].NAxis2()) {
      mff_[fib].Close();
      vfilenum_[fib]+=istep_;
      if (vfilenum_[fib]>imax_)  return true;
      char flnm[1024];
      sprintf(flnm,"%s/signal%d.fits",dirs_[fib].c_str(),vfilenum_[fib]);
      if (prtlev_ > 1)
        cout << " BRMultiFitsReader::ReadNext() opening" << flnm << endl;
      mff_[fib].Open(flnm, MF_Read);
      if (mff_[fib].NAxis1() != packsize_)  {
        cout << " BRMultiFitsReader::ReadNext(fib=" << fib << " File=" << flnm <<
          " NAxis1()= " << mff_[fib].NAxis1() << " <> PaqSize()=" << packsize_ << endl;
        throw BAORadioException("BRMultiFitsReader::ReadNext()/ mff.NAxis1() != memgr_.PaqSize() "); 
      }
      if (fib==0) {  // updating current date from file (fiber 0) 
        string fkvs=mff_[fib].GetKeyValue("DATEOBS");
        if (fkvs.length()>0)  cdateobs_.Set(fkvs);
        fkvs=mff_[fib].GetKeyValue("TMSTART");
        if (fkvs.length()>0) {
          ctmstart_.Set(fkvs);
          cout << " BRMultiFitsReader::ReadNext TMSTART=" << fkvs << endl;
          fkvs=mff_[fib].GetKeyValue("TMEND");
          SOPHYA::TimeStamp tmend_=ctmstart_;
          if (fkvs.length()>0) tmend_.Set(fkvs);
          cpaqdeltatime_=((double)(tmend_.DaysPart()-cdateobs_.DaysPart())*86400.+
                          (tmend_.SecondsPart()-cdateobs_.SecondsPart()))/(double)mff_[fib].NAxis2();
        }
      } 
      vfpos_[fib]=0;
    }
    mff_[fib].ReadB(vpaq_[fib].Begin(), packsize_, vfpos_[fib]*packsize_);
    vfpos_[fib]++;
    totnbytesrd_+=packsize_;
    totnpqrd_[fib]++;
    fggood = vpchk_[fib].Check(vpaq_[fib],curfc_[fib]);
  }
  totnpqok_[fib]++;
  return false;
}

/* --Methode-- */
bool BRMultiFitsReader::MZoneManage(bool fgclean)    // Retourne true si probleme 
{
  /* Pour debug 
  cout << " BRMultiFitsReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_  
       << " max_targ_npaq=" << max_targ_npaq << endl;
  */
  if (mid_ >= 0)  { 
    if (fgclean) memgr_.FreeMemZone(mid_, MemZS_Free);    
    else memgr_.FreeMemZone(mid_, MemZS_Filled);
  }
  mmbuf_ = NULL;  targ_npaq_ = 0;  mid_ = -2;
  for (int fib=0;fib<(int)memgr_.NbFibres() ;fib++) mmbufib_[fib]=NULL;
  if (fgclean)  return false;
  mid_ = memgr_.FindMemZoneId(MemZA_Fill);
  mmbuf_ = memgr_.GetMemZone(mid_);
  if (mmbuf_==NULL)   return true;
  for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)
    mmbufib_[fib]=memgr_.GetMemZone(mid_,fib);
  // Definition temps pour la zone a remplir et numeros de sequence des fichiers en cours de lecture
  memgr_.GetAuxData(mid_)->FillTime().Set(ctmstart_.ToDays()+cpaqdeltatime_*(double)vfpos_[0]/86400.);
  memgr_.GetAuxData(mid_)->FileSequenceNumVec()=vfilenum_; 
  return false;
}

//-------------------------------------------------------
// Classe thread de lecture de fichiers fits BAORadio 
//-------------------------------------------------------

/* --Methode-- */
BRFitsReader::BRFitsReader(RAcqMemZoneMgr& mem, vector<string>& infiles, bool fgnotrl)
  :  memgr(mem), infiles_(infiles), fgnotrl_(fgnotrl)
{
}

/* --Methode-- */
void BRFitsReader::run()
{
  setRC(1);     

  try {
    TimeStamp ts; 
    Timer tm("BRFitsReader", false);
    BRPaqChecker pcheck(!fgnotrl_);  // Verification/comptage des paquets 
    
    size_t totnbytesrd = 0;
    cout << " BRFitsReader::run() - Starting " << ts << " NbFiles=" << infiles_.size() 
         << "  memgr.PaqSize() = " << memgr.PaqSize() << endl;

    uint_4  nfileok = 0; 
    uint_8 nbytesrd = 0;  
    /*  Variables pour la logique des zones memoire et numeros de paquets dans la zone memoire */
    int mid = -2;
    Byte* buff = NULL;  
    int kmp = 0;   
    int kmpmax=memgr.NbPaquets(); 

    int paqsz = 0;
    for(int ifile=0; ifile<infiles_.size(); ifile++) {
      string ffname = infiles_[ifile];
// -------------- Lecture de bytes      
      cout << "BRFitsReader::run() [" << ifile <<"]Ouverture/lecture fichier " << ffname << endl;
      MiniFITSFile mff(ffname, MF_Read);
      cout << "... Type=" << mff.DataTypeToString() << " NAxis1=" << mff.NAxis1() 
           << " NAxis2=" << mff.NAxis2() << endl;
      if (mff.DataType() != MF_Byte) {
        cout << "BRFitsReader::run() PB : DataType!=MF_Byte --> skipping " << endl;
        continue;
      }
// Les fichier FITS contiennent l'entet (24 bytes), mais pas le trailer (16 bytes) si fgnotrl=true
      int incpaqsz=0;
      if (fgnotrl_) {
        incpaqsz=16;
        if (ifile==0) cout << " Warning : FITS files without frame trailers ..." << endl;
      }
      if (paqsz == 0)  {  // premier passage, on fixe la taille de paquet et on verifie compatibilite avec memgr
        paqsz = mff.NAxis1()+incpaqsz;
        if (paqsz != memgr.PaqSize()) {
         cout << "BRFitsReader::run() mff.NAxis1() incompatible with memgr.PaqSize() -> exception " << endl;
         throw SzMismatchError(" fits file size incompatible with memgr.PaqSize()"); 
        }
      }
      else {
        if (paqsz != mff.NAxis1()+incpaqsz) {
        cout << " PB : paqsz=" << paqsz << " != mff.NAxis1()+" << incpaqsz << " --> skipping " << endl;
        continue;
        }
      }   
      if (mid < 0) {
        mid = memgr.FindMemZoneId(MemZA_Fill);  
        buff = memgr.GetMemZone(mid);
        if (buff == NULL) {
              cout << " BRFitsReader::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
              setRC(2); 
               return;  
         }
         kmp=0;
      }
      size_t sx = mff.NAxis1(); 
      size_t sy = mff.NAxis2();      
      int nprt=0;
      for(int j=0; j<sy; j++) {
        mff.ReadB(buff+kmp*paqsz, sx, j*sx);
        BRPaquet paq(NULL, buff+kmp*paqsz, paqsz);
        bool pqok = pcheck.Check(paq);   // Verification du paquet / FrameCounter
        if (!pqok && (nprt < 10)) {
          cout << "--BUG-- i=" << ifile << " mid=" << mid << " j=" << j << " kmp=" << kmp 
               << " paqsz=" << paqsz << endl;
          nprt++; 
          paq.Print();  
        }
        kmp++;
        if (kmp >= kmpmax)  {   // Zone memoire rempli !
          memgr.FreeMemZone(mid, MemZS_Filled);
              mid = -2;
          if (j<sy) { 
            mid = memgr.FindMemZoneId(MemZA_Fill);  
            buff = memgr.GetMemZone(mid);
            if (buff == NULL) {
                  cout << " BRFitsReader::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
                  setRC(2);     
                   return;      
             }
             kmp=0;
          }
        } 
      }
      nfileok++;   
      size_t nbytesrd = sx*sy;
      totnbytesrd += nbytesrd;
    }   // Fin de la boucle sur les fichiers
// Gestion d'une zone partiellement remplie
    if (mid>=0) {
      for(int k=kmp;k<kmpmax;k++) {
        Byte* bp=buff+k*paqsz;
        for(int l=0;l<paqsz;l++) bp[l]=0;
      }
      memgr.FreeMemZone(mid, MemZS_Filled);
    }

//    sprintf(fname,"%s/.log",path_.c_str());
//    ofstream filog(fname);
//    filog << " DataProc::run() - starting log file " << ts << endl;                  
//    filog << " NbFiles=" << nfiles_ << " NBloc/File="  << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
        


    cout << " ------------------  BRFitsReader::run() END ----------------- " << endl;
    ts.SetNow();
    tm.SplitQ();
    cout << "  END reading " << ts << " NFileOK=" << nfileok << endl; 
    cout << "  TotalDiskRead= " << totnbytesrd/(1024*1024) << " MBytes Disk-Read rate= " 
         << (double)(totnbytesrd)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;    
    pcheck.Print(cout);
    cout << " BRFitsReader::run()/Timing: \n"; 
    tm.Print();
    cout << " ---------------------------------------------------------- " << endl;

  }  // Fin du bloc try 
  catch (MiniFITSException& exc) {
    cout << " BRFitsReader::run()/catched MiniFITSException " << exc.Msg() << endl;
    setRC(3);   
    return; 
  }
  catch(...) {
    cout << " BRFitsReader::run()/catched unknown ... exception " << endl;
    setRC(4);   
    return; 
  }
  setRC(0);
  return;
}