#include "racquproc.h"

/* ----------------------------------------
     Projet BAORadio --- LAL 
     2008 - 2010 
-------------------------------------------  */

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

#include "pexceptions.h"
#include "tvector.h"
#include "fioarr.h"
#include "timestamp.h"
#include "fftpserver.h"
#include "fftwserver.h"
#include "histos.h"
#include "histos2.h"

#include "FFTW/fftw3.h"


#include "pciewrap.h"
#include "brpaqu.h"
#include "minifits.h"

/* Fonction, module^2 des nombres complexes */
static inline r_4 Zmod2(complex<r_4> z) 
{ return (z.real()*z.real()+z.imag()*z.imag()); }
static inline r_8 Zmod2(complex<r_8> z) 
{ return (z.real()*z.real()+z.imag()*z.imag()); }
static inline r_4 Zmod2(TwoByteComplex z) 
{ return ((r_4)z.realB()*(r_4)z.realB()+(r_4)z.imagB()*(r_4)z.imagB()); }

//---------------------------------------------------------------------
// Classe thread de traitement/monitoring multifibre Raw/FFT 
//---------------------------------------------------------------------
/* --Methode-- */
MonitorProc::MonitorProc(RAcqMemZoneMgr& mem)
  :  memgr(mem) 
{
  BRAcqConfig bpar;
  par_.Set(bpar.GetParams());
  nmax_ = par_.nmaxProc; 
  if (nmax_==0) nmax_=par_.MaxNbBlocs();
  nmean_ = par_.nmeanProc;
  step_ = par_.stepProc;
  stop_ = false;	
  path_ = bpar.OutputDirectory();
  nfiles_ = 0;
  nblocproc_ = 0;
  nprocpaq_=0;
  npaqsamefc_=0;
  totnprocpaq_=0;
  totnpaqsamefc_=0;
  curfc_.SetSize(memgr.NbFibres(), memgr.NbPaquets());
  cpaqok_.SetSize(memgr.NbFibres(), memgr.NbPaquets());
}

/* --Methode-- */
MonitorProc::~MonitorProc()
{
  //  cout << " **** DBG ***** MonitorProc::~MonitorProc() " << endl;
}

/* --Methode-- */
void MonitorProc::Stop()
{
 stop_=true;
}

/* --Methode-- */
void MonitorProc::run()
{
  setRC(1);	
  int rc=0;
  try {
    TimeStamp ts; 
    cout << " MonitorProc::run() - Starting " << ts << " NMaxMemZones=" << nmax_ 
	 << " NMean=" << nmean_ << " Step=" << step_ << endl;	
    cout << " MonitorProc::run()... - Output Data Path: " << path_ << endl;
    char fname[512];
    sprintf(fname,"%s/monproc.log",path_.c_str());
    ofstream filog(fname);
    filog << " MonitorProc::run() - starting log file " << ts << endl;		       
    filog << " ... NMaxMemZones=" << nmax_ << " NMean=" << nmean_ << " Step=" << step_ << endl;	
    uint_4 paqsz = memgr.PaqSize();
    BRPaquet pq(paqsz); 
    if (par_.fgsinglechannel) {
      spectre_.SetSize(memgr.NbFibres(), pq.DataSize()/2);
      for(int kc=0; kc<memgr.NbFibres(); kc++)  nzm_.push_back(0);
      rc=procData1C(filog);
    }
    else {
      spectre_.SetSize(2*memgr.NbFibres(), pq.DataSize()/4);
      for(int kc=0; kc<2*memgr.NbFibres(); kc++)  nzm_.push_back(0);
      rc=procData2C(filog);
    }
    cout << " ---- MonitorProc::run()/End NBlocProcessed=" << nblocproc_ 
	  << " NFiles=" << nfiles_ << " Rc=" << rc << endl;
    ts.SetNow();
    filog << " ---- MonitorProc::run()/End " << ts << endl;
    filog << " --------- MonitorProc::run()/End NBlocProcessed=" << nblocproc_ 
	  << " NFiles=" << nfiles_ << " Rc=" << rc << endl;
  }
  catch (std::exception& exc) {
    cout << " MonitorProc::run()/catched std::exception " << exc.what() << endl;
    setRC(98);	
    return; 
  }
  catch(...) {
    cout << " MonitorProc::run()/catched unknown ... exception " << endl;
    setRC(99);	
    return; 
  }

  setRC(rc);
  return;
}   

