Coverage for tests/test_centroiding.py: 98%

1#!/usr/bin/env python 

2# This file is part of atmospec. 

3# 

4# Developed for the LSST Data Management System. 

5# This product includes software developed by the LSST Project 

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

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

8# for details of code ownership. 

9# 

10# This program is free software: you can redistribute it and/or modify 

11# it under the terms of the GNU General Public License as published by 

12# the Free Software Foundation, either version 3 of the License, or 

13# (at your option) any later version. 

14# 

15# This program is distributed in the hope that it will be useful, 

16# but WITHOUT ANY WARRANTY; without even the implied warranty of 

17# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

18# GNU General Public License for more details. 

19# 

20# You should have received a copy of the GNU General Public License 

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

22"""Test cases for the atmospec SingleStarCentroidTask.""" 

23 

24import unittest 

25from unittest import mock 

26 

27import lsst.utils 

28import lsst.utils.tests 

29import lsst.geom as geom 

30import lsst.afw.geom as afwGeom 

31import lsst.afw.image as afwImage 

32import lsst.pipe.base as pipeBase 

33from lsst.pipe.base.task import TaskError 

34from lsst.pipe.tasks.quickFrameMeasurement import QuickFrameMeasurementTask 

35from lsst.pipe.tasks.peekExposure import PeekExposureTask 

36 

37from lsst.atmospec.centroiding import SingleStarCentroidTask, SingleStarCentroidTaskConfig 

38 

39 

40def makeExposureWithWcs(target="HD 12345"): 

41 """Build a small ExposureF with a valid WCS and a visitInfo object.""" 

42 exp = afwImage.ExposureF(16, 16) 

43 wcs = afwGeom.makeSkyWcs(geom.Point2D(0, 0), 

44 geom.SpherePoint(10*geom.degrees, 20*geom.degrees), 

45 afwGeom.makeCdMatrix(1e-4*geom.degrees)) 

46 exp.setWcs(wcs) 

47 exp.info.setVisitInfo(afwImage.VisitInfo(object=target)) 

48 return exp 

49 

50 

51class SingleStarCentroidConfigTestCase(lsst.utils.tests.TestCase): 

52 """Tests for the task config, including setDefaults and validate.""" 

53 

54 def testSetDefaultsAndValidate(self): 

55 # The default fallback task is PeekExposureTask, which is valid. 

56 config = SingleStarCentroidTaskConfig() 

57 self.assertEqual(config.astromRefObjLoader.pixelMargin, 1000) 

58 self.assertEqual(config.referenceFilterOverride, "phot_g_mean") 

59 # Should not raise. 

60 config.validate() 

61 

62 def testValidateWithQuickFrameMeasurement(self): 

63 config = SingleStarCentroidTaskConfig() 

64 config.centroidingFallbackTask.retarget(QuickFrameMeasurementTask) 

65 # quickFrameMeasurementTask is an accepted fallback. 

66 config.validate() 

67 

68 def testValidateUnknownFallbackRaises(self): 

69 config = SingleStarCentroidTaskConfig() 

70 # Retarget to a task whose _DefaultName is not accepted. 

71 config.centroidingFallbackTask.retarget(SingleStarCentroidTask) 

72 with self.assertRaises(ValueError): 

73 config.validate() 

74 

75 

76class SingleStarCentroidTaskTestCase(lsst.utils.tests.TestCase): 

77 """Tests for the SingleStarCentroidTask itself.""" 

78 

79 def makeTask(self): 

80 config = SingleStarCentroidTaskConfig() 

81 return SingleStarCentroidTask(config=config) 

82 

83 def testInit(self): 

84 task = self.makeTask() 

85 self.assertIsInstance(task, SingleStarCentroidTask) 

86 self.assertIsInstance(task.centroidingFallbackTask, PeekExposureTask) 

87 

88 def testRunSuccessfulFit(self): 

89 task = self.makeTask() 

90 exp = makeExposureWithWcs() 

91 

92 astromResult = mock.MagicMock() 

93 astromResult.scatterOnSky.asArcseconds.return_value = 0.5 

94 task.astrometry = mock.MagicMock() 

95 task.astrometry.run.return_value = astromResult 

96 

97 with mock.patch("lsst.atmospec.centroiding.getTargetCentroidFromWcs", 

98 return_value=geom.Point2D(3.0, 4.0)): 

99 result = task.run(inputExp=exp, inputSources=mock.MagicMock()) 

100 

101 self.assertTrue(result.atmospecCentroid["astrometricMatch"]) 

102 self.assertEqual(result.atmospecCentroid["centroid"], (3.0, 4.0)) 

103 

104 def testRunSuccessfulFitNoCentroidFallsBack(self): 

105 task = self.makeTask() 

106 exp = makeExposureWithWcs() 

107 

108 astromResult = mock.MagicMock() 

109 astromResult.scatterOnSky.asArcseconds.return_value = 0.5 

110 task.astrometry = mock.MagicMock() 

111 task.astrometry.run.return_value = astromResult 

112 

113 # WCS lookup returns None, forcing the fallback path. 

114 peek = mock.MagicMock(spec=PeekExposureTask) 

115 peek.run.return_value = pipeBase.Struct(brightestCentroid=geom.Point2D(7.0, 8.0)) 

116 task.centroidingFallbackTask = peek 

