Coverage for tests / test_quickLook.py: 18%
153 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-16 08:22 +0000
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/>.
22import contextlib
23import tempfile
24import unittest
26import lsst.afw.image as afwImage
27import lsst.daf.butler.tests as butlerTests
28import lsst.ip.isr as ipIsr
29import lsst.ip.isr.isrMockLSST as isrMock
30import lsst.pex.exceptions
31import lsst.pipe.base as pipeBase
32import lsst.pipe.base.testUtils
33import lsst.utils.tests
34from lsst.summit.utils.quickLook import QuickLookIsrTask, QuickLookIsrTaskConfig
37class QuickLookIsrTaskTestCase(unittest.TestCase):
38 """Tests of the run method with fake data."""
40 def setUp(self):
41 self.mockConfig = isrMock.IsrMockLSSTConfig()
42 self.camera = isrMock.IsrMockLSST(config=self.mockConfig).getCamera()
44 self.ccdExposure = isrMock.RawMockLSST(config=self.mockConfig).run()
45 self.detector = self.ccdExposure.getDetector()
46 amps = self.detector.getAmplifiers()
47 ampNames = [amp.getName() for amp in amps]
49 # # Mock other optional parameters
50 self.bias = isrMock.BiasMockLSST(config=self.mockConfig).run()
51 self.dark = isrMock.DarkMockLSST(config=self.mockConfig).run()
52 self.flat = isrMock.FlatMockLSST(config=self.mockConfig).run()
53 self.defects = isrMock.DefectMockLSST(config=self.mockConfig).run()
54 self.ptc = ipIsr.PhotonTransferCurveDataset(ampNames=ampNames) # Mock PTC dataset
55 for amp, gain in self.mockConfig.gainDict.items():
56 self.ptc.gain[amp] = 1.0
57 self.bfKernel = isrMock.BfKernelMockLSST(config=self.mockConfig).run()
58 self.task = QuickLookIsrTask(config=QuickLookIsrTaskConfig())
60 def test_runQuickLook(self):
61 # Execute the run method with the mock data
62 result = self.task.run(
63 self.ccdExposure,
64 camera=self.camera,
65 bias=self.bias,
66 dark=self.dark,
67 flat=self.flat,
68 defects=self.defects,
69 linearizer=None,
70 crosstalk=None,
71 bfKernel=self.bfKernel,
72 ptc=self.ptc,
73 )
74 self.assertIsNotNone(result, "Result of run method should not be None")
75 self.assertIsInstance(result, pipeBase.Struct, "Result should be of type lsst.pipe.base.Struct")
76 self.assertIsInstance(
77 result.exposure,
78 afwImage.Exposure,
79 "Resulting exposure should be an instance of lsst.afw.image.Exposure",
80 )
82 def test_runQuickLookMissingData(self):
83 # Test without any inputs other than the exposure. And the PTC.
84 result = self.task.run(self.ccdExposure, ptc=self.ptc)
85 self.assertIsInstance(result.exposure, afwImage.Exposure)
87 def test_runQuickLookBadDark(self):
88 # Test with an incorrect dark frame
89 bbox = self.ccdExposure.getBBox()
90 bbox.grow(-20)
91 with self.assertRaises(lsst.pex.exceptions.wrappers.LengthError):
92 self.task.run(
93 self.ccdExposure,
94 camera=self.camera,
95 bias=self.bias,
96 dark=self.dark[bbox],
97 flat=self.flat,
98 defects=self.defects,
99 )
102class QuickLookIsrTaskRunQuantumTests(lsst.utils.tests.TestCase):
103 """Tests of ``QuickLookIsrTask.runQuantum``, which need a test butler,
104 but do not need real images.
106 Adapted from the unit tests of ``CalibrateImageTask.runQuantum``
107 """
109 def setUp(self):
110 # These need to be real, not empty:
111 self.mockConfig = isrMock.IsrMockLSSTConfig()
112 self.camera = isrMock.IsrMockLSST(config=self.mockConfig).getCamera()
114 self.ccdExposure = isrMock.RawMockLSST(config=self.mockConfig).run()
115 self.bias = isrMock.BiasMockLSST(config=self.mockConfig).run()
116 self.dark = isrMock.DarkMockLSST(config=self.mockConfig).run()
117 self.flat = isrMock.FlatMockLSST(config=self.mockConfig).run()
118 self.defects = isrMock.DefectMockLSST(config=self.mockConfig).run()
120 amps = self.ccdExposure.getDetector().getAmplifiers()
121 ampNames = [amp.getName() for amp in amps]
122 self.ptc = ipIsr.PhotonTransferCurveDataset(ampNames=ampNames) # Mock PTC dataset
123 for amp, gain in self.mockConfig.gainDict.items():
124 self.ptc.gain[amp] = 1.0
125 self.crosstalk = lsst.ip.isr.crosstalk.CrosstalkCalib(nAmp=len(ampNames))
126 self.crosstalk.hasCrosstalk = True
127 self.cti = isrMock.DeferredChargeMockLSST(config=self.mockConfig).run()
128 self.mockConfig.doDeferredCharge = False # TODO: DM-54880
129 self.bfKernel = isrMock.BfKernelMockLSST(config=self.mockConfig).run()
131 # dataId values:
132 instrument = self.camera.getName()
133 exposureId = 100
134 visit = 100101
135 detector = 0
136 physical_filter = "testCam_filter"
137 band = "X"
139 # Map the isrTask connection names to the names of the Butler dataset
140 # inputs
141 ccdExposure = "raw"
142 camera = "camera"
143 bias = "bias"
144 dark = "dark"
145 flat = "flat"
146 defects = "defects"
147 bfKernel = "bfk"
148 ptc = "ptc"
149 deferredChargeCalib = "cti"
150 crosstalk = "crosstalk"
151 linearizer = "linearizer"
152 gainCorrection = "gain_correction"
154 # outputs
155 outputExposure = "postISRCCD"
156 outputStatistics = "isrStatistics"
158 # quickLook-only outputs
159 exposure = "quickLookExp"
161 # Create a and populate a test butler for runQuantum tests.
162 self.repo_path = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
163 self.repo = butlerTests.makeTestRepo(self.repo_path.name)
165 # dataIds for fake data
166 butlerTests.addDataIdValue(self.repo, "instrument", instrument)
167 butlerTests.addDataIdValue(self.repo, "physical_filter", physical_filter, band=band)
168 butlerTests.addDataIdValue(self.repo, "detector", detector)
169 butlerTests.addDataIdValue(self.repo, "exposure", exposureId, physical_filter=physical_filter)
170 butlerTests.addDataIdValue(self.repo, "visit", visit)
172 # inputs
173 butlerTests.addDatasetType(self.repo, ccdExposure, {"instrument", "exposure", "detector"}, "Exposure")
174 butlerTests.addDatasetType(self.repo, camera, {"instrument"}, "Camera")
175 butlerTests.addDatasetType(self.repo, bias, {"instrument", "detector"}, "Exposure")
176 butlerTests.addDatasetType(self.repo, dark, {"instrument", "detector"}, "Exposure")
177 butlerTests.addDatasetType(self.repo, flat, {"instrument", "physical_filter", "detector"}, "Exposure")
178 butlerTests.addDatasetType(self.repo, defects, {"instrument", "detector"}, "Defects")
179 butlerTests.addDatasetType(self.repo, linearizer, {"instrument", "detector"}, "Linearizer")
180 butlerTests.addDatasetType(self.repo, crosstalk, {"instrument", "detector"}, "CrosstalkCalib")
181 butlerTests.addDatasetType(self.repo, bfKernel, {"instrument", "detector"}, "BrighterFatterKernel")
182 butlerTests.addDatasetType(self.repo, ptc, {"instrument", "detector"}, "PhotonTransferCurveDataset")
183 butlerTests.addDatasetType(self.repo, deferredChargeCalib, {"instrument", "detector"}, "IsrCalib")
184 butlerTests.addDatasetType(self.repo, gainCorrection, {"instrument", "detector"}, "IsrCalib")
186 # outputs
187 butlerTests.addDatasetType(
188 self.repo, outputExposure, {"instrument", "exposure", "detector"}, "Exposure"
189 )
190 butlerTests.addDatasetType(self.repo, exposure, {"instrument", "exposure", "detector"}, "Exposure")
191 butlerTests.addDatasetType(
192 self.repo, outputStatistics, {"instrument", "exposure", "detector"}, "StructuredDataDict"
193 )
195 # dataIds
196 self.exposure_id = self.repo.registry.expandDataId(
197 {
198 "instrument": instrument,
199 "exposure": exposureId,
200 "detector": detector,
201 "physical_filter": physical_filter,
202 }
203 )
204 self.instrument_id = self.repo.registry.expandDataId({"instrument": instrument})
205 self.flat_id = self.repo.registry.expandDataId(
206 {"instrument": instrument, "physical_filter": physical_filter, "detector": detector}
207 )
208 self.detector_id = self.repo.registry.expandDataId({"instrument": instrument, "detector": detector})
209 self.filter_id = self.repo.registry.expandDataId(
210 {"instrument": instrument, "physical_filter": physical_filter}
211 )
213 # put empty data
214 self.butler = butlerTests.makeTestCollection(self.repo)
215 self.butler.put(self.ccdExposure, ccdExposure, self.exposure_id)
216 self.butler.put(self.camera, camera, self.instrument_id)
217 self.butler.put(self.bias, bias, self.detector_id)
218 self.butler.put(self.dark, dark, self.detector_id)
219 self.butler.put(self.flat, flat, self.flat_id)
220 self.butler.put(self.defects, defects, self.detector_id)
221 self.butler.put(self.bfKernel, bfKernel, self.detector_id)
222 self.butler.put(self.ptc, ptc, self.detector_id)
223 self.butler.put(self.cti, deferredChargeCalib, self.detector_id)
224 self.butler.put(self.crosstalk, crosstalk, self.detector_id)
225 self.butler.put(lsst.ip.isr.linearize.Linearizer(), linearizer, self.detector_id)
226 self.butler.put(
227 lsst.ip.isr.GainCorrection(
228 ampNames=ampNames,
229 gainAdjustments=[1.0 for x in ampNames],
230 ),
231 gainCorrection,
232 self.detector_id,
233 )
235 def tearDown(self):
236 del self.repo_path # this removes the temporary directory
238 def test_runQuantum(self):
239 config = ipIsr.IsrTaskConfig()
240 # Remove some outputs
241 config.doBinnedExposures = False
242 config.doSaveInterpPixels = False
243 config.qa.doThumbnailOss = False
244 config.qa.doThumbnailFlattened = False
245 config.doCalculateStatistics = False
247 # Turn on all optional inputs, except CTI, as that isn't
248 # defined for LATISS.
249 config.doDeferredCharge = False
251 config.usePtcReadNoise = True
252 config.doCrosstalk = True
253 config.doBrighterFatter = True
255 # Override a method in IsrTask that is executed early, to instead raise
256 # a custom exception called ExitMock that we can catch and ignore.
257 isrTask = ipIsr.IsrTask
258 isrTask.ensureExposure = raiseExitMockError
259 task = QuickLookIsrTask(isrTask=isrTask)
260 lsst.pipe.base.testUtils.assertValidInitOutput(task)
262 # Use the names of the connections here, not the Butler dataset name
263 quantum = lsst.pipe.base.testUtils.makeQuantum(
264 task,
265 self.butler,
266 self.exposure_id,
267 {
268 "ccdExposure": self.exposure_id,
269 "camera": self.instrument_id,
270 "bias": self.detector_id,
271 "dark": self.detector_id,
272 "flat": self.flat_id,
273 "defects": self.detector_id,
274 "bfKernel": self.detector_id,
275 "ptc": self.detector_id,
276 "deferredChargeCalib": self.detector_id,
277 "crosstalk": self.detector_id,
278 "linearizer": self.detector_id,
279 "fringes": self.flat_id,
280 "gainCorrection": self.detector_id,
281 # outputs
282 "outputExposure": self.exposure_id,
283 "outputStatistics": self.exposure_id,
284 "exposure": self.exposure_id,
285 },
286 )
287 # Check that the proper kwargs are passed to run().
288 with contextlib.suppress(ExitMockError):
289 lsst.pipe.base.testUtils.runTestQuantum(task, self.butler, quantum, mockRun=False)
292def raiseExitMockError(*args):
293 """Raise a custom exception."""
294 raise ExitMockError
297class ExitMockError(Exception):
298 """A custom exception to catch during a unit test."""
300 pass
303class TestMemory(lsst.utils.tests.MemoryTestCase):
304 pass
307def setup_module(module):
308 lsst.utils.tests.init()
311if __name__ == "__main__": 311 ↛ 312line 311 didn't jump to line 312 because the condition on line 311 was never true
312 lsst.utils.tests.init()
313 unittest.main()