Coverage for python / lsst / analysis / tools / atools / diaFakeMetrics.py: 57%

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__ = ( 

24 "FractionFoundFakesDiaSnrMetric", 

25 "FractionFoundFakesDiaMagMetric", 

26 "FractionFoundFakesAssocDiaSnrMetric", 

27 "FractionFoundFakesAssocDiaMagMetric", 

28) 

29 

30import numpy as np 

31from lsst.pex.config import Field, ListField 

32 

33from ..actions.scalar import CountAction, FracThreshold 

34from ..actions.vector import FlagSelector, MultiCriteriaDownselectVector, RangeSelector 

35from ..interfaces import AnalysisTool 

36 

37 

38class FractionFoundFakesDiaSnrMetric(AnalysisTool): 

39 """Calculate the fraction of fake DIA Sources found within the 

40 given SNR range""" 

41 

42 parameterizedBand: bool = False 

43 

44 snrMin = Field[float](doc="Minimum SNR for fake sources metric calculation.", default=0) 

45 

46 snrMax = Field[float](doc="Maximum SNR for fake sources metric calculation.", default=np.inf) 

47 

48 fluxType = Field[str]( 

49 "Flux type for fake sources metric calculation.", default="forced_base_PsfFlux_instFlux" 

50 ) 

51 

52 fakeFlagsWhenTrue = ListField[str]( 

53 "Flags for fake source cleaning before metrics calculation. Select sources when flags are true.", 

54 default=[], 

55 ) 

56 

57 fakeFlagsWhenFalse = ListField[str]( 

58 "Flags for fake source cleaning before metrics calculation. Select sources when flags are false.", 

59 default=[ 

60 "forced_base_PixelFlags_flag_bad", 

61 "forced_base_LocalBackground_flag", 

62 "forced_base_PixelFlags_flag_interpolated", 

63 "forced_base_PixelFlags_flag_edgeCenter", 

64 ], 

65 ) 

66 

67 def setDefaults(self): 

68 super().setDefaults() 

69 

70 def finalize(self): 

71 # There is no need to calculate the SNR as it is already estimated 

72 # Select the fake sources using their SNR values for the given range 

73 # and flux type estimation 

74 self.process.filterActions.fakeSourcesDiaSrcId = MultiCriteriaDownselectVector( 

75 vectorKey="diaSourceId" 

76 ) 

77 

78 self.process.filterActions.fakeSourcesDiaSrcId.selectors.snrange = RangeSelector( 

79 vectorKey=f"{self.fluxType}_SNR", maximum=self.snrMax, minimum=self.snrMin 

80 ) 

81 

82 self.process.filterActions.fakeSourcesDiaSrcId.selectors.fakeFlags = FlagSelector( 

83 selectWhenFalse=self.fakeFlagsWhenFalse, selectWhenTrue=self.fakeFlagsWhenTrue 

84 ) 

85 

86 self.process.calculateActions.numTotalFakeSources = CountAction(vectorKey="fakeSourcesDiaSrcId") 

87 

88 self.process.calculateActions.numFoundFakeSources = CountAction( 

89 vectorKey="fakeSourcesDiaSrcId", op="gt", threshold=0 

90 ) 

91 

92 self.process.calculateActions.fractionFoundFakesDiaAll = FracThreshold( 

93 vectorKey="fakeSourcesDiaSrcId", op="gt", threshold=0 

94 ) 

95 

96 # the units for the quantity (count, an astropy quantity) 

97 self.produce.metric.units = { 

98 "numTotalFakeSources": "ct", 

99 "numFoundFakeSources": "ct", 

100 "fractionFoundFakesDiaAll": "", 

101 } 

102 

103 

104class FractionFoundFakesDiaMagMetric(AnalysisTool): 

105 """Calculate the fraction of fake DIA Sources found within the given 

106 magnitude range""" 

107 

108 parameterizedBand: bool = False 

109 

110 magMin = Field[float](doc="Minimum magnitude for fake sources metric calculation.", default=18) 

111 

112 magMax = Field[float](doc="Maximum magnitude for fake sources metric calculation.", default=22) 