/* --Methode-- */
int MonitorProc::procData1C(ofstream& logf)
{
  cout << " MonitorProc::procData1C() - NOT IMPLEMENTED -> Rc=67" << endl;
  logf << " MonitorProc::procData1C() - NOT IMPLEMENTED -> Rc=67" << endl;
  return 67;  
}

/* --Methode-- */
int MonitorProc::procData2C(ofstream& filog)
{
  BRPaqChecker pcheck[MAXNBFIB];  // Verification/comptage des paquets 
// Initialisation pour calcul FFT 
  uint_4 paqsz = memgr.PaqSize();
  BRPaquet pq(paqsz); 
  TVector<r_4> vx(pq.DataSize()/2);
  vx = (r_4)(0.);
  TVector< complex<r_4> > cfour(pq.DataSize()/4+1);  // composant TF
  
  fftwf_plan plan = fftwf_plan_dft_r2c_1d(vx.Size(), vx.Data(), 
					  (fftwf_complex *)cfour.Data(), FFTW_ESTIMATE); 

  // Histo pour valeurs des bytes dans les paquets 
  Histo* phist[2*MAXNBFIB] = {NULL,NULL,NULL,NULL, NULL,NULL,NULL,NULL};
  Histo2D* ph2[2*MAXNBFIB] = {NULL,NULL,NULL,NULL, NULL,NULL,NULL,NULL};
  Histo* phexp[2*MAXNBFIB] = {NULL,NULL,NULL,NULL, NULL,NULL,NULL,NULL};
  if (par_.fgdatafft) {    // data de type FFT   
    for(sa_size_t lc=0; lc<2*memgr.NbFibres(); lc++) {
      phist[lc] = new Histo(-128.5,+128.5,257);
      ph2[lc]   = new Histo2D(-128.5,+128.5,257, -128.5,+128.5,257);
      phexp[lc] = new Histo(-128.5,+128.5,257);
    }
  }
  else {
    for(sa_size_t lc=0; lc<2*memgr.NbFibres(); lc++)  
      phist[lc] = new Histo(-0.5,256.5,257);    
  }

  TimeStamp ts;   
  char fname[512];
  nfiles_ = 0;  			        
  for (uint_4 kmz=0; kmz<nmax_; kmz++) {
    if (stop_) break;
    if (memgr.GetRunState() == MemZR_Stopped) break;

    int mid = memgr.FindMemZoneId(MemZA_Proc);
    Byte* buffg = memgr.GetMemZone(mid);
    if (buffg == NULL) {
      cout << " MonitorProc::procData2C()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
      break;	
    }
    if ((step_>1)&&(kmz%step_ != 0))  {
      memgr.FreeMemZone(mid, MemZS_Proc);
      continue;
    }
    sa_size_t lc=0;
    Byte* fbuff[MAXNBFIB];
    for(uint_4 fib=0; fib<memgr.NbFibres(); fib++)  {  // Boucle sur les fibres 
      fbuff[fib] = memgr.GetMemZone(mid,fib);
      if (fbuff[fib] == NULL) { // cela ne devrait pas arriver 
	cout << " MonitorProc::procData2C()/ERROR memgr.GetMemZone(" << mid << "," << fib << ") -> NULL" << endl;
	return 9;	
      }
    }
    
    cpaqok_ = (uint_1)0;
    curfc_ = (uint_8)0;
    for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
      for(uint_4 i=0; i<memgr.NbPaquets(); i++) {
	BRPaquet paq(fbuff[fib]+i*paqsz, paqsz); 
	bool pqok=pcheck[fib].Check(paq,curfc_(fib,i));   // Verification du paquet / FrameCounter 
	if (!pqok)  continue;
	cpaqok_(fib,i) = 1;
	sa_size_t lc=2*fib;
	if (par_.fgdatafft) {    // Traitement data de type FFT 
	  TwoByteComplex* tbcp=paq.Data1C();
	  phexp[lc]->Add((r_4)tbcp[0].realD());
	  for(sa_size_t j=1; j<spectre_.NCols(); j++) {
	    phist[lc]->Add((r_4)tbcp[j].realD());
	    phist[lc]->Add((r_4)tbcp[j].imagD());
	    ph2[lc]->Add((r_4)tbcp[j].realD(), (r_4)tbcp[j].imagD());
	    spectre_(lc,j) += Zmod2(tbcp[j]);
	  }
	  nzm_[lc]++;  	
	  tbcp=paq.Data2C();   lc++;
	  phexp[lc]->Add((r_4)tbcp[0].realD());
	  for(sa_size_t j=1; j<spectre_.NCols(); j++) {
	    phist[lc]->Add((r_4)tbcp[j].realD());
	    phist[lc]->Add((r_4)tbcp[j].imagD());
	    ph2[lc]->Add((r_4)tbcp[j].realD(), (r_4)tbcp[j].imagD());
	    spectre_(lc,j) += Zmod2(tbcp[j]);
	  }
	  nzm_[lc]++;
	}
	else {   // Traitement RawData 
	  for(sa_size_t j=0; j<vx.Size(); j++) {
	    phist[lc]->Add((r_8)(*(paq.Data1()+j)));
	    vx(j) = (r_4)(*(paq.Data1()+j))-127.5;
	  }
	  fftwf_execute(plan); 
	  //   ffts_.FFTForward(vx, cfour_);
	  for(sa_size_t j=0; j<spectre_.NCols(); j++) 
	    spectre_(lc,j) += Zmod2(cfour(j+1));
	  nzm_[lc]++;      lc++;
	  for(sa_size_t j=0; j<vx.Size(); j++) {
	    phist[lc]->Add((r_8)(*(paq.Data2()+j)));
	    vx(j) = (r_4)(*(paq.Data2()+j))-127.5;
	  }
	  fftwf_execute(plan); 
	  //    ffts_.FFTForward(vx, cfour_);
	  for(sa_size_t j=0; j<spectre_.NCols(); j++) 
	    spectre_(lc,j) += Zmod2(cfour(j+1));
	  nzm_[lc]++;
	}	
      }  // FIN de la boucle sur les paquets 
    }   // Boucle sur les fibres 
    memgr.FreeMemZone(mid, MemZS_Proc);
    CheckFrameCounters();

    nblocproc_ ++;   
    totnprocpaq_ += memgr.NbPaquets();  nprocpaq_ += memgr.NbPaquets();
    bool fgnzm=true;
    for(int lc=0; lc<2*memgr.NbFibres(); lc++)  
      if (nzm_[lc]<nmean_) fgnzm=false;
    
    if (fgnzm) {
      char buff[32];
      for(sa_size_t lc=0; lc<2*memgr.NbFibres(); lc++)  {
	spectre_.Row(lc) /= (r_4)nzm_[lc];
	sprintf(buff,"NPaqMoy%d",(int)lc);
	spectre_.Info()[buff] = nzm_[lc];
      }
      sprintf(fname,"%s/meanspec%d.ppf",path_.c_str(),(int)nfiles_);
      nfiles_++;
      POutPersist po(fname);
      po << PPFNameTag("spectre") << spectre_; 
      spectre_ = (r_4)(0.);
      char buftag[32];
      for(int lc=0; lc<2*memgr.NbFibres(); lc++)  { 
	sprintf(buftag,"hvalV%d",(int)lc);
	po << PPFNameTag(buftag) << (*phist[lc]);
	phist[lc]->Zero();
	if (par_.fgdatafft) {    // data de type FFT   
	  sprintf(buftag,"h2dV%d",(int)lc);
	  po << PPFNameTag(buftag) << (*ph2[lc]);
	  ph2[lc]->Zero();
	  sprintf(buftag,"hexpV%d",(int)lc);
	  po << PPFNameTag(buftag) << (*phexp[lc]);
	  phexp[lc]->Zero();
	}
	nzm_[lc]=0;
      }
      ts.SetNow();
      // Calcul / impression fraction des paquets avec same-framecounter
      int fracsame=0;
      if (nprocpaq_>0)  fracsame=100*npaqsamefc_/nprocpaq_;
      int totfracsame=0;
      if (totnprocpaq_>0)  totfracsame=100*totnpaqsamefc_/totnprocpaq_;
      filog << ts << " :  proc file (meanspectra) " << fname << endl;
      filog << "  NBlocProcessed=" << nblocproc_ << " NSameFC=" << totnpaqsamefc_ << " / " << totnprocpaq_
	    << " -> " << totfracsame << " % (LastPqs: " << npaqsamefc_ <<  " / " << nprocpaq_ 
	    << " -> " << fracsame << " % )" << endl;
      cout << " MonitorProc::procData2C() " << ts << " : created file  " << fname << endl;
      cout << "  NBlocProcessed=" << nblocproc_ << " NSameFC=" << totnpaqsamefc_ << " / " << totnprocpaq_
	   << " -> " << totfracsame << " % (LastPqs: " << npaqsamefc_ <<  " / " << nprocpaq_ 
	   << " -> " << fracsame << " % )" << endl;
      nprocpaq_=npaqsamefc_=0;
    }
  }   // Fin de boucle sur les kmz ( bloc MemZoneMgr  a traiter )

  bool fgnzm=false;
  for(int lc=0; lc<2*memgr.NbFibres(); lc++)  
    if (nzm_[lc]>0) fgnzm=true;
  
  if (fgnzm) {
    char buff[32];
    for(sa_size_t lc=0; lc<2*memgr.NbFibres(); lc++)  {
      if (nzm_[lc]>0)  spectre_.Row(lc) /= (r_4)nzm_[lc];
      sprintf(buff,"NPaqMoy%d",(int)lc);
      spectre_.Info()[buff] = nzm_[lc];
    }
    sprintf(fname,"%s/meanspec%d.ppf",path_.c_str(),(int)nfiles_);
    POutPersist po(fname);
    po << PPFNameTag("spectre") << spectre_; 
    spectre_ = (r_4)(0.);
    char buftag[32];
    for(int lc=0; lc<2*memgr.NbFibres(); lc++)  { 
      sprintf(buftag,"hvalV%d",(int)lc);
      po << PPFNameTag(buftag) << (*phist[lc]);
      delete phist[lc];
      if (par_.fgdatafft) {    // data de type FFT   
	sprintf(buftag,"h2dV%d",(int)lc);
	po << PPFNameTag(buftag) << (*ph2[lc]);
	delete ph2[lc];
	sprintf(buftag,"hexpV%d",(int)lc);
	po << PPFNameTag(buftag) << (*phexp[lc]);
	delete phexp[lc];
      }
      nzm_[lc]=0;
    }
    ts.SetNow();
    // Calcul / impression fraction des paquets avec same-framecounter
    int fracsame=0;
    if (nprocpaq_>0)  fracsame=100*npaqsamefc_/nprocpaq_;
    int totfracsame=0;
    if (totnprocpaq_>0)  totfracsame=100*totnpaqsamefc_/totnprocpaq_;
    filog << ts << " :  proc file (meanspectra) " << fname << endl;		       
    filog << "  NBlocProcessed=" << nblocproc_ << " NSameFC=" << totnpaqsamefc_ << " / " << totnprocpaq_
	  << " -> " << totfracsame << " % (LastPqs: " << npaqsamefc_ <<  " / " << nprocpaq_ 
	  << " -> " << fracsame << " % )" << endl;
    cout << " MonitorProc::procData2C() " << ts << " : created file  " << fname << endl;
    cout << "  NBlocProcessed=" << nblocproc_ << " NSameFC=" << totnpaqsamefc_ << " / " << totnprocpaq_
	 << " -> " << totfracsame << " % (LastPqs: " << npaqsamefc_ <<  " / " << nprocpaq_ 
	 << " -> " << fracsame << " % )" << endl;
  }
  return 0;
}

