Coverage for tests/test_nightAnalysis.py: 98%

1#!/usr/bin/env python 

2# This file is part of atmospec. 

3# 

4# Developed for the LSST Data Management System. 

5# This product includes software developed by the LSST Project 

6# (https://www.lsst.org). 

7# See the COPYRIGHT file at the top-level directory of this distribution 

8# for details of code ownership. 

9# 

10# This program is free software: you can redistribute it and/or modify 

11# it under the terms of the GNU General Public License as published by 

12# the Free Software Foundation, either version 3 of the License, or 

13# (at your option) any later version. 

14# 

15# This program is distributed in the hope that it will be useful, 

16# but WITHOUT ANY WARRANTY; without even the implied warranty of 

17# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

18# GNU General Public License for more details. 

19# 

20# You should have received a copy of the GNU General Public License 

21# along with this program. If not, see <https://www.gnu.org/licenses/>. 

22"""Test cases for atmospec.nightAnalysis.""" 

23 

24import unittest 

25from types import SimpleNamespace 

26from unittest.mock import patch 

27 

28import numpy as np 

29from astropy.table import Table 

30 

31import lsst.utils 

32import lsst.utils.tests 

33 

34import lsst.atmospec.nightAnalysis as nightAnalysis 

35from lsst.atmospec.nightAnalysis import ( 

36 NightStellarSpectra, 

37 getLineValue, 

38 _getRowNum, 

39 LINE_NAMES, 

40) 

41 

42 

43def makeStrTable(): 

44 """Build a str-keyed table for the free-function tests. 

45 

46 Row 0 is deliberately a valid line so that the ``not rowNum`` 

47 branch (rowNum == 0 is falsy) can be exercised. 

48 """ 

49 table = Table() 

50 table["name"] = [LINE_NAMES["O2"], LINE_NAMES["H_alpha"], LINE_NAMES["H_beta"]] 

51 table["value"] = [10.0, 20.0, 30.0] 

52 return table 

53 

54 

55def makeByteTable(lineNames): 

56 """Build a byte-string keyed table like a real extraction file.""" 

57 table = Table() 

58 encoded = np.array([LINE_NAMES[name].encode("utf-8") for name in lineNames]) 

59 table["name"] = encoded 

60 table["value"] = np.arange(len(lineNames), dtype=float) 

61 return table 

62 

63 

64class FakeQuery: 

65 """Iterable stand-in for a butler query result with order_by.""" 

66 

67 def __init__(self, records): 

68 self._records = records 

69 

70 def __iter__(self): 

71 return iter(self._records) 

72 

73 def order_by(self, *args, **kwargs): 

74 return self 

75 

76 

77def makeRecords(): 

78 """Exposure records covering the seq_nums used in the tests.""" 

79 records = [] 

80 for seqNum, physFilter, mjd, za in ( 

81 (1, "filtA~disp1", 60000.1, 20.0), 

82 (2, "filtB~disp2", 60000.2, 40.0), 

83 (3, "filtA~disp1", 60000.3, 25.0), 

84 (4, "filtB~disp2", 60000.4, 30.0), 

85 ): 

86 records.append(SimpleNamespace( 

87 seq_num=seqNum, 

88 physical_filter=physFilter, 

89 zenith_angle=za, 

90 timespan=SimpleNamespace(begin=SimpleNamespace(mjd=mjd)), 

91 )) 

92 return records 

93 

94 

95def makeButlerClass(records, extractionMap): 

96 """Create a fake Butler class. 

97 

98 ``extractionMap`` maps seq_num -> table, ``None`` (returns None) or 

99 the string ``"lookup"`` (raises LookupError inside get). 

100 """ 

101 

102 class FakeButler: 

103 def __init__(self, *args, **kwargs): 

104 self.registry = SimpleNamespace( 

105 queryDimensionRecords=lambda *a, **k: FakeQuery(records)) 

106 

107 def get(self, datasetType, *, seq_num, day_obs): 

108 result = extractionMap.get(seq_num, None) 

109 if result == "lookup": 

110 raise LookupError("no such dataset") 

111 return result 

112 

113 return FakeButler 

114 

115 

116def makeInstance(*, dayObs=20200101, target="HD12345", ignoreSeqNums=[], 

117 records=None, extractionMap=None, dispersedSeqNums=(1, 2, 3), 

118 passInstance=False): 