113 

114 fakeFlagsWhenTrue = ListField[str]( 

115 "Flags for fake source cleaning before metrics calculation.. Select sources when flags are true.", 

116 default=[], 

117 ) 

118 

119 fakeFlagsWhenFalse = ListField[str]( 

120 "Flags for fake source cleaning before metrics calculation. Select sources when flags are false.", 

121 default=[ 

122 "forced_base_PixelFlags_flag_interpolated", 

123 "forced_base_LocalBackground_flag", 

124 "forced_base_PixelFlags_flag_bad", 

125 "forced_base_PixelFlags_flag_edgeCenter", 

126 ], 

127 ) 

128 

129 def finalize(self): 

130 # Selecting the fake sources using the truth magnitude values. 

131 self.process.filterActions.fakeSourcesDiaSrcId = MultiCriteriaDownselectVector( 

132 vectorKey="diaSourceId" 

133 ) 

134 

135 self.process.filterActions.fakeSourcesDiaSrcId.selectors.magrange = RangeSelector( 

136 vectorKey="mag", maximum=self.magMax, minimum=self.magMin 

137 ) 

138 

139 self.process.filterActions.fakeSourcesDiaSrcId.selectors.fakeFlags = FlagSelector( 

140 selectWhenFalse=self.fakeFlagsWhenFalse, selectWhenTrue=self.fakeFlagsWhenTrue 

141 ) 

142 

143 self.process.calculateActions.numTotalFakeSources = CountAction(vectorKey="fakeSourcesDiaSrcId") 

144 

145 self.process.calculateActions.numFoundFakeSources = CountAction( 

146 vectorKey="fakeSourcesDiaSrcId", op="gt", threshold=0 

147 ) 

148 

149 self.process.calculateActions.fractionFoundFakesDiaAll = FracThreshold( 

150 vectorKey="fakeSourcesDiaSrcId", op="gt", threshold=0 

151 ) 

152 

153 # the units for the quantity (count, an astropy quantity) 

154 self.produce.metric.units = { 

155 "numTotalFakeSources": "ct", 

156 "numFoundFakeSources": "ct", 

157 "fractionFoundFakesDiaAll": "", 

158 } 

159 

160 

161class FractionFoundFakesAssocDiaSnrMetric(AnalysisTool): 

162 """Calculate the fraction of fake DIA Sources found within the 

163 given SNR range""" 

164 

165 parameterizedBand: bool = False 

166 

167 snrMin = Field[float](doc="Minimum SNR for fake sources metric calculation.", default=0) 

168 

169 snrMax = Field[float](doc="Maximum SNR for fake sources metric calculation.", default=np.inf) 

170 

171 fluxType = Field[str]( 

172 "Flux type for fake sources metric calculation.", default="forced_base_PsfFlux_instFlux" 

173 ) 

174 

175 fakeFlagsWhenTrue = ListField[str]( 

176 "Flags for fake source cleaning before metrics calculation. Select sources when flags are true.", 

177 default=[], 

178 ) 

179 

180 fakeFlagsWhenFalse = ListField[str]( 

181 "Flags for fake source cleaning before metrics calculation. Select sources when flags are false.", 

182 default=[ 

183 "forced_base_PixelFlags_flag_bad", 

184 "forced_base_LocalBackground_flag", 

185 "forced_base_PixelFlags_flag_interpolated", 

186 "forced_base_PixelFlags_flag_edgeCenter", 

187 ], 

188 ) 

189 

190 def setDefaults(self): 

191 super().setDefaults() 

192 

193 def finalize(self): 

194 # There is no need to calculate the SNR as it is already estimated 

195 # Select the fake sources using their SNR values for the given range 

196 # and flux type estimation 

197 

198 self.process.filterActions.fakeSourcesAssocDiaSrcId = MultiCriteriaDownselectVector( 

199 vectorKey="isAssocDiaSource" 

200 ) 

201 

