Coverage for tests/test_assemble_cell_coadd.py: 25%
121 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-10-09 04:36 -0700
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.drp.tasks.assemble_cell_coadd import (
36 AssembleCellCoaddConfig,
37 AssembleCellCoaddTask,
38 EmptyCellCoaddError,
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 = pipeBase.Struct(
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 tearDown(self) -> None:
156 del self.result
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=obsId):
200 self.assertGreaterEqual(obsId.visit, visit)
201 if visit == obsId.visit:
202 with self.subTest(detector_number=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=cellId.x, y=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=cellId.x, y=cellId.y):
221 self.assertTrue(singleCellCoadd.aperture_correction_map is not None)
222 # Check that the inputs are sorted.
223 with self.subTest(x=cellId.x, y=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(EmptyCellCoaddError, 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=cellId.x, y=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)
291class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase):
292 pass
295def setup_module(module):
296 lsst.utils.tests.init()
299if __name__ == "__main__": 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 lsst.utils.tests.init()
301 unittest.main()