119 """Construct a NightStellarSpectra with a fully mocked butler.""" 

120 if records is None: 120 ↛ 122line 120 didn't jump to line 122 because the condition on line 120 was always true

121 records = makeRecords() 

122 if extractionMap is None: 

123 extractionMap = {1: makeByteTable(["H_alpha", "H_beta"]), 

124 2: makeByteTable(["H_alpha", "H_gamma"])} 

125 

126 FakeButler = makeButlerClass(records, extractionMap) 

127 

128 def fakeDispersed(dataId, butler): 

129 return dataId["seq_num"] in dispersedSeqNums 

130 

131 butlerInput = FakeButler() if passInstance else object() 

132 

133 with patch.object(nightAnalysis.dafButler, "Butler", FakeButler), \ 

134 patch.object(nightAnalysis, "isDispersedDataId", side_effect=fakeDispersed): 

135 instance = NightStellarSpectra(butlerInput, dayObs, target, 

136 ignoreSeqNums=ignoreSeqNums) 

137 return instance 

138 

139 

140class NightAnalysisFunctionsTestCase(lsst.utils.tests.TestCase): 

141 """Tests for the free functions _getRowNum and getLineValue.""" 

142 

143 def test_getRowNum_found(self): 

144 table = makeStrTable() 

145 self.assertEqual(_getRowNum(table, "O2"), 0) 

146 self.assertEqual(_getRowNum(table, "H_alpha"), 1) 

147 self.assertEqual(_getRowNum(table, "H_beta"), 2) 

148 

149 def test_getRowNum_missingLine(self): 

150 table = makeStrTable() 

151 # Valid line name but not present in the table -> None. 

152 self.assertIsNone(_getRowNum(table, "water")) 

153 

154 def test_getRowNum_unknownName(self): 

155 table = makeStrTable() 

156 with self.assertRaises(RuntimeError): 

157 _getRowNum(table, "not_a_real_line") 

158 

159 def test_getLineValue_found(self): 

160 table = makeStrTable() 

161 # H_alpha is at row 1 (truthy) so the real value is returned. 

162 self.assertEqual(getLineValue(table, "H_alpha", "value"), 20.0) 

163 

164 def test_getLineValue_rowZeroTreatedAsMissing(self): 

165 table = makeStrTable() 

166 # O2 is at row 0 which is falsy, so it is treated as missing. 

167 self.assertTrue(np.isnan(getLineValue(table, "O2", "value"))) 

168 with self.assertRaises(ValueError): 

169 getLineValue(table, "O2", "value", nanForMissingValues=False) 

170 

171 def test_getLineValue_missingNan(self): 

172 table = makeStrTable() 

173 self.assertTrue(np.isnan(getLineValue(table, "water", "value"))) 

174 

175 def test_getLineValue_missingRaises(self): 

176 table = makeStrTable() 

177 with self.assertRaises(ValueError): 

178 getLineValue(table, "water", "value", nanForMissingValues=False) 

179 

180 

181class NightStellarSpectraTestCase(lsst.utils.tests.TestCase): 

182 """Tests for the NightStellarSpectra class.""" 

183 

184 def test_construction_elseBranch(self): 

185 night = makeInstance(ignoreSeqNums=[3]) 

186 # 1, 2 dispersed and kept; 3 dispersed but ignored; 4 not dispersed. 

187 self.assertEqual(night.seqNums, [1, 2]) 

188 self.assertEqual(set(night.data.keys()), {1, 2}) 

189 self.assertEqual(night.dayObs, 20200101) 

190 self.assertEqual(night.targetName, "HD12345") 

191 

192 def test_construction_isinstanceBranch(self): 

193 # Passing an actual FakeButler instance exercises the isinstance 

194 # True branch of __init__. 

195 night = makeInstance(passInstance=True) 

196 self.assertEqual(night.seqNums, [1, 2]) 

197 

198 def test_construction_missingExtraction(self): 

199 # seq_num 2 raises LookupError inside get -> not loaded. 

200 extractionMap = {1: makeByteTable(["H_alpha"]), 2: "lookup"} 

201 night = makeInstance(extractionMap=extractionMap) 

202 self.assertEqual(night.seqNums, [1]) 

