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

Last change on this file since 3978 was 3978, checked in by ansari, 14 years ago

Correction bug getObsTime() pour le premier fichier FITS, Reza 03/05/2011
File size: 14.9 KB
Line 
1//----------------------------------------------------------------
2// Projet BAORadio - (C) LAL/IRFU 2008-2011
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), stop_(false), 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 cpaqdeltatime_=0.;
49
50 char flnm[1024];
51 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
52 sprintf(flnm,"%s/signal%d.fits",dirs_[fib].c_str(),imin_);
53 mff_[fib].Open(flnm, MF_Read);
54 cout << " BRMultiFitsReader::BRMultiFitsReader() opening " << flnm << endl;
55 if (mff_[fib].NAxis1() != memgr_.PaqSize()) {
56 cout << " BRMultiFitsReader::BRMultiFitsReader/ fib=" << fib << " File=" << flnm <<
57 " NAxis1()= " << mff_[fib].NAxis1() << " <> PaqSize()=" << memgr_.PaqSize() << endl;
58 throw BAORadioException("BRMultiFitsReader::BRMultiFitsReader/ mff.NAxis1() != memgr_.PaqSize() ");
59 }
60 // Extraction de qques parametres utiles depuis les fichiers FITS
61 string fkvs;
62 if (fib==0) {
63 fkvs=mff_[fib].GetKeyValue("DATEOBS");
64 if (fkvs.length()>0) cdateobs_.Set(fkvs);
65 fkvs=mff_[fib].GetKeyValue("TMSTART");
66 if (fkvs.length()>0) {
67 ctmstart_.Set(fkvs);
68 fkvs=mff_[fib].GetKeyValue("TMEND");
69 SOPHYA::TimeStamp tmend_=ctmstart_;
70 if (fkvs.length()>0) tmend_.Set(fkvs);
71 cpaqdeltatime_=((double)(tmend_.DaysPart()-cdateobs_.DaysPart())*86400.+
72 (tmend_.SecondsPart()-cdateobs_.SecondsPart()))/(double)mff_[fib].NAxis2();
73 cout << " BRMultiFitsReader/First file (for fiber 0) TMSTART=" << fkvs << " TMEND-START="
74 << cpaqdeltatime_*(double)mff_[fib].NAxis2() << endl;
75 }
76 }
77 fkvs=mff_[fib].GetKeyValue("FIBERNUM");
78 memgr_.FiberId(fib) = atoi( fkvs.c_str() );
79
80 vfilenum_.push_back(imin_);
81 vfpos_.push_back(0);
82 vpaq_.push_back(BRPaquet(NULL,memgr_.PaqSize()));
83 vpchk_.push_back(BRPaqChecker(true,0));
84 curfc_.push_back(0);
85 totnpqrd_.push_back(0);
86 totnpqok_.push_back(0);
87 }
88}
89
90
91/* --Methode-- */
92void BRMultiFitsReader::run()
93{
94 setRC(1);
95 try {
96 TimeStamp ts;
97 Timer tm("BRMultiFitsReader", false);
98 cout << " BRMultiFitsReader::run() - Starting " << ts << " NbFibres()=" << memgr_.NbFibres()
99 << " PaqSize() = " << memgr_.PaqSize() << endl;
100 cout << " ...ReadMode: " << ((rdsamefc_)?"Paquets With SameFrameCounter":"All OK paquets")
101 << " signalII.fits IMin=" << imin_ << " IMax=" << imax_ << " IStep=" << istep_ << endl;
102
103 uint_8 prtcnt=0;
104 Byte* nextpaq=NULL;
105 bool fgok=true;
106 while (fgok) {
107 if (stop_) break;
108 if ( MoveToNextTarget() ) {
109 cout << "BRMultiFitsReader::run()/Error-A- MoveToNextTarget() returned true ->STOP 9" << endl;
110 setRC(7); fgok=false; break;
111 }
112 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
113 nextpaq=GetPaquetTarget(fib);
114 if (nextpaq == NULL) { // Cela ne devrait pas arriver
115 cout << "BRMultiFitsReader::run()/Error-A2- GetPaquetTarget(fib) returned NULL ->STOP 9" << endl;
116 setRC(9); fgok=false; break;
117 }
118 vpaq_[fib].Set(nextpaq);
119 }
120 if (ReadNextAllFibers()) { fgok=false; break; }
121 prtcnt++;
122 if ((prtlev_>0)&&(prtcnt%prtmodulo_==0)) {
123 cout << "BRMultiFitsReader: NbPaqMFRead=" << prtcnt << " NSameFC="
124 << totsamefc_ << " / NPaqFib0Read=" << totnpqrd_[0]
125 << " FracSameFC=" << 100*totsamefc_/totnpqrd_[0] << " %" << endl;
126 }
127 }
128
129 MoveToNextTarget(); // Pour faire traiter le dernier paquet si plein
130 MZoneManage(true); // Nettoyage final
131
132 cout << " ------------------ BRMultiFitsReader::run() END ----------------- " << endl;
133 ts.SetNow();
134 tm.SplitQ();
135 cout << "BRMultiFitsReader::run(): END reading : " << ts << endl;
136 cout << "... NbPaqMFRead=" << prtcnt << " NSameFC="
137 << totsamefc_ << " / NPaqFib0Read=" << totnpqrd_[0]
138 << " FracSameFC=" << 100*totsamefc_/totnpqrd_[0] << " %" << endl;
139
140 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
141 int perc=0;
142 if (totnpqrd_[fib]>0) perc=100*totsamefc_/totnpqrd_[fib];
143 cout << " Fiber" << fib << " TotNPaqRd=" << totnpqrd_[fib] << " TotNPaqOK=" << totnpqok_[fib]
144 << " FracSameFC=" << perc << " %" << endl;
145 if (prtlev_ > 0) vpchk_[fib].Print(cout);
146 }
147 cout << " TotalDiskRead= " << totnbytesrd_/(1024*1024) << " MBytes Disk-Read rate= "
148 << (double)(totnbytesrd_)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;
149 cout << " BRMultiFitsReader::run()/Timing: \n";
150 tm.Print();
151 cout << " ---------------------------------------------------------- " << endl;
152
153 usleep(250000); // Attente de traitement du dernier paquet
154 memgr_.Stop(); // Arret
155
156 } // Fin du bloc try
157 catch (std::exception& exc) {
158 cout << " BRMultiFitsReader::run()/catched execption msg= " << exc.what() << endl;
159 setRC(3);
160 return;
161 }
162 catch(...) {
163 cout << " BRMultiFitsReader::run()/catched unknown ... exception " << endl;
164 setRC(4);
165 return;
166 }
167 setRC(0);
168 return;
169}
170
171/* --Methode-- */
172bool BRMultiFitsReader::ReadNextAllFibers()
173{
174 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++) {
175 if (ReadNext(fib)) return true; // probleme
176 }
177 if (!rdsamefc_ || (memgr_.NbFibres()<2)) {
178 uint_8 cfc=curfc_[0];
179 bool fgsamefc=true;
180 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
181 if (curfc_[fib]!=cfc) fgsamefc=false;
182 }
183 if (fgsamefc) totsamefc_++;
184 return false; // c'est OK
185 }
186 // On va essayer de lire jusqu'a avoir same_frame_counter
187 bool echec=true;
188 while (echec) {
189 uint_8 cfc=curfc_[0];
190 bool fgsamefc=true;
191 for(size_t fib=1; fib<memgr_.NbFibres(); fib++) {
192 if (curfc_[fib]!=cfc) {
193 fgsamefc=false;
194 if (curfc_[fib] > cfc) cfc=curfc_[fib];
195 }
196 }
197 if (fgsamefc) {
198 totsamefc_++; echec=false; return false; // c'est OK , same framecounter
199 }
200 else { // else !fgsame
201 for(uint_4 fib=0; fib<memgr_.NbFibres(); fib++) {
202 while (curfc_[fib]<cfc) {
203 if (ReadNext(fib)) return true; // probleme
204 }
205 }
206 } // fin de else !fgsame
207 } // fin de while(echec): lecture jusqu'a same_frame_counter
208
209 return true; // probleme
210}
211
212/* --Methode-- */
213bool BRMultiFitsReader::ReadNext(int fib)
214{
215 if (!mff_[fib].IsOpen()) return true;
216 bool fggood=false;
217 while(!fggood) {
218 if (vfpos_[fib] >= mff_[fib].NAxis2()) {
219 mff_[fib].Close();
220 vfilenum_[fib]+=istep_;
221 if (vfilenum_[fib]>imax_) return true;
222 char flnm[1024];
223 sprintf(flnm,"%s/signal%d.fits",dirs_[fib].c_str(),vfilenum_[fib]);
224 if (prtlev_ > 1)
225 cout << " BRMultiFitsReader::ReadNext() opening" << flnm << endl;
226 mff_[fib].Open(flnm, MF_Read);
227 if (mff_[fib].NAxis1() != packsize_) {
228 cout << " BRMultiFitsReader::ReadNext(fib=" << fib << " File=" << flnm <<
229 " NAxis1()= " << mff_[fib].NAxis1() << " <> PaqSize()=" << packsize_ << endl;
230 throw BAORadioException("BRMultiFitsReader::ReadNext()/ mff.NAxis1() != memgr_.PaqSize() ");
231 }
232 if (fib==0) { // updating current date from file (fiber 0)
233 string fkvs=mff_[fib].GetKeyValue("DATEOBS");
234 if (fkvs.length()>0) cdateobs_.Set(fkvs);
235 fkvs=mff_[fib].GetKeyValue("TMSTART");
236 if (fkvs.length()>0) {
237 ctmstart_.Set(fkvs);
238 cout << " BRMultiFitsReader::ReadNext TMSTART=" << fkvs << endl;
239 fkvs=mff_[fib].GetKeyValue("TMEND");
240 SOPHYA::TimeStamp tmend_=ctmstart_;
241 if (fkvs.length()>0) tmend_.Set(fkvs);
242 cpaqdeltatime_=((double)(tmend_.DaysPart()-cdateobs_.DaysPart())*86400.+
243 (tmend_.SecondsPart()-cdateobs_.SecondsPart()))/(double)mff_[fib].NAxis2();
244 }
245 }
246 vfpos_[fib]=0;
247 }
248 mff_[fib].ReadB(vpaq_[fib].Begin(), packsize_, vfpos_[fib]*packsize_);
249 vfpos_[fib]++;
250 totnbytesrd_+=packsize_;
251 totnpqrd_[fib]++;
252 fggood = vpchk_[fib].Check(vpaq_[fib],curfc_[fib]);
253 }
254 totnpqok_[fib]++;
255 return false;
256}
257
258/* --Methode-- */
259bool BRMultiFitsReader::MZoneManage(bool fgclean) // Retourne true si probleme
260{
261 /* Pour debug
262 cout << " BRMultiFitsReader::MZoneManage() mid_=" << mid_ << " arg_npaq_= " << targ_npaq_
263 << " max_targ_npaq=" << max_targ_npaq << endl;
264 */
265 if (mid_ >= 0) {
266 if (fgclean) memgr_.FreeMemZone(mid_, MemZS_Free);
267 else memgr_.FreeMemZone(mid_, MemZS_Filled);
268 }
269 mmbuf_ = NULL; targ_npaq_ = 0; mid_ = -2;
270 for (int fib=0;fib<(int)memgr_.NbFibres() ;fib++) mmbufib_[fib]=NULL;
271 if (fgclean) return false;
272 mid_ = memgr_.FindMemZoneId(MemZA_Fill);
273 mmbuf_ = memgr_.GetMemZone(mid_);
274 if (mmbuf_==NULL) return true;
275 for(size_t fib=0; fib<(size_t)memgr_.NbFibres(); fib++)
276 mmbufib_[fib]=memgr_.GetMemZone(mid_,fib);
277 // Definition temps pour la zone a remplir
278 memgr_.GetAuxData(mid_)->FillTime().Set(ctmstart_.ToDays()+cpaqdeltatime_*(double)vfpos_[0]/86400.);
279 return false;
280}
281
282//-------------------------------------------------------
283// Classe thread de lecture de fichiers fits BAORadio
284//-------------------------------------------------------
285
286/* --Methode-- */
287BRFitsReader::BRFitsReader(RAcqMemZoneMgr& mem, vector<string>& infiles, bool fgnotrl)
288 : memgr(mem), infiles_(infiles), fgnotrl_(fgnotrl)
289{
290}
291
292/* --Methode-- */
293void BRFitsReader::run()
294{
295 setRC(1);
296
297 try {
298 TimeStamp ts;
299 Timer tm("BRFitsReader", false);
300 BRPaqChecker pcheck(!fgnotrl_); // Verification/comptage des paquets
301
302 size_t totnbytesrd = 0;
303 cout << " BRFitsReader::run() - Starting " << ts << " NbFiles=" << infiles_.size()
304 << " memgr.PaqSize() = " << memgr.PaqSize() << endl;
305
306 uint_4 nfileok = 0;
307 uint_8 nbytesrd = 0;
308 /* Variables pour la logique des zones memoire et numeros de paquets dans la zone memoire */
309 int mid = -2;
310 Byte* buff = NULL;
311 int kmp = 0;
312 int kmpmax=memgr.NbPaquets();
313
314 int paqsz = 0;
315 for(int ifile=0; ifile<infiles_.size(); ifile++) {
316 string ffname = infiles_[ifile];
317// -------------- Lecture de bytes
318 cout << "BRFitsReader::run() [" << ifile <<"]Ouverture/lecture fichier " << ffname << endl;
319 MiniFITSFile mff(ffname, MF_Read);
320 cout << "... Type=" << mff.DataTypeToString() << " NAxis1=" << mff.NAxis1()
321 << " NAxis2=" << mff.NAxis2() << endl;
322 if (mff.DataType() != MF_Byte) {
323 cout << "BRFitsReader::run() PB : DataType!=MF_Byte --> skipping " << endl;
324 continue;
325 }
326// Les fichier FITS contiennent l'entet (24 bytes), mais pas le trailer (16 bytes) si fgnotrl=true
327 int incpaqsz=0;
328 if (fgnotrl_) {
329 incpaqsz=16;
330 if (ifile==0) cout << " Warning : FITS files without frame trailers ..." << endl;
331 }
332 if (paqsz == 0) { // premier passage, on fixe la taille de paquet et on verifie compatibilite avec memgr
333 paqsz = mff.NAxis1()+incpaqsz;
334 if (paqsz != memgr.PaqSize()) {
335 cout << "BRFitsReader::run() mff.NAxis1() incompatible with memgr.PaqSize() -> exception " << endl;
336 throw SzMismatchError(" fits file size incompatible with memgr.PaqSize()");
337 }
338 }
339 else {
340 if (paqsz != mff.NAxis1()+incpaqsz) {
341 cout << " PB : paqsz=" << paqsz << " != mff.NAxis1()+" << incpaqsz << " --> skipping " << endl;
342 continue;
343 }
344 }
345 if (mid < 0) {
346 mid = memgr.FindMemZoneId(MemZA_Fill);
347 buff = memgr.GetMemZone(mid);
348 if (buff == NULL) {
349 cout << " BRFitsReader::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
350 setRC(2);
351 return;
352 }
353 kmp=0;
354 }
355 size_t sx = mff.NAxis1();
356 size_t sy = mff.NAxis2();
357 int nprt=0;
358 for(int j=0; j<sy; j++) {
359 mff.ReadB(buff+kmp*paqsz, sx, j*sx);
360 BRPaquet paq(NULL, buff+kmp*paqsz, paqsz);
361 bool pqok = pcheck.Check(paq); // Verification du paquet / FrameCounter
362 if (!pqok && (nprt < 10)) {
363 cout << "--BUG-- i=" << ifile << " mid=" << mid << " j=" << j << " kmp=" << kmp
364 << " paqsz=" << paqsz << endl;
365 nprt++;
366 paq.Print();
367 }
368 kmp++;
369 if (kmp >= kmpmax) { // Zone memoire rempli !
370 memgr.FreeMemZone(mid, MemZS_Filled);
371 mid = -2;
372 if (j<sy) {
373 mid = memgr.FindMemZoneId(MemZA_Fill);
374 buff = memgr.GetMemZone(mid);
375 if (buff == NULL) {
376 cout << " BRFitsReader::run()/ERROR memgr.GetMemZone(" << mid << ") -> NULL" << endl;
377 setRC(2);
378 return;
379 }
380 kmp=0;
381 }
382 }
383 }
384 nfileok++;
385 size_t nbytesrd = sx*sy;
386 totnbytesrd += nbytesrd;
387 } // Fin de la boucle sur les fichiers
388// Gestion d'une zone partiellement remplie
389 if (mid>=0) {
390 for(int k=kmp;k<kmpmax;k++) {
391 Byte* bp=buff+k*paqsz;
392 for(int l=0;l<paqsz;l++) bp[l]=0;
393 }
394 memgr.FreeMemZone(mid, MemZS_Filled);
395 }
396
397// sprintf(fname,"%s/.log",path_.c_str());
398// ofstream filog(fname);
399// filog << " DataProc::run() - starting log file " << ts << endl;
400// filog << " NbFiles=" << nfiles_ << " NBloc/File=" << nblocperfile_ << " NMaxMemZones=" << nmax_ << endl;
401
402
403
404 cout << " ------------------ BRFitsReader::run() END ----------------- " << endl;
405 ts.SetNow();
406 tm.SplitQ();
407 cout << " END reading " << ts << " NFileOK=" << nfileok << endl;
408 cout << " TotalDiskRead= " << totnbytesrd/(1024*1024) << " MBytes Disk-Read rate= "
409 << (double)(totnbytesrd)/1024./tm.PartialElapsedTimems() << " MB/s" << endl;
410 pcheck.Print(cout);
411 cout << " BRFitsReader::run()/Timing: \n";
412 tm.Print();
413 cout << " ---------------------------------------------------------- " << endl;
414
415 } // Fin du bloc try
416 catch (MiniFITSException& exc) {
417 cout << " BRFitsReader::run()/catched MiniFITSException " << exc.Msg() << endl;
418 setRC(3);
419 return;
420 }
421 catch(...) {
422 cout << " BRFitsReader::run()/catched unknown ... exception " << endl;
423 setRC(4);
424 return;
425 }
426 setRC(0);
427 return;
428}
Note: See TracBrowser for help on using the repository browser.