Coverage for tests / test_pipelines.py: 34%
95 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-11 02:07 -0800
1#!/usr/bin/env python
3#
4# LSST Data Management System
5#
6# This product includes software developed by the
7# LSST Project (http://www.lsst.org/).
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 LSST License Statement and
20# the GNU General Public License along with this program. If not,
21# see <https://www.lsstcorp.org/LegalNotices/>.
22#
23"""Test cases for cp_pipe pipelines."""
25import glob
26import os
27import unittest
29from lsst.pipe.base import Pipeline, PipelineGraph
30import lsst.utils
32try:
33 import lsst.obs.lsst
34 has_obs_lsst = True
35except ImportError:
36 has_obs_lsst = False
38try:
39 import lsst.obs.subaru
40 has_obs_subaru = True
41except ImportError:
42 has_obs_subaru = False
44try:
45 import lsst.obs.decam
46 has_obs_decam = True
47except ImportError:
48 has_obs_decam = False
51class CalibrationPipelinesTestCase(lsst.utils.tests.TestCase):
52 """Test case for building the pipelines."""
54 def setUp(self):
55 self.pipeline_path = os.path.join(lsst.utils.getPackageDir("cp_pipe"), "pipelines")
57 def _get_pipelines(self, exclude=[]):
58 pipelines = {
59 "cpBfk.yaml",
60 "cpBias.yaml",
61 "cpCrosstalk.yaml",
62 "cpCti.yaml",
63 "cpDarkForDefects.yaml",
64 "cpDark.yaml",
65 "cpDefectsIndividual.yaml",
66 "cpDefects.yaml",
67 "cpFilterScan.yaml",
68 "cpFlatSingleChip.yaml",
69 "cpFlat.yaml",
70 "cpFringe.yaml",
71 "cpLinearizer.yaml",
72 "cpMonochromatorScan.yaml",
73 "cpPlotPtc.yaml",
74 "cpPtc.yaml",
75 "cpSky.yaml",
76 "cpBiasBootstrap.yaml",
77 "cpDarkBootstrap.yaml",
78 "cpFlatBootstrap.yaml",
79 "cpSpectroFlat.yaml",
80 # TODO DM-52883: Remove cpPtcFixupGainRatios and cpPtcRename.
81 "cpPtcFixupGainRatios.yaml",
82 "cpPtcRename.yaml",
83 "cpIlluminationCorrection.yaml",
84 "cpFlatAnaglyph.yaml",
85 "cpFlatGradientReference.yaml",
86 "cpQuadNotch.yaml",
87 "cpGainCorrection.yaml",
88 }
90 for ex in exclude:
91 pipelines.remove(ex)
93 return pipelines
95 def _check_pipeline(self, pipeline_file, overrides={}):
96 # Confirm that the file is there.
97 self.assertTrue(os.path.isfile(pipeline_file), msg=f"Could not find {pipeline_file}")
99 # The following loads the pipeline and confirms that it can parse all
100 # the configs.
101 try:
102 pipeline = Pipeline.fromFile(pipeline_file)
104 if overrides:
105 for label, value in overrides.items():
106 pipeline.addConfigOverride(label, value[0], value[1])
108 graph = pipeline.to_graph()
109 except Exception as e:
110 raise RuntimeError(f"Could not process {pipeline_file} {e}") from e
112 self.assertIsInstance(graph, PipelineGraph)
114 def test_ingredients(self):
115 """Check that all pipelines in pipelines/_ingredients are tested."""
116 glob_str = os.path.join(self.pipeline_path, "_ingredients", "*.yaml")
117 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
118 # and are not to be tested on their own.
119 ingredients = set(
120 [os.path.basename(pipeline) for pipeline in glob.glob(glob_str) if "LSST.yaml" not in pipeline]
121 )
122 # The *Bootstrap* pipelines are used by LATISS/LSSTComCam/LSSTCam
123 # but are renamed on import.
124 expected = set([pipeline for pipeline in self._get_pipelines() if "Bootstrap" not in pipeline])
125 # These pipelines have only an "LSST" version.
126 expected.discard("cpIlluminationCorrection.yaml")
127 expected.discard("cpFlatAnaglyph.yaml")
128 expected.discard("cpFlatGradientReference.yaml")
129 expected.discard("cpGainCorrection.yaml")
130 self.assertEqual(ingredients, expected)
132 def test_cameras(self):
133 """Check that all the cameras in pipelines are tested."""
134 glob_str = os.path.join(self.pipeline_path, "*")
135 paths = set(
136 [os.path.basename(path) for path in glob.glob(glob_str)]
137 )
138 expected = {
139 "DECam",
140 "HSC",
141 "_ingredients",
142 "LATISS",
143 "LSSTCam",
144 "LSSTCam-imSim",
145 "LSSTComCam",
146 "LSSTComCamSim",
147 "LSST-TS8",
148 "README.md",
149 }
150 self.assertEqual(paths, expected)
152 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
153 def test_latiss_pipelines(self):
154 for pipeline in self._get_pipelines(exclude=[
155 # The following two tasks are not part of the new pipelines.
156 "cpDarkForDefects.yaml",
157 "cpDefectsIndividual.yaml",
158 # The following tasks are not defined for LATISS.
159 "cpMonochromatorScan.yaml",
160 "cpIlluminationCorrection.yaml",
161 "cpFlatAnaglyph.yaml",
162 "cpFlatGradientReference.yaml",
163 "cpGainCorrection.yaml",
164 # The following tasks will be added in the future.
165 "cpCrosstalk.yaml",
166 "cpFringe.yaml",
167 # TODO: DM-46426
168 "cpCti.yaml",
169 ]):
170 self._check_pipeline(os.path.join(self.pipeline_path, "LATISS", pipeline))
172 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
173 def test_lsstcam_pipelines(self):
174 for pipeline in self._get_pipelines(exclude=[
175 "cpFilterScan.yaml",
176 "cpMonochromatorScan.yaml",
177 "cpSpectroFlat.yaml",
178 "cpDarkForDefects.yaml",
179 "cpDefectsIndividual.yaml",
180 "cpIlluminationCorrection.yaml",
181 # Unsupported pipelines.
182 "cpCrosstalk.yaml",
183 "cpFringe.yaml",
184 "cpQuadNotch.yaml",
185 ]):
186 if pipeline == "cpFlatAnaglyph.yaml":
187 overrides = {
188 "cpFlatBlueNormalize": ("downSelectionValue", "test1"),
189 "cpFlatRedNormalize": ("downSelectionValue", "test2"),
190 }
191 else:
192 overrides = {}
193 self._check_pipeline(
194 os.path.join(self.pipeline_path, "LSSTCam", pipeline),
195 overrides=overrides,
196 )
198 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
199 def test_lsstcam_imsim_pipelines(self):
200 for pipeline in self._get_pipelines(exclude=[
201 "cpDarkForDefects.yaml",
202 "cpFilterScan.yaml",
203 "cpMonochromatorScan.yaml",
204 "cpSpectroFlat.yaml",
205 "cpBiasBootstrap.yaml",
206 "cpDarkBootstrap.yaml",
207 "cpFlatBootstrap.yaml",
208 "cpPtcFixupGainRatios.yaml",
209 "cpPtcRename.yaml",
210 "cpIlluminationCorrection.yaml",
211 "cpFlatAnaglyph.yaml",
212 "cpFlatGradientReference.yaml",
213 "cpQuadNotch.yaml",
214 "cpGainCorrection.yaml",
215 ]):
216 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam-imSim", pipeline))
218 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
219 def test_lsstcomcam_pipelines(self):
220 for pipeline in self._get_pipelines(exclude=[
221 # The following tasks are not part of the new pipelines.
222 "cpDarkForDefects.yaml",
223 "cpDefectsIndividual.yaml",
224 # The following tasks are not for ComCam.
225 "cpFilterScan.yaml",
226 "cpMonochromatorScan.yaml",
227 "cpSpectroFlat.yaml",
228 "cpCrosstalk.yaml",
229 "cpFringe.yaml",
230 "cpFlatAnaglyph.yaml",
231 "cpFlatGradientReference.yaml",
232 "cpQuadNotch.yaml",
233 "cpGainCorrection.yaml",
234 # TODO: DM-46426
235 "cpCti.yaml",
236 ]):
237 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCam", pipeline))
239 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
240 def test_lsstcomcamsim_pipelines(self):
241 for pipeline in self._get_pipelines(exclude=[
242 # The following tasks are not part of the new pipelines.
243 "cpDarkForDefects.yaml",
244 "cpDefectsIndividual.yaml",
245 # The following tasks are not for ComCamSim.
246 "cpFilterScan.yaml",
247 "cpMonochromatorScan.yaml",
248 "cpSpectroFlat.yaml",
249 "cpFringe.yaml",
250 "cpLinearizer.yaml",
251 "cpCrosstalk.yaml",
252 "cpCti.yaml",
253 "cpPtcFixupGainRatios.yaml",
254 "cpPtcRename.yaml",
255 "cpIlluminationCorrection.yaml",
256 "cpFlatAnaglyph.yaml",
257 "cpFlatGradientReference.yaml",
258 "cpQuadNotch.yaml",
259 "cpGainCorrection.yaml",
260 ]):
261 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCamSim", pipeline))
263 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSST-TS8 pipelines without obs_lsst")
264 def test_lsst_ts8_pipelines(self):
265 for pipeline in self._get_pipelines(exclude=[
266 "cpFilterScan.yaml",
267 "cpMonochromatorScan.yaml",
268 "cpSpectroFlat.yaml",
269 "cpBiasBootstrap.yaml",
270 "cpDarkBootstrap.yaml",
271 "cpFlatBootstrap.yaml",
272 "cpPtcFixupGainRatios.yaml",
273 "cpPtcRename.yaml",
274 "cpIlluminationCorrection.yaml",
275 "cpFlatAnaglyph.yaml",
276 "cpFlatGradientReference.yaml",
277 "cpQuadNotch.yaml",
278 "cpGainCorrection.yaml",
279 ]):
280 self._check_pipeline(os.path.join(self.pipeline_path, "LSST-TS8", pipeline))
282 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
283 def test_decam_pipelines(self):
284 for pipeline in self._get_pipelines(exclude=[
285 "cpDarkForDefects.yaml",
286 "cpFilterScan.yaml",
287 "cpMonochromatorScan.yaml",
288 "cpSpectroFlat.yaml",
289 "cpBiasBootstrap.yaml",
290 "cpDarkBootstrap.yaml",
291 "cpFlatBootstrap.yaml",
292 "cpPtcFixupGainRatios.yaml",
293 "cpPtcRename.yaml",
294 "cpIlluminationCorrection.yaml",
295 "cpFlatAnaglyph.yaml",
296 "cpFlatGradientReference.yaml",
297 "cpQuadNotch.yaml",
298 "cpGainCorrection.yaml",
299 ]):
300 self._check_pipeline(os.path.join(self.pipeline_path, "DECam", pipeline))
302 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
303 def test_hsc_pipelines(self):
304 for pipeline in self._get_pipelines(exclude=[
305 "cpDarkForDefects.yaml",
306 "cpFilterScan.yaml",
307 "cpMonochromatorScan.yaml",
308 "cpSpectroFlat.yaml",
309 "cpBiasBootstrap.yaml",
310 "cpDarkBootstrap.yaml",
311 "cpFlatBootstrap.yaml",
312 "cpPtcFixupGainRatios.yaml",
313 "cpPtcRename.yaml",
314 "cpIlluminationCorrection.yaml",
315 "cpFlatAnaglyph.yaml",
316 "cpFlatGradientReference.yaml",
317 "cpQuadNotch.yaml",
318 "cpGainCorrection.yaml",
319 ]):
320 self._check_pipeline(os.path.join(self.pipeline_path, "HSC", pipeline))
323class TestMemory(lsst.utils.tests.MemoryTestCase):
324 pass
327def setup_module(module):
328 lsst.utils.tests.init()
331if __name__ == "__main__": 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true
332 lsst.utils.tests.init()
333 unittest.main()