Coverage for tests/test_NoiseReplacer.py: 98%

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/>. 

21 

22import unittest 

23import unittest.mock 

24 

25import numpy as np 

26 

27import lsst.geom 

28import lsst.afw.geom 

29import lsst.afw.detection 

30import lsst.afw.image 

31import lsst.afw.table 

32import lsst.meas.base.tests 

33import lsst.utils.tests 

34from lsst.daf.base import PropertyList 

35from lsst.utils.tests import methodParameters 

36 

37 

38@lsst.meas.base.register("test_NoiseReplacer") 

39class NoiseReplacerTestPlugin(lsst.meas.base.SingleFramePlugin): 

40 """Plugin that sums instFlux inside and outside the source footprint. 

41 """ 

42 

43 @staticmethod 

44 def getExecutionOrder(): 

45 return 2.0 

46 

47 def __init__(self, config, name, schema, metadata): 

48 lsst.meas.base.SingleFramePlugin.__init__(self, config, name, schema, metadata) 

49 self.insideKey = schema.addField("%s_inside" % (name,), type=np.float64, 

50 doc="instFlux inside footprint") 

51 self.outsideKey = schema.addField("%s_outside" % (name,), type=np.float64, 

52 doc="instFlux outside footprint") 

53 

54 def measure(self, measRecord, exposure): 

55 footprint = measRecord.getFootprint() 

56 fullArray = exposure.image.array 

57 insideArray = np.zeros(footprint.getArea(), dtype=fullArray.dtype) 

58 footprint.spans.flatten(insideArray, fullArray, exposure.getXY0()) 

59 insideFlux = float(insideArray.sum()) 

60 outsideFlux = float(fullArray.sum()) - insideFlux 

61 measRecord.set(self.insideKey, insideFlux) 

62 measRecord.set(self.outsideKey, outsideFlux) 

63 

64 

65class NoiseReplacerTestCase(lsst.meas.base.tests.AlgorithmTestCase, lsst.utils.tests.TestCase): 

66 

67 def setUp(self): 

68 self.bbox = lsst.geom.Box2I(lsst.geom.Point2I(-20, -30), 

69 lsst.geom.Extent2I(240, 260)) 

70 self.dataset = lsst.meas.base.tests.TestDataset(self.bbox) 

71 # first source is a point 

72 self.dataset.addSource(100000.0, lsst.geom.Point2D(50.1, 49.8)) 

73 # second source is extended 

74 self.dataset.addSource(120000.0, lsst.geom.Point2D(149.9, 50.3), 

75 lsst.afw.geom.Quadrupole(8, 9, 3)) 

76 with self.dataset.addBlend() as family: 

77 family.addChild(110000.0, lsst.geom.Point2D(65.2, 150.7), 

78 lsst.afw.geom.Quadrupole(7, 5, -1)) 

79 family.addChild(140000.0, lsst.geom.Point2D(72.3, 149.1)) 

80 family.addChild(90000.0, lsst.geom.Point2D(68.5, 156.9)) 

81 

82 @methodParameters(noiseSource=['measure', 'variance', 'meta', 'variance_median'], 

83 variance=[1.0, 1.0, 2.0, 1.0]) 

84 def testNoiseReplacer(self, noiseSource, variance): 

85 """Test noise replacement in SFM with ''measure'' mode. 

86 

87 We compare the instFlux inside and outside source Footprints on an 

88 extremely high S/N image. 

89 """ 

90 # We choose a random seed which causes the test to pass. 

91 task = self.makeSingleFrameMeasurementTask("test_NoiseReplacer") 

92 task.config.noiseReplacer.noiseSource = noiseSource 

93 exposure, catalog = self.dataset.realize(variance, task.schema, randomSeed=0) 

94 if noiseSource == 'meta': 

95 md = PropertyList() 

96 md['BGMEAN'] = variance 

97 exposure.setMetadata(md) 

98 task.run(catalog, exposure) 

99 sumVariance = exposure.variance.array.sum() 

100 for record in catalog: 

101 self.assertFloatsAlmostEqual(record.get("test_NoiseReplacer_inside"), 

102 record.get("truth_instFlux"), rtol=1E-3) 

