Coverage for tests / test_ApplyApCorr.py: 16%
166 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-12 01:19 -0800
1# This file is part of meas_base.
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/>.
22import unittest
24import astropy.table
25import numpy as np
27import lsst.utils.tests
28import lsst.geom
29import lsst.meas.base.tests
30import lsst.afw.image as afwImage
31import lsst.afw.table as afwTable
32import lsst.meas.base.applyApCorr as applyApCorr
33from lsst.afw.math import ChebyshevBoundedField
34from lsst.meas.base.apCorrRegistry import addApCorrName
37def initializeSourceCatalog(schema=None, name=None, instFlux=None, sigma=None, centroid=None):
38 instFluxName = name + "_instFlux"
39 instFluxErrName = name + "_instFluxErr"
40 instFluxKey = schema.find(instFluxName).key
41 centroidKey = afwTable.Point2DKey(schema["slot_Centroid"])
42 sourceCat = afwTable.SourceCatalog(schema)
43 source = sourceCat.addNew()
44 source.set(instFluxKey, instFlux)
45 source.set(instFluxErrName, sigma)
46 source.set(centroidKey, centroid)
47 return sourceCat
50class ApplyApCorrTestCase(lsst.meas.base.tests.AlgorithmTestCase, lsst.utils.tests.TestCase):
52 def setUp(self):
53 schema = afwTable.SourceTable.makeMinimalSchema()
54 names = ["test2", "test"]
55 for name in names:
56 addApCorrName(name)
57 schema.addField(name + "_instFlux", type=np.float64)
58 schema.addField(name + "_instFluxErr", type=np.float64)
59 schema.addField(name + "_flag", type=np.float64)
60 schema.addField(name + "_Centroid_x", type=np.float64)
61 schema.addField(name + "_Centroid_y", type=np.float64)
62 schema.getAliasMap().set('slot_Centroid', name + '_Centroid')
63 self.ap_corr_task = applyApCorr.ApplyApCorrTask(schema=schema)
64 self.name = name # just use 'test' prefix for most of the tests
65 self.schema = schema
67 def tearDown(self):
68 del self.schema
69 del self.ap_corr_task
71 def testAddFields(self):
72 # Check that the required fields have been added to the schema
73 self.assertIn(self.name + "_apCorr", self.schema.getNames())
74 self.assertIn(self.name + "_apCorrErr", self.schema.getNames())
75 self.assertIn(self.name + "_flag_apCorr", self.schema.getNames())
76 self.assertLess(self.schema.find("test_apCorr").key.getOffset(),
77 self.schema.find("test2_apCorr").key.getOffset())
79 def _catalogToAstropy(self, catalog):
80 # Convert a SourceCatalog to an Astropy Table
81 sourceTable = catalog.asAstropy(copy=True)
82 # Add the x and y object table columns for the centroid
83 sourceTable["slot_Centroid_x"] = sourceTable["slot_Centroid_x"]
84 sourceTable["slot_Centroid_y"] = sourceTable["slot_Centroid_y"]
85 return sourceTable
87 def testSuccessUnflagged(self):
88 # Check that the aperture correction flag is set to False if aperture
89 # correction was successfully run
90 flagName = self.name + "_flag_apCorr"
91 flagKey = self.schema.find(flagName).key
92 source_test_instFlux = 5.1
93 source_test_centroid = lsst.geom.Point2D(5, 7.1)
94 sourceCat = initializeSourceCatalog(schema=self.schema, name=self.name, instFlux=source_test_instFlux,
95 sigma=0, centroid=source_test_centroid)
96 instFluxName = self.name + "_instFlux"
97 instFluxErrName = self.name + "_instFluxErr"
99 apCorrMap = afwImage.ApCorrMap()
100 bbox = lsst.geom.Box2I(lsst.geom.Point2I(0, 0), lsst.geom.ExtentI(10, 10))
101 coefficients = np.ones((1, 1), dtype=np.float64)
102 coefficients_sigma = np.zeros((1, 1), dtype=np.float64)
103 apCorrMap[instFluxName] = ChebyshevBoundedField(bbox, coefficients)
104 apCorrMap[instFluxErrName] = ChebyshevBoundedField(bbox, coefficients_sigma)
106 sourceTable = self._catalogToAstropy(sourceCat)
107 self.ap_corr_task.run(sourceCat, apCorrMap)
108 self.assertFalse(sourceCat[flagKey])
109 self.ap_corr_task.run(sourceTable, apCorrMap)
110 self.assertFalse(sourceTable[flagName])
112 def testFailureFlagged(self):
113 # Check that aperture correction flag is set to True if aperture
114 # correction is invalid (negative)
115 flagName = self.name + "_flag_apCorr"
116 flagKey = self.schema.find(flagName).key
117 source_test_instFlux = 5.2
118 source_test_centroid = lsst.geom.Point2D(5, 7.1)
119 sourceCat = initializeSourceCatalog(schema=self.schema, name=self.name, instFlux=source_test_instFlux,
120 sigma=0, centroid=source_test_centroid)
121 instFluxName = self.name + "_instFlux"
122 instFluxErrName = self.name + "_instFluxErr"
124 apCorrMap = afwImage.ApCorrMap()
125 bbox = lsst.geom.Box2I(lsst.geom.Point2I(0, 0), lsst.geom.ExtentI(10, 10))
126 coefficients = -(np.ones((1, 1), dtype=np.float64))
127 coefficients_sigma = np.zeros((1, 1), dtype=np.float64)
128 apCorrMap[instFluxName] = ChebyshevBoundedField(bbox, coefficients)
129 apCorrMap[instFluxErrName] = ChebyshevBoundedField(bbox, coefficients_sigma)
130 sourceTable = self._catalogToAstropy(sourceCat)
131 self.ap_corr_task.run(sourceCat, apCorrMap)
132 self.assertTrue(sourceCat[flagKey])
133 self.ap_corr_task.run(sourceTable, apCorrMap)
134 self.assertTrue(sourceTable[flagName])
136 def testCatFluxUnchanged(self):
137 # Pick arbitrary but unique values for the test case
138 source_test_instFlux = 5.3
139 source_test_centroid = lsst.geom.Point2D(5, 7.1)
140 sourceCat = initializeSourceCatalog(schema=self.schema, name=self.name, instFlux=source_test_instFlux,
141 sigma=0, centroid=source_test_centroid)
142 instFluxName = self.name + "_instFlux"
143 instFluxErrName = self.name + "_instFluxErr"
144 instFluxKey = self.schema.find(instFluxName).key
146 apCorrMap = afwImage.ApCorrMap()
147 bbox = lsst.geom.Box2I(lsst.geom.Point2I(0, 0), lsst.geom.ExtentI(10, 10))
148 coefficients = np.ones((1, 1), dtype=np.float64)
149 coefficients_sigma = np.zeros((1, 1), dtype=np.float64)
150 apCorrMap[instFluxName] = ChebyshevBoundedField(bbox, coefficients)
151 apCorrMap[instFluxErrName] = ChebyshevBoundedField(bbox, coefficients_sigma)
153 sourceTable = self._catalogToAstropy(sourceCat)
154 self.ap_corr_task.run(sourceCat, apCorrMap)
155 self.assertEqual(sourceCat[instFluxKey], source_test_instFlux)
156 self.ap_corr_task.run(sourceTable, apCorrMap)
157 self.assertEqual(sourceTable[instFluxName], source_test_instFlux)
159 def testCatFluxHalf(self):
160 # Pick arbitrary but unique values for the test case
161 source_test_instFlux = 5.4
162 source_test_centroid = lsst.geom.Point2D(5, 7.1)
163 sourceCat = initializeSourceCatalog(schema=self.schema, name=self.name, instFlux=source_test_instFlux,
164 sigma=0, centroid=source_test_centroid)
165 instFluxName = self.name + "_instFlux"
166 instFluxErrName = self.name + "_instFluxErr"
167 instFluxKey = self.schema.find(instFluxName).key
169 apCorrMap = afwImage.ApCorrMap()
170 bbox = lsst.geom.Box2I(lsst.geom.Point2I(0, 0), lsst.geom.ExtentI(10, 10))
171 coefficients = np.ones((1, 1), dtype=np.float64)
172 coefficients /= 2.
173 coefficients_sigma = np.zeros((1, 1), dtype=np.float64)
174 apCorrMap[instFluxName] = ChebyshevBoundedField(bbox, coefficients)
175 apCorrMap[instFluxErrName] = ChebyshevBoundedField(bbox, coefficients_sigma)
177 sourceTable = self._catalogToAstropy(sourceCat)
178 self.ap_corr_task.run(sourceCat, apCorrMap)
179 self.assertFloatsAlmostEqual(sourceCat[instFluxKey], source_test_instFlux / 2)
180 self.ap_corr_task.run(sourceTable, apCorrMap)
181 self.assertFloatsAlmostEqual(sourceTable[instFluxName], source_test_instFlux / 2)
183 def testCatFluxErr(self):
184 """Test catalog flux errors.
186 Notes
187 -----
188 This test will break if ``UseNaiveFluxErr = False``~
190 The alternate method significantly overestimates noise, causing this
191 test to fail. It is likely that this test will need to be modified if
192 the noise calculation is updated.
193 """
194 # Pick arbitrary but unique values for the test case
195 source_test_instFlux = 5.5
196 source_test_sigma = 0.23
197 source_test_centroid = lsst.geom.Point2D(5, 7.3)
198 sourceCat = initializeSourceCatalog(schema=self.schema, name=self.name, instFlux=source_test_instFlux,
199 sigma=source_test_sigma, centroid=source_test_centroid)
201 instFluxName = self.name + "_instFlux"
202 instFluxErrName = self.name + "_instFluxErr"
203 instFluxErrKey = self.schema.find(instFluxErrName).key
205 apCorrMap = afwImage.ApCorrMap()
206 bbox = lsst.geom.Box2I(lsst.geom.Point2I(0, 0), lsst.geom.ExtentI(10, 10))
207 coefficients = np.ones((1, 1), dtype=np.float64)
208 coefficients_sigma = np.ones((1, 1), dtype=np.float64)
209 apCorrMap[instFluxName] = ChebyshevBoundedField(bbox, coefficients)
210 apCorrMap[instFluxErrName] = ChebyshevBoundedField(bbox, coefficients_sigma)
212 sourceTable = self._catalogToAstropy(sourceCat)
213 self.ap_corr_task.run(sourceCat, apCorrMap)
214 self.assertFloatsAlmostEqual(sourceCat[instFluxErrKey], source_test_sigma)
215 self.ap_corr_task.run(sourceTable, apCorrMap)
216 self.assertFloatsAlmostEqual(sourceTable[instFluxErrName], source_test_sigma)
218 def testSourceTable(self):
219 """Test that the task can handle a SourceTable without columns."""
220 # Create an empty SourceTable
221 source_test_instFlux = 5.3
222 source_test_centroid = lsst.geom.Point2D(5, 7.1)
223 instFluxName = self.name + "_instFlux"
224 instFluxErrName = self.name + "_instFluxErr"
226 sourceTable = astropy.table.Table(
227 {
228 "id": [1],
229 "slot_Centroid_x": [source_test_centroid.x],
230 "slot_Centroid_y": [source_test_centroid.y],
231 instFluxName: [source_test_instFlux],
232 instFluxErrName: [0.0],
233 }
234 )
236 apCorrMap = afwImage.ApCorrMap()
237 bbox = lsst.geom.Box2I(lsst.geom.Point2I(0, 0), lsst.geom.ExtentI(10, 10))
238 coefficients = np.ones((1, 1), dtype=np.float64)
239 coefficients /= 2.
240 coefficients_sigma = np.zeros((1, 1), dtype=np.float64)
241 apCorrMap[instFluxName] = ChebyshevBoundedField(bbox, coefficients)
242 apCorrMap[instFluxErrName] = ChebyshevBoundedField(bbox, coefficients_sigma)
244 self.ap_corr_task.run(sourceTable, apCorrMap)
245 self.assertFloatsAlmostEqual(sourceTable[instFluxName], source_test_instFlux / 2)
248class TestMemory(lsst.utils.tests.MemoryTestCase):
249 pass
252def setup_module(module):
253 lsst.utils.tests.init()
256if __name__ == "__main__": 256 ↛ 257line 256 didn't jump to line 257 because the condition on line 256 was never true
257 lsst.utils.tests.init()
258 unittest.main()