source: trunk/test/test_scantable.py@ 2801

Last change on this file since 2801 was 2801, checked in by Malte Marquarding, 13 years ago

A failed test to check jenkins email output. Remove after
File size: 8.8 KB
Line 
1import sys
2import os
3import shutil
4import datetime
5from nose.tools import *
6import asap
7from asap import scantable, selector, mask_not
8from asap.logging import asaplog
9# no need for log messages
10asaplog.disable()
11
12def tempdir_setup():
13 os.makedirs("test_temp")
14
15def tempdir_teardown():
16 shutil.rmtree("test_temp", True)
17
18class TestScantable(object):
19 def setup(self):
20 pth = os.path.dirname(__file__)
21 s = scantable(os.path.join(pth, "data", "MOPS.rpf"), average=True)
22 sel = selector()
23 # make sure this order is always correct - in can be random
24 sel.set_order(["SCANNO", "POLNO"])
25 s.set_selection(sel)
26 self.st = s.copy()
27 restfreqs = [86.243] # 13CO-1/0, SiO the two IF
28 self.st.set_restfreqs(restfreqs,"GHz")
29
30 def test_init(self):
31 fname = os.path.join(os.path.dirname(__file__), "data", "MOPS.rpf")
32 st = scantable(fname, average=False)
33 assert_equal(st.ncycle(), 32)
34 st = scantable(fname, average=True)
35 assert_equal(st.ncycle(), 2)
36 st = scantable(fname, unit="Jy")
37 assert_equal(st.get_fluxunit(), "Jy")
38 st = scantable(fname, unit="K")
39 assert_equal(st.get_fluxunit(), "K")
40 assert_raises(RuntimeError, scantable, fname, unit="junk")
41 st = scantable([fname,fname], average=False)
42 assert_equal(st.nscan(), 4)
43
44 def test_jenkins_email(self):
45 raise RuntimeError("Just check that the email has the log attached")
46
47 def test_copy(self):
48 st = self.st.copy()
49 assert_not_equal(id(st), id(self.st))
50
51 def test_drop_scan(self):
52 st = self.st.drop_scan([1])
53 assert_equal(st.nscan(), 1)
54
55 def test_get_scan(self):
56 st = self.st.get_scan([1])
57 assert_equal(st.nscan(), 1)
58 st = self.st.get_scan("Orion_SiO_R")
59 assert_equal(st.get_sourcename()[-1], "Orion_SiO_R")
60 assert_equal(st.nscan(), 1)
61
62 def test_get_spectrum(self):
63 spec = self.st.get_spectrum(0)
64 assert_almost_equal(max(spec), 215.279830933)
65
66 def test_get_mask(self):
67 spec = self.st.get_mask(0)
68 assert_equal(len(spec), 4096)
69
70 def test_set_spectrum(self):
71 spec = [ 1.0 for i in range(self.st.nchan()) ]
72 self.st.set_spectrum(spec, 0)
73 spec1 = self.st.get_spectrum(0)
74 assert_almost_equal(max(spec1), 1.0)
75
76 def test_selection(self):
77 sel = selector()
78 sel.set_polarisations("YY")
79 self.st.set_selection(sel)
80 assert_equal(self.st.getpolnos(), [1])
81 sel1 = self.st.get_selection()
82 assert_equal(sel1.get_pols(), [1])
83 self.st.set_selection(pols="XX")
84 assert_equal(self.st.getpolnos(), [0])
85
86
87 def stats(self, key, value, mask=False):
88 msk = None
89 if mask:
90 msk = self.st.create_mask([0,100], [3900,4096])
91 sval = self.st.stats(stat=key, mask=msk)
92 assert_almost_equal(sval[0], value)
93
94 def test_masked_stats(self):
95 stats = { 'min': 113.767166138,
96 'max':128.21571350, 'sumsq':4180516.75,
97 'sum':35216.87890625, 'mean':118.5753479,
98 'var':15.75608253, 'stddev':3.9693932533,
99 'avdev':3.395271778, 'rms':118.6415405,
100 'median':117.5024261}
101 for k,v in stats.items():
102 yield self.stats, k, v, True
103
104 def test_stats(self):
105 stats = { 'min': 113.767166138,
106 'max':215.279830933, 'sumsq':128759200.0,
107 'sum':720262.375, 'mean':175.845306396,
108 'var':513.95324707, 'stddev':22.6705360413,
109 'avdev':16.3966751099, 'rms':177.300170898,
110 'median':182.891845703}
111 for k,v in stats.items():
112 yield self.stats, k, v
113
114 def test_get_column_names(self):
115 cnames = ['SCANNO', 'CYCLENO', 'BEAMNO', 'IFNO',
116 'POLNO', 'FREQ_ID', 'MOLECULE_ID', 'REFBEAMNO', 'FLAGROW',
117 'TIME', 'INTERVAL', 'SRCNAME', 'SRCTYPE',
118 'FIELDNAME', 'SPECTRA', 'FLAGTRA', 'TSYS',
119 'DIRECTION', 'AZIMUTH', 'ELEVATION',
120 'OPACITY', 'TCAL_ID', 'FIT_ID',
121 'FOCUS_ID', 'WEATHER_ID', 'SRCVELOCITY',
122 'SRCPROPERMOTION', 'SRCDIRECTION',
123 'SCANRATE']
124 assert_equal(self.st.get_column_names(), cnames)
125
126 def test_get_tsys(self):
127 assert_almost_equal(self.st.get_tsys()[0], 175.830429077)
128
129 def test_set_tsys(self):
130 s = self.st.copy()
131 newval = 100.0
132 s.set_tsys(newval, 0)
133 assert_almost_equal(s.get_tsys()[0], newval)
134 s2 = self.st.copy()
135 s2.set_tsys(newval)
136 out = s2.get_tsys()
137 for i in xrange(len(out)):
138 assert_almost_equal(out[i], newval)
139
140 def test_get_time(self):
141 assert_equal(self.st.get_time(0), '2008/03/12/09:32:50')
142 dt = datetime.datetime(2008,3,12,9,32,50)
143 assert_equal(self.st.get_time(0, True), dt)
144
145 def test_get_inttime(self):
146 assert_almost_equal(self.st.get_inttime()[0], 30.720016479)
147
148 def test_get_sourcename(self):
149 assert_equal(self.st.get_sourcename(0), 'Orion_SiO_R')
150 assert_equal(self.st.get_sourcename(),
151 ['Orion_SiO_R', 'Orion_SiO_R',
152 'Orion_SiO', 'Orion_SiO'])
153
154 def test_get_azimuth(self):
155 assert_almost_equal(self.st.get_azimuth()[0], 5.628767013)
156
157 def test_get_elevation(self):
158 assert_almost_equal(self.st.get_elevation()[0], 1.01711678504)
159
160 def test_get_parangle(self):
161 assert_almost_equal(self.st.get_parangle()[0], 2.5921990871)
162
163 def test_get_direction(self):
164 assert_equal(self.st.get_direction()[0], 'J2000 05:35:14.5 -04.52.29.5')
165
166 def test_get_directionval(self):
167 dv = self.st.get_directionval()[0]
168 assert_almost_equal(dv[0], 1.4627692699)
169 assert_almost_equal(dv[1], -0.0850824415)
170
171 def test_unit(self):
172 self.st.set_unit('')
173 self.st.set_unit('GHz')
174 self.st.set_unit('km/s')
175 assert_raises(RuntimeError, self.st.set_unit, 'junk')
176 assert_equals(self.st.get_unit(), 'km/s')
177
178 def test_average_pol(self):
179 ap = self.st.average_pol()
180 assert_equal(ap.npol(), 1)
181
182 def test_drop_scan(self):
183 s0 = self.st.drop_scan(1)
184 assert_equal(s0.getscannos(), [0])
185 s1 = self.st.drop_scan([0])
186 assert_equal(s1.getscannos(), [1])
187
188 def test_flag(self):
189 q = self.st.auto_quotient()
190 q.set_unit('km/s')
191 q0 = q.copy()
192 q1 = q.copy()
193 msk = q0.create_mask([-10,20])
194 q0.flag(mask=mask_not(msk))
195 q1.flag(mask=msk)
196 assert_almost_equal(q0.stats(stat='max')[0], 95.62171936)
197 assert_almost_equal(q1.stats(stat='max')[0], 2.66563416)
198
199 @with_setup(tempdir_setup, tempdir_teardown)
200 def test_save(self):
201 fname = os.path.join("test_temp", 'scantable_test.%s')
202 formats = [(fname % 'sdfits', 'SDFITS', True),
203 (fname % 'ms', 'MS2', True),
204 (fname % 'class.fits', 'CLASS', False),
205 (fname % 'fits', 'FITS', False),
206 (fname % 'txt', 'ASCII', False),
207 ]
208 for format in formats:
209 yield self.save, format
210
211 def save(self, args):
212 fname = args[0]
213 self.st.save(fname, args[1], True)
214 # do some verification args[2] == True
215 if args[-1]:
216 s = scantable(fname)
217 ds = self.st - s
218 assert_equals(self.st.getpolnos(), s.getpolnos())
219 assert_equals(self.st.getscannos(), s.getscannos())
220 assert_equals(self.st.getifnos(), s.getifnos())
221 assert_equals(self.st.getbeamnos(), s.getbeamnos())
222 # see if the residual spectra are ~ 0.0
223 for spec in ds:
224 assert_almost_equals(sum(spec)/len(spec), 0.0, 5)
225
226 def test_auto_poly_baseline(self):
227 q = self.st.auto_quotient()
228 b = q.auto_poly_baseline(insitu=False)
229 res_rms = (q-b).stats('rms')
230 assert_almost_equals(res_rms[0], 0.38370, 5)
231 assert_almost_equals(res_rms[1], 0.38780, 5)
232
233
234 def test_poly_baseline(self):
235 q = self.st.auto_quotient()
236 msk = q.create_mask([0.0, 1471.0], [1745.0, 4095.0])
237 b = q.poly_baseline(order=0, mask=msk,insitu=False)
238 res_rms = (q-b).stats('rms')
239 assert_almost_equals(res_rms[0], 0.38346, 5)
240 assert_almost_equals(res_rms[1], 0.38780, 5)
241
242
243 def test_reshape(self):
244 cp = self.st.copy()
245 n = cp.nchan()
246 rs = cp.reshape(10,-10,False)
247 assert_equals(n-20, rs.nchan())
248 assert_equals(cp.nchan(), n)
249 rs = cp.reshape(10,n-11,False)
250 assert_equals(n-20, rs.nchan())
251 cp.reshape(10,-10,True)
252 assert_equals(n-20, cp.nchan())
Note: See TracBrowser for help on using the repository browser.