Coverage for tests / test_coadd_ap_corr_map.py: 17%

1# This file is part of cell_coadds. 

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 

22import unittest 

23 

24import numpy as np 

25 

26import lsst.afw.image as afwImage 

27import lsst.afw.math as afwMath 

28import lsst.geom as geom 

29import lsst.utils.tests 

30from lsst.cell_coadds import CoaddApCorrMapStacker 

31 

32 

33class TestCoaddApCorrMapStacker(lsst.utils.tests.TestCase): 

34 """Test the CoaddApCorrMap stacker.""" 

35 

36 @classmethod 

37 def setUpClass(cls): # noqa: D102 

38 cls.bbox = geom.Box2I( 

39 minimum=geom.Point2I(-10, -10), 

40 maximum=geom.Point2I(10, 10), 

41 ) 

42 cls.evaluation_point = geom.Point2D(0, 0) 

43 cls.visit_count = 5 

44 cls.ap_corr_map_list = [afwImage.ApCorrMap() for _ in range(cls.visit_count)] 

45 # Set them up to the spatially constant BoundedFields. 

46 for idx, ap_corr_map in enumerate(cls.ap_corr_map_list, start=2): 

47 ap_corr_map.set( 

48 "base_PsfFlux_instFlux", afwMath.ChebyshevBoundedField(cls.bbox, np.array([[1 / idx]])) 

49 ) 

50 ap_corr_map.set( 

51 "base_PsfFlux_instFluxErr", afwMath.ChebyshevBoundedField(cls.bbox, np.array([[0.05 / idx]])) 

52 ) 

53 ap_corr_map.set( 

54 "base_GaussianFlux_instFlux", afwMath.ChebyshevBoundedField(cls.bbox, np.array([[1.5 / idx]])) 

55 ) 

56 ap_corr_map.set( 

57 "base_GaussianFlux_instFluxErr", 

58 afwMath.ChebyshevBoundedField(cls.bbox, np.array([[0.6 / idx]])), 

59 ) 

60 

61 def test_add_direct_ap_corr(self): 

62 """Test that the values are what we expect when adding directly.""" 

63 stacker = CoaddApCorrMapStacker( 

64 evaluation_point=self.evaluation_point, 

65 do_coadd_inverse_ap_corr=False, 

66 ) 

67 for idx in range(self.visit_count): 

68 stacker.add( 

69 self.ap_corr_map_list[idx], 

70 weight=1.0, 

71 ) 

72 

73 final_ap_corr_map = stacker.final_ap_corr_map 

74 

75 self.assertFloatsAlmostEqual( 

76 final_ap_corr_map["base_PsfFlux_instFlux"], 

77 0.29, # mean of 1/2, 1/3, 1/4, 1/5, 1/6 

78 atol=1e-8, 

79 ) 

80 self.assertFloatsAlmostEqual( 

81 final_ap_corr_map["base_PsfFlux_instFluxErr"], 

82 0.05 * 0.14019827229875395, 

83 atol=1e-8, 

84 ) 

85 self.assertFloatsAlmostEqual( 

86 final_ap_corr_map["base_GaussianFlux_instFlux"], 

87 1.5 * 0.29, 

88 atol=1e-8, 

89 ) 

90 self.assertFloatsAlmostEqual( 

91 final_ap_corr_map["base_GaussianFlux_instFluxErr"], 

92 0.6 * 0.14019827229875395, 

93 atol=1e-8, 

94 ) 

95 

96 def test_add_inverse_ap_corr(self): 

97 """Test that the values are what we expect when adding inverse.""" 

98 stacker = CoaddApCorrMapStacker( 

99 evaluation_point=self.evaluation_point, 

100 do_coadd_inverse_ap_corr=True, 

101 ) 

102 for idx in range(self.visit_count): 

103 stacker.add( 

104 self.ap_corr_map_list[idx], 

105 weight=1.0, 

106 ) 

107 

108 final_ap_corr_map = stacker.final_ap_corr_map 

