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