103 

104 # N.B. Next line checks that a random value is correct to a 

105 # statistical 1-sigma prediction; some RNG seeds may cause it to 

106 # fail (indeed, 67% should) 

107 self.assertLess(record.get("test_NoiseReplacer_outside"), np.sqrt(sumVariance)) 

108 

109 @methodParameters(noiseSource=['measure', 'variance', 'meta', 'variance_median'], 

110 variance=[1.0, 1.0, 2.0, 1.0]) 

111 def testNoiseReplacerExternalFootprints(self, noiseSource, variance): 

112 # We choose a random seed which causes the test to pass. 

113 task = self.makeSingleFrameMeasurementTask("test_NoiseReplacer") 

114 task.config.noiseReplacer.noiseSource = noiseSource 

115 exposure, catalog = self.dataset.realize(variance, task.schema, randomSeed=0) 

116 if noiseSource == 'meta': 

117 md = PropertyList() 

118 md['BGMEAN'] = variance 

119 exposure.setMetadata(md) 

120 footprints = task.getFootprintsFromCatalog(catalog) 

121 task.run(catalog, exposure, footprints=footprints) 

122 sumVariance = exposure.variance.array.sum() 

123 for record in catalog: 

124 self.assertFloatsAlmostEqual(record.get("test_NoiseReplacer_inside"), 

125 record.get("truth_instFlux"), rtol=1E-3) 

126 

127 # N.B. Next line checks that a random value is correct to a 

128 # statistical 1-sigma prediction; some RNG seeds may cause it to 

129 # fail (indeed, 67% should) 

130 self.assertLess(record.get("test_NoiseReplacer_outside"), np.sqrt(sumVariance)) 

131 

132 def _countBits(self, mask, planeName): 

133 """Return the number of pixels with ``planeName`` set, or 0 if the 

134 plane is not registered. 

135 """ 

136 if planeName not in mask.getMaskPlaneDict(): 

137 return 0 

138 return np.count_nonzero(mask.array & mask.getPlaneBitMask(planeName)) 

139 

140 def testPreregisteredPlanes(self): 

141 """Test that measurement leaves no THISDET or OTHERDET bits set when 

142 those planes were registered before it started. 

143 """ 

144 for planeName in ["THISDET", "OTHERDET"]: 

145 lsst.afw.image.Mask.addMaskPlane(planeName) 

146 self.addCleanup(lsst.afw.image.Mask.removeMaskPlane, planeName) 

147 task = self.makeSingleFrameMeasurementTask("test_NoiseReplacer") 

148 exposure, catalog = self.dataset.realize(1.0, task.schema, randomSeed=0) 

149 task.run(catalog, exposure) 

150 for planeName in ["THISDET", "OTHERDET"]: 

151 self.assertEqual(self._countBits(exposure.mask, planeName), 0) 

152 

153 def testPlanesRemovedOnFailure(self): 

154 """Test that the temporary mask planes are removed when measurement 

155 raises. 

156 """ 

157 task = self.makeSingleFrameMeasurementTask("test_NoiseReplacer") 

158 exposure, catalog = self.dataset.realize(1.0, task.schema, randomSeed=0) 

159 error = lsst.meas.base.FatalAlgorithmError("Abort measurement.") 

160 # A registered failing plugin would break tests that run every plugin. 

161 with unittest.mock.patch.object(task, "callMeasure", side_effect=error): 

162 with self.assertRaises(lsst.meas.base.FatalAlgorithmError): 

163 task.run(catalog, exposure) 

164 for planeName in ["THISDET", "OTHERDET"]: 

165 self.assertNotIn(planeName, exposure.mask.getMaskPlaneDict()) 

166 self.assertEqual(self._countBits(exposure.mask, planeName), 0) 

167 

168 def tearDown(self): 

169 del self.bbox 

170 del self.dataset 

171 

172 

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

174 pass 

175 

176 

177def setup_module(module): 

178 lsst.utils.tests.init() 

179 

180 

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

182 lsst.utils.tests.init() 

183 unittest.main()