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

Last change on this file since 3684 was 3683, checked in by ansari, 16 years ago

Mise a jour et ajout de fichier pour taritement multifibres apres

prise de donnees de Nov2009 a Pittsburgh

  • Introduction des classes BRMultiFitsReader et BRBaseProcessor Reza, 27/11/2009
File size: 12.5 KB
Line 
1//----------------------------------------------------------------
2// Projet BAORadio - (C) LAL/IRFU 2008-2010
3// Classes de threads pour lecture fichiers fits BAORadio
4//----------------------------------------------------------------
5
6
7#include "brfitsrd.h"
8
9#include <stdlib.h>
10#include <unistd.h>
11#include <fstream>
12#include <exception>
13
14#include "pexceptions.h"
15#include "timestamp.h"
16#include "ctimer.h"
17
18#include "brpaqu.h"
19
20#include "resusage.h" // Pour mesure temps elapsed/CPU ...
21#include <sys/time.h> // pour gettimeofday
22
23using namespace SOPHYA;
24
25//---------------------------------------------------------------------
26// Classe thread de lecture de Multi-fibres de fichiers FITS BAORadio
27//---------------------------------------------------------------------
28
29/* --Methode-- */
30BRMultiFitsReader::BRMultiFitsReader(RAcqMemZoneMgr& mem, vector<string>& dirs, bool rdsamefc,
31 uint_4 imin, uint_4 imax, uint_4 istep)
32 : memgr_(mem), dirs_(dirs), rdsamefc_(rdsamefc), imin_(imin), imax_(imax), istep_(istep)
33{
34 SetPrintLevel();
35 totnbytesrd_ = 0;
36 totsamefc_ = 0;
37 if (memgr_.NbFibres() > MAXANAFIB)
38 throw BAORadioException("BRMultiFitsReader::BRMultiFitsReader/ NbFibres>MAXANAFIB ");
39 if (dirs_.size() != memgr_.NbFibres())
40 throw BAORadioException("BRMultiFitsReader::BRMultiFitsReader/ NbFibres != Nb Data Directories");
41
42 packsize_=memgr_.PaqSize();
43 mid_=-2;
44 mmbuf_=NULL;
45 max_targ_npaq = memgr_.NbPaquets();
46 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) mmbufib_[fib]=NULL;
47
48 char flnm[1024];
49 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
50 sprintf(flnm,"%s/signal%d.fits",dirs_[fib].c_str(),imin_);
51 mff_[fib].Open(flnm, MF_Read);
52 if (mff_[fib].NAxis1() != memgr_.PaqSize()) {
53 cout << " BRMultiFitsReader::BRMultiFitsReader/ fib=" << fib << " File=" << flnm <<
54 " NAxis1()= " << mff_[fib].NAxis1() << " <> PaqSize()=" << memgr_.PaqSize() << endl;
55 throw BAORadioException("BRMultiFitsReader::BRMultiFitsReader/ mff.NAxis1() != memgr_.PaqSize() ");
56 }
57 vfilenum_.push_back(imin_);
58 vfpos_.push_back(0);
59 vpaq_.push_back(BRPaquet(NULL,memgr_.PaqSize()));
60 vpchk_.push_back(BRPaqChecker(true,0));
61 curfc_.push_back(0);
62 totnpqrd_.push_back(0);
63 totnpqok_.push_back(0);
64 }
65}
66
67
68/* --Methode-- */
69void BRMultiFitsReader::run()
70{
71 setRC(1);
72 try {
73 TimeStamp ts;
74 Timer tm("BRMultiFitsReader", false);
75 cout << " BRMultiFitsReader::run() - Starting " << ts << " NbFibres()=" << memgr_.NbFibres()
76 << " PaqSize() = " << memgr_.PaqSize() << endl;
77 cout << " ...ReadMode: " << ((rdsamefc_)?"Paquets With SameFrameCounter":"All OK paquets")
78 << " signalII.fits IMin=" << imin_ << " IMax=" << imax_ << " IStep=" << istep_ << endl;
79
80 Byte* nextpaq=NULL;
81 bool fgok=true;
82 while (fgok) {
83 if (stop_) break;
84 if ( MoveToNextTarget() ) {
85 cout << "BRMultiFitsReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
86 setRC(7); fgok=false; break;
87 }
88 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
89 nextpaq=GetPaquetTarget(fib);
90 if (nextpaq == NULL) { // Cela ne devrait pas arriver
91 cout << "BRMultiFitsReader::run()/Error-A2- GetPaquetTarget(fib) returned NULL ->STOP 9" << endl;
92 setRC(9); fgok=false; break;
93 }
94 vpaq_[fib].Set(nextpaq);
95 }
96 if (ReadNextAllFibers()) { fgok=false; break; }
97 }
98
99 MoveToNextTarget(); // Pour faire traiter le dernier paquet si plein
100 MZoneManage(true); // Nettoyage final
101 usleep(50000); // Attente de traitement du dernier paquet
102 memgr_.Stop(); // Arret
103
104 cout << " ------------------ BRMultiFitsReader::run() END ----------------- " << endl;
105 ts.SetNow();
106 tm.SplitQ();
107 cout << " END reading : " << ts ;
108 if (rdsamefc_) cout << " NSameFC=" << totsamefc_ << endl;
109 else cout << endl;
110 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
111 int perc=0;
112 if (totnpqrd_[fib]>0) perc=100*totsamefc_/totnpqrd_[fib];
113 cout << " Fiber" << fib << " TotNPaqRd=" << totnpqrd_[fib] << " TotNPaqOK=" << totnpqok_[fib]
114 << " FracSameFC=" << perc << " %" << endl;
115 }
116 cout << " TotalDiskRead= " << totnbytesrd_/(1024*1024) << " MBytes Disk-Read rate= "
117 << (double)(totnbytesrd_)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;
118 cout << " BRMultiFitsReader::run()/Timing: \n";
119 tm.Print();
120 cout << " ---------------------------------------------------------- " << endl;
121
122 } // Fin du bloc try
123 catch (std::exception& exc) {
124 cout << " BRMultiFitsReader::run()/catched execption msg= " << exc.what() << endl;
125 setRC(3);
126 return;
127 }
128 catch(...) {
129 cout << " BRMultiFitsReader::run()/catched unknown ... exception " << endl;
130 setRC(4);
131 return;
132 }
133 setRC(0);
134 return;
135}
136
137/* --Methode-- */
138bool BRMultiFitsReader::ReadNextAllFibers()
139{
140 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
141 if (ReadNext(fib)) return true; // probleme
142 }
143 if (!rdsamefc_ || (memgr_.NbFibres()<2)) {
144 totsamefc_++; return false; // c'est OK
145 }
146 uint_8 cfc=curfc_[0];
147 bool fgsamefc=true;
148 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
149 if (curfc_[fib]!=cfc) {
150 fgsamefc=false;
151 if (curfc_[fib] > cfc) cfc=curfc_[fib];
152 }
153 }
154 if (fgsamefc) {
155 totsamefc_++; return false; // c'est OK , same framecounter
156 }
157 else { // else !fgsame
158 for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) {
159 while (curfc_[fib]<cfc) {
160 if (ReadNext(fib)) return true; // probleme
161 }
162 }
163 } // fin de else !fgsame
164 totsamefc_++;
165 return false; // c'est OK
166}
167
168/* --Methode-- */
169bool BRMultiFitsReader::ReadNext(int fib)
170{
171 if (!mff_[fib].IsOpen()) return true;
172 bool fggood=false;
173 while(!fggood) {
174 if (vfpos_[fib] >= mff_[fib].NAxis2()) {
175 mff_[fib].Close();
176 vfilenum_[fib]++;
177 if (vfilenum_[fib]>imax_) return true;
178 char flnm[1024];
179 sprintf(flnm,"%s/signal%d.fits",dirs_[fib].c_str(),vfilenum_[fib]++);
180 if (prtlev_ > 0)
181 cout << " BRMultiFitsReader::ReadNext() opening" << flnm << endl;
182 mff_[fib].Open(flnm, MF_Read);
183 if (mff_[fib].NAxis1() != packsize_) {
184 cout << " BRMultiFitsReader::ReadNext(fib=" << fib << " File=" << flnm <<
185 " NAxis1()= " << mff_[fib].NAxis1() << " <> PaqSize()=" << packsize_ << endl;
186 throw BAORadioException("BRMultiFitsReader::ReadNext()/ mff.NAxis1() != memgr_.PaqSize() ");
187 }
188 vfpos_[fib]=0;
189 }
190 mff_[fib].ReadB(vpaq_[fib].Begin(), packsize_, vfpos_[fib]*packsize_);
191 vfpos_[fib]++;
192 totnbytesrd_+=packsize_;
193 totnpqrd_[fib]++;
194 fggood = vpchk_[fib].Check(vpaq_[fib],curfc_[fib]);
195 }
196 totnpqok_[fib]++;
197 return false;
198}
199
200/* --Methode-- */
201bool BRMultiFitsReader::MZoneManage(bool fgclean) // Retourne true si probleme
202{
203 /* Pour debug
204 cout << " BRMultiFitsReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
205 << " max_targ_npaq=" << max_targ_npaq << endl;
206 */
207 if (mid_ >= 0) {
208 if (fgclean) memgr_.FreeMemZone(mid_, MemZS_Free);
209 memgr_.FreeMemZone(mid_, MemZS_Filled);
210 }
211 mmbuf_ = NULL; targ_npaq_ = 0; mid_ = -2;
212 for (int fib=0;fib<(int)memgr_.NbFibres() ;fib++) mmbufib_[fib]=NULL;
213 if (fgclean) return false;
214 mid_ = memgr_.FindMemZoneId(MemZA_Fill);
215 mmbuf_ = memgr_.GetMemZone(mid_);
216 if (mmbuf_==NULL) return true;
217 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)
218 mmbufib_[fib]=memgr_.GetMemZone(mid_,fib);
219 return false;
220}
221
222//-------------------------------------------------------
223// Classe thread de lecture de fichiers fits BAORadio
224//-------------------------------------------------------
225
226/* --Methode-- */
227BRFitsReader::BRFitsReader(RAcqMemZoneMgr& mem, vector<string>& infiles, bool fgnotrl)
228 : memgr(mem), infiles_(infiles), fgnotrl_(fgnotrl)
229{
230}
231
232/* --Methode-- */
233void BRFitsReader::run()
234{
235 setRC(1);
236
237 try {
238 TimeStamp ts;
239 Timer tm("BRFitsReader", false);
240 BRPaqChecker pcheck(!fgnotrl_); // Verification/comptage des paquets
241
242 size_t totnbytesrd = 0;
243 cout << " BRFitsReader::run() - Starting " << ts << " NbFiles=" << infiles_.size()
244 << " memgr.PaqSize() = " << memgr.PaqSize() << endl;
245
246 uint_4 nfileok = 0;
247 uint_8 nbytesrd = 0;
248 /* Variables pour la logique des zones memoire et numeros de paquets dans la zone memoire */
249 int mid = -2;
250 Byte* buff = NULL;
251 int kmp = 0;
252 int kmpmax=memgr.NbPaquets();
253
254 int paqsz = 0;
255 for(int ifile=0; ifile<infiles_.size(); ifile++) {
256 string ffname = infiles_[ifile];
257// -------------- Lecture de bytes
258 cout << "BRFitsReader::run() [" << ifile <<"]Ouverture/lecture fichier " << ffname << endl;
259 MiniFITSFile mff(ffname, MF_Read);
260 cout << "... Type=" << mff.DataTypeToString() << " NAxis1=" << mff.NAxis1()
261 << " NAxis2=" << mff.NAxis2() << endl;
262 if (mff.DataType() != MF_Byte) {
263 cout << "BRFitsReader::run() PB : DataType!=MF_Byte --> skipping " << endl;
264 continue;
265 }
266// Les fichier FITS contiennent l'entet (24 bytes), mais pas le trailer (16 bytes) si fgnotrl=true
267 int incpaqsz=0;
268 if (fgnotrl_) {
269 incpaqsz=16;
270 if (ifile==0) cout << " Warning : FITS files without frame trailers ..." << endl;
271 }
272 if (paqsz == 0) { // premier passage, on fixe la taille de paquet et on verifie compatibilite avec memgr
273 paqsz = mff.NAxis1()+incpaqsz;
274 if (paqsz != memgr.PaqSize()) {
275 cout << "BRFitsReader::run() mff.NAxis1() incompatible with memgr.PaqSize() -> exception " << endl;
276 throw SzMismatchError(" fits file size incompatible with memgr.PaqSize()");
277 }
278 }
279 else {
280 if (paqsz != mff.NAxis1()+incpaqsz) {
281 cout << " PB : paqsz=" << paqsz << " != mff.NAxis1()+" << incpaqsz << " --> skipping " << endl;
282 continue;
283 }
284 }
285 if (mid < 0) {
286 mid = memgr.FindMemZoneId(MemZA_Fill);
287 buff = memgr.GetMemZone(mid);
288 if (buff == NULL) {
289 cout << " BRFitsReader::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
290 setRC(2);
291 return;
292 }
293 kmp=0;
294 }
295 size_t sx = mff.NAxis1();
296 size_t sy = mff.NAxis2();
297 int nprt=0;
298 for(int j=0; j<sy; j++) {
299 mff.ReadB(buff+kmp*paqsz, sx, j*sx);
300 BRPaquet paq(NULL, buff+kmp*paqsz, paqsz);
301 bool pqok = pcheck.Check(paq); // Verification du paquet / FrameCounter
302 if (!pqok && (nprt < 10)) {
303 cout << "--BUG-- i=" << ifile << " mid=" << mid << " j=" << j << " kmp=" << kmp
304 << " paqsz=" << paqsz << endl;
305 nprt++;
306 paq.Print();
307 }
308 kmp++;
309 if (kmp >= kmpmax) { // Zone memoire rempli !
310 memgr.FreeMemZone(mid, MemZS_Filled);
311 mid = -2;
312 if (j<sy) {
313 mid = memgr.FindMemZoneId(MemZA_Fill);
314 buff = memgr.GetMemZone(mid);
315 if (buff == NULL) {
316 cout << " BRFitsReader::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
317 setRC(2);
318 return;
319 }
320 kmp=0;
321 }
322 }
323 }
324 nfileok++;
325 size_t nbytesrd = sx*sy;
326 totnbytesrd += nbytesrd;
327 } // Fin de la boucle sur les fichiers
328// Gestion d'une zone partiellement remplie
329 if (mid>=0) {
330 for(int k=kmp;k<kmpmax;k++) {
331 Byte* bp=buff+k*paqsz;
332 for(int l=0;l<paqsz;l++) bp[l]=0;
333 }
334 memgr.FreeMemZone(mid, MemZS_Filled);
335 }
336
337// sprintf(fname,"%s/.log",path_.c_str());
338// ofstream filog(fname);
339// filog << " DataProc::run() - starting log file " << ts << endl;
340// filog << " NbFiles=" << nfiles_ << " NBloc/File=" << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
341
342
343
344 cout << " ------------------ BRFitsReader::run() END ----------------- " << endl;
345 ts.SetNow();
346 tm.SplitQ();
347 cout << " END reading " << ts << " NFileOK=" << nfileok << endl;
348 cout << " TotalDiskRead= " << totnbytesrd/(1024*1024) << " MBytes Disk-Read rate= "
349 << (double)(totnbytesrd)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;
350 pcheck.Print(cout);
351 cout << " BRFitsReader::run()/Timing: \n";
352 tm.Print();
353 cout << " ---------------------------------------------------------- " << endl;
354
355 } // Fin du bloc try
356 catch (MiniFITSException& exc) {
357 cout << " BRFitsReader::run()/catched MiniFITSException " << exc.Msg() << endl;
358 setRC(3);
359 return;
360 }
361 catch(...) {
362 cout << " BRFitsReader::run()/catched unknown ... exception " << endl;
363 setRC(4);
364 return;
365 }
366 setRC(0);
367 return;
368}
Note: See TracBrowser for help on using the repository browser.