/* --Methode-- */
int MonitorProc::CheckFrameCounters()
{
  if (memgr.NbFibres()<2)  { 
    npaqsamefc_++;  totnpaqsamefc_++;
    return 99;
  }
  sa_size_t  pidx[MAXNBFIB];
  sa_size_t maxidx=curfc_.NCols();
  uint_8 cfc=0;
  for(uint_4 fib=0; fib<curfc_.NRows(); fib++) {
    pidx[fib]=0;
    while((pidx[fib]<maxidx)&&(cpaqok_(fib,pidx[fib])==0)) pidx[fib]++;
  }

  bool fgsuite=true;
  while (fgsuite) {   // Boucle sur l'ensemble des paquets
    for(uint_4 fib=0; fib<curfc_.NRows(); fib++)  {
      if ((pidx[fib]>=maxidx)||(cpaqok_(fib,pidx[fib])==0))  { fgsuite=false; break; } 
    }
    if (!fgsuite) break;
    cfc=curfc_(0,pidx[0]);
    bool fgsame=true;
    for(uint_4 fib=1; fib<curfc_.NRows(); fib++) {
      if (curfc_(fib,pidx[fib])!=cfc) {
	fgsame=false;
	if (curfc_(fib,pidx[fib]) > cfc)  cfc=curfc_(fib,pidx[fib]);
      }
    }
    if (fgsame) {
      npaqsamefc_++;  totnpaqsamefc_++;
      for(uint_4 fib=0; fib<curfc_.NRows(); fib++) {
	pidx[fib]++;
	while((pidx[fib]<maxidx)&&(cpaqok_(fib,pidx[fib])==0)) pidx[fib]++;
      }
    }   // fin if (fgsame) 
    else {  // else !fgsame
      for(uint_4 fib=0; fib<memgr.NbFibres(); fib++) {
	if (curfc_(fib,pidx[fib])<cfc) {
	  pidx[fib]++;
	  while((pidx[fib]<maxidx)&&(cpaqok_(fib,pidx[fib])==0)) pidx[fib]++;
	}
      }
    }   // fin de  else !fgsame 
  }  // Fin de while sur l'ensemble des paquets
  return 0;
}


