[1840] | 1 | import sys |
---|
| 2 | import os |
---|
| 3 | import shutil |
---|
[1600] | 4 | import datetime |
---|
[1903] | 5 | from nose.tools import * |
---|
[2797] | 6 | import asap |
---|
[1840] | 7 | from asap import scantable, selector, mask_not |
---|
| 8 | from asap.logging import asaplog |
---|
[1903] | 9 | # no need for log messages |
---|
[1824] | 10 | asaplog.disable() |
---|
[1544] | 11 | |
---|
[1840] | 12 | def tempdir_setup(): |
---|
| 13 | os.makedirs("test_temp") |
---|
| 14 | |
---|
| 15 | def tempdir_teardown(): |
---|
| 16 | shutil.rmtree("test_temp", True) |
---|
| 17 | |
---|
| 18 | class TestScantable(object): |
---|
| 19 | def setup(self): |
---|
[2797] | 20 | pth = os.path.dirname(__file__) |
---|
| 21 | s = scantable(os.path.join(pth, "data", "MOPS.rpf"), average=True) |
---|
[1824] | 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() |
---|
[1725] | 27 | restfreqs = [86.243] # 13CO-1/0, SiO the two IF |
---|
| 28 | self.st.set_restfreqs(restfreqs,"GHz") |
---|
[1544] | 29 | |
---|
| 30 | def test_init(self): |
---|
[2797] | 31 | fname = os.path.join(os.path.dirname(__file__), "data", "MOPS.rpf") |
---|
| 32 | st = scantable(fname, average=False) |
---|
[1840] | 33 | assert_equal(st.ncycle(), 32) |
---|
[2797] | 34 | st = scantable(fname, average=True) |
---|
[1840] | 35 | assert_equal(st.ncycle(), 2) |
---|
[2797] | 36 | st = scantable(fname, unit="Jy") |
---|
[1840] | 37 | assert_equal(st.get_fluxunit(), "Jy") |
---|
[2797] | 38 | st = scantable(fname, unit="K") |
---|
[1840] | 39 | assert_equal(st.get_fluxunit(), "K") |
---|
[2797] | 40 | assert_raises(RuntimeError, scantable, fname, unit="junk") |
---|
| 41 | st = scantable([fname,fname], average=False) |
---|
[1840] | 42 | assert_equal(st.nscan(), 4) |
---|
[1544] | 43 | |
---|
[2801] | 44 | def test_jenkins_email(self): |
---|
| 45 | raise RuntimeError("Just check that the email has the log attached") |
---|
| 46 | |
---|
[1544] | 47 | def test_copy(self): |
---|
| 48 | st = self.st.copy() |
---|
[1840] | 49 | assert_not_equal(id(st), id(self.st)) |
---|
[1592] | 50 | |
---|
[1544] | 51 | def test_drop_scan(self): |
---|
| 52 | st = self.st.drop_scan([1]) |
---|
[1840] | 53 | assert_equal(st.nscan(), 1) |
---|
[1544] | 54 | |
---|
| 55 | def test_get_scan(self): |
---|
| 56 | st = self.st.get_scan([1]) |
---|
[1840] | 57 | assert_equal(st.nscan(), 1) |
---|
[1544] | 58 | st = self.st.get_scan("Orion_SiO_R") |
---|
[1840] | 59 | assert_equal(st.get_sourcename()[-1], "Orion_SiO_R") |
---|
| 60 | assert_equal(st.nscan(), 1) |
---|
[1544] | 61 | |
---|
| 62 | def test_get_spectrum(self): |
---|
| 63 | spec = self.st.get_spectrum(0) |
---|
[1840] | 64 | assert_almost_equal(max(spec), 215.279830933) |
---|
[1544] | 65 | |
---|
| 66 | def test_get_mask(self): |
---|
| 67 | spec = self.st.get_mask(0) |
---|
[1840] | 68 | assert_equal(len(spec), 4096) |
---|
[1592] | 69 | |
---|
[1544] | 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) |
---|
[1840] | 74 | assert_almost_equal(max(spec1), 1.0) |
---|
[1544] | 75 | |
---|
| 76 | def test_selection(self): |
---|
| 77 | sel = selector() |
---|
| 78 | sel.set_polarisations("YY") |
---|
| 79 | self.st.set_selection(sel) |
---|
[1903] | 80 | assert_equal(self.st.getpolnos(), [1]) |
---|
[1544] | 81 | sel1 = self.st.get_selection() |
---|
[1840] | 82 | assert_equal(sel1.get_pols(), [1]) |
---|
[1600] | 83 | self.st.set_selection(pols="XX") |
---|
[1903] | 84 | assert_equal(self.st.getpolnos(), [0]) |
---|
[1544] | 85 | |
---|
[1842] | 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 | |
---|
[1544] | 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, |
---|
[1592] | 109 | 'avdev':16.3966751099, 'rms':177.300170898, |
---|
| 110 | 'median':182.891845703} |
---|
[1842] | 111 | for k,v in stats.items(): |
---|
| 112 | yield self.stats, k, v |
---|
[1544] | 113 | |
---|
| 114 | def test_get_column_names(self): |
---|
[1592] | 115 | cnames = ['SCANNO', 'CYCLENO', 'BEAMNO', 'IFNO', |
---|
[1824] | 116 | 'POLNO', 'FREQ_ID', 'MOLECULE_ID', 'REFBEAMNO', 'FLAGROW', |
---|
[1592] | 117 | 'TIME', 'INTERVAL', 'SRCNAME', 'SRCTYPE', |
---|
[1544] | 118 | 'FIELDNAME', 'SPECTRA', 'FLAGTRA', 'TSYS', |
---|
[1592] | 119 | 'DIRECTION', 'AZIMUTH', 'ELEVATION', |
---|
| 120 | 'OPACITY', 'TCAL_ID', 'FIT_ID', |
---|
[1544] | 121 | 'FOCUS_ID', 'WEATHER_ID', 'SRCVELOCITY', |
---|
| 122 | 'SRCPROPERMOTION', 'SRCDIRECTION', |
---|
| 123 | 'SCANRATE'] |
---|
[1840] | 124 | assert_equal(self.st.get_column_names(), cnames) |
---|
[1544] | 125 | |
---|
[1600] | 126 | def test_get_tsys(self): |
---|
[1840] | 127 | assert_almost_equal(self.st.get_tsys()[0], 175.830429077) |
---|
[1600] | 128 | |
---|
[2791] | 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 | |
---|
[1600] | 140 | def test_get_time(self): |
---|
[1840] | 141 | assert_equal(self.st.get_time(0), '2008/03/12/09:32:50') |
---|
[1600] | 142 | dt = datetime.datetime(2008,3,12,9,32,50) |
---|
[1840] | 143 | assert_equal(self.st.get_time(0, True), dt) |
---|
[1600] | 144 | |
---|
| 145 | def test_get_inttime(self): |
---|
[1840] | 146 | assert_almost_equal(self.st.get_inttime()[0], 30.720016479) |
---|
[1600] | 147 | |
---|
| 148 | def test_get_sourcename(self): |
---|
[1840] | 149 | assert_equal(self.st.get_sourcename(0), 'Orion_SiO_R') |
---|
| 150 | assert_equal(self.st.get_sourcename(), |
---|
[1824] | 151 | ['Orion_SiO_R', 'Orion_SiO_R', |
---|
| 152 | 'Orion_SiO', 'Orion_SiO']) |
---|
[1600] | 153 | |
---|
| 154 | def test_get_azimuth(self): |
---|
[1840] | 155 | assert_almost_equal(self.st.get_azimuth()[0], 5.628767013) |
---|
[1600] | 156 | |
---|
| 157 | def test_get_elevation(self): |
---|
[1840] | 158 | assert_almost_equal(self.st.get_elevation()[0], 1.01711678504) |
---|
[1600] | 159 | |
---|
| 160 | def test_get_parangle(self): |
---|
[1840] | 161 | assert_almost_equal(self.st.get_parangle()[0], 2.5921990871) |
---|
[1600] | 162 | |
---|
| 163 | def test_get_direction(self): |
---|
[2634] | 164 | assert_equal(self.st.get_direction()[0], 'J2000 05:35:14.5 -04.52.29.5') |
---|
[1600] | 165 | |
---|
| 166 | def test_get_directionval(self): |
---|
| 167 | dv = self.st.get_directionval()[0] |
---|
[1840] | 168 | assert_almost_equal(dv[0], 1.4627692699) |
---|
| 169 | assert_almost_equal(dv[1], -0.0850824415) |
---|
[1600] | 170 | |
---|
| 171 | def test_unit(self): |
---|
| 172 | self.st.set_unit('') |
---|
| 173 | self.st.set_unit('GHz') |
---|
| 174 | self.st.set_unit('km/s') |
---|
[1840] | 175 | assert_raises(RuntimeError, self.st.set_unit, 'junk') |
---|
| 176 | assert_equals(self.st.get_unit(), 'km/s') |
---|
[1600] | 177 | |
---|
[1592] | 178 | def test_average_pol(self): |
---|
| 179 | ap = self.st.average_pol() |
---|
[1840] | 180 | assert_equal(ap.npol(), 1) |
---|
[1544] | 181 | |
---|
[1600] | 182 | def test_drop_scan(self): |
---|
| 183 | s0 = self.st.drop_scan(1) |
---|
[1903] | 184 | assert_equal(s0.getscannos(), [0]) |
---|
[1600] | 185 | s1 = self.st.drop_scan([0]) |
---|
[1903] | 186 | assert_equal(s1.getscannos(), [1]) |
---|
[1592] | 187 | |
---|
[1725] | 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)) |
---|
[1824] | 195 | q1.flag(mask=msk) |
---|
[1840] | 196 | assert_almost_equal(q0.stats(stat='max')[0], 95.62171936) |
---|
| 197 | assert_almost_equal(q1.stats(stat='max')[0], 2.66563416) |
---|
[1600] | 198 | |
---|
[1840] | 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) |
---|
[1844] | 214 | # do some verification args[2] == True |
---|
[1840] | 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()) |
---|
[1844] | 222 | # see if the residual spectra are ~ 0.0 |
---|
[1840] | 223 | for spec in ds: |
---|
| 224 | assert_almost_equals(sum(spec)/len(spec), 0.0, 5) |
---|
[1925] | 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) |
---|
[2787] | 240 | assert_almost_equals(res_rms[1], 0.38780, 5) |
---|
[2672] | 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()) |
---|