Coverage for python / lsst / analysis / tools / atools / deblenderMetric.py: 25%

1# This file is part of analysis_tools. 

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

21from __future__ import annotations 

22 

23__all__ = ("ParentDeblenderMetrics", "SkippedDeblenderMetrics", "BlendMetrics", "IsolatedDeblenderMetrics") 

24 

25from ..actions.scalar.scalarActions import CountAction, DivideScalar, MeanAction, SumAction 

26from ..actions.vector.mathActions import AddVector, SubtractVector 

27from ..actions.vector.selectors import ( 

28 ChildObjectSelector, 

29 FlagSelector, 

30 ParentObjectSelector, 

31 ThresholdSelector, 

32) 

33from ..actions.vector.vectorActions import LoadVector 

34from ..interfaces import AnalysisTool 

35 

36 

37class ParentDeblenderMetrics(AnalysisTool): 

38 """Calculate metrics based on the performance of the deblender""" 

39 

40 def setDefaults(self): 

41 super().setDefaults() 

42 

43 # Only select parents 

44 self.prep.selectors.parentSelector = ParentObjectSelector() 

45 

46 # Subtract the number of children+isolated from the number of peaks 

47 # to get the number of peakDropouts. Ideally, this should be zero. 

48 self.process.buildActions.peakDropouts = SubtractVector() 

49 self.process.buildActions.peakDropouts.actionA = LoadVector(vectorKey="deblend_nPeaks") 

50 self.process.buildActions.peakDropouts.actionB = AddVector( 

51 actionA=LoadVector(vectorKey="deblend_nChild"), 

52 actionB=LoadVector(vectorKey="deblend_skipped_isolatedParent"), 

53 ) 

54 

55 # Statistics for parent blends 

56 self.process.calculateActions.numParents = CountAction(vectorKey="parentObjectId") 

57 self.process.calculateActions.numDeblendFailed = SumAction(vectorKey="deblend_failed") 

58 self.process.calculateActions.numIncompleteData = SumAction(vectorKey="deblend_incompleteData") 

59 

60 # Total number of detected peaks 

61 self.process.calculateActions.numDetectedPeaks = SumAction(vectorKey="deblend_nPeaks") 

62 

63 # Total number of deblended children 

64 self.process.calculateActions.numDeblendedChildren = SumAction(vectorKey="deblend_nChild") 

65 

66 # Total number of peak dropouts 

67 self.process.calculateActions.numPeakDropouts = SumAction(vectorKey="peakDropouts") 

68 

69 # Total number of peak dropouts as proportion of total number of peaks 

70 self.process.calculateActions.propPeakDropouts = DivideScalar() 

71 self.process.calculateActions.propPeakDropouts.actionA = SumAction(vectorKey="peakDropouts") 

72 self.process.calculateActions.propPeakDropouts.actionB = SumAction(vectorKey="deblend_nPeaks") 

73 

74 self.produce.metric.units = { 

75 "numParents": "", 

76 "numDeblendFailed": "", 

77 "numIncompleteData": "", 

78 "numDetectedPeaks": "", 

79 "numDeblendedChildren": "", 

80 "numPeakDropouts": "", 

81 "propPeakDropouts": "", 

82 } 

83 

84 

85class SkippedDeblenderMetrics(AnalysisTool): 

86 """Calculate metrics based on blends skipped by the deblender""" 

87 

88 def setDefaults(self): 

89 super().setDefaults() 

90 

91 # Only select non-sky object parents that were skipped but did not fail 

92 # This also excludes isolated objects that were skipped 

93 # if isolated objects are not being deblended 

94 self.prep.selectors.parentSelector = ParentObjectSelector() 

95 self.prep.selectors.skippedSelector = FlagSelector() 

96 self.prep.selectors.skippedSelector.selectWhenTrue = ["deblend_skipped"] 

97 self.prep.selectors.skippedSelector.selectWhenFalse = [ 

98 "deblend_failed", 

99 "deblend_skipped_isolatedParent", 

100 ] 

