Coverage for tests / test_quickLook.py: 17%

1# This file is part of summit_utils. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

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

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

7# for details of code ownership. 

8# 

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

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

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

12# (at your option) any later version. 

13# 

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

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

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

17# GNU General Public License for more details. 

18# 

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

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

21 

22import contextlib 

23import tempfile 

24import unittest 

25 

26import numpy as np 

27 

28import lsst.afw.cameraGeom.testUtils as afwTestUtils 

29import lsst.afw.image as afwImage 

30import lsst.daf.butler.tests as butlerTests 

31import lsst.ip.isr as ipIsr 

32import lsst.ip.isr.isrMock as isrMock 

33import lsst.pex.exceptions 

34import lsst.pipe.base as pipeBase 

35import lsst.pipe.base.testUtils 

36import lsst.utils.tests 

37from lsst.afw.image import TransmissionCurve 

38from lsst.summit.utils.quickLook import QuickLookIsrTask, QuickLookIsrTaskConfig 

39 

40 

41class QuickLookIsrTaskTestCase(unittest.TestCase): 

42 """Tests of the run method with fake data.""" 

43 

44 def setUp(self): 

45 self.mockConfig = isrMock.IsrMockConfig() 

46 self.dataContainer = isrMock.MockDataContainer(config=self.mockConfig) 

47 self.camera = isrMock.IsrMock(config=self.mockConfig).getCamera() 

48 

49 self.ccdExposure = isrMock.RawMock(config=self.mockConfig).run() 

50 self.detector = self.ccdExposure.getDetector() 

51 amps = self.detector.getAmplifiers() 

52 ampNames = [amp.getName() for amp in amps] 

53 

54 # # Mock other optional parameters 

55 self.bias = self.dataContainer.get("bias") 

56 self.dark = self.dataContainer.get("dark") 

57 self.flat = self.dataContainer.get("flat") 

58 self.defects = self.dataContainer.get("defects") 

59 self.ptc = ipIsr.PhotonTransferCurveDataset(ampNames=ampNames) # Mock PTC dataset 

60 self.bfKernel = self.dataContainer.get("bfKernel") 

61 self.newBFKernel = pipeBase.Struct(gain={}) 

62 for amp_i, amp in enumerate(ampNames): 

63 self.newBFKernel.gain[amp] = 0.9 + 0.1 * amp_i 

64 self.task = QuickLookIsrTask(config=QuickLookIsrTaskConfig()) 

65 

66 def test_runQuickLook(self): 

67 # Execute the run method with the mock data 

68 result = self.task.run( 

69 self.ccdExposure, 

70 camera=self.camera, 

71 bias=self.bias, 

72 dark=self.dark, 

73 flat=self.flat, 

74 defects=self.defects, 

75 linearizer=None, 

76 crosstalk=None, 

77 bfKernel=self.bfKernel, 

78 newBFKernel=self.newBFKernel, 

79 ptc=self.ptc, 

80 crosstalkSources=None, 

81 ) 

82 self.assertIsNotNone(result, "Result of run method should not be None") 

83 self.assertIsInstance(result, pipeBase.Struct, "Result should be of type lsst.pipe.base.Struct") 

84 self.assertIsInstance( 

85 result.exposure, 

86 afwImage.Exposure, 

87 "Resulting exposure should be an instance of lsst.afw.image.Exposure", 

88 ) 

89 

90 def test_runQuickLookMissingData(self): 

91 # Test without any inputs other than the exposure 

92 result = self.task.run(self.ccdExposure) 

93 self.assertIsInstance(result.exposure, afwImage.Exposure) 

94 

95 def test_runQuickLookBadDark(self): 

96 # Test with an incorrect dark frame 

97 bbox = self.ccdExposure.getBBox() 

98 bbox.grow(-20) 

99 with self.assertRaises(lsst.pex.exceptions.wrappers.LengthError): 

100 self.task.run( 

101 self.ccdExposure, 

102 camera=self.camera, 

103 bias=self.bias, 

104 dark=self.dark[bbox], 

105 flat=self.flat, 

106 defects=self.defects, 

107 ) 

108 

109 

110class QuickLookIsrTaskRunQuantumTests(lsst.utils.tests.TestCase): 

111 """Tests of ``QuickLookIsrTask.runQuantum``, which need a test butler, 

112 but do not need real images. 

113 

114 Adapted from the unit tests of ``CalibrateImageTask.runQuantum`` 

115 """ 

116 

117 def setUp(self): 

118 instrument = "testCam" 

119 exposureId = 100 

120 visit = 100101 

121 detector = 0 

122 physical_filter = "testCam_filter" 

123 band = "X" 

124 

125 # Map the isrTask connection names to the names of the Butler dataset 

126 # inputs 

127 ccdExposure = "raw" 

128 camera = "camera" 

129 bias = "bias" 

130 dark = "dark" 

131 flat = "flat" 

132 defects = "defects" 

133 bfKernel = "bfKernel" 

134 newBFKernel = "brighterFatterKernel" 

135 ptc = "ptc" 

136 filterTransmission = "transmission_filter" 

