Coverage for tests / test_assemble_cell_coadd.py: 24%
135 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-16 00:38 -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 galsim
30import numpy as np
31from assemble_coadd_test_utils import MockCoaddTestData, makeMockSkyInfo
33import lsst.afw.image as afwImage
34import lsst.pipe.base as pipeBase
35import lsst.utils.tests
36from lsst.cell_coadds import CommonComponents
37from lsst.drp.tasks.assemble_cell_coadd import (
38 AssembleCellCoaddConfig,
39 AssembleCellCoaddTask,
40 WarpInputs,
41)
43if TYPE_CHECKING:
44 from lsst.cell_coadds import ObservationIdentifiers
46__all__ = (
47 "MockAssembleCellCoaddConfig",
48 "MockAssembleCellCoaddTask",
49)
52class MockAssembleCellCoaddConfig(AssembleCellCoaddConfig):
53 def setDefaults(self):
54 super().setDefaults()
55 self.require_artifact_mask = False
58class MockAssembleCellCoaddTask(AssembleCellCoaddTask):
59 """Lightly modified version of `AssembleCellCoaddTask` for unit tests.
61 The modifications bypass the usual middleware for loading data and setting
62 up the Task, and instead supply in-memory mock data references to the `run`
63 method so that the coaddition algorithms can be tested without a Butler.
64 """
66 ConfigClass = MockAssembleCellCoaddConfig
68 def runQuantum(
69 self,
70 mockSkyInfo,
71 warpRefList,
72 maskedFractionRefList,
73 noise0RefList,
74 visitSummaryList=None,
75 ):
76 """Modified interface for testing coaddition algorithms without a
77 Butler.
79 Parameters
80 ----------
81 mockSkyInfo : `lsst.pipe.base.Struct`
82 A simple container that supplies a bounding box and WCS in the
83 same format as the output of
84 `lsst.pipe.tasks.CoaddBaseTask.getSkyInfo`
85 warpRefList : `list` of `lsst.pipe.tasks.MockExposureReference`
86 Data references to the test exposures that will be coadded,
87 using the Gen 3 API.
89 Returns
90 -------
91 retStruct : `lsst.pipe.base.Struct`
92 The coadded exposure and associated metadata.
93 """
95 self.common = CommonComponents(
96 units=None,
97 wcs=mockSkyInfo.wcs,
98 band="i",
99 identifiers=pipeBase.Struct(skymap=None, tract=0, patch=42, band="i"),
100 )
102 inputs = {}
103 for warpInput, maskedFractionInput, noise0Input in zip(
104 warpRefList,
105 maskedFractionRefList,
106 noise0RefList,
107 ):
108 inputs[warpInput.dataId] = WarpInputs(
109 warp=warpInput,
110 masked_fraction=maskedFractionInput,
111 noise_warps=[noise0Input],
112 )
114 retStruct = self.run(
115 inputs=inputs,
116 skyInfo=mockSkyInfo,
117 visitSummaryList=visitSummaryList,
118 )
120 return retStruct
123class AssembleCellCoaddTestCase(lsst.utils.tests.TestCase):
124 """Tests of AssembleCellCoaddTask.
126 These tests bypass the middleware used for accessing data and managing Task
127 execution.
128 """
130 @classmethod
131 def setUpClass(cls) -> None:
132 np.random.seed(42)
133 rng = np.random.Generator(np.random.MT19937(42))
134 patch = 42
135 tract = 0
136 testData = MockCoaddTestData(fluxRange=1e4)
137 exposures = {}
138 matchedExposures = {}
139 masked_fraction_images = {}
140 noise0_masked_images = {}
141 for expId in range(100, 110):
142 exposures[expId], matchedExposures[expId] = testData.makeTestImage(expId)
143 masked_fraction_images[expId] = afwImage.ImageF(bbox=exposures[expId].getBBox())
144 masked_fraction_images[expId].array[:, :] = rng.random(masked_fraction_images[expId].array.shape)
145 noise0_masked_images[expId] = afwImage.MaskedImageF(bbox=exposures[expId].getBBox())
146 noise0_masked_images[expId].image.array[:, :] = rng.normal(
147 0, 1, noise0_masked_images[expId].image.array.shape
148 ) * (exposures[expId].variance.array**0.5)
150 cls.handleList = testData.makeDataRefList(exposures, patch=patch, tract=tract)
151 cls.maskedFractionRefList = testData.makeDataRefList(masked_fraction_images, patch=patch, tract=tract)
152 cls.noise0RefList = testData.makeDataRefList(noise0_masked_images, patch=patch, tract=tract)
153 cls.visitSummaryList = [
154 testData.makeVisitSummaryTableHandle(warpHandle) for warpHandle in cls.handleList
155 ]
156 cls.skyInfo = makeMockSkyInfo(testData.bbox, testData.wcs, patch=patch)
158 def runTask(
159 self,
160 config=None,
161 warpRefList=None,
162 maskedFractionRefList=None,
163 noise0RefList=None,
164 visitSummaryList=None,
165 ) -> None:
166 if config is None:
167 config = MockAssembleCellCoaddConfig()
168 assembleTask = MockAssembleCellCoaddTask(config=config)
169 if warpRefList is None:
170 warpRefList = self.handleList
171 if maskedFractionRefList is None:
172 maskedFractionRefList = self.maskedFractionRefList
173 if noise0RefList is None:
174 noise0RefList = self.noise0RefList
175 if visitSummaryList is None:
176 visitSummaryList = self.visitSummaryList
178 self.result = assembleTask.runQuantum(
179 self.skyInfo,
180 warpRefList=warpRefList,
181 maskedFractionRefList=maskedFractionRefList,
182 noise0RefList=noise0RefList,
183 visitSummaryList=visitSummaryList,
184 )
186 def checkSortOrder(self, inputs: Iterable[ObservationIdentifiers]) -> None:
187 """Check that the inputs are sorted.
189 The inputs must be sorted first by visit, and within the same visit,
190 by detector.
192 Parameters
193 ----------
194 inputs : `Iterable` [`ObservationIdentifiers`]
195 The inputs to be checked.
196 """
197 visit, detector = -np.inf, -np.inf # Previous visit, detector IDs.
198 for _, obsId in enumerate(inputs):
199 with self.subTest(input_number=repr(obsId)):
200 self.assertGreaterEqual(obsId.visit, visit)
201 if visit == obsId.visit:
202 with self.subTest(detector_number=repr(obsId.detector)):
203 self.assertGreaterEqual(obsId.detector, detector)
205 visit, detector = obsId.visit, obsId.detector
207 def checkRun(self, assembleTask):
208 """Check that the task runs successfully."""
209 result = assembleTask.runQuantum(self.skyInfo, self.handleList)
211 # Check that we produced an exposure.
212 self.assertTrue(result.multipleCellCoadd is not None)
213 # Check that the visit_count method returns a number less than or equal
214 # to the total number of input exposures available.
215 max_visit_count = len(self.handleList)
216 for cellId, singleCellCoadd in result.multipleCellCoadd.cells.items():
217 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
218 self.assertLessEqual(singleCellCoadd.visit_count, max_visit_count)
219 # Check that the aperture correction maps are not None.
220 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
221 self.assertTrue(singleCellCoadd.aperture_correction_map is not None)
222 # Check that the inputs are sorted.
223 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
224 self.checkSortOrder(singleCellCoadd.inputs)
226 def test_assemble_basic(self):
227 """Test that AssembleCellCoaddTask runs successfully without errors.
229 This test does not check the correctness of the coaddition algorithms.
230 This is intended to prevent the code from bit rotting.
231 """
232 self.runTask()
233 # Check that we produced an exposure.
234 self.assertTrue(self.result.multipleCellCoadd is not None)
236 def test_assemble_empty(self):
237 """Test that AssembleCellCoaddTask runs successfully without errors
238 when no input exposures are provided."""
239 self.result = None # so tearDown has something.
240 with self.assertRaises(pipeBase.NoWorkFound, msg="No cells could be populated for the cell coadd."):
241 self.runTask(warpRefList=[], maskedFractionRefList=[], noise0RefList=[], visitSummaryList=[])
243 def test_assemble_without_visitSummary(self):
244 """Test that AssembleCellCoaddTask calculates detector weights and
245 runs successfully without errors when no visit summaries are provided.
246 """
247 self.runTask(visitSummaryList=[])
248 # Check that we produced an exposure.
249 self.assertTrue(self.result.multipleCellCoadd is not None)
251 # TODO: Remove this test in DM-49401
252 @lsst.utils.tests.methodParameters(do_scale_zero_point=[False, True])
253 def test_do_scale_zero_point(self, do_scale_zero_point):
254 config = MockAssembleCellCoaddConfig()
255 with warnings.catch_warnings():
256 warnings.filterwarnings("ignore", category=FutureWarning)
257 config.do_scale_zero_point = do_scale_zero_point
258 self.runTask(config)
259 # Check that we produced an exposure.
260 self.assertTrue(self.result.multipleCellCoadd is not None)
262 @lsst.utils.tests.methodParameters(do_calculate_weight_from_warp=[False, True])
263 def test_do_calculate_weight_from_warp(self, do_calculate_weight_from_warp):
264 config = MockAssembleCellCoaddConfig()
265 config.do_calculate_weight_from_warp = do_calculate_weight_from_warp
266 self.runTask(config)
267 # Check that we produced an exposure.
268 self.assertTrue(self.result.multipleCellCoadd is not None)
270 def test_visit_count(self):
271 """Check that the visit_count method returns a number less than or
272 equal to the total number of input exposures available.
273 """
274 self.runTask()
275 max_visit_count = len(self.handleList)
276 for cellId, singleCellCoadd in self.result.multipleCellCoadd.cells.items():
277 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
278 self.assertLessEqual(singleCellCoadd.visit_count, max_visit_count)
280 def test_inputs_sorted(self):
281 """Check that the inputs are sorted.
283 The ordering is that inputs are sorted first by visit, and within the
284 same visit, they are ordered by detector.
285 """
286 self.runTask()
287 for _, singleCellCoadd in self.result.multipleCellCoadd.cells.items():
288 self.checkSortOrder(singleCellCoadd.inputs)
290 def test_psf_normalization(self):
291 """Check that the sum of PSF images is close to 1."""
292 self.runTask()
293 for cellId, singleCellCoadd in self.result.multipleCellCoadd.cells.items():
294 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
295 self.assertFloatsAlmostEqual(singleCellCoadd.psf_image.array.sum(), 1.0, rtol=None, atol=1e-7)
297 def test_psf_centering(self):
298 """Check that the PSF images are centered."""
299 self.runTask()
300 for cellId, singleCellCoadd in self.result.multipleCellCoadd.cells.items():
301 with self.subTest(x=repr(cellId.x), y=repr(cellId.y)):
302 shape = galsim.hsm.FindAdaptiveMom(galsim.Image(singleCellCoadd.psf_image.array))
303 self.assertFloatsAlmostEqual(
304 shape.moments_centroid.x, round(shape.moments_centroid.x), rtol=None, atol=0.01
305 )
306 self.assertFloatsAlmostEqual(
307 shape.moments_centroid.y, round(shape.moments_centroid.y), rtol=None, atol=0.01
308 )
311class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase):
312 pass
315def setup_module(module):
316 lsst.utils.tests.init()
319if __name__ == "__main__": 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 lsst.utils.tests.init()
321 unittest.main()