Coverage for /wheeldirectory/casa-6.7.0-12-py3.10.el8/lib/py/lib/python3.10/site-packages/casatasks/private/callibrary.py: 8%
229 statements
« prev ^ index » next coverage.py v7.6.4, created at 2024-10-31 17:39 +0000
3class callibrary(object):
5 def __init__(self):
6 self.cld={} # as dict
8 def __repr__(self):
9 return "<callibrary>"
11 def cldefinstance(self):
12 definst={
13 "field" :"",
14 "intent":"",
15 "spw": "",
16 "obs": "",
17 "fldmap" : [],
18 "obsmap" : [],
19 "spwmap" : [],
20 "antmap" : [],
21 #"calwt" : False,
22 "tinterp" : "",
23 "finterp" : "",
24 "reach" : ""
25 }
26 return definst
29 def clear(self):
30 self.__init__()
32 def len(self):
33 return len(self.cld)
35 def addold(self,field='',spw='',intent='',
36 gaintable='',gainfield='',interp='',spwmap=[],calwt=False):
38 if len(gaintable)<1:
39 raise Exception('Please specify at least a gaintable.')
41 # insist all cal params are lists
42 # NB: data selection params are _not_ lists
43 if (not isinstance(gaintable,list)):
44 gaintable=[gaintable]
46 if (not isinstance(gainfield,list)):
47 gainfield=[gainfield]
48 if (not isinstance(interp,list)):
49 interp=[interp]
50 if (not isinstance(calwt,list)):
51 calwt=[calwt]
52 if isinstance(spwmap,list) and len(spwmap)>0:
53 if (not isinstance(spwmap[0],list)):
54 spwmap=[spwmap] # nest it
55 else:
56 spwmap=[]
59 for itab in range(len(gaintable)):
60 tint='linear'
61 fint=''
62 sinterp=interp[itab].split(',') if itab<len(interp) else []
63 if len(sinterp)>0 and len(sinterp[0])>0:
64 tint=sinterp[0]
65 fint=sinterp[1] if (len(sinterp)>1) else ''
66 self.add(caltable=gaintable[itab],
67 field=field,
68 spw=spw,
69 intent=intent,
70 tinterp=tint,
71 finterp=fint,
72 calwt=calwt[itab] if itab<len(calwt) else calwt[len(calwt)-1],
73 fldmap=gainfield[itab] if itab<len(gainfield) else '',
74 spwmap=spwmap[itab] if itab<len(spwmap) else []
75 )
78 #
79 def add(self,caltable,
80 field='',intent='',spw='',obs='',
81 tinterp='',finterp='',reach='',calwt=True,
82 obsmap=[],fldmap=[],spwmap=[],antmap=[]):
84 # if caltable is a list, insist others are, too
85 if (isinstance(caltable,list)):
87 if (not isinstance(field,list)):
88 field=[field]
89 if (not isinstance(intent,list)):
90 intent=[intent]
91 if (not isinstance(spw,list)):
92 spw=[spw]
93 if (not isinstance(obs,list)):
94 obs=[obs]
96 if (not isinstance(tinterp,list)):
97 tinterp=[tinterp]
98 if (not isinstance(finterp,list)):
99 finterp=[finterp]
100 if (not isinstance(reach,list)):
101 reach=[reach]
102 if (not isinstance(calwt,list)):
103 calwt=[calwt]
105 if (len(obsmap)>0):
106 if not isinstance(obsmap[0],list):
107 obsmap=[obsmap] # nested list
108 if (len(fldmap)>0):
109 if not isinstance(fldmap[0],list):
110 fldmap=[fldmap] # nested list
111 if (len(spwmap)>0):
112 if not isinstance(spwmap[0],list):
113 spwmap=[spwmap] # nested list
114 if (len(antmap)>0):
115 if not isinstance(antmap[0],list):
116 antmap=[antmap] # nested list
118 igt=0
119 for ct in caltable:
120 self.parsetorec(caltable=ct,
121 field=field[igt] if (igt<len(field)) else "",
122 intent=intent[igt] if (igt<len(intent)) else "",
123 spw=spw[igt] if (igt<len(spw)) else "",
124 obs=obs[igt] if (igt<len(obs)) else "",
125 tinterp=tinterp[igt] if (igt<len(tinterp)) else "",
126 finterp=finterp[igt] if (igt<len(finterp)) else "",
127 reach=reach[igt] if (igt<len(reach)) else "",
128 calwt=calwt[igt] if (igt<len(calwt)) else calwt[(len(calwt)-1)],
129 obsmap=obsmap[igt] if (igt<len(obsmap)) else [],
130 fldmap=fldmap[igt] if (igt<len(fldmap)) else [],
131 spwmap=spwmap[igt] if (igt<len(spwmap)) else [],
132 antmap=antmap[igt] if (igt<len(antmap)) else [],
133 )
134 igt+=1
135 else:
136 self.parsetorec(caltable=caltable,
137 field=field,intent=intent,spw=spw,obs=obs,
138 tinterp=tinterp,finterp=finterp,
139 reach=reach,calwt=calwt,
140 obsmap=obsmap,fldmap=fldmap,spwmap=spwmap,antmap=antmap)
142 def parsetorec(self,caltable,
143 field='',intent='',spw='',obs='',
144 tinterp='linear',finterp='',reach='',calwt=True,
145 obsmap=[],fldmap=[],spwmap=[],antmap=[]):
147 d0=self.cldefinstance()
148 d0["field"]=field
149 d0["intent"]=intent
150 d0["spw"]=spw
151 d0["obs"]=obs
153 d0["obsmap"]=obsmap
154 d0["fldmap"]=fldmap
155 d0["spwmap"]=spwmap
156 d0["antmap"]=antmap
158 d0["tinterp"]=tinterp
159 d0["finterp"]=finterp
160 d0["reach"]=reach
162 self.addrec({caltable : d0},calwt)
164 def addrec(self,crec,calwt):
166 ctname=list(crec.keys())[0]
168 irec=0
169 if (ctname in self.cld):
170 # ctname exists, will add a new instance
171 irec=len(self.cld[ctname])-1
173 # prefer already-set calwt
174 calwt0=self.cld[ctname]["calwt"]
175 if calwt!=calwt0:
176 print('WARNING: For caltable=\''+ctname+'\' using already-set calwt='+str(calwt0)+'.')
177 else:
178 # ctname does not yet exist, add it
179 self.cld[ctname] = {}
180 self.cld[ctname]["calwt"]=calwt
183 self.cld[ctname][str(irec)]=crec[ctname]
185 def list(self):
186 print('There are '+str(len(self.cld))+' caltables in the cal library:')
187 keys=list(self.cld.keys())
188 keys.sort()
189 for ct in keys:
190 print(ct+': calwt='+str(self.cld[ct]['calwt'])+str(' (')+str(len(self.cld[ct])-1)+str(' instance[s]):'))
191 for ims in self.cld[ct].keys():
192 if (isinstance(self.cld[ct][ims],dict)):
193 print(' field=\''+str(self.cld[ct][ims]['field'])+'\'', end=' ')
194 print(' intent=\''+str(self.cld[ct][ims]['intent'])+'\'', end=' ')
195 print(' spw=\''+str(self.cld[ct][ims]['spw'])+'\'', end=' ')
196 print(' obs=\''+str(self.cld[ct][ims]['obs'])+'\'')
197 print(' tinterp=\''+str(self.cld[ct][ims]['tinterp'])+'\'', end=' ')
198 print(' finterp=\''+str(self.cld[ct][ims]['finterp'])+'\'')
199 #print( ' reach=\''+str(self.cld[ct][ims]['reach'])+'\'')
200 print(' obsmap='+str(self.cld[ct][ims]['obsmap']), end=' ')
201 print(' fldmap='+str(self.cld[ct][ims]['fldmap']), end=' ')
202 print(' spwmap='+str(self.cld[ct][ims]['spwmap']), end=' ')
203 print(' antmap='+str(self.cld[ct][ims]['antmap']))
206 def write(self,filename,append=False):
207 if len(filename)<1:
208 raise Exception('Please specify a filename')
209 if len(self.cld)<1:
210 raise Exception('There is no cal library to write')
212 fw="w"
213 if append:
214 fw="a"
216 f=open(filename,fw)
217 keys0=list(self.cld.keys())
218 keys0.sort()
219 for ct in keys0:
220 ict0=self.cld[ct]
221 keys1=list(ict0.keys())
222 keys1.sort()
223 for ims in keys1:
224 ict1=ict0[ims]
225 if isinstance(ict1,dict):
226 print('caltable=\''+ct+'\'', end=' ', file=f)
227 print('calwt='+str(ict0['calwt']), end=' ', file=f)
228 if len(ict1['field'])>0:
229 print('field=\''+str(ict1['field'])+'\'', end=' ', file=f)
230 if len(ict1['intent'])>0:
231 print('intent=\''+str(ict1['intent'])+'\'', end=' ', file=f)
232 if len(ict1['spw'])>0:
233 print('spw=\''+str(ict1['spw'])+'\'', end=' ', file=f)
234 if len(ict1['obs'])>0:
235 print('obs=\''+str(ict1['obs'])+'\'', end=' ', file=f)
237 if len(ict1['tinterp'])>0:
238 print('tinterp=\''+str(ict1['tinterp'])+'\'', end=' ', file=f)
239 if len(ict1['finterp'])>0:
240 print('finterp=\''+str(ict1['finterp'])+'\'', end=' ', file=f)
241 if len(ict1['reach'])>0:
242 print('reach=\''+str(ict1['reach'])+'\'', end=' ', file=f)
244 if len(ict1['obsmap'])>0:
245 print('obsmap='+str(ict1['obsmap']), end=' ', file=f)
246 if len(ict1['fldmap'])>0:
247 if isinstance(ict1['fldmap'],str):
248 print('fldmap=\''+str(ict1['fldmap'])+'\'', end=' ', file=f)
249 else:
250 print('fldmap='+str(ict1['fldmap']), end=' ', file=f)
251 if len(ict1['spwmap'])>0:
252 print('spwmap='+str(ict1['spwmap']), end=' ', file=f)
253 if len(ict1['antmap'])>0:
254 print('antmap='+str(ict1['antmap']), end=' ', file=f)
256 print('', file=f)
258 f.close()
260 def read(self,callibr):
262 lines=[]
263 if isinstance(callibr,list):
264 # a python list of lines has been specified
265 lines=callibr
266 else:
267 # assume a filename has been specified
268 lines=open(callibr)
270 for line in lines:
271 line2=line.strip() # remove leading/trailing whitespace
273 # Attempt to parse if it has content
274 if len(line2)>0:
275 if line2[0]=='#':
276 # Ignore lines that are comments (or turned off with #)
277 print('Found comment (not parsed): ',line2)
278 else:
279 # A nominally parsable line, apparently
281 # reduce whitespace to only singles
282 line2=' '.join(line2.split())
284 # absorb remaining spaces adjacent to =
285 line2=line2.replace(' =','=')
286 line2=line2.replace('= ','=')
288 # sub , for spaces to delimit keys in the parsed command
289 line2=line2.replace(' ',',')
290 line2=line2.replace(',,',',') # ~corrects likely comma replacement within quotes
292 # add parsetorec() command syntax
293 parsecmd='self.parsetorec('+line2+')'
295 # cope with bool recognition for calwt
296 parsecmd=parsecmd.replace('calwt=T,','calwt=True,')
297 parsecmd=parsecmd.replace('calwt=T)','calwt=True)')
298 parsecmd=parsecmd.replace('calwt=F,','calwt=False,')
299 parsecmd=parsecmd.replace('calwt=F)','calwt=False)')
301 # execute, and trap/report any errors that occur
302 try:
303 exec(parsecmd)
304 except Exception as errline:
305 self.clear()
306 print('Error: ',errline)
307 print('Problem parsing cal library line (check for typos): "'+line+'"')
308 raise Exception('Problem parsing cal library line (check for typos): '+line)
310 def compare(self,other):
311 return self.cld==other.cld
314def applycaltocallib(filename,append=False,field='',spw='',intent='',
315 gaintable='',gainfield='',interp='',spwmap=[],calwt=True):
317 if len(filename)<1:
318 raise Exception('Please specify a filename')
320 if len(gaintable)<1:
321 raise Exception('No caltable specified in gaintable')
323 c=callibrary()
324 c.addold(field=field,spw=spw,intent=intent,gaintable=gaintable,
325 gainfield=gainfield,interp=interp,spwmap=spwmap,calwt=calwt)
326 c.write(filename,append)
327 c.clear()
330def testcallib0():
332 c=callibrary()
333 c.add(caltable="G0",field="0",tinterp="nearest")
334 c.add(caltable="G0",field="1,2",tinterp="linear")
335 c.add(caltable="B0",tinterp="nearest",finterp="linear")
336 c.list()
337 return c
340def testcallib1():
342 c1=callibrary()
343 c1.add(caltable=['B','phase','flux'],field='0~1,3~4',
344 tinterp=['nearest','linear'],ctfield=['0',''])
345 c1.add(caltable=['B','phase','flux'],field='2',
346 finterp=['nearest','linear'],ctfield=['0','3,4'])
347 print('')
348 print('c1: -----------------------------------------------')
349 c1.list()
351 c2=callibrary()
352 c2.add(gaintable=['B','flux'],field='0~4',
353 tinterp=['nearest'],ctfield=['0'])
354 c2.add(gaintable='phase',field='0~1,3~4',
355 tinterp='linear',ctfield='')
356 c2.add(gaintable='phase',field='2',
357 tinterp='linear',ctfield='3,4')
358 print('')
359 print('c2: -----------------------------------------------')
360 c2.list()
362 return (c1,c2)
365def testcallib2():
367 c1=callibrary()
368 c1.add(caltable=['B','flux'],field='0~4',
369 tinterp=['nearest'],ctfield=['0'])
370 c1.add(caltable='phase',field='0~1,3~4',
371 tinterp='linear',ctfield='')
372 c1.add(caltable='phase',field='2',
373 tinterp='linear',ctfield='3,4')
374 c1.write('testcallib2.txt')
376 c2=callibrary()
377 c2.read('testcallib2.txt')
379 print('Cal libraries match?', c2.cld==c1.cld)
381 return (c1,c2)