137 deferredChargeCalib = "cpCtiCalib" 

138 opticsTransmission = "transmission_optics" 

139 strayLightData = "yBackground" 

140 atmosphereTransmission = "transmission_atmosphere" 

141 crosstalk = "crosstalk" 

142 illumMaskedImage = "illum" 

143 linearizer = "linearizer" 

144 fringes = "fringe" 

145 sensorTransmission = "transmission_sensor" 

146 crosstalkSources = "isrOverscanCorrected" 

147 

148 # outputs 

149 outputExposure = "postISRCCD" 

150 

151 # quickLook-only outputs 

152 exposure = "quickLookExp" 

153 

154 # Create a and populate a test butler for runQuantum tests. 

155 self.repo_path = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) 

156 self.repo = butlerTests.makeTestRepo(self.repo_path.name) 

157 

158 # dataIds for fake data 

159 butlerTests.addDataIdValue(self.repo, "instrument", instrument) 

160 butlerTests.addDataIdValue(self.repo, "physical_filter", physical_filter, band=band) 

161 butlerTests.addDataIdValue(self.repo, "detector", detector) 

162 butlerTests.addDataIdValue(self.repo, "exposure", exposureId, physical_filter=physical_filter) 

163 butlerTests.addDataIdValue(self.repo, "visit", visit) 

164 

165 # inputs 

166 butlerTests.addDatasetType(self.repo, ccdExposure, {"instrument", "exposure", "detector"}, "Exposure") 

167 butlerTests.addDatasetType(self.repo, camera, {"instrument"}, "Camera") 

168 butlerTests.addDatasetType(self.repo, bias, {"instrument", "detector"}, "Exposure") 

169 butlerTests.addDatasetType(self.repo, dark, {"instrument", "detector"}, "Exposure") 

170 butlerTests.addDatasetType(self.repo, flat, {"instrument", "physical_filter", "detector"}, "Exposure") 

171 butlerTests.addDatasetType(self.repo, defects, {"instrument", "detector"}, "Defects") 

172 butlerTests.addDatasetType(self.repo, linearizer, {"instrument", "detector"}, "Linearizer") 

173 butlerTests.addDatasetType(self.repo, crosstalk, {"instrument", "detector"}, "CrosstalkCalib") 

174 butlerTests.addDatasetType(self.repo, bfKernel, {"instrument"}, "NumpyArray") 

175 butlerTests.addDatasetType(self.repo, newBFKernel, {"instrument", "detector"}, "BrighterFatterKernel") 

176 butlerTests.addDatasetType(self.repo, ptc, {"instrument", "detector"}, "PhotonTransferCurveDataset") 

177 butlerTests.addDatasetType( 

178 self.repo, filterTransmission, {"instrument", "physical_filter"}, "TransmissionCurve" 

179 ) 

180 butlerTests.addDatasetType(self.repo, opticsTransmission, {"instrument"}, "TransmissionCurve") 

181 butlerTests.addDatasetType(self.repo, deferredChargeCalib, {"instrument", "detector"}, "IsrCalib") 

182 butlerTests.addDatasetType( 

183 self.repo, strayLightData, {"instrument", "physical_filter", "detector"}, "Exposure" 

184 ) 

185 butlerTests.addDatasetType(self.repo, atmosphereTransmission, {"instrument"}, "TransmissionCurve") 

186 butlerTests.addDatasetType( 

187 self.repo, illumMaskedImage, {"instrument", "physical_filter", "detector"}, "MaskedImage" 

188 ) 

189 butlerTests.addDatasetType( 

190 self.repo, fringes, {"instrument", "physical_filter", "detector"}, "Exposure" 

191 ) 

192 butlerTests.addDatasetType( 

193 self.repo, sensorTransmission, {"instrument", "detector"}, "TransmissionCurve" 

194 ) 

195 butlerTests.addDatasetType( 

196 self.repo, crosstalkSources, {"instrument", "exposure", "detector"}, "Exposure" 

197 ) 

198 

199 # outputs 

200 butlerTests.addDatasetType( 

201 self.repo, outputExposure, {"instrument", "exposure", "detector"}, "Exposure" 

202 ) 

203 butlerTests.addDatasetType(self.repo, exposure, {"instrument", "exposure", "detector"}, "Exposure") 

204 

205 # dataIds 

206 self.exposure_id = self.repo.registry.expandDataId( 

207 { 

208 "instrument": instrument, 

209 "exposure": exposureId, 

210 "detector": detector, 

211 "physical_filter": physical_filter, 

212 } 

213 ) 

214 self.instrument_id = self.repo.registry.expandDataId({"instrument": instrument}) 

215 self.flat_id = self.repo.registry.expandDataId( 

216 {"instrument": instrument, "physical_filter": physical_filter, "detector": detector} 

217 ) 

218 self.detector_id = self.repo.registry.expandDataId({"instrument": instrument, "detector": detector}) 

219 self.filter_id = self.repo.registry.expandDataId( 

220 {"instrument": instrument, "physical_filter": physical_filter} 

221 ) 

222 

223 # put empty data 