109 

110 self.assertFloatsAlmostEqual( 

111 final_ap_corr_map["base_PsfFlux_instFlux"], 

112 0.25, 

113 atol=1e-8, 

114 ) 

115 

116 self.assertFloatsAlmostEqual( 

117 final_ap_corr_map["base_GaussianFlux_instFlux"], 

118 1.5 * 0.25, 

119 atol=1e-8, 

120 ) 

121 

122 self.assertFloatsAlmostEqual( 

123 final_ap_corr_map["base_PsfFlux_instFluxErr"], 

124 0.05 * (0.25**2) * np.sqrt(90) / 5, 

125 atol=1e-8, 

126 ) 

127 

128 self.assertFloatsAlmostEqual( 

129 final_ap_corr_map["base_GaussianFlux_instFluxErr"], 

130 0.6 / (1.5**2) * ((1.5 * 0.25) ** 2) * np.sqrt(90) / 5, 

131 atol=1e-8, 

132 ) 

133 

134 @lsst.utils.tests.methodParameters( 

135 do_coadd_inverse_ap_corr=[True, False], 

136 ) 

137 def test_weight_normalization(self, do_coadd_inverse_ap_corr): 

138 """Test that the coadd aperture corrections stay the same when the 

139 relative weights are the same. 

140 """ 

141 stacker = CoaddApCorrMapStacker( 

142 evaluation_point=self.evaluation_point, 

143 do_coadd_inverse_ap_corr=do_coadd_inverse_ap_corr, 

144 ) 

145 

146 normalized_weights = np.random.rand(self.visit_count) 

147 normalized_weights /= np.sum(normalized_weights) 

148 

149 for idx in range(self.visit_count): 

150 stacker.add( 

151 self.ap_corr_map_list[idx], 

152 weight=normalized_weights[idx], 

153 ) 

154 

155 reference_ap_corr_map = stacker.final_ap_corr_map 

156 

157 for scale in ( 

158 1.0, 

159 0.5, 

160 2.0, 

161 ): 

162 stacker = CoaddApCorrMapStacker( 

163 evaluation_point=self.evaluation_point, 

164 do_coadd_inverse_ap_corr=do_coadd_inverse_ap_corr, 

165 ) 

166 weights = scale * normalized_weights 

167 for idx in range(self.visit_count): 

168 stacker.add( 

169 self.ap_corr_map_list[idx], 

170 weight=weights[idx], 

171 ) 

172 final_ap_corr_map = stacker.final_ap_corr_map 

173 

174 for field_name in reference_ap_corr_map: 

175 with self.subTest(scale=scale, field_name=field_name): 

176 self.assertFloatsAlmostEqual( 

177 final_ap_corr_map[field_name], 

178 reference_ap_corr_map[field_name], 

179 atol=1e-9, 

180 ) 

181 

182 @lsst.utils.tests.methodParameters( 

183 do_coadd_inverse_ap_corr=[True, False], 

184 ) 

185 def test_constant_weight(self, do_coadd_inverse_ap_corr): 

186 """Test that the coadd aperture corrections stay the same when the 

187 weights are all equal. 

188 """ 

189 stacker = CoaddApCorrMapStacker( 

190 evaluation_point=self.evaluation_point, 

191 do_coadd_inverse_ap_corr=do_coadd_inverse_ap_corr, 

192 ) 

193 

194 for idx in range(self.visit_count): 

195 stacker.add( 

196 self.ap_corr_map_list[idx], 

197 weight=1.0, 

198 ) 

199 

200 reference_ap_corr_map = stacker.final_ap_corr_map 

201 

202 stacker = CoaddApCorrMapStacker( 

203 evaluation_point=self.evaluation_point, 

204 do_coadd_inverse_ap_corr=do_coadd_inverse_ap_corr, 

205 ) 

206 for idx in range(self.visit_count): 

207 stacker.add( 

208 self.ap_corr_map_list[idx], 

209 weight=2.0, 

210 ) 

