source: Sophya/trunk/ArchTOIPipe/Kernel/fitstoirdr.cc@ 2063

Last change on this file since 2063 was 2063, checked in by ansari, 23 years ago

Debug de setSepFlagFile - Reza 18/6/2002
File size: 12.5 KB
Line 
1// ArchTOIPipe (C) CEA/DAPNIA/SPP IN2P3/LAL
2// Eric Aubourg
3// Christophe Magneville
4// Reza Ansari
5// $Id: fitstoirdr.cc,v 1.33 2002-06-18 18:43:50 ansari Exp $
6
7#include "fitstoirdr.h"
8#include "toimanager.h"
9#include <sched.h>
10
11
12FITSTOIReader::FITSTOIReader(string fn,int buff_sz) {
13 fname = fn;
14 allfn.push_back(fn);
15 Buff_Sz = (buff_sz>0) ? buff_sz: 1000;
16 cout<<"FITSTOIReader::FITSTOIReader"<<endl;
17 cout<<"FITSTOIReader::inited "<<inited<<" bsz="<<Buff_Sz<<endl;
18 name = "rdr";
19 fptr = NULL;
20 totnscount = 0;
21
22 implicitSN = false;
23 implicitSNStart = 0;
24
25 // Variables rajoutee pour gerer les fichiers de flag de LevelS Reza 18/6/2002
26 sepFlagfile = false;
27 fptrflg = NULL;
28}
29
30FITSTOIReader::~FITSTOIReader() {
31}
32
33void FITSTOIReader::setImplicitSN(long snStart) {
34 implicitSN = true;
35 implicitSNStart = snStart;
36}
37
38void FITSTOIReader::setBufferSize(int buffsz)
39{
40 Buff_Sz = (buffsz>0) ? buffsz: 1024;
41 return;
42}
43
44pthread_mutex_t fits_mutex = PTHREAD_MUTEX_INITIALIZER;
45void fits_lock();
46void fits_unlock();
47
48void fits_lock() {
49 pthread_mutex_lock(&fits_mutex);
50}
51void fits_unlock() {
52 pthread_mutex_unlock(&fits_mutex);
53}
54
55void FITSTOIReader::openFile(string fn) {
56 fits_lock();
57 if (fptr) {
58 fits_close_file(fptr,&fstatus);
59 fptr = NULL;
60 }
61 fname = fn;
62 cout << "FITSTOIReader::open FileName=" << fname << endl;
63 fstatus = 0;
64 // Open file
65 fits_open_file(&fptr,fname.c_str(),READONLY,&fstatus);
66 if (fstatus != 0) {
67 fits_report_error(stderr, fstatus);
68 fits_unlock();
69 throw IOExc("FITSTOIReader::openFile() fitsio error");
70 }
71
72 // Go to first extension, which should be a BINTABLE
73
74 int simple, bitpix, naxis;
75 long naxes;
76 long pcount, gcount;
77 int extend;
78 fits_read_imghdr(fptr, 1, &simple, &bitpix,
79 &naxis, &naxes, &pcount, &gcount, &extend, &fstatus);
80
81 fits_movabs_hdu(fptr, 2, NULL, &fstatus);
82
83 fits_get_num_cols(fptr,&ncols,&fstatus);
84 fits_get_num_rows(fptr,&nrows,&fstatus);
85
86 cout << "FITSTOIReader cols = " << ncols << " rows=" << nrows << endl;
87 if (implicitSN) {
88 firstSn = implicitSNStart;
89 } else {
90 int anyNul;
91 double y;
92 fits_read_col_dbl(fptr,1,1,1,1,0,&y,&anyNul,&fstatus);
93 firstSn = (int) (y+.1);
94 }
95
96 // Ouverture fichier de flag separe de LevelS (Reza 18/6/2002)
97 if (sepFlagfile) {
98 cout << " DBG - Opening separate flag file " << sepFlagFileName
99 << " NColFlags= " << sepFlagCols.size() << endl;
100 fits_open_file(&fptrflg,sepFlagFileName.c_str(),READONLY,&fstatus);
101 if (fstatus != 0) {
102 fits_report_error(stderr, fstatus);
103 fits_unlock();
104 throw IOExc("FITSTOIReader::openFile() - sepFlagfile open fitsio error");
105 }
106 fits_movabs_hdu(fptrflg, 2, NULL, &fstatus);
107 if (fstatus != 0) {
108 fits_report_error(stderr, fstatus);
109 fits_unlock();
110 throw IOExc("FITSTOIReader::openFile() - sepFlagfile fits_movabs_hdu(2) fitsio error");
111 }
112 long nrowsflg;
113 fits_get_num_rows(fptrflg,&nrowsflg,&fstatus);
114 if (nrows != nrowsflg) {
115 cerr << " FITSTOIReader::openFile()/Error: Different NRows in flag and data files!" << endl;
116 fits_unlock();
117 throw ParmError("FITSTOIReader::openFile() Different NRows in flag and data files");
118 }
119 }
120 //
121 fits_unlock();
122}
123
124void FITSTOIReader::init() {
125
126 // Modif pour fichiers de flag separe de LevelS (Reza 18/6/2002)
127 if (sepFlagfile && (allfn.size() > 1) ) {
128 cerr << "FITSTOIReader::init()/Error- Multiple files and separate flag file not allowed !"<<endl;
129 throw ParmError("FITSTOIReader::init() Multiple files and separate flag file not allowed");
130 }
131
132 openFile(allfn.front());
133
134 fits_lock();
135 // si pas implicitSN, la premiere colonne est le sampleNum.
136 // Sinon, le samplenum est la fitsline + offset.
137 int itoi=-1;
138 int col1 = implicitSN ? 0 : 1;
139 for (int i=col1; i<ncols; i++) {
140 char templt[10];
141 sprintf(templt, "%d", i+1);
142 char colname[200];
143 int colnum;
144 fits_get_colname(fptr, CASESEN, templt, colname, &colnum, &fstatus);
145 cout << "FITSTOIReader col " << colname << endl;
146 // On verifie si c'est une colonne de flag
147 if (!strncmp(colname, "fg_", 3) || !strncmp(colname, "FLAG_", 5) ) {
148 colsinput[itoi].second=true;
149 } else {
150 declareOutput(colname);
151 itoi++;
152 colsinput[itoi] = pair<int,bool>(i,false);
153 }
154 }
155 fits_unlock();
156
157 snBegin = firstSn;
158 if (forcedMinIn > 0 && forcedMinIn > snBegin) {
159 snBegin = forcedMinIn;
160 }
161 openFile(allfn.back());
162 snEnd = firstSn+nrows-1;
163 if (forcedMaxIn > 0 && forcedMaxIn < snEnd) {
164 snEnd = forcedMaxIn;
165 }
166 cout << "FITSTOIReader range " << snBegin << " -> " << snEnd << endl;
167}
168
169int FITSTOIReader::calcMinOut() {
170 chkinit();
171 TOIManager* mgr = TOIManager::getManager();
172 int firstReq = mgr->getRequestedBegin();
173 return snBegin > firstReq ? snBegin : firstReq;
174}
175
176int FITSTOIReader::calcMaxOut() {
177 chkinit();
178 TOIManager* mgr = TOIManager::getManager();
179 int lastReq = mgr->getRequestedEnd();
180 return snEnd < lastReq ? snEnd : lastReq;
181}
182
183void FITSTOIReader::addFile(string fn) {
184 allfn.push_back(fn);
185}
186
187void FITSTOIReader::setFlagFile(string fn, vector<FlagToiDef> flags)
188{
189 if (flags.size() < 1) {
190 cerr << " FITSTOIReader::setFlagFile()/Error flag.size() = 0 ! " << endl;
191 throw ParmError("FITSTOIReader::setFlagFile() flag.size() = 0");
192 }
193
194 sepFlagfile = true;
195 sepFlagFileName = fn;
196 sepFlagCols = flags;
197}
198
199void FITSTOIReader::run() {
200 for (vector<string>::iterator i=allfn.begin(); i!=allfn.end(); i++) {
201 openFile(*i);
202 if (Buff_Sz > 1) run2(); // Lecture bufferise
203 else run1(); // Lecture un echantillon a la fois
204 }
205 fits_lock();
206 if (fptr) {
207 fits_close_file(fptr,&fstatus);
208 fptr = NULL;
209 }
210 if (sepFlagfile && fptrflg) {
211 fits_close_file(fptrflg,&fstatus);
212 fptrflg = NULL;
213 }
214 fits_unlock();
215}
216
217// run 1 : deprecated. NON MAINTENU. Incompatible avec implicit SN.
218// ^^^^^^^^^^^^ Reza , 13/5/2002 , Je viens de rajouter
219// implicitSNStart et switvh run1/run2
220// suivant buffersize
221
222void FITSTOIReader::run1() {
223 // Il faudrait optimiser en fonction de ce qui a ete demande comme samplenum,
224 // mais cela implique de gerer aussi bien echant uniforme que non.
225 // On pourrait aussi lire plusieurs elements d'un coup.
226 int ncols = outIx.size();
227 cout << "reader reading... NRows=" << nrows << " firstSn= "
228 << firstSn << endl;
229
230 double* tabdata = new double[getNOut()];
231 uint_8* tabflag = new uint_8[getNOut()];
232
233 for (long i=0; i<nrows; i++) {
234 int anyNul;
235 double y;
236 fits_lock();
237 long sn;
238 if (implicitSN) {
239 fits_read_col_dbl(fptr,1,i+1,1,1,0,&y,&anyNul,&fstatus);
240 sn = (long) (y+.1);
241 }
242 else sn = implicitSNStart+i;
243 //fits_unlock();
244 TOIManager* mgr = TOIManager::getManager();
245 if (sn > mgr->getRequestedEnd()) {fits_unlock(); break;}
246 if (sn < mgr->getRequestedBegin()) {fits_unlock(); continue;}
247 // if (sn < mgr->getRequestedBegin()+10) cout << "rdr out " << sn << endl;
248 int k;
249 for (k=0; k<getNOut(); k++) {
250 int j = colsinput[k].first;
251 if ( !checkOutputTOIIndex(k) ) continue; // Reza - Si TOI non connecte
252 //fits_lock();
253 fits_read_col_dbl(fptr,j+1,i+1,1,1,0,&y,&anyNul,&fstatus);
254 tabdata[k] = y;
255 long flg = 0;
256 if (colsinput[k].second) {
257 fits_read_col_lng(fptr,j+2,i+1,1,1,0,&flg,&anyNul,&fstatus);
258 }
259
260 if (sepFlagfile) { // Ajout Reza (18/6/2002) pour fichier de flags separe
261 int sflg;
262 flg = 0;
263 for(int skf=0; skf<sepFlagCols.size(); skf++) {
264 fits_read_col_int(fptrflg, skf+1, i+1,1,1,0,&sflg,&anyNul,&fstatus);
265 if (sflg) flg |= sepFlagCols[skf];
266 }
267 } // Fin modif pour fichier de flags separe (18/6/2002)
268
269 tabflag[k] = flg;
270 // fits_unlock();
271 //putData(k, sn, y, flg);
272 }
273 fits_unlock();
274 for (k=0; k<getNOut(); k++) {
275 putData(k, sn, tabdata[k], tabflag[k]);
276 }
277 /* if (i%50==0) {
278 // fits_unlock();
279 sched_yield();
280 //fits_lock();
281 }*/
282 totnscount++;
283 }
284 //fits_unlock();
285 delete[] tabflag;
286 delete[] tabdata;
287 cout << "reader done reading... " << pthread_self() << endl;
288}
289
290void FITSTOIReader::run2()
291// ---- Version bufferisee (Christophe 20/10/2001)
292{
293 int ncols = outIx.size();
294 cout<<"reader reading... NRows="<<nrows<<" firstSn= " <<firstSn<<endl;
295
296 //////// Prepare buffer, allocate memory
297 double *samplenum = new double[Buff_Sz];
298 vector<double*> colval;
299 vector<long*> colflg;
300 for(int k=0; k<getNOut(); k++) {
301 colval.push_back(NULL);
302 colflg.push_back(NULL);
303 if(!checkOutputTOIIndex(k)) continue;
304 colval[k] = new double[Buff_Sz];
305 if(colsinput[k].second) colflg[k] = new long[Buff_Sz];
306 }
307 uint_8 * tmpflg = new uint_8[Buff_Sz];
308
309 // Ajout Reza (18/6/2002) pour fichier de flags separe
310 int* stmpflg = NULL;
311 if (sepFlagfile) stmpflg = new int[Buff_Sz];
312 long nonzeroflg = 0;
313 long nbckflg = 0;
314 // Fin modif pour fichier de flags separe (18/6/2002)
315
316
317 TOIManager* mgr = TOIManager::getManager();
318
319 //////// Read data and put into pipe
320 long ideb=-1,ifin=-1;
321 for(long i=0; i<nrows; i+=Buff_Sz) {
322 fits_lock(); // lock en debut de boucle de lecture - Reza 13/05/2002
323 // faut-il lire dans le fichier fits ?
324 if(i<ideb || i>ifin) { // Toujours vrai avec le += Buff_Sz
325 ideb = i;
326 ifin = ideb+Buff_Sz-1;
327 if(ifin>=nrows) ifin=nrows-1;
328 long n = ifin-ideb+1;
329 int anyNul;
330 if (implicitSN) {
331 if (ideb+implicitSNStart > mgr->getRequestedEnd()
332 || (forcedMaxIn > 0 && ideb+implicitSNStart > forcedMaxIn)) {
333 fits_unlock(); // unlock avant de casser la boucle - Reza 16/05/2002
334 break;
335 }
336 if (ifin+implicitSNStart < mgr->getRequestedBegin()
337 || (forcedMinIn > 0 && ifin+implicitSNStart < forcedMinIn)) {
338 fits_unlock(); // unlock avant de continuer - Reza 16/05/2002
339 continue;
340 }
341 for (long j=0; j<Buff_Sz; j++) {
342 samplenum[j] = j+ideb+implicitSNStart;
343 }
344 //---- Pas la peine, fait en debut de boucle - Reza 13/05/2002 fits_lock();
345 } else {
346 //---- Pas la peine, fait en debut de boucle - Reza 13/05/2002 fits_lock();
347 fits_read_col_dbl(fptr,1,ideb+1,1,n,0,samplenum,&anyNul,&fstatus);
348 if (samplenum[0] > mgr->getRequestedEnd()
349 || (forcedMaxIn > 0 && samplenum[0] > forcedMaxIn)) {
350 fits_unlock(); // unlock avant de casser la boucle - Reza 13/05/2002
351 break;
352 }
353 if (samplenum[n-1] < mgr->getRequestedBegin()
354 || (forcedMinIn > 0 && samplenum[n-1] < forcedMinIn)) {
355 fits_unlock();
356 continue;
357 }
358 }
359 for(int k=0; k<getNOut(); k++) {
360 if(colval[k]==NULL) continue;
361 int j = colsinput[k].first;
362 fits_read_col_dbl(fptr,j+1,ideb+1,1,n,0,colval[k],&anyNul,&fstatus);
363
364 //--- Ajout Reza (18/6/2002) pour fichier de flags separe
365 if (sepFlagfile) {
366 uint_8 sflg = 0;
367 int sjj;
368 for(sjj=0; sjj<n; sjj++) colflg[k][sjj] = 0;
369 for(int skf=0; skf<sepFlagCols.size(); skf++) {
370 fits_read_col_int(fptrflg, skf+1, ideb+1,1,n,0,stmpflg,&anyNul,&fstatus);
371 for(sjj=0; sjj<n; sjj++) {
372 nbckflg++;
373 if (stmpflg) {
374 nonzeroflg++;
375 colflg[k][sjj] |= sepFlagCols[skf];
376 }
377 }
378 }
379 } //--- Fin modif pour fichier de flags separe (18/6/2002)
380
381 if(colflg[k]==NULL) continue;
382 fits_read_col_lng(fptr,j+2,ideb+1,1,n,0,colflg[k],&anyNul,&fstatus);
383 }
384 if(fstatus!=0) {
385 fits_report_error(stderr,fstatus);
386 fits_unlock(); // On supprime le lock avant le throw - Reza 13/05/2002
387 throw RangeCheckError("FITSTOIReader::run2: Error Reading Fits file\n");
388 }
389 }
390 fits_unlock(); // unlock en fin de boucle de lecture - Reza 13/05/2002
391
392 long ip = i-ideb; // pointeurs dans les buffers
393 long sn = (long) (samplenum[ip]+.1);
394 if (ip == 0) {
395 for(int k=0; k<getNOut(); k++) {
396 if(colval[k]==NULL) continue;
397 if (colflg[k] != NULL) {
398 for (int ii=0; ii<Buff_Sz; ii++) {
399 tmpflg[ii] = colflg[k][ii];
400 }
401 putData(k, sn, ifin-ideb+1, colval[k], tmpflg);
402 } else {
403 putData(k, sn, ifin-ideb+1, colval[k]);
404 }
405 }
406 }
407 totnscount += (ifin-ideb+1);
408 }
409
410 //////// des-allocate buffers
411 delete [] samplenum;
412 delete [] tmpflg;
413 if (sepFlagfile) delete [] stmpflg;
414 {for(int k=0; k<getNOut(); k++) {
415 if(colval[k]!=NULL) delete [] colval[k];
416 if(colflg[k]!=NULL) delete [] colflg[k];
417 }}
418 colval.resize(0); colflg.resize(0);
419
420 cout << "reader (buffered) done reading... " << pthread_self() << endl;
421 cout << " --- CkRz: nbckflg=" << nbckflg << " nonzeroflg= " << nonzeroflg << endl;
422
423}
Note: See TracBrowser for help on using the repository browser.