224 transmissionCurve = TransmissionCurve.makeSpatiallyConstant( 

225 np.ones(2), np.linspace(0, 1, 2), 0.0, 0.0 

226 ) 

227 self.butler = butlerTests.makeTestCollection(self.repo) 

228 self.butler.put(afwImage.ExposureF(), ccdExposure, self.exposure_id) 

229 self.butler.put(afwTestUtils.CameraWrapper().camera, camera, self.instrument_id) 

230 self.butler.put(afwImage.ExposureF(), bias, self.detector_id) 

231 self.butler.put(afwImage.ExposureF(), dark, self.detector_id) 

232 self.butler.put(afwImage.ExposureF(), flat, self.flat_id) 

233 self.butler.put(lsst.ip.isr.Defects(), defects, self.detector_id) 

234 self.butler.put(np.zeros(2), bfKernel, self.instrument_id) 

235 self.butler.put( 

236 lsst.ip.isr.brighterFatterKernel.BrighterFatterKernel(), newBFKernel, self.detector_id 

237 ) 

238 self.butler.put(ipIsr.PhotonTransferCurveDataset(), ptc, self.detector_id) 

239 self.butler.put(transmissionCurve, filterTransmission, self.filter_id) 

240 self.butler.put(lsst.ip.isr.calibType.IsrCalib(), deferredChargeCalib, self.detector_id) 

241 self.butler.put(transmissionCurve, opticsTransmission, self.instrument_id) 

242 self.butler.put(afwImage.ExposureF(), strayLightData, self.flat_id) 

243 self.butler.put(transmissionCurve, atmosphereTransmission, self.instrument_id) 

244 self.butler.put(lsst.ip.isr.crosstalk.CrosstalkCalib(), crosstalk, self.detector_id) 

245 self.butler.put(afwImage.ExposureF().maskedImage, illumMaskedImage, self.flat_id) 

246 self.butler.put(lsst.ip.isr.linearize.Linearizer(), linearizer, self.detector_id) 

247 self.butler.put(afwImage.ExposureF(), fringes, self.flat_id) 

248 self.butler.put(transmissionCurve, sensorTransmission, self.detector_id) 

249 self.butler.put(afwImage.ExposureF(), crosstalkSources, self.exposure_id) 

250 

251 def tearDown(self): 

252 del self.repo_path # this removes the temporary directory 

253 

254 def test_runQuantum(self): 

255 config = ipIsr.IsrTaskConfig() 

256 # Remove some outputs 

257 config.doBinnedExposures = False 

258 config.doSaveInterpPixels = False 

259 config.qa.doThumbnailOss = False 

260 config.qa.doThumbnailFlattened = False 

261 config.doCalculateStatistics = False 

262 

263 # Turn on all optional inputs 

264 config.doAttachTransmissionCurve = True 

265 config.doIlluminationCorrection = True 

266 config.doStrayLight = True 

267 config.doDeferredCharge = True 

268 config.usePtcReadNoise = True 

269 config.doCrosstalk = True 

270 config.doBrighterFatter = True 

271 

272 # Override a method in IsrTask that is executed early, to instead raise 

273 # a custom exception called ExitMock that we can catch and ignore. 

274 isrTask = ipIsr.IsrTask 

275 isrTask.ensureExposure = raiseExitMockError 

276 task = QuickLookIsrTask(isrTask=isrTask) 

277 lsst.pipe.base.testUtils.assertValidInitOutput(task) 

278 

279 # Use the names of the connections here, not the Butler dataset name 

280 quantum = lsst.pipe.base.testUtils.makeQuantum( 

281 task, 

282 self.butler, 

283 self.exposure_id, 

284 { 

285 "ccdExposure": self.exposure_id, 

286 "camera": self.instrument_id, 

287 "bias": self.detector_id, 

288 "dark": self.detector_id, 

289 "flat": self.flat_id, 

290 "defects": self.detector_id, 

291 "bfKernel": self.instrument_id, 

292 "newBFKernel": self.detector_id, 

293 "ptc": self.detector_id, 

294 "filterTransmission": self.filter_id, 

295 "deferredChargeCalib": self.detector_id, 

296 "opticsTransmission": self.instrument_id, 

297 "strayLightData": self.flat_id, 

298 "atmosphereTransmission": self.instrument_id, 

299 "crosstalk": self.detector_id, 

300 "illumMaskedImage": self.flat_id, 

301 "linearizer": self.detector_id, 

302 "fringes": self.flat_id, 

303 "sensorTransmission": self.detector_id, 

304 "crosstalkSources": [self.exposure_id, self.exposure_id], 

305 # outputs 

306 "outputExposure": self.exposure_id, 

307 "exposure": self.exposure_id, 

308 }, 

309 ) 

310 # Check that the proper kwargs are passed to run(). 

311 with contextlib.suppress(ExitMockError): 

312 lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum, mockRun=False) 

313 

314 

315def raiseExitMockError(*args): 

316 """Raise a custom exception.""" 

317 raise ExitMockError 

318 

319 

320class ExitMockError(Exception): 

321 """A custom exception to catch during a unit test.""" 

322 

323 pass