1 | import unittest
|
---|
2 | import datetime
|
---|
3 | from asap import scantable, selector, rcParams, mask_not, asaplog
|
---|
4 | asaplog.disable()
|
---|
5 |
|
---|
6 | class ScantableTest(unittest.TestCase):
|
---|
7 | def setUp(self):
|
---|
8 | s = scantable("data/MOPS.rpf", average=True)
|
---|
9 | sel = selector()
|
---|
10 | # make sure this order is always correct - in can be random
|
---|
11 | sel.set_order(["SCANNO", "POLNO"])
|
---|
12 | s.set_selection(sel)
|
---|
13 | self.st = s.copy()
|
---|
14 | restfreqs = [86.243] # 13CO-1/0, SiO the two IF
|
---|
15 | self.st.set_restfreqs(restfreqs,"GHz")
|
---|
16 |
|
---|
17 | def test_init(self):
|
---|
18 | st = scantable("data/MOPS.rpf", average=False)
|
---|
19 | self.assertEqual(st.ncycle(), 32)
|
---|
20 | st = scantable("data/MOPS.rpf", average=True)
|
---|
21 | self.assertEqual(st.ncycle(), 2)
|
---|
22 | st = scantable("data/MOPS.rpf", unit="Jy")
|
---|
23 | self.assertEqual(st.get_fluxunit(), "Jy")
|
---|
24 | st = scantable("data/MOPS.rpf", unit="K")
|
---|
25 | self.assertEqual(st.get_fluxunit(), "K")
|
---|
26 | self.assertRaises(RuntimeError, scantable, "data/MOPS.rpf", unit="junk")
|
---|
27 | st = scantable(["data/MOPS.rpf","data/MOPS.rpf"], average=False)
|
---|
28 | self.assertEqual(st.nscan(), 4)
|
---|
29 |
|
---|
30 | def test_copy(self):
|
---|
31 | st = self.st.copy()
|
---|
32 | self.assertNotEqual(id(st), id(self.st))
|
---|
33 |
|
---|
34 | def test_drop_scan(self):
|
---|
35 | st = self.st.drop_scan([1])
|
---|
36 | self.assertEqual(st.nscan(), 1)
|
---|
37 |
|
---|
38 | def test_get_scan(self):
|
---|
39 | st = self.st.get_scan([1])
|
---|
40 | self.assertEqual(st.nscan(), 1)
|
---|
41 | st = self.st.get_scan("Orion_SiO_R")
|
---|
42 | self.assertEqual(st.get_sourcename()[-1], "Orion_SiO_R")
|
---|
43 | self.assertEqual(st.nscan(), 1)
|
---|
44 |
|
---|
45 | def test_get_spectrum(self):
|
---|
46 | spec = self.st.get_spectrum(0)
|
---|
47 | self.assertAlmostEqual(max(spec), 215.279830933)
|
---|
48 |
|
---|
49 | def test_get_mask(self):
|
---|
50 | spec = self.st.get_mask(0)
|
---|
51 | self.assertEqual(len(spec), 4096)
|
---|
52 |
|
---|
53 | def test_set_spectrum(self):
|
---|
54 | spec = [ 1.0 for i in range(self.st.nchan()) ]
|
---|
55 | self.st.set_spectrum(spec, 0)
|
---|
56 | spec1 = self.st.get_spectrum(0)
|
---|
57 | self.assertAlmostEqual(max(spec1), 1.0)
|
---|
58 |
|
---|
59 | def test_selection(self):
|
---|
60 | sel = selector()
|
---|
61 | sel.set_polarisations("YY")
|
---|
62 | self.st.set_selection(sel)
|
---|
63 | self.assertEqual(self.st.getpolnos(), (1,))
|
---|
64 | sel1 = self.st.get_selection()
|
---|
65 | self.assertEqual(sel1.get_pols(), [1])
|
---|
66 | self.st.set_selection(pols="XX")
|
---|
67 | self.assertEqual(self.st.getpolnos(), (0,))
|
---|
68 |
|
---|
69 | def test_stats(self):
|
---|
70 | stats = { 'min': 113.767166138,
|
---|
71 | 'max':215.279830933, 'sumsq':128759200.0,
|
---|
72 | 'sum':720262.375, 'mean':175.845306396,
|
---|
73 | 'var':513.95324707, 'stddev':22.6705360413,
|
---|
74 | 'avdev':16.3966751099, 'rms':177.300170898,
|
---|
75 | 'median':182.891845703}
|
---|
76 | for k,v in stats.iteritems():
|
---|
77 | sval = self.st.stats(stat=k)
|
---|
78 | self.assertAlmostEqual(sval[0], v)
|
---|
79 | msk = self.st.create_mask([0,100], [3900,4096])
|
---|
80 | self.assertAlmostEqual(self.st.stats("sum", msk)[0], 35216.87890625)
|
---|
81 |
|
---|
82 | def test_get_column_names(self):
|
---|
83 | cnames = ['SCANNO', 'CYCLENO', 'BEAMNO', 'IFNO',
|
---|
84 | 'POLNO', 'FREQ_ID', 'MOLECULE_ID', 'REFBEAMNO', 'FLAGROW',
|
---|
85 | 'TIME', 'INTERVAL', 'SRCNAME', 'SRCTYPE',
|
---|
86 | 'FIELDNAME', 'SPECTRA', 'FLAGTRA', 'TSYS',
|
---|
87 | 'DIRECTION', 'AZIMUTH', 'ELEVATION',
|
---|
88 | 'OPACITY', 'TCAL_ID', 'FIT_ID',
|
---|
89 | 'FOCUS_ID', 'WEATHER_ID', 'SRCVELOCITY',
|
---|
90 | 'SRCPROPERMOTION', 'SRCDIRECTION',
|
---|
91 | 'SCANRATE']
|
---|
92 | self.assertEqual(self.st.get_column_names(), cnames)
|
---|
93 |
|
---|
94 | def test_get_tsys(self):
|
---|
95 | self.assertAlmostEqual(self.st.get_tsys()[0], 175.830429077)
|
---|
96 |
|
---|
97 | def test_get_time(self):
|
---|
98 | self.assertEqual(self.st.get_time(0), '2008/03/12/09:32:50')
|
---|
99 | dt = datetime.datetime(2008,3,12,9,32,50)
|
---|
100 | self.assertEqual(self.st.get_time(0, True), dt)
|
---|
101 |
|
---|
102 | def test_get_inttime(self):
|
---|
103 | self.assertAlmostEqual(self.st.get_inttime()[0], 30.720016479)
|
---|
104 |
|
---|
105 | def test_get_sourcename(self):
|
---|
106 | self.assertEqual(self.st.get_sourcename(0), 'Orion_SiO_R')
|
---|
107 | self.assertEqual(self.st.get_sourcename(),
|
---|
108 | ['Orion_SiO_R', 'Orion_SiO_R',
|
---|
109 | 'Orion_SiO', 'Orion_SiO'])
|
---|
110 |
|
---|
111 | def test_get_azimuth(self):
|
---|
112 | self.assertAlmostEqual(self.st.get_azimuth()[0], 5.628767013)
|
---|
113 |
|
---|
114 | def test_get_elevation(self):
|
---|
115 | self.assertAlmostEqual(self.st.get_elevation()[0], 1.01711678504)
|
---|
116 |
|
---|
117 | def test_get_parangle(self):
|
---|
118 | self.assertAlmostEqual(self.st.get_parangle()[0], 2.5921990871)
|
---|
119 |
|
---|
120 | def test_get_direction(self):
|
---|
121 | self.assertEqual(self.st.get_direction()[0], '05:35:14.5 -04.52.29.5')
|
---|
122 |
|
---|
123 | def test_get_directionval(self):
|
---|
124 | dv = self.st.get_directionval()[0]
|
---|
125 | self.assertAlmostEqual(dv[0], 1.4627692699)
|
---|
126 | self.assertAlmostEqual(dv[1], -0.0850824415)
|
---|
127 |
|
---|
128 | def test_unit(self):
|
---|
129 | self.st.set_unit('')
|
---|
130 | self.st.set_unit('GHz')
|
---|
131 | self.st.set_unit('km/s')
|
---|
132 | self.assertRaises(RuntimeError, self.st.set_unit, 'junk')
|
---|
133 | self.assertEquals(self.st.get_unit(), 'km/s')
|
---|
134 |
|
---|
135 | def test_average_pol(self):
|
---|
136 | ap = self.st.average_pol()
|
---|
137 | self.assertEqual(ap.npol(), 1)
|
---|
138 |
|
---|
139 | def test_drop_scan(self):
|
---|
140 | s0 = self.st.drop_scan(1)
|
---|
141 | self.assertEqual(s0.getscannos(), (0,))
|
---|
142 | s1 = self.st.drop_scan([0])
|
---|
143 | self.assertEqual(s1.getscannos(), (1,))
|
---|
144 |
|
---|
145 | def test_flag(self):
|
---|
146 | q = self.st.auto_quotient()
|
---|
147 | q.set_unit('km/s')
|
---|
148 | q0 = q.copy()
|
---|
149 | q1 = q.copy()
|
---|
150 | msk = q0.create_mask([-10,20])
|
---|
151 | q0.flag(mask=mask_not(msk))
|
---|
152 | q1.flag(mask=msk)
|
---|
153 | self.assertAlmostEqual(q0.stats(stat='max')[0], 95.62171936)
|
---|
154 | self.assertAlmostEqual(q1.stats(stat='max')[0], 2.66563416)
|
---|
155 |
|
---|
156 | if __name__ == '__main__':
|
---|
157 | unittest.main()
|
---|