101 

102 # Statistics for skipped blends 

103 self.process.calculateActions.numSkippedBlends = CountAction(vectorKey="parentObjectId") 

104 self.process.calculateActions.numBlendParentTooBig = SumAction( 

105 vectorKey="deblend_skipped_parentTooBig" 

106 ) 

107 self.process.calculateActions.numBlendTooManyPeaks = SumAction( 

108 vectorKey="deblend_skipped_tooManyPeaks" 

109 ) 

110 self.process.calculateActions.numBlendTooManyMasked = SumAction(vectorKey="deblend_skipped_masked") 

111 

112 # Total number of skipped peaks 

113 self.process.calculateActions.numSkippedPeaks = SumAction(vectorKey="deblend_nPeaks") 

114 

115 self.produce.metric.units = { 

116 "numSkippedBlends": "", 

117 "numBlendParentTooBig": "", 

118 "numBlendTooManyPeaks": "", 

119 "numBlendTooManyMasked": "", 

120 "numSkippedPeaks": "", 

121 } 

122 

123 

124class BlendMetrics(AnalysisTool): 

125 """Calculate metrics based on the performance of the deblender for blends 

126 with multiple children 

127 """ 

128 

129 def setDefaults(self): 

130 super().setDefaults() 

131 

132 # Only select parents that were successfully deblended 

133 # with more than one child 

134 self.prep.selectors.parentSelector = ParentObjectSelector() 

135 self.prep.selectors.blendSelector = ThresholdSelector() 

136 self.prep.selectors.blendSelector.vectorKey = "deblend_nChild" 

137 self.prep.selectors.blendSelector.op = "gt" 

138 self.prep.selectors.blendSelector.threshold = 1 

139 

140 # Statistics for blended parents 

141 self.process.calculateActions.numBlends = CountAction(vectorKey="parentObjectId") 

142 self.process.calculateActions.meanBlendIterations = MeanAction(vectorKey="deblend_iterations") 

143 self.process.calculateActions.meanBlendChi2 = MeanAction(vectorKey="deblend_chi2") 

144 

145 self.produce.metric.units = { 

146 "numBlends": "", 

147 "meanBlendIterations": "", 

148 "meanBlendChi2": "", 

149 } 

150 

151 

152class IsolatedDeblenderMetrics(AnalysisTool): 

153 """Calculate metrics based on the performance of the deblender for 

154 parents with only a single child peak. 

155 """ 

156 

157 def setDefaults(self): 

158 super().setDefaults() 

159 

160 # Only select parents that were successfully deblended with one child 

161 self.prep.selectors.parentSelector = ParentObjectSelector() 

162 self.prep.selectors.blendSelector = ThresholdSelector() 

163 self.prep.selectors.blendSelector.vectorKey = "deblend_nChild" 

164 self.prep.selectors.blendSelector.op = "eq" 

165 self.prep.selectors.blendSelector.threshold = 1 

166 

167 # Statistics for isolated parent scarlet_lite models 

168 self.process.calculateActions.numIsolated = CountAction(vectorKey="parentObjectId") 

169 self.process.calculateActions.meanIsolatedIterations = MeanAction(vectorKey="deblend_iterations") 

170 self.process.calculateActions.meanIsolatedChi2 = MeanAction(vectorKey="deblend_chi2") 

171 

172 self.produce.metric.units = { 

173 "numIsolated": "", 

174 "meanIsolatedIterations": "", 

175 "meanIsolatedChi2": "", 

176 } 

177 

178 

179class ChildDeblenderMetrics(AnalysisTool): 

180 """Calculate metrics based on the performance of the deblender for 

181 single sources. 

182 """ 

183 

184 def setDefaults(self): 

185 super().setDefaults() 

186 self.prep.selectors.childSelector = ChildObjectSelector() 

187 self.process.calculateActions.zeroFlux = SumAction(vectorKey="deblend_zeroFlux") 

188 

189 self.produce.metric.units = { 

190 "zeroFlux": "", 

191 }