static struct sigaction act;
//-------------------------------------------------------
// Classe thread de traitement  avec 1 voie par frame
//-------------------------------------------------------

void DataProcSignal(int s)
{
  cout <<"............................................... receive CtrlC" << endl;

}

DataProc::DataProc(RAcqMemZoneMgr& mem, string& path, uint_4 nmean, uint_4 step, uint_4 nmax)
  :  memgr(mem) 
{
  nmax_ = nmax;	
  nmean_ = nmean;
  step_ = step;
  stop_ = false;	
  path_ = path;	
  act.sa_handler=DataProcSignal;
}


void DataProc::Stop()
{
 stop_=true;
 //  cout <<" DataProc::Stop ... > STOP " << endl;
}

void DataProc::run()
{
  
  // sigaddset(&act.sa_mask,SIGINT);  // pour proteger le transfert DMA
  // sigaction(SIGINT,&act,NULL);  	
  setRC(1);	
  try {
    TimeStamp ts; 
    cout << " DataProc::run() - Starting " << ts << " NMaxMemZones=" << nmax_ 
	 << " NMean=" << nmean_ << " Step=" << step_ << endl;	
    cout << " DataProc::run()... - Output Data Path: " << path_ << endl;
    char fname[512];
    sprintf(fname,"%s/proc.log",path_.c_str());
    ofstream filog(fname);
    filog << " DataProc::run() - starting log file " << ts << endl;		       
    filog << " ... NMaxMemZones=" << nmax_ << " NMean=" << nmean_ << " Step=" << step_ << endl;	

// Initialisation pour clcul FFT 
    TVector< complex<r_4> > cfour;  // composant TF
    uint_4 paqsz = memgr.PaqSize();
    BRPaquet pq(NULL, NULL, paqsz); 
    TVector<r_4> vx(pq.DataSize());
    vx = (r_4)(0.);
    FFTPackServer ffts;
    ffts.FFTForward(vx, cfour);
    TVector<r_4> spectre;
    spectre.ReSize(cfour.Size());
    
    fftwf_plan plan = fftwf_plan_dft_r2c_1d(vx.Size(), vx.Data(), 
	  		       	       (fftwf_complex *)cfour.Data(), FFTW_ESTIMATE); 

    uint_4 ifile = 0;  			        
    uint_4 nzm = 0;
    for (uint_4 kmz=0; kmz<nmax_; kmz++) {
      if (stop_) break;
      int mid = memgr.FindMemZoneId(MemZA_Proc);
      Byte* buff = memgr.GetMemZone(mid);
      if (buff == NULL) {
  	cout << " DataProc::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
  	setRC(2);	
  	return;	
      }
      BRPaquet paq0(NULL, buff, paqsz); 
      for(uint_4 i=0; i<memgr.NbPaquets(); i+=step_) {
  	BRPaquet paq(NULL, buff+i*paqsz, paqsz); 
  	Byte min = 255;
  	Byte max = 0;
  	    
  	for(sa_size_t j=0; j<vx.Size(); j++) 
  	  vx(j) = (r_4)(*(paq.Data1()+j))-127.5;
  	fftwf_execute(plan); 
	//        ffts_.FFTForward(vx, cfour_);
        for(sa_size_t j=0; j<spectre.Size(); j++) 
          spectre(j) += Zmod2(cfour(j)); 
        nzm++;
      }
      if ((nzm >= nmean_) ||(kmz==(nmax_-1))) {
        spectre /= (r_4)(nzm);
        spectre.Info()["NPaqMoy"] = nzm;
    	sprintf(fname,"%s/spectre%d.ppf",path_.c_str(),(int)ifile);
        POutPersist po(fname);
        po << spectre; 
        spectre = (r_4)(0.);
        nzm = 0;  ifile++; 
        ts.SetNow();
        filog << ts << " :  proc file  " << fname << endl;		       
        cout << " DataProc::run() " << ts << " : created file  " << fname << endl;
      }
      
      memgr.FreeMemZone(mid, MemZS_Proc);
    }
  }
  catch (std::exception& exc) {
    cout << " DataProc::run()/catched std::exception " << exc.what() << endl;
    setRC(3);	
    return; 
  }
  catch(...) {
    cout << " DataProc::run()/catched unknown ... exception " << endl;
    setRC(4);	
    return; 
  }
  setRC(0);
  return;
}   