203 self.assertEqual(set(night.data.keys()), {1}) 

204 

205 def test_construction_noneExtraction(self): 

206 # get returns None -> falsy -> not loaded. 

207 extractionMap = {1: makeByteTable(["H_alpha"]), 2: None} 

208 night = makeInstance(extractionMap=extractionMap) 

209 self.assertEqual(night.seqNums, [1]) 

210 

211 def test_isDispersed(self): 

212 night = makeInstance() 

213 with patch.object(nightAnalysis, "isDispersedDataId", 

214 side_effect=lambda dataId, butler: dataId["seq_num"] == 5): 

215 self.assertTrue(night.isDispersed(5)) 

216 self.assertFalse(night.isDispersed(6)) 

217 

218 def test_readOneExtractionFile_lookupError(self): 

219 night = makeInstance() 

220 with patch.object(night.butler, "get", side_effect=LookupError): 

221 self.assertIsNone(night._readOneExtractionFile(999)) 

222 

223 def test_removeSeqNums(self): 

224 night = makeInstance(ignoreSeqNums=[3]) 

225 night.removeSeqNums([2, 42]) # 42 not present -> no-op 

226 self.assertEqual(night.seqNums, [1]) 

227 self.assertNotIn(2, night.data) 

228 

229 def test_getFilterDisperserSet(self): 

230 night = makeInstance(ignoreSeqNums=[3]) 

231 self.assertEqual(night.getFilterDisperserSet(), 

232 {"filtA~disp1", "filtB~disp2"}) 

233 

234 def test_getFilterSet(self): 

235 night = makeInstance(ignoreSeqNums=[3]) 

236 self.assertEqual(night.getFilterSet(), {"filtA", "filtB"}) 

237 

238 def test_getDisperserSet(self): 

239 night = makeInstance(ignoreSeqNums=[3]) 

240 self.assertEqual(night.getDisperserSet(), {"disp1", "disp2"}) 

241 

242 def test_getLineValue_instanceMethod(self): 

243 night = makeInstance(ignoreSeqNums=[3]) 

244 # Byte-string tables never match the str line names, so this is nan. 

245 self.assertTrue(np.isnan(night.getLineValue(1, "H_alpha", "value"))) 

246 

247 def test_getLineValues(self): 

248 night = makeInstance(ignoreSeqNums=[3]) 

249 # Add a seq_num that is not in data to exercise the else branch. 

250 night.seqNums = [1, 2, 99] 

251 values = night.getLineValues("H_alpha", "value") 

252 self.assertEqual(len(values), 3) 

253 self.assertTrue(all(np.isnan(v) for v in values)) 

254 

255 def test_getAllTableLines_intermittent(self): 

256 night = makeInstance(ignoreSeqNums=[3]) 

257 lines = night.getAllTableLines(includeIntermittentLines=True) 

258 self.assertEqual(set(lines), {"H_alpha", "H_beta", "H_gamma"}) 

259 

260 def test_getAllTableLines_common(self): 

261 night = makeInstance(ignoreSeqNums=[3]) 

262 lines = night.getAllTableLines(includeIntermittentLines=False) 

263 self.assertEqual(set(lines), {"H_alpha"}) 

264 

265 def test_getObsTimes(self): 

266 night = makeInstance(ignoreSeqNums=[3]) 

267 self.assertEqual(night.getObsTimes(), [60000.1, 60000.2]) 

268 

269 def test_getAirmasses(self): 

270 night = makeInstance(ignoreSeqNums=[3]) 

271 airmasses = night.getAirmasses() 

272 self.assertEqual(len(airmasses), 2) 

273 expected = 1.0 / np.cos(np.deg2rad(20.0)) 

274 self.assertFloatsAlmostEqual(airmasses[0], expected) 

275 

276 def test_printObservationTable(self): 

277 night = makeInstance(ignoreSeqNums=[3]) 

278 # Simply exercise the code path. 

279 night.printObservationTable() 

280 

281 

282class TestMemory(lsst.utils.tests.MemoryTestCase): 

283 pass 

284 

285 

286def setup_module(module): 

287 lsst.utils.tests.init() 

288 

289 

290if __name__ == "__main__": 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true

291 lsst.utils.tests.init() 

292 unittest.main()