117 

118 with mock.patch("lsst.atmospec.centroiding.getTargetCentroidFromWcs", return_value=None): 

119 result = task.run(inputExp=exp, inputSources=mock.MagicMock()) 

120 

121 self.assertFalse(result.atmospecCentroid["astrometricMatch"]) 

122 self.assertEqual(result.atmospecCentroid["centroid"], (7.0, 8.0)) 

123 

124 def testRunScatterTooLargeFallsBack(self): 

125 task = self.makeTask() 

126 exp = makeExposureWithWcs() 

127 

128 astromResult = mock.MagicMock() 

129 astromResult.scatterOnSky.asArcseconds.return_value = 5.0 # >= 1, so not successful 

130 task.astrometry = mock.MagicMock() 

131 task.astrometry.run.return_value = astromResult 

132 

133 peek = mock.MagicMock(spec=PeekExposureTask) 

134 peek.run.return_value = pipeBase.Struct(brightestCentroid=geom.Point2D(1.0, 2.0)) 

135 task.centroidingFallbackTask = peek 

136 

137 result = task.run(inputExp=exp, inputSources=mock.MagicMock()) 

138 self.assertFalse(result.atmospecCentroid["astrometricMatch"]) 

139 self.assertEqual(result.atmospecCentroid["centroid"], (1.0, 2.0)) 

140 

141 def testRunAstrometryRaisesFallsBack(self): 

142 task = self.makeTask() 

143 exp = makeExposureWithWcs() 

144 originalWcs = exp.getWcs() 

145 

146 task.astrometry = mock.MagicMock() 

147 task.astrometry.run.side_effect = TaskError("no solution") 

148 

149 peek = mock.MagicMock(spec=PeekExposureTask) 

150 peek.run.return_value = pipeBase.Struct(brightestCentroid=geom.Point2D(9.0, 10.0)) 

151 task.centroidingFallbackTask = peek 

152 

153 result = task.run(inputExp=exp, inputSources=mock.MagicMock()) 

154 self.assertFalse(result.atmospecCentroid["astrometricMatch"]) 

155 self.assertEqual(result.atmospecCentroid["centroid"], (9.0, 10.0)) 

156 # The WCS should have been restored after the failure. 

157 self.assertIsNotNone(exp.getWcs()) 

158 self.assertEqual(exp.getWcs(), originalWcs) 

159 

160 def testRunFallbackTaskQuickFrameMeasurement(self): 

161 task = self.makeTask() 

162 qfm = mock.MagicMock(spec=QuickFrameMeasurementTask) 

163 qfm.run.return_value = pipeBase.Struct(brightestObjCentroid=(11.0, 12.0)) 

164 task.centroidingFallbackTask = qfm 

165 

166 centroid = task.runFallbackTask(makeExposureWithWcs()) 

167 self.assertEqual(centroid, (11.0, 12.0)) 

168 

169 def testRunFallbackTaskPeekExposure(self): 

170 task = self.makeTask() 

171 peek = mock.MagicMock(spec=PeekExposureTask) 

172 peek.run.return_value = pipeBase.Struct(brightestCentroid=geom.Point2D(13.0, 14.0)) 

173 task.centroidingFallbackTask = peek 

174 

175 centroid = task.runFallbackTask(makeExposureWithWcs()) 

176 self.assertEqual(centroid, (13.0, 14.0)) 

177 

178 def testRunFallbackTaskUnsupportedRaises(self): 

179 task = self.makeTask() 

180 # A mock that is neither QFM nor PeekExposureTask. 

181 task.centroidingFallbackTask = mock.MagicMock() 

182 with self.assertRaises(ValueError): 

183 task.runFallbackTask(makeExposureWithWcs()) 

184 

185 def testRunQuantum(self): 

186 task = self.makeTask() 

187 task.astrometry = mock.MagicMock() 

188 

189 exp = makeExposureWithWcs() 

190 sources = mock.MagicMock() 

191 

192 butlerQC = mock.MagicMock() 

193 butlerQC.get.return_value = {"inputExp": exp, 

194 "inputSources": sources, 

195 "astromRefCat": [mock.MagicMock()]} 

196 

197 inputRefs = mock.MagicMock() 

198 inputRefs.astromRefCat = [mock.MagicMock()] 

199 outputRefs = mock.MagicMock() 

200 

201 fakeOutputs = pipeBase.Struct(atmospecCentroid={"centroid": (1.0, 2.0), 

202 "astrometricMatch": True}) 

203 

204 with mock.patch("lsst.atmospec.centroiding.ReferenceObjectLoader") as mockLoader, \ 

205 mock.patch.object(task, "run", return_value=fakeOutputs) as mockRun: 

206 task.runQuantum(butlerQC, inputRefs, outputRefs) 

207 

208 mockLoader.assert_called_once() 

209 task.astrometry.setRefObjLoader.assert_called_once() 

210 mockRun.assert_called_once_with(inputExp=exp, inputSources=sources) 

211 butlerQC.put.assert_called_once_with(fakeOutputs, outputRefs) 

212 

213 

214class TestMemory(lsst.utils.tests.MemoryTestCase): 

215 pass 

216 

217 

218def setup_module(module): 

219 lsst.utils.tests.init() 

220 

221 

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

223 lsst.utils.tests.init() 

224 unittest.main()