//---------------------------------------------------------------
// Classe thread de traitement donnees ADC avec 2 voies par frame
//---------------------------------------------------------------

DataProc2C::DataProc2C(RAcqMemZoneMgr& mem, string& path, uint_4 nmean, uint_4 step, uint_4 nmax)
  :  memgr(mem) 
{
  nmax_ = nmax;	
  nmean_ = nmean;
  step_ = step;
  stop_ = false;	
  path_ = path;	
  act.sa_handler=DataProcSignal;
}
void DataProc2C::Stop()
{
 stop_=true;
 // cout <<" DataProc2C::Stop ... > STOP " << endl;
}

void DataProc2C::run()
{
  //   sigaction(SIGINT,&act,NULL);  	
  setRC(1);	
  try {
    TimeStamp ts; 
    cout << " DataProc2C::run() - Starting " << ts << " NMaxMemZones=" << nmax_ 
	 << " NMean=" << nmean_ << " Step=" << step_ << endl;	
    cout << " DataProc::run()... - Output Data Path: " << path_ << endl;
    char fname[512];
    sprintf(fname,"%s/proc.log",path_.c_str());
    ofstream filog(fname);
    filog << " DataProc2C::run() - starting log file " << ts << endl;		       
    filog << " ... NMaxMemZones=" << nmax_ << " NMean=" << nmean_ << " Step=" << step_ << endl;	

// Initialisation pour clcul FFT 
    TVector< complex<r_4> > cfour;  // composant TF
    uint_4 paqsz = memgr.PaqSize();
    BRPaquet pq(NULL, NULL, paqsz); 
    TVector<r_4> vx(pq.DataSize()/2);
    vx = (r_4)(0.);
    FFTPackServer ffts;
    ffts.FFTForward(vx, cfour);
    TVector<r_4> spectreV1, spectreV2;
    spectreV1.ReSize(cfour.Size());
    spectreV2.ReSize(cfour.Size());
    
    fftwf_plan plan = fftwf_plan_dft_r2c_1d(vx.Size(), vx.Data(), 
	  			       (fftwf_complex *)cfour.Data(), FFTW_ESTIMATE); 

    uint_4 ifile = 0;  			        
    uint_4 nzm = 0;
    for (uint_4 kmz=0; kmz<nmax_; kmz++) {
      if (stop_) break;
      int mid = memgr.FindMemZoneId(MemZA_Proc);
      Byte* buff = memgr.GetMemZone(mid);
      if (buff == NULL) {
  	cout << " DataProc2C::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
  	setRC(2);	
  	return;	
      }
      BRPaquet paq0(NULL, buff, paqsz); 
      for(uint_4 i=0; i<memgr.NbPaquets(); i+=step_) {
  	BRPaquet paq(NULL, buff+i*paqsz, paqsz); 
  	Byte min = 255;
  	Byte max = 0;
  	    
  	for(sa_size_t j=0; j<vx.Size(); j++) 
  	  vx(j) = (r_4)(*(paq.Data1()+j))-127.5;
  	fftwf_execute(plan); 
        //   ffts_.FFTForward(vx, cfour_);
        for(sa_size_t j=0; j<spectreV1.Size(); j++) 
          spectreV1(j) += Zmod2(cfour(j));
 
   	for(sa_size_t j=0; j<vx.Size(); j++) 
 	  vx(j) = (r_4)(*(paq.Data2()+j))-127.5;
  	fftwf_execute(plan); 
        //    ffts_.FFTForward(vx, cfour_);
        for(sa_size_t j=0; j<spectreV2.Size(); j++) 
          spectreV2(j) += Zmod2(cfour(j)); 

        nzm++;
      }
      if ((nzm >= nmean_) ||(kmz==(nmax_-1))) {
        spectreV1 /= (r_4)(nzm);
        spectreV2 /= (r_4)(nzm);
        spectreV1.Info()["NPaqMoy"] = nzm;
        spectreV2.Info()["NPaqMoy"] = nzm;
	{
    	sprintf(fname,"%s/spectre2C_%d.ppf",path_.c_str(),(int)ifile);
        POutPersist po(fname);
        po << PPFNameTag("specV1") << spectreV1; 
        po << PPFNameTag("specV2") << spectreV2; 
	}
        spectreV1 = (r_4)(0.);
        spectreV2 = (r_4)(0.);
        nzm = 0;  ifile++; 
        ts.SetNow();
        filog << ts << " :  proc file  " << fname << endl;		       
        cout << " DataProc2C::run() " << ts << " : created file  " << fname << endl;
      }
      
      memgr.FreeMemZone(mid, MemZS_Proc);
    }
  }
  catch (std::exception& exc) {
    cout << " DataProc::run()/catched std::exception " << exc.what() << endl;
    setRC(3);	
    return; 
  }
  catch(...) {
    cout << " DataProc2C::run()/catched unknown ... exception " << endl;
    setRC(4);	
    return; 
  }
  setRC(0);
  return;
}   




