Coverage for tests/test_finalizeCharacterization.py: 99%
208 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-23 09:30 +0000
1# This file is part of pipe_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#
22"""Test FinalizeCharacterizationTask.
23"""
24import logging
25import unittest
27import astropy.table.table
28import numpy as np
30import lsst.utils.tests
31import lsst.afw.detection as afwDetection
32import lsst.afw.image as afwImage
33import lsst.afw.table as afwTable
34import lsst.pipe.base as pipeBase
36from lsst.pipe.tasks.finalizeCharacterization import (
37 FinalizeCharacterizationConfig,
38 FinalizeCharacterizationTask,
39 FinalizeCharacterizationDetectorConfig,
40 FinalizeCharacterizationDetectorTask,
41 ConsolidateFinalizeCharacterizationDetectorConfig,
42 ConsolidateFinalizeCharacterizationDetectorTask,
43)
46def _make_dummy_psf_and_ap_corr_map():
47 # Make dummy versions of these products, including required fields.
48 psf = afwDetection.GaussianPsf(15, 15, 2.0)
49 ap_corr_map = afwImage.ApCorrMap()
50 schema = afwTable.SourceTable.makeMinimalSchema()
51 schema.addField("visit", type=np.int64, doc="Visit number for the sources.")
52 schema.addField("detector", type=np.int32, doc="Detector number for the sources.")
53 measured_src = afwTable.SourceCatalog(schema)
54 measured_src.resize(10)
55 measured_src["id"] = np.arange(10)
57 return psf, ap_corr_map, measured_src
60class MockFinalizeCharacterizationTask(FinalizeCharacterizationTask):
61 """A derived class which skips the initialization routines.
62 """
63 def __init__(self, **kwargs):
64 pipeBase.PipelineTask.__init__(self, **kwargs)
66 # This ensures that the schema-loading isn't triggered on run.
67 self.schema = afwTable.SourceTable.makeMinimalSchema()
69 self.makeSubtask('reserve_selection')
70 self.makeSubtask('source_selector')
72 def compute_psf_and_ap_corr_map(
73 self,
74 visit,
75 detector,
76 exposure,
77 src,
78 isolated_src_table,
79 fgcm_standard_star_cat,
80 use_super=False,
81 ):
82 """A mocked version of this method."""
83 if use_super:
84 return super().compute_psf_and_ap_corr_map(visit, detector, exposure, src,
85 isolated_src_table, fgcm_standard_star_cat)
87 return _make_dummy_psf_and_ap_corr_map()
90class MockFinalizeCharacterizationDetectorTask(FinalizeCharacterizationDetectorTask):
91 """A derived class which skips the initialization routines.
92 """
93 def __init__(self, **kwargs):
94 pipeBase.PipelineTask.__init__(self, **kwargs)
96 self.schema = afwTable.SourceTable.makeMinimalSchema()
98 self.makeSubtask('reserve_selection')
99 self.makeSubtask('source_selector')
101 def compute_psf_and_ap_corr_map(
102 self,
103 visit,
104 detector,
105 exposure,
106 src,
107 isolated_src_table,
108 fgcm_standard_star_cat,
109 use_super=False,
110 ):
111 """A mocked version of this method."""
112 return _make_dummy_psf_and_ap_corr_map()
115class FinalizeCharacterizationTestCase(lsst.utils.tests.TestCase):
116 """Tests of some functionality of FinalizeCharacterizationTask.
118 Full testing comes from integration tests such as ci_hsc and ci_imsim.
120 These tests bypass the middleware used for accessing data and
121 managing Task execution.
122 """
123 def setUp(self):
124 config = FinalizeCharacterizationConfig()
126 self.finalizeCharacterizationTask = MockFinalizeCharacterizationTask(
127 config=config,
128 )
130 config_det = FinalizeCharacterizationDetectorConfig()
132 self.finalizeCharacterizationDetectorTask = MockFinalizeCharacterizationDetectorTask(
133 config=config_det,
134 )
136 def _make_isocats(self, visit=0, detectors=[0, 0, 0]):
137 """Make test isolated star catalogs.
139 Returns
140 -------
141 isolated_star_cat_dict : `dict`
142 Per-"tract" dict of isolated star catalogs.
143 isolate_star_source_dict : `dict`
144 Per-"tract" dict of isolated source catalogs.
145 visit : `int`, optional
146 Visit to set in the source catalogs.
147 detectors : `list` [`int`], optional
148 Detectors to set in the source catalogs; should be 3 items.
149 """
150 dtype_cat = [('isolated_star_id', 'i8'),
151 ('ra', 'f8'),
152 ('dec', 'f8'),
153 ('primary_band', 'U2'),
154 ('source_cat_index', 'i4'),
155 ('nsource', 'i4'),
156 ('source_cat_index_i', 'i4'),
157 ('nsource_i', 'i4'),
158 ('source_cat_index_r', 'i4'),
159 ('nsource_r', 'i4'),
160 ('source_cat_index_z', 'i4'),
161 ('nsource_z', 'i4')]
163 dtype_source = [
164 ('sourceId', 'i8'),
165 ('obj_index', 'i4'),
166 ('visit', 'i8'),
167 ('detector', 'i4'),
168 ]
170 dtype_fgcm = [
171 ('ra', 'f8'),
172 ('dec', 'f8'),
173 ('mag_g', 'f8'),
174 ('mag_i', 'f8'),
175 ]
177 isolated_star_cat_dict = {}
178 isolated_star_source_dict = {}
179 fgcm_cat_dict = {}
181 np.random.seed(12345)
183 # There are 90 stars in both r, i. 10 individually in each.
184 nstar = 110
185 nsource_per_band_per_star = 2
186 self.nstar_total = nstar
187 self.nstar_per_band = nstar - 10
189 # This is a brute-force assembly of a star catalog and matched sources.
190 for tract in [0, 1, 2]:
191 ra = np.random.uniform(low=tract, high=tract + 1.0, size=nstar)
192 dec = np.random.uniform(low=0.0, high=1.0, size=nstar)
194 cat = np.zeros(nstar, dtype=dtype_cat)
195 cat['isolated_star_id'] = tract*nstar + np.arange(nstar)
196 cat['ra'] = ra
197 cat['dec'] = dec
198 if tract < 2:
199 cat['primary_band'][0: 100] = 'i'
200 cat['primary_band'][100:] = 'r'
201 else:
202 # Tract 2 only has z band.
203 cat['primary_band'][:] = 'z'
205 source_cats = []
206 counter = 0
207 for i in range(cat.size):
208 cat['source_cat_index'][i] = counter
209 if tract < 2:
210 if i < 90:
211 cat['nsource'][i] = 2*nsource_per_band_per_star
212 bands = ['r', 'i']
213 else:
214 cat['nsource'][i] = nsource_per_band_per_star
215 if i < 100:
216 bands = ['i']
217 else:
218 bands = ['r']
219 else:
220 cat['nsource'][i] = nsource_per_band_per_star
221 bands = ['z']
223 for j, band in enumerate(bands):
224 cat[f'source_cat_index_{band}'][i] = counter
225 cat[f'nsource_{band}'][i] = nsource_per_band_per_star
226 source_cat = np.zeros(nsource_per_band_per_star, dtype=dtype_source)
227 source_cat['sourceId'] = np.arange(
228 tract*nstar + counter,
229 tract*nstar + counter + nsource_per_band_per_star
230 )
231 source_cat['obj_index'] = i
233 source_cat['visit'] = visit
234 source_cat['detector'] = detectors[j]
236 source_cats.append(source_cat)
238 counter += nsource_per_band_per_star
240 fgcm_cat = np.zeros(nstar, dtype=dtype_fgcm)
241 fgcm_cat['ra'] = ra
242 fgcm_cat['dec'] = dec
243 fgcm_cat['mag_g'] = np.random.uniform(low=16, high=20, size=nstar)
244 fgcm_cat['mag_i'] = np.random.uniform(low=16, high=20, size=nstar)
246 source_cat = np.concatenate(source_cats)
248 isolated_star_cat_dict[tract] = pipeBase.InMemoryDatasetHandle(astropy.table.Table(cat),
249 storageClass="ArrowAstropy")
250 isolated_star_source_dict[tract] = pipeBase.InMemoryDatasetHandle(astropy.table.Table(source_cat),
251 storageClass="ArrowAstropy")
252 fgcm_cat_dict[tract] = pipeBase.InMemoryDatasetHandle(astropy.table.Table(fgcm_cat),
253 storageClass="ArrowAstropy")
255 return isolated_star_cat_dict, isolated_star_source_dict, fgcm_cat_dict
257 def test_concat_isolated_star_cats(self):
258 """Test concatenation and reservation of the isolated star catalogs.
259 """
261 isolated_star_cat_dict, isolated_star_source_dict, fgcm_cat = self._make_isocats()
263 for band in ['r', 'i']:
264 iso, iso_src = self.finalizeCharacterizationTask.concat_isolated_star_cats(
265 band,
266 isolated_star_cat_dict,
267 isolated_star_source_dict
268 )
270 # There are two tracts, so double everything.
271 self.assertEqual(len(iso), 2*self.nstar_per_band)
273 reserve_fraction = self.finalizeCharacterizationTask.config.reserve_selection.reserve_fraction
274 self.assertEqual(np.sum(iso['reserved']),
275 int(reserve_fraction*len(iso)))
277 # 2 tracts, 4 observations per tract per star, minus 2*10 not in the given band.
278 self.assertEqual(len(iso_src), 2*(4*len(iso)//2 - 20))
280 # Check that every star is properly matched to the sources.
281 for i in range(len(iso)):
282 np.testing.assert_array_equal(
283 iso_src['obj_index'][iso[f'source_cat_index_{band}'][i]:
284 iso[f'source_cat_index_{band}'][i] + iso[f'nsource_{band}'][i]],
285 i
286 )
288 # Check that every reserved star is marked as a reserved source.
289 res_star, = np.where(iso['reserved'])
290 for i in res_star:
291 np.testing.assert_array_equal(
292 iso_src['reserved'][iso[f'source_cat_index_{band}'][i]:
293 iso[f'source_cat_index_{band}'][i] + iso[f'nsource_{band}'][i]],
294 True
295 )
297 # Check that every reserved source is marked as a reserved star.
298 res_src, = np.where(iso_src['reserved'])
299 np.testing.assert_array_equal(
300 iso['reserved'][iso_src['obj_index'][res_src]],
301 True
302 )
304 def test_concat_isolate_star_cats_no_sources(self):
305 """Test concatenation when there are no sources in a tract."""
307 isolated_star_cat_dict, isolated_star_source_dict, fgcm_cat = self._make_isocats()
309 iso, iso_src = self.finalizeCharacterizationTask.concat_isolated_star_cats(
310 'z',
311 isolated_star_cat_dict,
312 isolated_star_source_dict
313 )
315 self.assertGreater(len(iso), 0)
317 def test_compute_psf_and_ap_corr_map_no_sources(self):
318 """Test log message when there are no good sources after selection."""
319 # Create an empty source catalog.
320 src_schema = afwTable.SourceTable.makeMinimalSchema()
321 src_schema.addField('base_GaussianFlux_instFlux', type='F', doc='Flux field')
322 src_schema.addField('base_GaussianFlux_instFluxErr', type='F', doc='Flux field')
323 src = afwTable.SourceCatalog(src_schema)
325 # Set defaults and placeholders for required positional arguments.
326 self.finalizeCharacterizationTask.config.source_selector['science'].flags.bad = []
327 visit = 0
328 detector = 0
329 exposure = None
330 isolated_source_table = None
331 fgcm_cat = None
332 with self.assertLogs(level=logging.WARNING) as cm:
333 psf, ap_corr_map, measured_src = self.finalizeCharacterizationTask.compute_psf_and_ap_corr_map(
334 visit,
335 detector,
336 exposure,
337 src,
338 isolated_source_table,
339 fgcm_cat,
340 use_super=True,
341 )
342 self.assertIn(
343 "No good sources remain after cuts for visit {}, detector {}".format(visit, detector),
344 cm.output[0]
345 )
347 def test_run_visit(self):
348 """Test the run method on a full visit."""
349 visit = 100
350 detector0 = 0
351 detector1 = 1
352 band = 'r'
354 isolated_star_cat_dict, isolated_star_source_dict, fgcm_cat = self._make_isocats(
355 visit=visit,
356 detectors=[detector0, detector1, detector1],
357 )
359 # src_dict should be a dictionary keyed by detector, with src handles.
360 # calexp_dict should be a dictionary keyed by detector, with calexp handles.
361 # These can be dummy objects.
363 src0 = afwTable.SourceCatalog(afwTable.SourceTable.makeMinimalSchema())
364 src1 = afwTable.SourceCatalog(afwTable.SourceTable.makeMinimalSchema())
365 calexp0 = afwImage.ExposureF()
366 calexp1 = afwImage.ExposureF()
368 src_dict = {
369 detector0: pipeBase.InMemoryDatasetHandle(src0),
370 detector1: pipeBase.InMemoryDatasetHandle(src1),
371 }
372 calexp_dict = {
373 detector0: pipeBase.InMemoryDatasetHandle(calexp0),
374 detector1: pipeBase.InMemoryDatasetHandle(calexp1),
375 }
377 results = self.finalizeCharacterizationTask.run(
378 visit,
379 band,
380 isolated_star_cat_dict,
381 isolated_star_source_dict,
382 src_dict,
383 calexp_dict,
384 fgcm_standard_star_dict=fgcm_cat,
385 )
387 # Get the dummy values.
388 psf, ap_corr_map, measured_src = _make_dummy_psf_and_ap_corr_map()
390 self.assertEqual(len(results.psf_ap_corr_cat), 2)
391 np.testing.assert_array_equal(results.psf_ap_corr_cat["id"], [detector0, detector1])
392 np.testing.assert_array_equal(results.psf_ap_corr_cat["visit"], visit)
393 row = results.psf_ap_corr_cat.find(detector0)
394 self.assertEqual(row.getPsf().getSigma(), psf.getSigma())
395 self.assertEqual(list(row.getApCorrMap()), list(ap_corr_map))
396 np.testing.assert_array_equal(results.output_table["visit"], visit)
397 table_len = len(results.output_table)
398 np.testing.assert_array_equal(results.output_table["detector"][0: table_len // 2], detector0)
399 np.testing.assert_array_equal(results.output_table["detector"][table_len // 2:], detector1)
401 def test_run_detectors(self):
402 """Test the run method on individual detectors."""
403 visit = 100
404 detector0 = 0
405 detector1 = 1
406 band = 'r'
408 isolated_star_cat_dict, isolated_star_source_dict, fgcm_cat = self._make_isocats(
409 visit=visit,
410 detectors=[detector0, detector1, detector1],
411 )
413 src0 = afwTable.SourceCatalog(afwTable.SourceTable.makeMinimalSchema())
414 src1 = afwTable.SourceCatalog(afwTable.SourceTable.makeMinimalSchema())
415 calexp0 = afwImage.ExposureF()
416 calexp1 = afwImage.ExposureF()
418 results0 = self.finalizeCharacterizationDetectorTask.run(
419 visit,
420 band,
421 detector0,
422 isolated_star_cat_dict,
423 isolated_star_source_dict,
424 src0,
425 calexp0,
426 fgcm_standard_star_dict=fgcm_cat,
427 )
429 results1 = self.finalizeCharacterizationDetectorTask.run(
430 visit,
431 band,
432 detector1,
433 isolated_star_cat_dict,
434 isolated_star_source_dict,
435 src1,
436 calexp1,
437 fgcm_standard_star_dict=fgcm_cat,
438 )
440 # Get the dummy values.
441 psf, ap_corr_map, measured_src = _make_dummy_psf_and_ap_corr_map()
443 # Compare to expected values.
444 self.assertEqual(len(results0.psf_ap_corr_cat), 1)
445 self.assertEqual(len(results1.psf_ap_corr_cat), 1)
446 np.testing.assert_array_equal(results0.psf_ap_corr_cat["id"], detector0)
447 np.testing.assert_array_equal(results1.psf_ap_corr_cat["id"], detector1)
448 np.testing.assert_array_equal(results0.psf_ap_corr_cat["visit"], visit)
449 np.testing.assert_array_equal(results1.psf_ap_corr_cat["visit"], visit)
450 row = results0.psf_ap_corr_cat.find(detector0)
451 self.assertEqual(row.getPsf().getSigma(), psf.getSigma())
452 self.assertEqual(list(row.getApCorrMap()), list(ap_corr_map))
453 row = results1.psf_ap_corr_cat.find(detector1)
454 self.assertEqual(row.getPsf().getSigma(), psf.getSigma())
455 self.assertEqual(list(row.getApCorrMap()), list(ap_corr_map))
456 np.testing.assert_array_equal(results0.output_table["visit"], visit)
457 np.testing.assert_array_equal(results1.output_table["visit"], visit)
458 np.testing.assert_array_equal(results0.output_table["detector"], detector0)
459 np.testing.assert_array_equal(results1.output_table["detector"], detector1)
461 # Now test the task to concatenate these together.
462 consolidate_task = ConsolidateFinalizeCharacterizationDetectorTask(
463 config=ConsolidateFinalizeCharacterizationDetectorConfig(),
464 )
466 psf_ap_corr_detector_dict = {
467 detector0: pipeBase.InMemoryDatasetHandle(results0.psf_ap_corr_cat),
468 detector1: pipeBase.InMemoryDatasetHandle(results1.psf_ap_corr_cat),
469 }
470 src_detector_table_dict = {
471 detector0: pipeBase.InMemoryDatasetHandle(results0.output_table, storageClass="ArrowAstropy"),
472 detector1: pipeBase.InMemoryDatasetHandle(results1.output_table, storageClass="ArrowAstropy"),
473 }
475 results = consolidate_task.run(
476 psf_ap_corr_detector_dict=psf_ap_corr_detector_dict,
477 src_detector_table_dict=src_detector_table_dict,
478 )
480 self.assertEqual(len(results.psf_ap_corr_cat), 2)
481 np.testing.assert_array_equal(results.psf_ap_corr_cat["id"], [detector0, detector1])
482 np.testing.assert_array_equal(results.psf_ap_corr_cat["visit"], visit)
483 row = results.psf_ap_corr_cat.find(detector0)
484 self.assertEqual(row.getPsf().getSigma(), psf.getSigma())
485 self.assertEqual(list(row.getApCorrMap()), list(ap_corr_map))
486 np.testing.assert_array_equal(results.output_table["visit"], visit)
487 table_len = len(results.output_table)
488 np.testing.assert_array_equal(results.output_table["detector"][0: table_len // 2], detector0)
489 np.testing.assert_array_equal(results.output_table["detector"][table_len // 2:], detector1)
492class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase):
493 pass
496def setup_module(module):
497 lsst.utils.tests.init()
500if __name__ == "__main__": 500 ↛ 501line 500 didn't jump to line 501 because the condition on line 500 was never true
501 lsst.utils.tests.init()
502 unittest.main()