Coverage for tests/test_produce_and_standardize_prompt_source.py: 93%

1# This file is part of drp_tasks. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

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 GNU General Public License 

20# along with this program. If not, see <https://www.gnu.org/licenses/>. 

21 

22"""End-to-end test of the prompt_source production chain. 

23 

24This exercises the sequence of tasks that ApPipe.yaml runs to turn a 

25calibrated visit image into the ``prompt_source`` table: 

26 

271. ``SingleFrameDetectAndMeasureTask`` detects and measures 

28 sources on a synthetic exposure, producing the 

29 ``single_visit_star_reprocessed_unstandardized`` catalog. 

302. ``TransformSourceTableTask`` applies the 

31 ``prompt_source.yaml`` functors from ``pipe_tasks``. 

323. ``ConsolidateSourceTableTask`` concatenates the per-detector tables. 

334. ``SplitPrimaryTask`` keeps the primary rows and drops sky sources. 

34""" 

35 

36import os 

37import unittest 

38 

39import numpy as np 

40 

41import lsst.afw.geom 

42import lsst.afw.image 

43import lsst.geom 

44import lsst.meas.algorithms 

45import lsst.meas.base.tests 

46import lsst.utils.tests 

47from lsst.drp.tasks.single_frame_detect_and_measure import ( 

48 SingleFrameDetectAndMeasureConfig, 

49 SingleFrameDetectAndMeasureTask, 

50) 

51from lsst.pipe.base import InMemoryDatasetHandle 

52from lsst.pipe.tasks.postprocess import ( 

53 TableVStack, 

54 TransformSourceTableConfig, 

55 TransformSourceTableTask, 

56) 

57from lsst.pipe.tasks.schemaUtils import ( 

58 checkSdmSchemaColumns, 

59 column_dtype, 

60 readSdmSchemaFile, 

61) 

62from lsst.pipe.tasks.split_primary import SplitPrimaryTask 

63from lsst.resources import ResourcePath 

64 

65FUNCTOR_FILE = ResourcePath("eups://pipe_tasks/schemas/prompt_source.yaml") 

66SCHEMA_FILE = os.path.join("${SDM_SCHEMAS_DIR}", "yml", "ap_extra.yaml") 

67TABLE_NAME = "PromptSource" 

68SPLIT_DISCARD_PRIMARY_COLUMNS = ["sky_source"] 

69 

70VISIT = 98765 

71DETECTOR = 42 

72 

73 

74def make_exposure_and_background(): 

75 """Build a synthetic calibrated exposure and its background model. 

76 

77 Modeled on ``test_reprocess_visit_image.py``: a `TestDataset` provides the 

78 PSF, WCS and PhotoCalib, a handful of point sources are added, and a 

79 background is fit to the realized image. 

80 

81 Returns 

82 ------- 

83 exposure : `lsst.afw.image.ExposureF` 

84 Calibrated exposure ready for detection and measurement. 

85 background : `lsst.afw.math.BackgroundList` 

86 Background model for the exposure. 

87 """ 

88 bbox = lsst.geom.Box2I(lsst.geom.Point2I(5, 4), lsst.geom.Point2I(205, 184)) 

89 dataset = lsst.meas.base.tests.TestDataset( 

90 bbox, 

91 crval=lsst.geom.SpherePoint(245.0, -45.0, lsst.geom.degrees), 

92 calibration=12.3, 

93 detector=DETECTOR, 

94 visitId=VISIT, 

95 ) 

96 psf_scale = np.sqrt(4 * np.pi * (dataset.psfShape.getDeterminantRadius()) ** 2) 

97 noise = 10.0 

98 for flux, centroid in [ 

99 (45 * noise * psf_scale, (40, 70)), 

100 (150 * noise * psf_scale, (50, 120)), 

101 (400 * noise * psf_scale, (92, 35)), 

102 (1000 * noise * psf_scale, (175, 154)), 

103 ]: 

104 dataset.addSource(instFlux=flux, centroid=lsst.geom.Point2D(*centroid)) 

105 truth_exposure, _ = dataset.realize(noise=noise, schema=dataset.makeMinimalSchema()) 

106 

107 # Build an input exposure carrying the calibrations the task reads. 

108 exposure = lsst.afw.image.ExposureF(truth_exposure.maskedImage.clone()) 

109 exposure.mask.clearMaskPlane(exposure.mask.getMaskPlane("DETECTED")) 

110 exposure.setPsf(truth_exposure.psf) 

111 exposure.setWcs(truth_exposure.wcs) 

112 exposure.setPhotoCalib(truth_exposure.photoCalib) 

113 exposure.info.setApCorrMap(lsst.afw.image.ApCorrMap()) 

114 exposure.info.setVisitInfo(truth_exposure.visitInfo) 

115 exposure.info.setDetector(truth_exposure.getDetector()) 

116 

117 bkgConfig = lsst.meas.algorithms.SubtractBackgroundTask.ConfigClass() 

