Source
xxxxxxxxxx
45
45
datapath = DATADIR
46
46
47
47
print('setjy tests will use data from '+datapath)
48
48
49
49
if 'BYPASS_PARALLEL_PROCESSING' in os.environ:
50
50
ParallelTaskHelper.bypassParallelProcessing(1)
51
51
52
52
53
53
class SetjyUnitTestBase(unittest.TestCase):
54
54
55
-
def setUpMS(self,MS):
55
+
def setUpMS(self,MS, ismms=False):
56
56
57
+
# Modified to test MMS case in the same test_setjy run (2020.05.12 TT)
58
+
# If ismms=True, look for MMS data is in the same root data directory path but under
59
+
# sub-directory, mms.
60
+
57
61
self.inpms = MS
58
62
59
63
# Create working area
60
64
# setjydatapath = 'unittest/setjy/'
61
65
# 2015-02-02 TT: this seems unnessary directory layer...
62
66
#if not os.path.exists(setjydatapath):
63
67
# print "\nCreate working area..."
64
68
# os.system('mkdir -p '+setjydatapath)
65
69
66
70
# Create a new fresh copy of the MS
67
71
print("\nCreate a new local copy of the MS...")
68
72
#print(" setjydatapath=",setjydatapath, " inpms=",self.inpms)
69
73
if testmms:
70
74
os.system('cp -rH ' + os.path.join(datapath,self.inpms) + ' ' + self.inpms)
75
+
elif ismms:
76
+
os.system('cp -rH ' + os.path.join(datapath+'/mms',self.inpms) + ' ' + self.inpms)
71
77
else:
72
78
os.system('cp -rf ' + os.path.join(datapath,self.inpms) + ' ' + self.inpms)
73
79
74
80
def resetMS(self):
75
81
76
82
if os.path.exists(self.inpms):
77
83
print("\nRemoving a local copy of MS from the previous test...")
78
84
#ret = os.system('rm -rf unittest/setjy/*')
79
85
shutil.rmtree(self.inpms)
80
86
1159
1165
for ky in sjran.keys():
1160
1166
if 'fieldName' in sjran[ky] and sjran[ky]['fieldName']==self.field:
1161
1167
outfldid = ky
1162
1168
break
1163
1169
ret = len(outfldid)
1164
1170
if not ret:
1165
1171
print("FAIL: missing field = %s in the returned dictionary" % self.field)
1166
1172
self.check_eq(sjran['12']['0']['fluxd'][0],0.99137,0.0001)
1167
1173
self.check_eq(sjran['12']['1']['fluxd'][0],0.99132,0.0001)
1168
1174
self.assertTrue(ret)
1175
+
print("ret=%s" % sjran)
1169
1176
1170
1177
def test_PB2017(self):
1171
1178
self.modelim = ""
1172
1179
sjran = setjy(vis=self.inpms,
1173
1180
field=self.field,
1174
1181
modimage=self.modelim,
1175
1182
standard='Perley-Butler 2017',
1176
1183
usescratch=True
1177
1184
)
1178
1185
ret = True
1185
1192
outfldid = ky
1186
1193
break
1187
1194
ret = len(outfldid)
1188
1195
if not ret:
1189
1196
print("FAIL: missing field = %s in the returned dictionary" % self.field)
1190
1197
#self.check_eq(sjran['12']['0']['fluxd'][0],1.15116881972,0.0001)
1191
1198
#self.check_eq(sjran['12']['1']['fluxd'][0],1.15111995508,0.0001)
1192
1199
self.check_eq(sjran['12']['0']['fluxd'][0],0.99137,0.0001)
1193
1200
self.check_eq(sjran['12']['1']['fluxd'][0],0.99132,0.0001)
1194
1201
self.assertTrue(ret)
1195
-
1202
+
print("ret=%s" % sjran)
1196
1203
1204
+
class test_newStandards_MMS(SetjyUnitTestBase):
1205
+
"""Test simple Stnadard Scaling with MMS data"""
1206
+
def setUp(self):
1207
+
# MMS version of the data is stored in the subdirectory, mms
1208
+
prefix = 'n1333_1'
1209
+
msname=prefix+'.ms'
1210
+
# this is MMS
1211
+
self.setUpMS(msname, True)
1212
+
self.field='0542+498_1' #3C147
1213
+
1214
+
def tearDown(self):
1215
+
self.resetMS()
1216
+
#pass
1217
+
1218
+
def test_PB2013_MMS(self):
1219
+
self.modelim = ""
1220
+
sjran = setjy(vis=self.inpms,
1221
+
field=self.field,
1222
+
modimage=self.modelim,
1223
+
standard='Perley-Butler 2013',
1224
+
usescratch=True
1225
+
)
1226
+
ret = True
1227
+
if type(sjran)!=dict:
1228
+
ret = False
1229
+
else:
1230
+
outfldid = ""
1231
+
for ky in sjran.keys():
1232
+
if 'fieldName' in sjran[ky] and sjran[ky]['fieldName']==self.field:
1233
+
outfldid = ky
1234
+
break
1235
+
ret = len(outfldid)
1236
+
if not ret:
1237
+
print("FAIL: missing field = %s in the returned dictionary" % self.field)
1238
+
self.check_eq(sjran['12']['0']['fluxd'][0],0.99137,0.0001)
1239
+
self.check_eq(sjran['12']['1']['fluxd'][0],0.99132,0.0001)
1240
+
self.assertTrue(ret)
1241
+
#print("ret=%s" % sjran)
1242
+
1243
+
def test_PB2017_MMS(self):
1244
+
self.modelim = ""
1245
+
sjran = setjy(vis=self.inpms,
1246
+
field=self.field,
1247
+
modimage=self.modelim,
1248
+
standard='Perley-Butler 2017',
1249
+
usescratch=True
1250
+
)
1251
+
ret = True
1252
+
if type(sjran)!=dict:
1253
+
ret = False
1254
+
else:
1255
+
outfldid = ""
1256
+
for ky in sjran.keys():
1257
+
if 'fieldName' in sjran[ky] and sjran[ky]['fieldName']==self.field:
1258
+
outfldid = ky
1259
+
break
1260
+
ret = len(outfldid)
1261
+
if not ret:
1262
+
print("FAIL: missing field = %s in the returned dictionary" % self.field)
1263
+
#self.check_eq(sjran['12']['0']['fluxd'][0],1.15116881972,0.0001)
1264
+
#self.check_eq(sjran['12']['1']['fluxd'][0],1.15111995508,0.0001)
1265
+
self.check_eq(sjran['12']['0']['fluxd'][0],0.99137,0.0001)
1266
+
self.check_eq(sjran['12']['1']['fluxd'][0],0.99132,0.0001)
1267
+
self.assertTrue(ret)
1268
+
#print("ret=%s" % sjran)
1269
+
1270
+
1271
+
1197
1272
class test_inputs(SetjyUnitTestBase):
1198
1273
"""Test input parameter checking"""
1199
1274
def setUp(self):
1200
1275
#self.setUpMS("unittest/setjy/2528.ms") # Uranus
1201
1276
#self.setUpMS("2528.ms") # Uranus
1202
1277
self.setUpMS("multiobs.ms")
1203
1278
1204
1279
def tearDown(self):
1205
1280
self.resetMS()
1206
1281
1971
2046
res = setjy(vis=self.inpms, field='1', spw='3', scan='5',
1972
2047
standard='Butler-JPL-Horizons 2012', intent="*CALIB*FLUX*",
1973
2048
usescratch=True)
1974
2049
1975
2050
self.assertEqual(res, False, "setjy did not return False for a null selection")
1976
2051
1977
2052
1978
2053
def suite():
1979
2054
return [test_SingleObservation,test_MultipleObservations,test_ModImage, test_inputs,
1980
2055
test_conesearch, test_fluxscaleStandard, test_setpol, test_ephemtbl,
1981
-
test_tpmAsteroid,test_NullSelection, test_newStandards]
2056
+
test_tpmAsteroid,test_NullSelection, test_newStandards, test_newStandards_MMS]
1982
2057
1983
2058
if is_CASA6:
1984
2059
if __name__ == '__main__':
1985
2060
unittest.main()