//---------------------------------------------------------------
// Classe thread de traitement donnees FFT avec 2 voies par frame
//---------------------------------------------------------------

inline r_4 Zmod2TwoByte(TwoByteComplex z) 
{ return (z.realD()*z.realD()+z.imagD()*z.imagD()); }

DataProcFFT2C::DataProcFFT2C(RAcqMemZoneMgr& mem, string& path, uint_4 nmean, uint_4 step, uint_4 nmax)
  :  memgr(mem) 
{
  nmax_ = nmax;	
  nmean_ = nmean;
  step_ = step;
  stop_ = false;	
  path_ = path;	
 act.sa_handler=DataProcSignal;
}
void DataProcFFT2C::Stop()
{
 stop_=true;
 // cout <<" DataProcFFT2C::Stop ... > STOP " << endl;
}

void DataProcFFT2C::run()
{
  // sigaction(SIGINT,&act,NULL);  	
  setRC(1);	
  try {
    TimeStamp ts; 
    cout << " DataProcFFT2C::run() - Starting " << ts << " NMaxMemZones=" << nmax_ 
	 << " NMean=" << nmean_ << " Step=" << step_ << endl;	
    cout << " DataProc::run()... - Output Data Path: " << path_ << endl;
    char fname[512];
    sprintf(fname,"%s/proc.log",path_.c_str());
    ofstream filog(fname);
    filog << " DataProcFFT2C::run() - starting log file " << ts << endl;		       
    filog << " ... NMaxMemZones=" << nmax_ << " NMean=" << nmean_ << " Step=" << step_ << endl;	

// Initialisation pour clcul FFT 
    TVector< complex<r_4> > cfour;  // composant TF
    uint_4 paqsz = memgr.PaqSize();
    BRPaquet pq(NULL, NULL, paqsz); 
 
    TVector<r_4> spectreV1(pq.DataSize()/4+1), spectreV2(pq.DataSize()/4+1);    

    uint_4 ifile = 0;  			        
    for (uint_4 kmz=0; kmz<nmax_; kmz++) {
      if (stop_ ) break;
      int mid = memgr.FindMemZoneId(MemZA_Proc);
      Byte* buff = memgr.GetMemZone(mid);
      if (buff == NULL) {
  	cout << " DataProcFFT2C::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
  	setRC(2);	
  	return;	
      }
      BRPaquet paq0(NULL, buff, paqsz); 
      uint_4 nzm = 0;
      for(uint_4 i=0; i<memgr.NbPaquets(); i+=step_) {
  	BRPaquet paq(NULL, buff+i*paqsz, paqsz); 
  	Byte min = 255;
  	Byte max = 0;
  	        
	TwoByteComplex* zz;
	// Traitement Voie 1
        zz = (TwoByteComplex*)paq.Data1();
        spectreV1(0) += zz[0].realD()*zz[0].realD();  // Composante continue, partie reelle uniquement
        for(sa_size_t j=1; j<spectreV1.Size()-1; j++)  spectreV1(j) += Zmod2TwoByte(zz[j]);

        spectreV1(spectreV1.Size()-1) += zz[0].imagD()*zz[0].imagD();  // Freq. Nyquist a N/2 

	// Traitement Voie 2
        zz = (TwoByteComplex*)paq.Data2();
        spectreV2(0) += zz[0].realD()*zz[0].realD();  // Composante continue, partie reelle uniquement
        for(sa_size_t j=1; j<spectreV2.Size()-1; j++)  spectreV2(j) += Zmod2TwoByte(zz[j]);

        spectreV2(spectreV2.Size()-1) += zz[0].imagD()*zz[0].imagD();  // Freq. Nyquist a N/2 

        nzm++;
      }
      if ((nzm >= nmean_) ||(kmz==(nmax_-1))) {
        spectreV1 /= (r_4)(nzm);
        spectreV2 /= (r_4)(nzm);
        spectreV1.Info()["NPaqMoy"] = nzm;
        spectreV2.Info()["NPaqMoy"] = nzm;
	{
    	sprintf(fname,"%s/spectreFFT2C_%d.ppf",path_.c_str(),(int)ifile);
        POutPersist po(fname);
        po << PPFNameTag("specV1") << spectreV1; 
        po << PPFNameTag("specV2") << spectreV2; 
	}
        spectreV1 = (r_4)(0.);
        spectreV2 = (r_4)(0.);
        nzm = 0;  ifile++; 
        ts.SetNow();
        filog << ts << " :  proc file  " << fname << endl;		       
        cout << " DataProcFFT2C::run() " << ts << " : created file  " << fname << endl;
      }
      
      memgr.FreeMemZone(mid, MemZS_Proc);
    }
  }
  catch (std::exception& exc) {
    cout << " DataProc::run()/catched std::exception " << exc.what() << endl;
    setRC(3);	
    return; 
  }
  catch(...) {
    cout << " DataProcFFT2C::run()/catched unknown ... exception " << endl;
    setRC(4);	
    return; 
  }
  setRC(0);
  return;
}