|
casa
$Rev:20696$
|
|
casa
$Rev:20696$
|
00001 import os 00002 import sys 00003 import shutil 00004 from __main__ import default 00005 from tasks import * 00006 from taskinit import * 00007 import unittest 00008 # 00009 import listing 00010 from numpy import array 00011 00012 import asap as sd 00013 from sdbaseline import sdbaseline 00014 from sdstat import sdstat 00015 00016 class sdbaseline_unittest_base: 00017 """ 00018 Base class for sdbaseline unit test 00019 """ 00020 # Data path of input/output 00021 datapath = os.environ.get('CASAPATH').split()[0] + \ 00022 '/data/regression/unittest/sdbaseline/' 00023 taskname = "sdbaseline" 00024 00025 #complist = ['max','min','rms','median','stddev'] 00026 00027 ### helper functions for tests ### 00028 def _checkfile( self, name ): 00029 isthere=os.path.exists(name) 00030 self.assertTrue(isthere, 00031 msg='Could not find, %s'%(name)) 00032 00033 def _getStats( self, filename, ifno=None ): 00034 if not ifno: 00035 ifno=[] 00036 self._checkfile(filename) 00037 sd.rcParams['scantable.storage'] = 'memory' 00038 retstat = sdstat(filename, iflist=ifno) 00039 return retstat 00040 00041 def _compareStats( self, currstat, refstat, reltol=1.0e-2, complist=None ): 00042 # test if the statistics of baselined spectra are equal to 00043 # the reference values 00044 printstat = False #True 00045 # In case currstat is filename 00046 if isinstance(currstat, str) and os.path.exists(currstat): 00047 #print "calculating statistics from '%s'" % currstat 00048 currstat = self._getStats(currstat) 00049 00050 self.assertTrue(isinstance(currstat,dict) and \ 00051 isinstance(refstat, dict),\ 00052 "Need to specify two dictionaries to compare") 00053 if complist: 00054 keylist = complist 00055 else: 00056 keylist = refstat.keys() 00057 #keylist = self.complist 00058 00059 for key in keylist: 00060 self.assertTrue(currstat.has_key(key),\ 00061 msg="%s is not defined in the current results."\ 00062 % key) 00063 self.assertTrue(refstat.has_key(key),\ 00064 msg="%s is not defined in the reference data."\ 00065 % key) 00066 refval = refstat[key] 00067 currval = currstat[key] 00068 # Quantum values 00069 if isinstance(refval,dict): 00070 if refval.has_key('unit') and currval.has_key('unit'): 00071 if printstat: 00072 print "Comparing unit of '%s': %s (current run), %s (reference)" %\ 00073 (key,currval['unit'],refval['unit']) 00074 self.assertEqual(refval['unit'],currval['unit'],\ 00075 "The units of '%s' differs: %s (expected: %s)" % \ 00076 (key, currval['unit'], refval['unit'])) 00077 refval = refval['value'] 00078 currval = currval['value'] 00079 else: 00080 raise Exception("Invalid quantum values. %s (current run) %s (reference)" %\ 00081 (str(currval),str(refval))) 00082 currval = self._to_list(currval) 00083 refval = self._to_list(refval) 00084 if printstat: 00085 print "Comparing '%s': %s (current run), %s (reference)" %\ 00086 (key,str(currval),str(refval)) 00087 self.assertTrue(len(currval)==len(refval),"Number of elemnets in '%s' differs." % key) 00088 for i in range(len(currval)): 00089 if isinstance(refval[i],str): 00090 self.assertTrue(currval[i]==refval[i],\ 00091 msg="%s[%d] differs: %s (expected: %s) " % \ 00092 (key, i, str(currval[i]), str(refval[i]))) 00093 else: 00094 self.assertTrue(self._isInAllowedRange(currval[i],refval[i],reltol),\ 00095 msg="%s[%d] differs: %s (expected: %s) " % \ 00096 (key, i, str(currval[i]), str(refval[i]))) 00097 del currval, refval 00098 00099 00100 def _isInAllowedRange( self, testval, refval, reltol=1.e-2 ): 00101 """ 00102 Check if a test value is within permissive relative difference from refval. 00103 Returns a boolean. 00104 testval & refval : two numerical values to compare 00105 reltol : allowed relative difference to consider the two 00106 values to be equal. (default 0.01) 00107 """ 00108 denom = refval 00109 if refval == 0: 00110 if testval == 0: 00111 return True 00112 else: 00113 denom = testval 00114 rdiff = (testval-refval)/denom 00115 del denom,testval,refval 00116 return (abs(rdiff) <= reltol) 00117 00118 def _to_list( self, input ): 00119 """ 00120 Convert input to a list 00121 If input is None, this method simply returns None. 00122 """ 00123 import numpy 00124 listtypes = (list, tuple, numpy.ndarray) 00125 if input == None: 00126 return None 00127 elif type(input) in listtypes: 00128 return list(input) 00129 else: 00130 return [input] 00131 00132 00133 def _compareBLparam( self, out, reference ): 00134 # test if baseline parameters are equal to the reference values 00135 # currently comparing every lines in the files 00136 # TO DO: compare only "Fitter range" and "Baseline parameters" 00137 self._checkfile(out) 00138 self._checkfile(reference) 00139 00140 blparse_out = BlparamFileParser( out ) 00141 blparse_out.parse() 00142 coeffs_out = blparse_out.coeff() 00143 rms_out = blparse_out.rms() 00144 blparse_ref = BlparamFileParser( reference ) 00145 blparse_ref.parse() 00146 coeffs_ref = blparse_ref.coeff() 00147 rms_ref = blparse_ref.rms() 00148 allowdiff = 0.01 00149 print 'Check baseline parameters:' 00150 for irow in xrange(len(rms_out)): 00151 print 'Row %s:'%(irow) 00152 print ' Reference rms = %s'%(rms_ref[irow]) 00153 print ' Calculated rms = %s'%(rms_out[irow]) 00154 print ' Reference coeffs = %s'%(coeffs_ref[irow]) 00155 print ' Calculated coeffs = %s'%(coeffs_out[irow]) 00156 r0 = rms_ref[irow] 00157 r1 = rms_out[irow] 00158 rdiff = ( r1 - r0 ) / r0 00159 self.assertTrue((abs(rdiff)<allowdiff), 00160 msg='row %s: rms is different'%(irow)) 00161 c0 = coeffs_ref[irow] 00162 c1 = coeffs_out[irow] 00163 for ic in xrange(len(c1)): 00164 rdiff = ( c1[ic] - c0[ic] ) / c0[ic] 00165 self.assertTrue((abs(rdiff)<allowdiff), 00166 msg='row %s: coefficient for order %s is different'%(irow,ic)) 00167 print '' 00168 # self.assertTrue(listing.compare(out,reference), 00169 # 'New and reference files are different. %s != %s. ' 00170 # %(out,reference)) 00171 00172 00173 class sdbaseline_basicTest( sdbaseline_unittest_base, unittest.TestCase ): 00174 """ 00175 Basic unit tests for task sdbaseline. No interactive testing. 00176 00177 The list of tests: 00178 test01 --- test polynominal baseline with maskmode = 'auto' 00179 test02 --- test polynominal baseline with maskmode = 'list' 00180 testwp00 --- test existing file as outfile with overwrite=False (raises an exception) 00181 testwp01 --- test no data after selection (raises an exception) 00182 00183 Note: input data is generated from a single dish regression data, 00184 'OrionS_rawACSmod', as follows: 00185 default(sdcal) 00186 sdcal(infile='OrionS_rawACSmod',scanlist=[20,21,22,23], 00187 calmode='ps',tau=0.09,outfile='temp.asap') 00188 default(sdcal) 00189 sdcal(infile='temp.asap',timeaverage=True, 00190 tweight='tintsys',outfile=self.infile) 00191 """ 00192 # Input and output names 00193 #infile = 'OrionS_rawACSmod_calTave.asap' 00194 infile = 'OrionS_rawACSmod_calave.asap' 00195 outroot = sdbaseline_unittest_base.taskname+'_test' 00196 blrefroot = sdbaseline_unittest_base.datapath+'refblparam' 00197 strefroot = sdbaseline_unittest_base.datapath+'refstats' 00198 00199 def setUp( self ): 00200 if os.path.exists(self.infile): 00201 shutil.rmtree(self.infile) 00202 shutil.copytree(self.datapath+self.infile, self.infile) 00203 00204 default(sdbaseline) 00205 00206 def tearDown( self ): 00207 if (os.path.exists(self.infile)): 00208 shutil.rmtree(self.infile) 00209 00210 def test01( self ): 00211 """Test 0: maskmode = 'auto'""" 00212 tid = "01" 00213 infile = self.infile 00214 outfile = self.outroot+tid+".asap" 00215 mode = "auto" 00216 iflist = [0,2] 00217 pollist=[0] 00218 result = sdbaseline(infile=infile,maskmode=mode,outfile=outfile, 00219 blfunc='poly',iflist=iflist,pollist=pollist) 00220 # sdbaseline returns None if it runs successfully 00221 self.assertEqual(result,None, 00222 msg="The task returned '"+str(result)+"' instead of None") 00223 self._compareBLparam(outfile+"_blparam.txt",self.blrefroot+tid) 00224 reference = {'rms': [0.42423143982887268, 0.19752366840839386], 00225 'min': [-19.465526580810547, -2.7562465667724609], 00226 'max': [14.881179809570312, 2.0289769172668457], 00227 'max_abscissa': {'value': array([ 21., 3045.]), 00228 'unit': 'channel'}, 00229 'median': [0.005268096923828125, 0.0032062530517578125], 00230 'min_abscissa': {'value': array([ 18., 0.]), 00231 'unit': 'channel'}, 00232 'stddev': [0.42413413524627686, 0.19690337777137756]} 00233 self._compareStats(outfile,reference) 00234 #self._compareStats(outfile,self.strefroot+tid) 00235 00236 def test02( self ): 00237 """Test 1: maskmode = 'list' and masklist=[] (all channels)""" 00238 tid = "02" 00239 infile = self.infile 00240 outfile = self.outroot+tid+".asap" 00241 mode = "list" 00242 iflist = [2] 00243 pollist = [1] 00244 result = sdbaseline(infile=infile,maskmode=mode,outfile=outfile, 00245 iflist=iflist,pollist=pollist) 00246 # sdbaseline returns None if it runs successfully 00247 self.assertEqual(result,None, 00248 msg="The task returned '"+str(result)+"' instead of None") 00249 self._compareBLparam(outfile+"_blparam.txt",self.blrefroot+tid) 00250 reference = {'rms': 3.4925737380981445, 00251 'min': -226.3941650390625, 00252 'max': 129.78572082519531, 00253 'max_abscissa': {'value': 8186.0, 'unit': 'channel'}, 00254 'median': -0.025681495666503906, 00255 'stddev': 3.4927871227264404, 00256 'min_abscissa': {'value': 8187.0, 'unit': 'channel'}} 00257 self._compareStats(outfile,reference) 00258 #self._compareStats(outfile,self.strefroot+tid) 00259 00260 def testwp00( self ): 00261 """Test wp00: Test existing file as outfile with overwrite=False""" 00262 infile = self.infile 00263 outfile = "Dummy_Empty.asap" 00264 mode = "list" 00265 masklist = [] 00266 os.mkdir(outfile) 00267 try: 00268 result = sdbaseline(infile=infile, outfile=outfile, overwrite=False, maskmode=mode, masklist=masklist) 00269 except Exception, e: 00270 pos = str(e).find("Output file 'Dummy_Empty.asap' exists.") 00271 self.assertNotEqual(pos, -1, msg='Unexpected exception was thrown: %s'%(str(e))) 00272 finally: 00273 shutil.rmtree(outfile) 00274 00275 def testwp01( self ): 00276 """Test wp01: Test no data after selection""" 00277 tid = "wp01" 00278 infile = self.infile 00279 outfile = self.outroot+tid+".asap" 00280 iflist = [10] # non-existent IF value 00281 mode = "list" 00282 masklist = [] 00283 try: 00284 result = sdbaseline(infile=infile, outfile=outfile, iflist=iflist, maskmode=mode, masklist=masklist) 00285 except Exception, e: 00286 pos = str(e).find('Selection contains no data. Not applying it.') 00287 self.assertNotEqual(pos, -1, msg='Unexpected exception was thrown: %s'%(str(e))) 00288 00289 00290 00291 00292 class sdbaseline_maskTest( sdbaseline_unittest_base, unittest.TestCase ): 00293 """ 00294 Unit tests for task sdbaseline. Test various mask selections. 00295 Polynominal baselining. No interactive testing. 00296 00297 The list of tests: 00298 masktest01-02 --- test masklist (list) 00299 masktest03-04 --- test masklist (string) 00300 masktest05-08 --- test specunit='GHz' 00301 masktest09-12 --- test specunit='km/s' 00302 00303 Note: input data is generated from a single dish regression data, 00304 'OrionS_rawACSmod', as follows: 00305 default(sdcal) 00306 sdcal(infile='OrionS_rawACSmod',scanlist=[20,21,22,23], 00307 calmode='ps',tau=0.09,outfile='temp.asap') 00308 default(sdcal) 00309 sdcal(infile='temp.asap',timeaverage=True, 00310 tweight='tintsys',outfile=self.infile) 00311 """ 00312 # Input and output names 00313 #infile = 'OrionS_rawACSmod_calTave.asap' 00314 infile = 'OrionS_rawACSmod_calave.asap' 00315 outroot = sdbaseline_unittest_base.taskname+'_masktest' 00316 blrefroot = sdbaseline_unittest_base.datapath+'refblparam_mask' 00317 #strefroot = sdbaseline_unittest_base.datapath+'refstats_mask' 00318 tid = None 00319 00320 # Channel range excluding bad edge 00321 search = [[200,7599]] 00322 # Baseline channels. should be identical to one selected by 'auto' mode 00323 blchan0 = [[200,3979],[4152,7599]] 00324 blchan2 = [[200,2959],[3120,7599]] 00325 00326 # reference values for specunit='channel' 00327 ref_pol0if0 = {'linemaxpos': 4102.0, 'linesum': 103.81604766845703, 00328 'linemax': 1.6280698776245117, 00329 'baserms': 0.15021507441997528, 00330 'basestd': 0.15022546052932739} 00331 ref_pol0if2 = {'linemaxpos': 3045.0, 'linesum': 127.79755401611328, 00332 'linemax': 2.0193681716918945, 00333 'baserms': 0.13134850561618805, 00334 'basestd': 0.1313575953245163} 00335 # reference values for specunit='GHz' 00336 ref_pol0if0f = {'linemaxpos': 4102.0, 'linesum': 101.49118804931641, 00337 'linemax': 1.6132903099060059, 00338 'baserms': 0.15236321091651917, 00339 'basestd': 0.15236999094486237} 00340 ref_pol0if2f = {'linemaxpos': 3045.0, 'linesum': 129.6041259765625, 00341 'linemax': 2.0308537483215332, 00342 'baserms': 0.13158561289310455, 00343 'basestd': 0.13159450888633728} 00344 00345 # reference values for specunit='km/s' 00346 ref_pol0if0v = {'linemaxpos': 4102.0, 'linesum': 103.81607055664062, 00347 'linemax': 1.6280698776245117, 00348 'baserms': 0.15021507441997528, 00349 'basestd': 0.15022547543048859} 00350 ref_pol0if2v = {'linemaxpos': 3045.0, 'linesum': 128.9298095703125, 00351 'linemax': 2.0264592170715332, 00352 'baserms': 0.13150280714035034, 00353 'basestd': 0.13151165843009949} 00354 00355 def setUp( self ): 00356 if os.path.exists(self.infile): 00357 shutil.rmtree(self.infile) 00358 shutil.copytree(self.datapath+self.infile, self.infile) 00359 00360 default(sdbaseline) 00361 00362 00363 def tearDown( self ): 00364 # self._compareBLparam(self.outroot+self.tid+'.asap_blparam.txt',\ 00365 # self.blrefroot+self.tid) 00366 if (os.path.exists(self.infile)): 00367 shutil.rmtree(self.infile) 00368 00369 00370 def testblmask01( self ): 00371 """Mask test 1: test masklist (list) with maskmode = 'auto'""" 00372 self.tid="01" 00373 infile = self.infile 00374 outfile = self.outroot+self.tid+".asap" 00375 mode = "auto" 00376 masklist = self.search 00377 iflist = [0,2] 00378 pollist=[0] 00379 00380 print "masklist =", masklist 00381 00382 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00383 outfile=outfile,iflist=iflist,pollist=pollist) 00384 # sdbaseline returns None if it runs successfully 00385 self.assertEqual(result,None, 00386 msg="The task returned '"+str(result)+"' instead of None") 00387 # Compare IF0 00388 testval = self._getStats(outfile,self.blchan0,0) 00389 self._compareStats(testval,self.ref_pol0if0) 00390 # Compare IF2 00391 testval = self._getStats(outfile,self.blchan2,2) 00392 self._compareStats(testval,self.ref_pol0if2) 00393 self._compareBLparam(self.outroot+self.tid+'.asap_blparam.txt',\ 00394 self.blrefroot+self.tid) 00395 00396 def testblmask02( self ): 00397 """Mask test 2: test masklist (list) with maskmode = 'list'""" 00398 self.tid="02" 00399 infile = self.infile 00400 outfile = self.outroot+self.tid+".asap" 00401 mode = "list" 00402 masklist = self.blchan2 00403 iflist = [2] 00404 pollist=[0] 00405 00406 print "masklist =", masklist 00407 00408 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00409 outfile=outfile,iflist=iflist,pollist=pollist) 00410 # sdbaseline returns None if it runs successfully 00411 self.assertEqual(result,None, 00412 msg="The task returned '"+str(result)+"' instead of None") 00413 # Compare IF2 00414 testval = self._getStats(outfile,self.blchan2,2) 00415 self._compareStats(testval,self.ref_pol0if2) 00416 self._compareBLparam(self.outroot+self.tid+'.asap_blparam.txt',\ 00417 self.blrefroot+self.tid) 00418 00419 def testblmask03( self ): 00420 """Mask test 3: test masklist (string) with maskmode = 'auto'""" 00421 self.tid="03" 00422 infile = self.infile 00423 outfile = self.outroot+self.tid+".asap" 00424 mode = "auto" 00425 #masklist = "0~2:200~7599" 00426 iflist = [0,2] 00427 pollist=[0] 00428 00429 searchrange = self._get_range_in_string(self.search[0]) 00430 masklist = "0~2:"+searchrange 00431 print "masklist =", masklist 00432 00433 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00434 outfile=outfile,iflist=iflist,pollist=pollist) 00435 # sdbaseline returns None if it runs successfully 00436 self.assertEqual(result,None, 00437 msg="The task returned '"+str(result)+"' instead of None") 00438 # Compare IF0 00439 testval = self._getStats(outfile,self.blchan0,0) 00440 self._compareStats(testval,self.ref_pol0if0) 00441 # Compare IF2 00442 testval = self._getStats(outfile,self.blchan2,2) 00443 self._compareStats(testval,self.ref_pol0if2) 00444 self._compareBLparam(self.outroot+self.tid+'.asap_blparam.txt',\ 00445 self.blrefroot+self.tid) 00446 00447 def testblmask04( self ): 00448 """Mask test 4: test masklist (string) with maskmode = 'list'""" 00449 self.tid="04" 00450 infile = self.infile 00451 outfile = self.outroot+self.tid+".asap" 00452 mode = "list" 00453 #masklist = "0:200~3979;4152~7599, 2:200~2959;3120~7599" 00454 iflist = [0,2] 00455 pollist=[0] 00456 00457 sblrange=[] 00458 chanlist = (self.blchan0, self.blchan2) 00459 for i in xrange(len(iflist)): 00460 sblrange.append("") 00461 for chanrange in chanlist[i]: 00462 if len(sblrange[i]): sblrange[i] += ";" 00463 sblrange[i] += self._get_range_in_string(chanrange) 00464 masklist = "0:"+sblrange[0]+", 2:"+sblrange[1] 00465 print "masklist =", masklist 00466 00467 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00468 outfile=outfile,iflist=iflist,pollist=pollist) 00469 # sdbaseline returns None if it runs successfully 00470 self.assertEqual(result,None, 00471 msg="The task returned '"+str(result)+"' instead of None") 00472 # Compare IF0 00473 testval = self._getStats(outfile,self.blchan0,0) 00474 self._compareStats(testval,self.ref_pol0if0) 00475 # Compare IF2 00476 testval = self._getStats(outfile,self.blchan2,2) 00477 self._compareStats(testval,self.ref_pol0if2) 00478 self._compareBLparam(self.outroot+self.tid+'.asap_blparam.txt',\ 00479 self.blrefroot+self.tid) 00480 00481 def testblmask05( self ): 00482 """Mask test 5: test specunit='GHz' with masklist (list) and maskmode = 'auto'""" 00483 self.tid="05" 00484 infile = self.infile 00485 outfile = self.outroot+self.tid+".asap" 00486 mode = "auto" 00487 specunit = "GHz" 00488 #masklist = [[44.0511472,44.0963125]] 00489 iflist = [2] 00490 pollist=[0] 00491 00492 masklist = self._get_chanval(infile,self.search,specunit,spw=iflist[0],addedge=True) 00493 print "masklist =", masklist 00494 00495 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00496 outfile=outfile,iflist=iflist,pollist=pollist, 00497 specunit=specunit) 00498 # sdbaseline returns None if it runs successfully 00499 self.assertEqual(result,None, 00500 msg="The task returned '"+str(result)+"' instead of None") 00501 # Compare IF2 00502 testval = self._getStats(outfile,self.blchan2,2) 00503 self._compareStats(testval,self.ref_pol0if2f) 00504 00505 def testblmask06( self ): 00506 """Mask test 6: test specunit='GHz' with masklist (list) and maskmode = 'list'""" 00507 self.tid="06" 00508 infile = self.infile 00509 outfile = self.outroot+self.tid+".asap" 00510 mode = "list" 00511 specunit = "GHz" 00512 #masklist = [[44.0511472,44.0679889],[44.0689716,44.0963125]] 00513 iflist = [2] 00514 pollist=[0] 00515 00516 masklist = self._get_chanval(infile,self.blchan2,specunit,spw=iflist[0],addedge=True) 00517 print "masklist =", masklist 00518 00519 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00520 outfile=outfile,iflist=iflist,pollist=pollist, 00521 specunit=specunit) 00522 # sdbaseline returns None if it runs successfully 00523 self.assertEqual(result,None, 00524 msg="The task returned '"+str(result)+"' instead of None") 00525 # Compare IF2 00526 testval = self._getStats(outfile,self.blchan2,2) 00527 self._compareStats(testval,self.ref_pol0if2f) 00528 00529 def testblmask07( self ): 00530 """Mask test 7: test specunit='GHz' with masklist (string) and maskmode = 'auto'""" 00531 self.tid="07" 00532 infile = self.infile 00533 outfile = self.outroot+self.tid+".asap" 00534 mode = "auto" 00535 specunit = "GHz" 00536 #masklist = "0:45.4655714~45.5107367, 2:44.0511472~44.0963125" 00537 iflist = [0,2] 00538 pollist=[0] 00539 00540 sblrange=[] 00541 chanlist = (self.search, self.search) 00542 for i in xrange(len(iflist)): 00543 sblrange.append("") 00544 mlist = self._get_chanval(infile,chanlist[i],specunit,spw=iflist[i],addedge=True) 00545 for valrange in mlist: 00546 if len(sblrange[i]): sblrange[i] += ";" 00547 sblrange[i] += self._get_range_in_string(valrange) 00548 masklist = "0:"+sblrange[0]+", 2:"+sblrange[1] 00549 print "masklist =", masklist 00550 00551 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00552 outfile=outfile,iflist=iflist,pollist=pollist, 00553 specunit=specunit) 00554 # sdbaseline returns None if it runs successfully 00555 self.assertEqual(result,None, 00556 msg="The task returned '"+str(result)+"' instead of None") 00557 # Compare IF0 00558 testval = self._getStats(outfile,self.blchan0,0) 00559 self._compareStats(testval,self.ref_pol0if0f) 00560 # Compare IF2 00561 testval = self._getStats(outfile,self.blchan2,2) 00562 self._compareStats(testval,self.ref_pol0if2f) 00563 00564 def testblmask08( self ): 00565 """Mask test 8: test specunit='GHz' with masklist (string) and maskmode = 'list'""" 00566 self.tid="08" 00567 infile = self.infile 00568 outfile = self.outroot+self.tid+".asap" 00569 mode = "list" 00570 specunit = "GHz" 00571 #masklist = "0:45.4655714~45.4886394;45.4896954~45.5107367, 2:44.0511472~44.0679889;44.0689716~44.0963125" 00572 iflist = [0,2] 00573 pollist=[0] 00574 00575 sblrange=[] 00576 chanlist = (self.blchan0, self.blchan2) 00577 for i in xrange(len(iflist)): 00578 sblrange.append("") 00579 mlist = self._get_chanval(infile,chanlist[i],specunit,spw=iflist[i],addedge=True) 00580 for valrange in mlist: 00581 if len(sblrange[i]): sblrange[i] += ";" 00582 sblrange[i] += self._get_range_in_string(valrange) 00583 masklist = "0:"+sblrange[0]+", 2:"+sblrange[1] 00584 print "masklist =", masklist 00585 00586 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00587 outfile=outfile,iflist=iflist,pollist=pollist, 00588 specunit=specunit) 00589 # sdbaseline returns None if it runs successfully 00590 self.assertEqual(result,None, 00591 msg="The task returned '"+str(result)+"' instead of None") 00592 # Compare IF0 00593 testval = self._getStats(outfile,self.blchan0,0) 00594 self._compareStats(testval,self.ref_pol0if0f) 00595 # Compare IF2 00596 testval = self._getStats(outfile,self.blchan2,2) 00597 self._compareStats(testval,self.ref_pol0if2f) 00598 00599 def testblmask09( self ): 00600 """Mask test 9: test specunit='km/s' with masklist (list) and maskmode = 'auto'""" 00601 self.tid="09" 00602 infile = self.infile 00603 outfile = self.outroot+self.tid+".asap" 00604 mode = "auto" 00605 specunit = "km/s" 00606 #masklist = [[9186.458, 9484.109]] 00607 iflist = [2] 00608 pollist=[0] 00609 00610 masklist = self._get_chanval(infile,self.search,specunit,spw=iflist[0],addedge=True) 00611 print "masklist =", masklist 00612 00613 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00614 outfile=outfile,iflist=iflist,pollist=pollist, 00615 specunit=specunit) 00616 # sdbaseline returns None if it runs successfully 00617 self.assertEqual(result,None, 00618 msg="The task returned '"+str(result)+"' instead of None") 00619 # Compare IF2 00620 testval = self._getStats(outfile,self.blchan2,2) 00621 self._compareStats(testval,self.ref_pol0if2v) 00622 00623 def testblmask10( self ): 00624 """Mask test 10: test specunit='km/s' with masklist (list) and maskmode = 'list'""" 00625 self.tid="10" 00626 infile = self.infile 00627 outfile = self.outroot+self.tid+".asap" 00628 mode = "list" 00629 specunit = "km/s" 00630 #masklist = [[9186.458, 9366.642],[9373.118, 9484.109]] 00631 iflist = [2] 00632 pollist=[0] 00633 00634 masklist = self._get_chanval(infile,self.blchan2,specunit,spw=iflist[0],addedge=True) 00635 print "masklist =", masklist 00636 00637 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00638 outfile=outfile,iflist=iflist,pollist=pollist, 00639 specunit=specunit) 00640 # sdbaseline returns None if it runs successfully 00641 self.assertEqual(result,None, 00642 msg="The task returned '"+str(result)+"' instead of None") 00643 # Compare IF2 00644 testval = self._getStats(outfile,self.blchan2,2) 00645 self._compareStats(testval,self.ref_pol0if2v) 00646 00647 def testblmask11( self ): 00648 """Mask test 11: test specunit='km/s' with masklist (string) and maskmode = 'auto'""" 00649 self.tid="11" 00650 infile = self.infile 00651 outfile = self.outroot+self.tid+".asap" 00652 mode = "auto" 00653 specunit = "km/s" 00654 #masklist = "0:-134.960~162.691, 2:9186.458~9484.109" 00655 iflist = [0,2] 00656 pollist=[0] 00657 00658 sblrange=[] 00659 chanlist = (self.search, self.search) 00660 for i in xrange(len(iflist)): 00661 sblrange.append("") 00662 mlist = self._get_chanval(infile,chanlist[i],specunit,spw=iflist[i],addedge=True) 00663 for valrange in mlist: 00664 if len(sblrange[i]): sblrange[i] += ";" 00665 sblrange[i] += self._get_range_in_string(valrange) 00666 masklist = "0:"+sblrange[0]+", 2:"+sblrange[1] 00667 print "masklist =", masklist 00668 00669 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00670 outfile=outfile,iflist=iflist,pollist=pollist, 00671 specunit=specunit) 00672 # sdbaseline returns None if it runs successfully 00673 self.assertEqual(result,None, 00674 msg="The task returned '"+str(result)+"' instead of None") 00675 # Compare IF0 00676 testval = self._getStats(outfile,self.blchan0,0) 00677 self._compareStats(testval,self.ref_pol0if0v) 00678 # Compare IF2 00679 testval = self._getStats(outfile,self.blchan2,2) 00680 self._compareStats(testval,self.ref_pol0if2v) 00681 00682 def testblmask12( self ): 00683 """Mask test 12: test specunit='km/s' with masklist (string) and maskmode = 'list'""" 00684 self.tid="12" 00685 infile = self.infile 00686 outfile = self.outroot+self.tid+".asap" 00687 mode = "list" 00688 specunit = "km/s" 00689 #masklist = "0:-134.960~3.708;10.667~162.691, 2:9373.118~9484.109;9186.458~9366.642" 00690 iflist = [0,2] 00691 pollist=[0] 00692 00693 sblrange=[] 00694 chanlist = (self.blchan0, self.blchan2) 00695 for i in xrange(len(iflist)): 00696 sblrange.append("") 00697 mlist = self._get_chanval(infile,chanlist[i],specunit,spw=iflist[i],addedge=True) 00698 for valrange in mlist: 00699 if len(sblrange[i]): sblrange[i] += ";" 00700 sblrange[i] += self._get_range_in_string(valrange) 00701 masklist = "0:"+sblrange[0]+", 2:"+sblrange[1] 00702 print "masklist =", masklist 00703 00704 result = sdbaseline(infile=infile,maskmode=mode,masklist=masklist, 00705 outfile=outfile,iflist=iflist,pollist=pollist, 00706 specunit=specunit) 00707 # sdbaseline returns None if it runs successfully 00708 self.assertEqual(result,None, 00709 msg="The task returned '"+str(result)+"' instead of None") 00710 # Compare IF0 00711 testval = self._getStats(outfile,self.blchan0,0) 00712 self._compareStats(testval,self.ref_pol0if0v) 00713 # Compare IF2 00714 testval = self._getStats(outfile,self.blchan2,2) 00715 self._compareStats(testval,self.ref_pol0if2v) 00716 00717 00718 def _get_range_in_string( self, valrange ): 00719 if isinstance(valrange, list) or isinstance(valrange, tuple): 00720 return str(valrange[0])+"~"+str(valrange[1]) 00721 else: 00722 return False 00723 00724 def _get_chanval( self,file, chanrange, unit, spw=0, addedge=False ): 00725 mylist = [] 00726 scan = sd.scantable(file, average=False) 00727 scan.set_unit(unit) 00728 scan.set_selection(ifs=[spw]) 00729 chanval = scan._getabcissa(0) 00730 edge = 0 00731 if addedge: 00732 # add 1/2 chan to both edges 00733 nchan = len(chanval) 00734 #edge = 0.0 00735 edge = 0.5*abs(chanval[nchan-1]-chanval[0])/float(nchan-1) 00736 for schan, echan in chanrange: 00737 lval = max(chanval[schan],chanval[echan]) 00738 sval = min(chanval[schan],chanval[echan]) 00739 mylist.append([sval-edge,lval+edge]) 00740 del scan, nchan, edge, lval, sval 00741 return mylist 00742 00743 def _getStats( self, filename, basechan, ispw ): 00744 self.assertTrue(os.path.exists(filename), 00745 msg=("Output file '%s' doesn't exist" % filename)) 00746 linechan = [basechan[0][1]+1,basechan[1][0]-1] 00747 scan = sd.scantable(filename,average=False) 00748 scan.set_selection(ifs=[ispw]) 00749 scan.set_unit('channel') 00750 linmsk = scan.create_mask(linechan) 00751 blmsk = scan.create_mask(basechan) 00752 # only the fist row is returned 00753 linmax = scan.stats('max',mask=linmsk)[0] 00754 linmaxpos = scan.stats('max_abc',mask=linmsk)[0] 00755 linesum = scan.stats('sum',mask=linmsk)[0] 00756 blrms = scan.stats('rms',mask=blmsk)[0] 00757 blstd = scan.stats('stddev',mask=blmsk)[0] 00758 del scan, linmsk, blmsk 00759 retdic = {'linemax': linmax, 'linemaxpos': linmaxpos, 00760 'linesum': linesum, 'baserms': blrms, 'basestd': blstd} 00761 del linmax, linmaxpos, linesum, blrms, blstd 00762 print 'Current run (IF',ispw,'):',retdic 00763 return retdic 00764 00765 00766 class sdbaseline_funcTest( unittest.TestCase ): 00767 """ 00768 Unit tests for task sdbaseline. No interactive testing. 00769 00770 The list of tests: 00771 testCSpline01 --- test cubic spline fitting with maskmode = 'list' 00772 testCSpline02 --- test cubic spline fitting with maskmode = 'auto' 00773 testSinusoid01 --- test sinusoidal fitting with maskmode = 'list' 00774 testSinusoid02 --- test sinusoidal fitting with maskmode = 'auto' 00775 00776 Note: (1) the rms noise of input data is 1.0. 00777 (2) the max_rms value comes from +3sigma level of Chi-square 00778 distribution with freedom of (2048-38) for cspline, (2048-27) 00779 for sinusoid, respectively. 00780 00781 created 19/04/2011 by Wataru Kawasaki 00782 """ 00783 # Data path of input/output 00784 datapath = os.environ.get('CASAPATH').split()[0] + \ 00785 '/data/regression/unittest/sdbaseline/' 00786 # Input and output names 00787 infile_cspline = 'Artificial_CubicSpline.asap' 00788 infile_sinusoid = 'Artificial_Sinusoid.asap' 00789 blparamfile_suffix = '_blparam.txt' 00790 outroot = 'sdbaseline_test' 00791 tid = None 00792 00793 def setUp( self ): 00794 if os.path.exists(self.infile_cspline): 00795 shutil.rmtree(self.infile_cspline) 00796 shutil.copytree(self.datapath+self.infile_cspline, self.infile_cspline) 00797 if os.path.exists(self.infile_sinusoid): 00798 shutil.rmtree(self.infile_sinusoid) 00799 shutil.copytree(self.datapath+self.infile_sinusoid, self.infile_sinusoid) 00800 00801 default(sdbaseline) 00802 00803 def tearDown( self ): 00804 if os.path.exists(self.infile_cspline): 00805 shutil.rmtree(self.infile_cspline) 00806 if os.path.exists(self.infile_sinusoid): 00807 shutil.rmtree(self.infile_sinusoid) 00808 00809 def testCSpline01( self ): 00810 """Test CSpline01: Cubic spline fitting with maskmode = 'list'""" 00811 self.tid = "CSpline01" 00812 infile = self.infile_cspline 00813 mode = "list" 00814 outfile = self.outroot+self.tid+".asap" 00815 blparamfile = outfile+self.blparamfile_suffix 00816 00817 result = sdbaseline(infile=infile,maskmode=mode,outfile=outfile,blfunc='cspline',npiece=35) 00818 self.assertEqual(result, None, msg="The task returned '"+str(result)+"' instead of None") 00819 self.checkRms(blparamfile, 1.038696) #the actual rms should be 1.02407 though 00820 00821 def testCSpline02( self ): 00822 """Test CSpline02: Cubic spline fitting with maskmode = 'auto'""" 00823 self.tid = "CSpline02" 00824 infile = self.infile_cspline 00825 mode = "auto" 00826 outfile = self.outroot+self.tid+".asap" 00827 blparamfile = outfile+self.blparamfile_suffix 00828 00829 result = sdbaseline(infile=infile,maskmode=mode,outfile=outfile,blfunc='cspline',npiece=35) 00830 self.assertEqual(result, None, msg="The task returned '"+str(result)+"' instead of None") 00831 self.checkRms(blparamfile, 1.038696) #the actual rms should be 1.02407 though 00832 00833 def testSinusoid01( self ): 00834 """Test Sinusoid01: Sinusoidal fitting with maskmode = 'list'""" 00835 self.tid = "Sinusoid01" 00836 infile = self.infile_sinusoid 00837 mode = "list" 00838 outfile = self.outroot+self.tid+".asap" 00839 blparamfile = outfile+self.blparamfile_suffix 00840 00841 result = sdbaseline(infile=infile,maskmode=mode,outfile=outfile,blfunc='sinusoid') 00842 self.assertEqual(result, None, msg="The task returned '"+str(result)+"' instead of None") 00843 self.checkRms(blparamfile, 1.10) #the actual rms should be 1.09705 though 00844 00845 def testSinusoid02( self ): 00846 """Test Sinusoid02: Sinusoidal fitting with maskmode = 'auto'""" 00847 self.tid = "Sinusoid02" 00848 infile = self.infile_sinusoid 00849 mode = "auto" 00850 outfile = self.outroot+self.tid+".asap" 00851 blparamfile = outfile+self.blparamfile_suffix 00852 00853 result = sdbaseline(infile=infile,maskmode=mode,outfile=outfile,blfunc='sinusoid') 00854 self.assertEqual(result, None, msg="The task returned '"+str(result)+"' instead of None") 00855 self.checkRms(blparamfile, 1.10) #the actual rms should be 1.09705 though 00856 00857 def checkRms( self, blparamfile, max_rms ): 00858 rms = 10000.0 00859 for line in open(blparamfile,'r'): 00860 if line.strip() != "": 00861 items = line.split() 00862 if len(items) >= 3: 00863 if (items[0] == 'rms') and (items[1] == '='): 00864 rms = float(items[2]) 00865 break 00866 00867 self.assertTrue((rms <= max_rms), msg = "CSpline fitting failed.") 00868 00869 00870 class sdbaseline_multi_IF_test( sdbaseline_unittest_base, unittest.TestCase ): 00871 """ 00872 Unit tests for task sdbaseline. No interactive testing. 00873 00874 This test intends to check whether sdbaseline task works fine 00875 for data that has multiple IFs whose nchan differ each other. 00876 00877 The list of tests: 00878 test0 --- test multi IF data input 00879 00880 created 24/02/2012 by Takeshi Nakazato 00881 """ 00882 # Input and output names 00883 infile = 'testMultiIF.asap' 00884 blparamfile_suffix = '_blparam.txt' 00885 outroot = sdbaseline_unittest_base.taskname+'_multi' 00886 refblparamfile = 'refblparam_multiIF' 00887 00888 def setUp( self ): 00889 if os.path.exists(self.infile): 00890 shutil.rmtree(self.infile) 00891 shutil.copytree(self.datapath+self.infile, self.infile) 00892 default(sdbaseline) 00893 00894 def tearDown( self ): 00895 if os.path.exists(self.infile): 00896 shutil.rmtree(self.infile) 00897 00898 def test01multi( self ): 00899 """test01multi: Test the task works with multi IF data""" 00900 infile = self.infile 00901 mode = "list" 00902 blfunc = "poly" 00903 order = 1 00904 outfile = self.outroot+".asap" 00905 blparamfile = outfile+self.blparamfile_suffix 00906 00907 result = sdbaseline(infile=infile,maskmode=mode,outfile=outfile,blfunc=blfunc,order=order) 00908 self.assertEqual(result, None, msg="The task returned '"+str(result)+"' instead of None") 00909 self._compareBLparam(blparamfile,self.datapath+self.refblparamfile) 00910 reference = {5: {'rms': 1.4250789880752563, 00911 'min': -4.2702846527099609, 00912 'max': 5.5566844940185547, 00913 'max_abscissa': {'value': 823.0, 00914 'unit': 'channel'}, 00915 'median': 0.017315864562988281, 00916 'min_abscissa': {'value': 520.0, 00917 'unit': 'channel'}, 00918 'stddev': 1.425775408744812}, 00919 7: {'rms': 1.4971292018890381, 00920 'min': -4.7103700637817383, 00921 'max': 5.4820127487182617, 00922 'max_abscissa': {'value': 1335.0, 00923 'unit': 'channel'}, 00924 'median': 0.027227401733398438, 00925 'min_abscissa': {'value': 1490.0, 00926 'unit': 'channel'}, 00927 'stddev': 1.4974949359893799}} 00928 # sdstat must run each IF separately 00929 for ifno in [5,7]: 00930 currstat = self._getStats(outfile,[ifno]) 00931 self._compareStats(currstat,reference[ifno]) 00932 00933 class sdbaseline_storageTest( sdbaseline_unittest_base, unittest.TestCase ): 00934 """ 00935 Unit tests for task sdbaseline. Test scantable sotrage and insitu 00936 parameters 00937 00938 The list of tests: 00939 testMT --- storage = 'memory', insitu = True 00940 testMF --- storage = 'memory', insitu = False 00941 testDT --- storage = 'disk', insitu = True 00942 testDF --- storage = 'disk', insitu = False 00943 testDTow --- infile=outfile on storage = 'disk', insitu = True 00944 00945 Note on handlings of disk storage: 00946 Task script copies scantable after setting selection when storage='disk' 00947 """ 00948 # Input and output names 00949 infile = 'OrionS_rawACSmod_calave.asap' 00950 outroot = sdbaseline_unittest_base.taskname+'_store' 00951 mode = "list" 00952 iflist = [2] 00953 pollist = [1] 00954 00955 blparamfile_suffix = '_blparam.txt' 00956 blreffile = sdbaseline_unittest_base.datapath+'refblparam02' 00957 00958 # Reference statistic values of IF=2, POL=1 00959 refstat = {'rms': 3.4925737380981445, 00960 'min': -226.3941650390625, 00961 'max': 129.78572082519531, 00962 'max_abscissa': {'value': 8186.0, 'unit': 'channel'}, 00963 'median': -0.025681495666503906, 00964 'stddev': 3.4927871227264404, 00965 'min_abscissa': {'value': 8187.0, 'unit': 'channel'}} 00966 00967 def setUp( self ): 00968 if os.path.exists(self.infile): 00969 shutil.rmtree(self.infile) 00970 shutil.copytree(self.datapath+self.infile, self.infile) 00971 default(sdbaseline) 00972 00973 def tearDown( self ): 00974 if os.path.exists(self.infile): 00975 shutil.rmtree(self.infile) 00976 00977 def testMT( self ): 00978 """Storage Test MT: storage='memory' and insitu=T""" 00979 tid = "MT" 00980 outfile = self.outroot+tid+".asap" 00981 00982 initstat = self._getStats(self.infile) 00983 00984 sd.rcParams['scantable.storage'] = 'memory' 00985 sd.rcParams['insitu'] = True 00986 print "Running test with storage='%s' and insitu=%s" % \ 00987 (sd.rcParams['scantable.storage'], str(sd.rcParams['insitu'])) 00988 result = sdbaseline(infile=self.infile,maskmode=self.mode,outfile=outfile, 00989 iflist=self.iflist,pollist=self.pollist) 00990 00991 # sdbaseline returns None if it runs successfully 00992 self.assertEqual(result,None, 00993 msg="The task returned '"+str(result)+"' instead of None") 00994 print "Testing OUTPUT statistics and baseline parameters" 00995 self._compareBLparam(outfile+self.blparamfile_suffix,self.blreffile) 00996 self._compareStats(outfile,self.refstat) 00997 # Test input data 00998 newinstat = self._getStats(self.infile) 00999 print "Comparing INPUT statistics before/after calculations" 01000 self._compareStats(newinstat,initstat) 01001 01002 def testMF( self ): 01003 """Storage Test MF: storage='memory' and insitu=F""" 01004 tid = "MF" 01005 outfile = self.outroot+tid+".asap" 01006 01007 initstat = self._getStats(self.infile) 01008 01009 sd.rcParams['scantable.storage'] = 'memory' 01010 sd.rcParams['insitu'] = False 01011 print "Running test with storage='%s' and insitu=%s" % \ 01012 (sd.rcParams['scantable.storage'], str(sd.rcParams['insitu'])) 01013 result = sdbaseline(infile=self.infile,maskmode=self.mode,outfile=outfile, 01014 iflist=self.iflist,pollist=self.pollist) 01015 01016 # sdbaseline returns None if it runs successfully 01017 self.assertEqual(result,None, 01018 msg="The task returned '"+str(result)+"' instead of None") 01019 print "Testing OUTPUT statistics and baseline parameters" 01020 self._compareBLparam(outfile+self.blparamfile_suffix,self.blreffile) 01021 self._compareStats(outfile,self.refstat) 01022 # Test input data 01023 newinstat = self._getStats(self.infile) 01024 print "Comparing INPUT statistics before/after calculations" 01025 self._compareStats(newinstat,initstat) 01026 01027 def testDT( self ): 01028 """Storage Test DT: storage='disk' and insitu=T""" 01029 tid = "DT" 01030 outfile = self.outroot+tid+".asap" 01031 01032 initstat = self._getStats(self.infile) 01033 01034 sd.rcParams['scantable.storage'] = 'disk' 01035 sd.rcParams['insitu'] = True 01036 print "Running test with storage='%s' and insitu=%s" % \ 01037 (sd.rcParams['scantable.storage'], str(sd.rcParams['insitu'])) 01038 result = sdbaseline(infile=self.infile,maskmode=self.mode,outfile=outfile, 01039 iflist=self.iflist,pollist=self.pollist) 01040 01041 # sdbaseline returns None if it runs successfully 01042 self.assertEqual(result,None, 01043 msg="The task returned '"+str(result)+"' instead of None") 01044 print "Testing OUTPUT statistics and baseline parameters" 01045 self._compareBLparam(outfile+self.blparamfile_suffix,self.blreffile) 01046 self._compareStats(outfile,self.refstat) 01047 # Test input data 01048 newinstat = self._getStats(self.infile) 01049 print "Comparing INPUT statistics before/after calculations" 01050 self._compareStats(newinstat,initstat) 01051 01052 def testDF( self ): 01053 """Storage Test DF: storage='disk' and insitu=F""" 01054 tid = "DF" 01055 outfile = self.outroot+tid+".asap" 01056 01057 initstat = self._getStats(self.infile) 01058 01059 sd.rcParams['scantable.storage'] = 'disk' 01060 sd.rcParams['insitu'] = False 01061 print "Running test with storage='%s' and insitu=%s" % \ 01062 (sd.rcParams['scantable.storage'], str(sd.rcParams['insitu'])) 01063 result = sdbaseline(infile=self.infile,maskmode=self.mode,outfile=outfile, 01064 iflist=self.iflist,pollist=self.pollist) 01065 01066 # sdbaseline returns None if it runs successfully 01067 self.assertEqual(result,None, 01068 msg="The task returned '"+str(result)+"' instead of None") 01069 print "Testing OUTPUT statistics and baseline parameters" 01070 self._compareBLparam(outfile+self.blparamfile_suffix,self.blreffile) 01071 self._compareStats(outfile,self.refstat) 01072 # Test input data 01073 newinstat = self._getStats(self.infile) 01074 print "Comparing INPUT statistics before/after calculations" 01075 self._compareStats(newinstat,initstat) 01076 01077 01078 def testDTow( self ): 01079 """Storage Test DTow: infile=outfile on storage='disk' and insitu=T""" 01080 tid = "DTow" 01081 infile = self.infile 01082 outfile = self.infile 01083 overwrite = True 01084 01085 sd.rcParams['scantable.storage'] = 'disk' 01086 sd.rcParams['insitu'] = True 01087 print "Running test with storage='%s' and insitu=%s" % \ 01088 (sd.rcParams['scantable.storage'], str(sd.rcParams['insitu'])) 01089 result = sdbaseline(infile=infile,maskmode=self.mode,outfile=outfile, 01090 iflist=self.iflist,pollist=self.pollist,overwrite=True) 01091 01092 # sdbaseline returns None if it runs successfully 01093 self.assertEqual(result,None, 01094 msg="The task returned '"+str(result)+"' instead of None") 01095 print "Testing OUTPUT statistics and baseline parameters" 01096 self._compareBLparam(outfile+self.blparamfile_suffix,self.blreffile) 01097 self._compareStats(outfile,self.refstat) 01098 01099 01100 def suite(): 01101 return [sdbaseline_basicTest, sdbaseline_maskTest, sdbaseline_funcTest, 01102 sdbaseline_multi_IF_test, sdbaseline_storageTest] 01103 01104 ### Utilities for reading blparam file 01105 class FileReader( object ): 01106 def __init__( self, filename ): 01107 self.__filename = filename 01108 self.__data = None 01109 self.__nline = None 01110 01111 def read( self ): 01112 if self.__data is None: 01113 f = open(self.__filename, 'r') 01114 self.__data = f.readlines() 01115 f.close() 01116 self.__nline = len( self.__data ) 01117 return 01118 01119 def nline( self ): 01120 self.read() 01121 return self.__nline 01122 01123 def index( self, txt, start ): 01124 return self.__data[start:].index( txt ) + 1 + start 01125 01126 def getline( self, idx ): 01127 return self.__data[idx] 01128 01129 class BlparamFileParser( FileReader ): 01130 def __init__( self, blfile ): 01131 FileReader.__init__( self, blfile ) 01132 self.__nrow = None 01133 self.__coeff = None 01134 self.__rms = None 01135 self.__ctxt = 'Baseline parameters\n' 01136 self.__rtxt = 'Results of baseline fit\n' 01137 01138 def nrow( self ): 01139 self.read() 01140 if self.__nrow is None: 01141 return self._nrow() 01142 else: 01143 return self.__nrow 01144 01145 def coeff( self ): 01146 self.read() 01147 if self.__coeff is None: 01148 self.parseCoeff() 01149 return self.__coeff 01150 01151 def rms( self ): 01152 self.read() 01153 if self.__rms is None: 01154 self.parseRms() 01155 return self.__rms 01156 01157 def _nrow( self ): 01158 self.__nrow = 0 01159 for i in xrange(self.nline()): 01160 if self.getline( i ) == self.__ctxt: 01161 self.__nrow += 1 01162 return self.__nrow 01163 01164 def parse( self ): 01165 self.read() 01166 self.parseCoeff() 01167 self.parseRms() 01168 return 01169 01170 def parseCoeff( self ): 01171 self.__coeff = [] 01172 nrow = self.nrow() 01173 idx = 0 01174 while ( len(self.__coeff) < nrow ): 01175 try: 01176 idx = self.index( self.__ctxt, idx ) 01177 coeffs = [] 01178 while( self.getline( idx ) != self.__rtxt ): 01179 coeff = self.__parseCoeff( idx ) 01180 coeffs += coeff 01181 idx += 1 01182 self.__coeff.append( coeffs ) 01183 except: 01184 break 01185 return 01186 01187 def parseRms( self ): 01188 self.__rms = [] 01189 nrow = self.nrow() 01190 idx = 0 01191 while ( len(self.__rms) < nrow ): 01192 try: 01193 idx = self.index( self.__rtxt, idx ) 01194 self.__rms.append( self.__parseRms( idx ) ) 01195 except: 01196 break 01197 return 01198 01199 def __parseCoeff( self, idx ): 01200 return parseCoeff( self.getline( idx ) ) 01201 01202 def __parseRms( self, idx ): 01203 return parseRms( self.getline( idx ) ) 01204 01205 def parseCoeff( txt ): 01206 clist = txt.rstrip( '\n' ).split(',') 01207 ret = [] 01208 for c in clist: 01209 ret.append( float( c.split('=')[1] ) ) 01210 return ret 01211 01212 def parseRms( txt ): 01213 t = txt.lstrip().rstrip( '\n' )[6:] 01214 return float( t ) 01215
1.8.0