202 self.process.filterActions.fakeSourcesAssocDiaSrcId.selectors.snrange = RangeSelector( 

203 vectorKey=f"{self.fluxType}_SNR", maximum=self.snrMax, minimum=self.snrMin 

204 ) 

205 

206 self.process.filterActions.fakeSourcesAssocDiaSrcId.selectors.fakeFlags = FlagSelector( 

207 selectWhenFalse=self.fakeFlagsWhenFalse, selectWhenTrue=self.fakeFlagsWhenTrue 

208 ) 

209 

210 self.process.calculateActions.numTotalFakeAssocSources = CountAction( 

211 vectorKey="fakeSourcesAssocDiaSrcId" 

212 ) 

213 

214 self.process.calculateActions.numFoundFakeAssocSources = CountAction( 

215 vectorKey="fakeSourcesAssocDiaSrcId", op="gt", threshold=0 

216 ) 

217 

218 self.process.calculateActions.fractionFoundFakesAssocDiaAll = FracThreshold( 

219 vectorKey="fakeSourcesAssocDiaSrcId", op="gt", threshold=0 

220 ) 

221 

222 # the units for the quantity (count, an astropy quantity) 

223 self.produce.metric.units = { 

224 "numTotalFakeAssocSources": "ct", 

225 "numFoundFakeAssocSources": "ct", 

226 "fractionFoundFakesAssocDiaAll": "", 

227 } 

228 

229 

230class FractionFoundFakesAssocDiaMagMetric(AnalysisTool): 

231 """Calculate the fraction of fake sources found in AssocDiasrcs within 

232 the given magnitude range""" 

233 

234 parameterizedBand: bool = False 

235 

236 magMin = Field[float](doc="Minimum magnitude for fake sources metric calculation.", default=18) 

237 

238 magMax = Field[float](doc="Maximum magnitude for fake sources metric calculation.", default=22) 

239 

240 fakeFlagsWhenTrue = ListField[str]( 

241 "Flags for fake source cleaning before metrics calculation. Select sources when flags are true.", 

242 default=[], 

243 ) 

244 

245 fakeFlagsWhenFalse = ListField[str]( 

246 "Flags for fake source cleaning before metrics calculation. Select sources when flags are false.", 

247 default=[ 

248 "forced_base_PixelFlags_flag_interpolated", 

249 "forced_base_LocalBackground_flag", 

250 "forced_base_PixelFlags_flag_bad", 

251 "forced_base_PixelFlags_flag_edgeCenter", 

252 ], 

253 ) 

254 

255 def finalize(self): 

256 # Selecting the fake sources using the truth magnitude values. 

257 self.process.filterActions.fakeSourcesAssocDiaSrcId = MultiCriteriaDownselectVector( 

258 vectorKey="isAssocDiaSource" 

259 ) 

260 

261 self.process.filterActions.fakeSourcesAssocDiaSrcId.selectors.magrange = RangeSelector( 

262 vectorKey="mag", maximum=self.magMax, minimum=self.magMin 

263 ) 

264 

265 self.process.filterActions.fakeSourcesAssocDiaSrcId.selectors.fakeFlags = FlagSelector( 

266 selectWhenFalse=self.fakeFlagsWhenFalse, selectWhenTrue=self.fakeFlagsWhenTrue 

267 ) 

268 

269 self.process.calculateActions.numTotalFakeAssocSources = CountAction( 

270 vectorKey="fakeSourcesAssocDiaSrcId" 

271 ) 

272 

273 self.process.calculateActions.numFoundFakeAssocSources = CountAction( 

274 vectorKey="fakeSourcesAssocDiaSrcId", op="gt", threshold=0 

275 ) 

276 

277 self.process.calculateActions.fractionFoundFakesAssocDiaAll = FracThreshold( 

278 vectorKey="fakeSourcesAssocDiaSrcId", op="gt", threshold=0 

279 ) 

280 

281 # the units for the quantity (count, an astropy quantity) 

282 self.produce.metric.units = { 

283 "numTotalFakeAssocSources": "ct", 

284 "numFoundFakeAssocSources": "ct", 

285 "fractionFoundFakesAssocDiaAll": "", 

286 }