Coverage for tests / test_assemble_cell_coadd.py: 25%
125 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-13 01:15 -0800
1# This file is part of drp_tasks.
2#
3# LSST Data Management System
4# This product includes software developed by the
5# LSST Project (http://www.lsst.org/).
6# See COPYRIGHT file at the top of the source tree.
7#
8# This program is free software: you can redistribute it and/or modify
9# it under the terms of the GNU General Public License as published by
10# the Free Software Foundation, either version 3 of the License, or
11# (at your option) any later version.
12#
13# This program is distributed in the hope that it will be useful,
14# but WITHOUT ANY WARRANTY; without even the implied warranty of
15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16# GNU General Public License for more details.
17#
18# You should have received a copy of the LSST License Statement and
19# the GNU General Public License along with this program. If not,
20# see <https://www.lsstcorp.org/LegalNotices/>.
21#
23from __future__ import annotations
25import unittest
26import warnings
27from typing import TYPE_CHECKING, Iterable
29import numpy as np
30from assemble_coadd_test_utils import MockCoaddTestData, makeMockSkyInfo
32import lsst.afw.image as afwImage
33import lsst.pipe.base as pipeBase
34import lsst.utils.tests
35from lsst.cell_coadds import CommonComponents
36from lsst.drp.tasks.assemble_cell_coadd import (
37 AssembleCellCoaddConfig,
38 AssembleCellCoaddTask,
39 WarpInputs,
40)
42if TYPE_CHECKING:
43 from lsst.cell_coadds import ObservationIdentifiers
45__all__ = (
46 "MockAssembleCellCoaddConfig",
47 "MockAssembleCellCoaddTask",
48)
51class MockAssembleCellCoaddConfig(AssembleCellCoaddConfig):
52 pass
55class MockAssembleCellCoaddTask(AssembleCellCoaddTask):
56 """Lightly modified version of `AssembleCellCoaddTask` for unit tests.
58 The modifications bypass the usual middleware for loading data and setting
59 up the Task, and instead supply in-memory mock data references to the `run`
60 method so that the coaddition algorithms can be tested without a Butler.
61 """
63 ConfigClass = MockAssembleCellCoaddConfig
65 def runQuantum(
66 self,
67 mockSkyInfo,
68 warpRefList,
69 maskedFractionRefList,
70 noise0RefList,
71 visitSummaryList=None,
72 ):
73 """Modified interface for testing coaddition algorithms without a
74 Butler.
76 Parameters
77 ----------
78 mockSkyInfo : `lsst.pipe.base.Struct`
79 A simple container that supplies a bounding box and WCS in the
80 same format as the output of
81 `lsst.pipe.tasks.CoaddBaseTask.getSkyInfo`
82 warpRefList : `list` of `lsst.pipe.tasks.MockExposureReference`
83 Data references to the test exposures that will be coadded,
84 using the Gen 3 API.
86 Returns
87 -------
88 retStruct : `lsst.pipe.base.Struct`
89 The coadded exposure and associated metadata.
90 """
92 self.common = CommonComponents(
93 units=None,
94 wcs=mockSkyInfo.wcs,
95 band="i",
96 identifiers=pipeBase.Struct(skymap=None, tract=0, patch=42, band="i"),
97 )
99 inputs = {}
100 for warpInput, maskedFractionInput, noise0Input in zip(
101 warpRefList,
102 maskedFractionRefList,
103 noise0RefList,
104 ):
105 inputs[warpInput.dataId] = WarpInputs(
106 warp=warpInput,
107 masked_fraction=maskedFractionInput,
108 noise_warps=[noise0Input],
109 )
111 retStruct = self.run(
112 inputs=inputs,
113 skyInfo=mockSkyInfo,
114 visitSummaryList=visitSummaryList,
115 )
117 return retStruct
120class AssembleCellCoaddTestCase(lsst.utils.tests.TestCase):
121 """Tests of AssembleCellCoaddTask.
123 These tests bypass the middleware used for accessing data and managing Task
124 execution.
125 """
127 @classmethod
128 def setUpClass(cls) -> None:
129 np.random.seed(42)
130 rng = np.random.Generator(np.random.MT19937(42))
131 patch = 42
132 tract = 0
133 testData = MockCoaddTestData(fluxRange=1e4)
134 exposures = {}
135 matchedExposures = {}
136 masked_fraction_images = {}
137 noise0_masked_images = {}
138 for expId in range(100, 110):
139 exposures[expId], matchedExposures[expId] = testData.makeTestImage(expId)
140 masked_fraction_images[expId] = afwImage.ImageF(bbox=exposures[expId].getBBox())
141 masked_fraction_images[expId].array[:, :] = rng.random(masked_fraction_images[expId].array.shape)
142 noise0_masked_images[expId] = afwImage.MaskedImageF(bbox=exposures[expId].getBBox())
143 noise0_masked_images[expId].image.array[:, :] = rng.normal(
144 0, 1, noise0_masked_images[expId].image.array.shape
145 ) * (exposures[expId].variance.array**0.5)
147 cls.handleList = testData.makeDataRefList(exposures, patch=patch, tract=tract)
148 cls.maskedFractionRefList = testData.makeDataRefList(masked_fraction_images, patch=patch, tract=tract)
149 cls.noise0RefList = testData.makeDataRefList(noise0_masked_images, patch=patch, tract=tract)
150 cls.visitSummaryList = [
151 testData.makeVisitSummaryTableHandle(warpHandle) for warpHandle in cls.handleList
152 ]
153 cls.skyInfo = makeMockSkyInfo(testData.bbox, testData.wcs, patch=patch)
155 def runTask(
156 self,
157 config=None,
158 warpRefList=None,
159 maskedFractionRefList=None,
160 noise0RefList=None,
161 visitSummaryList=None,
162 ) -> None:
163 if config is None:
164 config = MockAssembleCellCoaddConfig()
165 assembleTask = MockAssembleCellCoaddTask(config=config)
166 if warpRefList is None:
167 warpRefList = self.handleList
168 if maskedFractionRefList is None:
169 maskedFractionRefList = self.maskedFractionRefList
170 if noise0RefList is None:
171 noise0RefList = self.noise0RefList
172 if visitSummaryList is None:
173 visitSummaryList = self.visitSummaryList
175 self.result = assembleTask.runQuantum(
176 self.skyInfo,
177 warpRefList=warpRefList,
178 maskedFractionRefList=maskedFractionRefList,
179 noise0RefList=noise0RefList,
180 visitSummaryList=visitSummaryList,
181 )
183 def checkSortOrder(self, inputs: Iterable[ObservationIdentifiers]) -> None:
184 """Check that the inputs are sorted.
186 The inputs must be sorted first by visit, and within the same visit,
187 by detector.
189 Parameters
190 ----------
191 inputs : `Iterable` [`ObservationIdentifiers`]
192 The inputs to be checked.
193 """
194 visit, detector = -np.inf, -np.inf # Previous visit, detector IDs.
195 for _, obsId in enumerate(inputs):
196 with self.subTest(input_number=repr(obsId)):
197 self.assertGreaterEqual(obsId.visit, visit)
198 if visit == obsId.visit:
199 with self.subTest(detector_number=repr(obsId.detector)):
200 self.assertGreaterEqual(obsId.detector, detector)
202 visit, detector = obsId.visit, obsId.detector
204 def checkRun(self, assembleTask):
205 """Check that the task runs successfully."""
206 result = assembleTask.runQuantum(self.skyInfo, self.handleList)
208 # Check that we produced an exposure.
209 self.assertTrue(result.multipleCellCoadd is not None)
210 # Check that the visit_count method returns a number less than or equal
211 # to the total number of input exposures available.
212 max_visit_count = len(self.handleList)
213 for cellId, singleCellCoadd in result.multipleCellCoadd.cells.items():
214 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
215 self.assertLessEqual(singleCellCoadd.visit_count, max_visit_count)
216 # Check that the aperture correction maps are not None.
217 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
218 self.assertTrue(singleCellCoadd.aperture_correction_map is not None)
219 # Check that the inputs are sorted.
220 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
221 self.checkSortOrder(singleCellCoadd.inputs)
223 def test_assemble_basic(self):
224 """Test that AssembleCellCoaddTask runs successfully without errors.
226 This test does not check the correctness of the coaddition algorithms.
227 This is intended to prevent the code from bit rotting.
228 """
229 self.runTask()
230 # Check that we produced an exposure.
231 self.assertTrue(self.result.multipleCellCoadd is not None)
233 def test_assemble_empty(self):
234 """Test that AssembleCellCoaddTask runs successfully without errors
235 when no input exposures are provided."""
236 self.result = None # so tearDown has something.
237 with self.assertRaises(pipeBase.NoWorkFound, msg="No cells could be populated for the cell coadd."):
238 self.runTask(warpRefList=[], maskedFractionRefList=[], noise0RefList=[], visitSummaryList=[])
240 def test_assemble_without_visitSummary(self):
241 """Test that AssembleCellCoaddTask calculates detector weights and
242 runs successfully without errors when no visit summaries are provided.
243 """
244 self.runTask(visitSummaryList=[])
245 # Check that we produced an exposure.
246 self.assertTrue(self.result.multipleCellCoadd is not None)
248 # TODO: Remove this test in DM-49401
249 @lsst.utils.tests.methodParameters(do_scale_zero_point=[False, True])
250 def test_do_scale_zero_point(self, do_scale_zero_point):
251 config = MockAssembleCellCoaddConfig()
252 with warnings.catch_warnings():
253 warnings.filterwarnings("ignore", category=FutureWarning)
254 config.do_scale_zero_point = do_scale_zero_point
255 self.runTask(config)
256 # Check that we produced an exposure.
257 self.assertTrue(self.result.multipleCellCoadd is not None)
259 @lsst.utils.tests.methodParameters(do_calculate_weight_from_warp=[False, True])
260 def test_do_calculate_weight_from_warp(self, do_calculate_weight_from_warp):
261 config = MockAssembleCellCoaddConfig()
262 config.do_calculate_weight_from_warp = do_calculate_weight_from_warp
263 self.runTask(config)
264 # Check that we produced an exposure.
265 self.assertTrue(self.result.multipleCellCoadd is not None)
267 def test_visit_count(self):
268 """Check that the visit_count method returns a number less than or
269 equal to the total number of input exposures available.
270 """
271 self.runTask()
272 max_visit_count = len(self.handleList)
273 for cellId, singleCellCoadd in self.result.multipleCellCoadd.cells.items():
274 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
275 self.assertLessEqual(singleCellCoadd.visit_count, max_visit_count)
277 def test_inputs_sorted(self):
278 """Check that the inputs are sorted.
280 The ordering is that inputs are sorted first by visit, and within the
281 same visit, they are ordered by detector.
282 """
283 self.runTask()
284 for _, singleCellCoadd in self.result.multipleCellCoadd.cells.items():
285 self.checkSortOrder(singleCellCoadd.inputs)
287 def test_psf_normalization(self):
288 """Check that the sum of PSF images is close to 1."""
289 self.runTask()
290 for cellId, singleCellCoadd in self.result.multipleCellCoadd.cells.items():
291 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
292 self.assertFloatsAlmostEqual(singleCellCoadd.psf_image.array.sum(), 1.0, rtol=None, atol=1e-7)
295class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase):
296 pass
299def setup_module(module):
300 lsst.utils.tests.init()
303if __name__ == "__main__": 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true
304 lsst.utils.tests.init()
305 unittest.main()