Coverage for tests/test_pipelines.py: 34%
94 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-07 02:49 -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 "cpPtcFixupGainRatios.yaml",
81 "cpPtcRename.yaml",
82 "cpIlluminationCorrection.yaml",
83 "cpFlatAnaglyph.yaml",
84 "cpFlatGradientReference.yaml",
85 "cpQuadNotch.yaml",
86 }
88 for ex in exclude:
89 pipelines.remove(ex)
91 return pipelines
93 def _check_pipeline(self, pipeline_file, overrides={}):
94 # Confirm that the file is there.
95 self.assertTrue(os.path.isfile(pipeline_file), msg=f"Could not find {pipeline_file}")
97 # The following loads the pipeline and confirms that it can parse all
98 # the configs.
99 try:
100 pipeline = Pipeline.fromFile(pipeline_file)
102 if overrides:
103 for label, value in overrides.items():
104 pipeline.addConfigOverride(label, value[0], value[1])
106 graph = pipeline.to_graph()
107 except Exception as e:
108 raise RuntimeError(f"Could not process {pipeline_file} {e}") from e
110 self.assertIsInstance(graph, PipelineGraph)
112 def test_ingredients(self):
113 """Check that all pipelines in pipelines/_ingredients are tested."""
114 glob_str = os.path.join(self.pipeline_path, "_ingredients", "*.yaml")
115 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
116 # and are not to be tested on their own.
117 ingredients = set(
118 [os.path.basename(pipeline) for pipeline in glob.glob(glob_str) if "LSST.yaml" not in pipeline]
119 )
120 # The *Bootstrap* pipelines are used by LATISS/LSSTComCam/LSSTCam
121 # but are renamed on import.
122 expected = set([pipeline for pipeline in self._get_pipelines() if "Bootstrap" not in pipeline])
123 # These pipelines have only an "LSST" version.
124 expected.discard("cpIlluminationCorrection.yaml")
125 expected.discard("cpFlatAnaglyph.yaml")
126 expected.discard("cpFlatGradientReference.yaml")
127 self.assertEqual(ingredients, expected)
129 def test_cameras(self):
130 """Check that all the cameras in pipelines are tested."""
131 glob_str = os.path.join(self.pipeline_path, "*")
132 paths = set(
133 [os.path.basename(path) for path in glob.glob(glob_str)]
134 )
135 expected = {
136 "DECam",
137 "HSC",
138 "_ingredients",
139 "LATISS",
140 "LSSTCam",
141 "LSSTCam-imSim",
142 "LSSTComCam",
143 "LSSTComCamSim",
144 "LSST-TS8",
145 "README.md",
146 }
147 self.assertEqual(paths, expected)
149 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
150 def test_latiss_pipelines(self):
151 for pipeline in self._get_pipelines(exclude=[
152 # The following two tasks are not part of the new pipelines.
153 "cpDarkForDefects.yaml",
154 "cpDefectsIndividual.yaml",
155 # The following tasks are not defined for LATISS.
156 "cpMonochromatorScan.yaml",
157 "cpIlluminationCorrection.yaml",
158 "cpFlatAnaglyph.yaml",
159 "cpFlatGradientReference.yaml",
160 # The following tasks will be added in the future.
161 "cpCrosstalk.yaml",
162 "cpFringe.yaml",
163 # TODO: DM-46426
164 "cpCti.yaml",
165 ]):
166 self._check_pipeline(os.path.join(self.pipeline_path, "LATISS", pipeline))
168 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
169 def test_lsstcam_pipelines(self):
170 for pipeline in self._get_pipelines(exclude=[
171 "cpFilterScan.yaml",
172 "cpMonochromatorScan.yaml",
173 "cpSpectroFlat.yaml",
174 "cpDarkForDefects.yaml",
175 "cpDefectsIndividual.yaml",
176 "cpIlluminationCorrection.yaml",
177 # Unsupported pipelines.
178 "cpCrosstalk.yaml",
179 "cpFringe.yaml",
180 "cpQuadNotch.yaml",
181 ]):
182 if pipeline == "cpFlatAnaglyph.yaml":
183 overrides = {
184 "cpFlatBlueNormalize": ("downSelectionValue", "test1"),
185 "cpFlatRedNormalize": ("downSelectionValue", "test2"),
186 }
187 else:
188 overrides = {}
189 self._check_pipeline(
190 os.path.join(self.pipeline_path, "LSSTCam", pipeline),
191 overrides=overrides,
192 )
194 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
195 def test_lsstcam_imsim_pipelines(self):
196 for pipeline in self._get_pipelines(exclude=[
197 "cpDarkForDefects.yaml",
198 "cpFilterScan.yaml",
199 "cpMonochromatorScan.yaml",
200 "cpSpectroFlat.yaml",
201 "cpBiasBootstrap.yaml",
202 "cpDarkBootstrap.yaml",
203 "cpFlatBootstrap.yaml",
204 "cpPtcFixupGainRatios.yaml",
205 "cpPtcRename.yaml",
206 "cpIlluminationCorrection.yaml",
207 "cpFlatAnaglyph.yaml",
208 "cpFlatGradientReference.yaml",
209 "cpQuadNotch.yaml",
210 ]):
211 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam-imSim", pipeline))
213 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
214 def test_lsstcomcam_pipelines(self):
215 for pipeline in self._get_pipelines(exclude=[
216 # The following tasks are not part of the new pipelines.
217 "cpDarkForDefects.yaml",
218 "cpDefectsIndividual.yaml",
219 # The following tasks are not for ComCam.
220 "cpFilterScan.yaml",
221 "cpMonochromatorScan.yaml",
222 "cpSpectroFlat.yaml",
223 "cpCrosstalk.yaml",
224 "cpFringe.yaml",
225 "cpFlatAnaglyph.yaml",
226 "cpFlatGradientReference.yaml",
227 "cpQuadNotch.yaml",
228 # TODO: DM-46426
229 "cpCti.yaml",
230 ]):
231 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCam", pipeline))
233 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
234 def test_lsstcomcamsim_pipelines(self):
235 for pipeline in self._get_pipelines(exclude=[
236 # The following tasks are not part of the new pipelines.
237 "cpDarkForDefects.yaml",
238 "cpDefectsIndividual.yaml",
239 # The following tasks are not for ComCamSim.
240 "cpFilterScan.yaml",
241 "cpMonochromatorScan.yaml",
242 "cpSpectroFlat.yaml",
243 "cpFringe.yaml",
244 "cpLinearizer.yaml",
245 "cpCrosstalk.yaml",
246 "cpCti.yaml",
247 "cpPtcFixupGainRatios.yaml",
248 "cpPtcRename.yaml",
249 "cpIlluminationCorrection.yaml",
250 "cpFlatAnaglyph.yaml",
251 "cpFlatGradientReference.yaml",
252 "cpQuadNotch.yaml",
253 ]):
254 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCamSim", pipeline))
256 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSST-TS8 pipelines without obs_lsst")
257 def test_lsst_ts8_pipelines(self):
258 for pipeline in self._get_pipelines(exclude=[
259 "cpFilterScan.yaml",
260 "cpMonochromatorScan.yaml",
261 "cpSpectroFlat.yaml",
262 "cpBiasBootstrap.yaml",
263 "cpDarkBootstrap.yaml",
264 "cpFlatBootstrap.yaml",
265 "cpPtcFixupGainRatios.yaml",
266 "cpPtcRename.yaml",
267 "cpIlluminationCorrection.yaml",
268 "cpFlatAnaglyph.yaml",
269 "cpFlatGradientReference.yaml",
270 "cpQuadNotch.yaml",
271 ]):
272 self._check_pipeline(os.path.join(self.pipeline_path, "LSST-TS8", pipeline))
274 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
275 def test_decam_pipelines(self):
276 for pipeline in self._get_pipelines(exclude=[
277 "cpDarkForDefects.yaml",
278 "cpFilterScan.yaml",
279 "cpMonochromatorScan.yaml",
280 "cpSpectroFlat.yaml",
281 "cpBiasBootstrap.yaml",
282 "cpDarkBootstrap.yaml",
283 "cpFlatBootstrap.yaml",
284 "cpPtcFixupGainRatios.yaml",
285 "cpPtcRename.yaml",
286 "cpIlluminationCorrection.yaml",
287 "cpFlatAnaglyph.yaml",
288 "cpFlatGradientReference.yaml",
289 "cpQuadNotch.yaml",
290 ]):
291 self._check_pipeline(os.path.join(self.pipeline_path, "DECam", pipeline))
293 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
294 def test_hsc_pipelines(self):
295 for pipeline in self._get_pipelines(exclude=[
296 "cpDarkForDefects.yaml",
297 "cpFilterScan.yaml",
298 "cpMonochromatorScan.yaml",
299 "cpSpectroFlat.yaml",
300 "cpBiasBootstrap.yaml",
301 "cpDarkBootstrap.yaml",
302 "cpFlatBootstrap.yaml",
303 "cpPtcFixupGainRatios.yaml",
304 "cpPtcRename.yaml",
305 "cpIlluminationCorrection.yaml",
306 "cpFlatAnaglyph.yaml",
307 "cpFlatGradientReference.yaml",
308 "cpQuadNotch.yaml",
309 ]):
310 self._check_pipeline(os.path.join(self.pipeline_path, "HSC", pipeline))
313class TestMemory(lsst.utils.tests.MemoryTestCase):
314 pass
317def setup_module(module):
318 lsst.utils.tests.init()
321if __name__ == "__main__": 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 lsst.utils.tests.init()
323 unittest.main()