211 final_ap_corr_map = stacker.final_ap_corr_map 

212 

213 for algorithm_name in stacker.ap_corr_names: 

214 field_name = f"{algorithm_name}_instFlux" 

215 with self.subTest(field_name=field_name): 

216 self.assertFloatsAlmostEqual( 

217 final_ap_corr_map[field_name], 

218 reference_ap_corr_map[field_name], 

219 atol=1e-8, 

220 ) 

221 

222 field_name = f"{algorithm_name}_instFluxErr" 

223 # Errors are down by sqrt(visit_count). 

224 conversion_factor = 1.0 # np.sqrt(self.visit_count) 

225 with self.subTest(field_name=field_name): 

226 self.assertFloatsAlmostEqual( 

227 final_ap_corr_map[field_name] / conversion_factor, 

228 reference_ap_corr_map[field_name], 

229 atol=1e-8, 

230 ) 

231 

232 @lsst.utils.tests.methodParameters( 

233 do_coadd_inverse_ap_corr=[True, False], 

234 ) 

235 def test_constant_apcorr(self, do_coadd_inverse_ap_corr): 

236 """Test that the coadd aperture corrections are constant when the 

237 input aperture corrections are constant. 

238 """ 

239 stacker = CoaddApCorrMapStacker( 

240 evaluation_point=self.evaluation_point, 

241 do_coadd_inverse_ap_corr=do_coadd_inverse_ap_corr, 

242 ) 

243 

244 weights = np.random.rand(self.visit_count) 

245 initial_ap_corr_map = self.ap_corr_map_list[0] 

246 for idx in range(self.visit_count): 

247 stacker.add( 

248 initial_ap_corr_map, # Add the same over and over again. 

249 weight=weights[idx], 

250 ) 

251 

252 final_ap_corr_map = stacker.final_ap_corr_map 

253 

254 for algorithm_name in stacker.ap_corr_names: 

255 field_name = f"{algorithm_name}_instFlux" 

256 with self.subTest(field_name=field_name): 

257 self.assertFloatsAlmostEqual( 

258 final_ap_corr_map[field_name], 

259 initial_ap_corr_map[field_name].evaluate(geom.Point2D(self.evaluation_point)), 

260 atol=1e-8, 

261 ) 

262 

263 field_name = f"{algorithm_name}_instFluxErr" 

264 # Errors are down by conversion_factor. 

265 conversion_factor = np.sqrt(np.sum(weights**2)) / np.sum(weights) 

266 with self.subTest(field_name=field_name): 

267 self.assertFloatsAlmostEqual( 

268 final_ap_corr_map[field_name] / conversion_factor, 

269 initial_ap_corr_map[field_name].evaluate(geom.Point2D(self.evaluation_point)), 

270 atol=1e-8, 

271 ) 

272 

273 # Evaluate aperture corrections at several points. 

274 evaluate_at_x = np.linspace(self.bbox.getMinX(), self.bbox.getMaxX(), 5) 

275 evaluate_at_y = np.linspace(self.bbox.getMinY(), self.bbox.getMaxY(), 5) 

276 for algorithm_name in stacker.ap_corr_names: 

277 field_name = f"{algorithm_name}_instFlux" 

278 reference_value = initial_ap_corr_map[field_name].evaluate(geom.Point2D(self.evaluation_point)) 

279 aperture_correction_values = initial_ap_corr_map[field_name].evaluate( 

280 evaluate_at_x, 

281 evaluate_at_y, 

282 ) 

283 with self.subTest(field_name=field_name): 

284 np.testing.assert_array_equal(aperture_correction_values, reference_value) 

285 

286 

287class TestCoaddApCorrMapMemoryTestCase(lsst.utils.tests.MemoryTestCase): 

288 """Test the CoaddApCorrMapStacker for memory leaks.""" 

289 

290 

291def setup_module(module): # noqa: D103 

292 lsst.utils.tests.init() 

293 

294 

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

296 lsst.utils.tests.init() 

297 unittest.main()