118 # Small test image; fit a simple background model. 

119 bkgConfig.approxOrderX = 1 

120 bkgTask = lsst.meas.algorithms.SubtractBackgroundTask(config=bkgConfig) 

121 background = bkgTask.run(truth_exposure).background 

122 

123 return exposure, background 

124 

125 

126class PromptSourceEndToEndTestCase(lsst.utils.tests.TestCase): 

127 """Run SingleFrameDetectAndMeasure output through the prompt_source steps 

128 and validate the final schema. 

129 """ 

130 

131 @classmethod 

132 def setUpClass(cls): 

133 super().setUpClass() 

134 

135 exposure, background = make_exposure_and_background() 

136 

137 # 1. Detect and measure direct sources. 

138 sfdmConfig = SingleFrameDetectAndMeasureConfig() 

139 sfdmConfig.detection.background.approxOrderX = 1 

140 sfdmConfig.sky_sources.nSources = 2 

141 sfdmTask = SingleFrameDetectAndMeasureTask(config=sfdmConfig) 

142 cls.sources = sfdmTask.run(exposure=exposure, input_background=background).sources 

143 

144 schemaFile = os.path.expandvars(SCHEMA_FILE) 

145 cls.schema = readSdmSchemaFile(schemaFile) 

146 

147 # 2. Run standardization 

148 transformConfig = TransformSourceTableConfig() 

149 with FUNCTOR_FILE.as_local() as local: 

150 transformConfig.functorFile = local.ospath 

151 transformTask = TransformSourceTableTask(config=transformConfig) 

152 detectorTable = transformTask.run( 

153 handle=InMemoryDatasetHandle(cls.sources.to_pandas(), storageClass="DataFrame"), 

154 funcs=transformTask.funcs, 

155 dataId={"visit": VISIT, "detector": DETECTOR, "band": "r", "physical_filter": "r_03"}, 

156 ).outputCatalog 

157 

158 # 3. Mimic the stacking done by the consolidate task 

159 consolidated = TableVStack.vstack_handles( 

160 [InMemoryDatasetHandle(detectorTable, storageClass="ArrowAstropy")] 

161 ) 

162 

163 # 4. Drop non-primary rows from blends, and sky sources 

164 splitConfig = SplitPrimaryTask.ConfigClass() 

165 splitConfig.discard_primary_columns = SPLIT_DISCARD_PRIMARY_COLUMNS 

166 splitTask = SplitPrimaryTask(config=splitConfig) 

167 cls.promptSource = splitTask.run(full=consolidated).primary 

168 

169 def testProducesPrimarySources(self): 

170 """prompt_source holds the primary rows minus the dropped columns.""" 

171 nPrimary = int(np.sum(self.sources["detect_isPrimary"])) 

172 self.assertGreater(nPrimary, 0) 

173 self.assertEqual(len(self.promptSource), nPrimary) 

174 self.assertNotIn("detect_isPrimary", self.promptSource.colnames) 

175 self.assertNotIn("sky_source", self.promptSource.colnames) 

176 

177 def testCalibratedFluxesComputed(self): 

178 """The calibrated flux functors ran and produced finite values.""" 

179 for column in ("psfFlux", "calibFlux"): 

180 self.assertIn(column, self.promptSource.colnames) 

181 self.assertTrue(np.isfinite(self.promptSource[column]).any()) 

182 

183 def testSchema(self): 

184 """Check that prompt_source column names and dtypes match the 

185 PromptSource schema. 

186 """ 

187 dataframe = self.promptSource.to_pandas() 

188 # There should be no columns beyond those defined for PromptSource. 

189 extra = checkSdmSchemaColumns(self.schema, list(dataframe.columns), TABLE_NAME) 

190 self.assertEqual(extra, [], f"prompt_source has columns absent from {TABLE_NAME}: {extra}") 

191 

192 # Every schema column is present with the expected dtype. 

193 mismatches = {} 

194 for columnDef in self.schema[TABLE_NAME].columns: 

195 # Extract the expected datatype for each column from the schema 

196 expected = column_dtype(columnDef.datatype) 

197 if columnDef.name not in dataframe.columns: 197 ↛ 198line 197 didn't jump to line 198 because the condition on line 197 was never true

198 mismatches[columnDef.name] = f"missing (expected {expected})" 

199 elif str(dataframe[columnDef.name].dtype) != expected: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true

200 mismatches[columnDef.name] = f"{dataframe[columnDef.name].dtype} != {expected}" 

201 self.assertEqual(mismatches, {}, f"prompt_source dtypes do not match {TABLE_NAME}: {mismatches}") 

202 

203 

204class MemoryTester(lsst.utils.tests.MemoryTestCase): 

205 pass 

206 

207 

208def setup_module(module): 

209 lsst.utils.tests.init() 

210 

211 

212if __name__ == "__main__": 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true

213 lsst.utils.tests.init() 

214 unittest.main()