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 

31 

32from lsst.pex.config import Field, ListField 

33 

34from ..actions.scalar import CountAction, FracThreshold 

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

36from ..interfaces import AnalysisTool 

37 

38 

39class FractionFoundFakesDiaSnrMetric(AnalysisTool): 

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

41 given SNR range""" 

42 

43 parameterizedBand: bool = False 

44 

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

46 

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

48 

49 fluxType = Field[str]( 

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

51 ) 

52 

53 fakeFlagsWhenTrue = ListField[str]( 

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

55 default=[], 

56 ) 

57 

58 fakeFlagsWhenFalse = ListField[str]( 

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

60 default=[ 

61 "forced_base_PixelFlags_flag_bad", 

62 "forced_base_LocalBackground_flag", 

63 "forced_base_PixelFlags_flag_interpolated", 

64 "forced_base_PixelFlags_flag_edgeCenter", 

65 ], 

66 ) 

67 

68 def setDefaults(self): 

69 super().setDefaults() 

70 

71 def finalize(self): 

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

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

74 # and flux type estimation 

75 self.process.filterActions.fakeSourcesDiaSrcId = MultiCriteriaDownselectVector( 

76 vectorKey="diaSourceId" 

77 ) 

78 

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

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

81 ) 

82 

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

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

85 ) 

86 

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

88 

89 self.process.calculateActions.numFoundFakeSources = CountAction( 

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

91 ) 

92 

93 self.process.calculateActions.fractionFoundFakesDiaAll = FracThreshold( 

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

95 ) 

96 

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

98 self.produce.metric.units = { 

99 "numTotalFakeSources": "ct", 

100 "numFoundFakeSources": "ct", 

101 "fractionFoundFakesDiaAll": "", 

102 } 

103 

104 

105class FractionFoundFakesDiaMagMetric(AnalysisTool): 

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

107 magnitude range""" 

108 

109 parameterizedBand: bool = False 

110 

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

112 

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

114 

115 fakeFlagsWhenTrue = ListField[str]( 

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

117 default=[], 

118 ) 

119 

120 fakeFlagsWhenFalse = ListField[str]( 

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

122 default=[ 

123 "forced_base_PixelFlags_flag_interpolated", 

124 "forced_base_LocalBackground_flag", 

125 "forced_base_PixelFlags_flag_bad", 

126 "forced_base_PixelFlags_flag_edgeCenter", 

127 ], 

128 ) 

129 

130 def finalize(self): 

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

132 self.process.filterActions.fakeSourcesDiaSrcId = MultiCriteriaDownselectVector( 

133 vectorKey="diaSourceId" 

134 ) 

135 

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

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

138 ) 

139 

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

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

142 ) 

143 

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

145 

146 self.process.calculateActions.numFoundFakeSources = CountAction( 

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

148 ) 

149 

150 self.process.calculateActions.fractionFoundFakesDiaAll = FracThreshold( 

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

152 ) 

153 

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

155 self.produce.metric.units = { 

156 "numTotalFakeSources": "ct", 

157 "numFoundFakeSources": "ct", 

158 "fractionFoundFakesDiaAll": "", 

159 } 

160 

161 

162class FractionFoundFakesAssocDiaSnrMetric(AnalysisTool): 

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

164 given SNR range""" 

165 

166 parameterizedBand: bool = False 

167 

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

169 

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

171 

172 fluxType = Field[str]( 

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

174 ) 

175 

176 fakeFlagsWhenTrue = ListField[str]( 

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

178 default=[], 

179 ) 

180 

181 fakeFlagsWhenFalse = ListField[str]( 

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

183 default=[ 

184 "forced_base_PixelFlags_flag_bad", 

185 "forced_base_LocalBackground_flag", 

186 "forced_base_PixelFlags_flag_interpolated", 

187 "forced_base_PixelFlags_flag_edgeCenter", 

188 ], 

189 ) 

190 

191 def setDefaults(self): 

192 super().setDefaults() 

193 

194 def finalize(self): 

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

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

197 # and flux type estimation 

198 

199 self.process.filterActions.fakeSourcesAssocDiaSrcId = MultiCriteriaDownselectVector( 

200 vectorKey="isAssocDiaSource" 

201 ) 

202 

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

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

205 ) 

206 

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

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

209 ) 

210 

211 self.process.calculateActions.numTotalFakeAssocSources = CountAction( 

212 vectorKey="fakeSourcesAssocDiaSrcId" 

213 ) 

214 

215 self.process.calculateActions.numFoundFakeAssocSources = CountAction( 

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

217 ) 

218 

219 self.process.calculateActions.fractionFoundFakesAssocDiaAll = FracThreshold( 

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

221 ) 

222 

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

224 self.produce.metric.units = { 

225 "numTotalFakeAssocSources": "ct", 

226 "numFoundFakeAssocSources": "ct", 

227 "fractionFoundFakesAssocDiaAll": "", 

228 } 

229 

230 

231class FractionFoundFakesAssocDiaMagMetric(AnalysisTool): 

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

233 the given magnitude range""" 

234 

235 parameterizedBand: bool = False 

236 

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

238 

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

240 

241 fakeFlagsWhenTrue = ListField[str]( 

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

243 default=[], 

244 ) 

245 

246 fakeFlagsWhenFalse = ListField[str]( 

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

248 default=[ 

249 "forced_base_PixelFlags_flag_interpolated", 

250 "forced_base_LocalBackground_flag", 

251 "forced_base_PixelFlags_flag_bad", 

252 "forced_base_PixelFlags_flag_edgeCenter", 

253 ], 

254 ) 

255 

256 def finalize(self): 

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

258 self.process.filterActions.fakeSourcesAssocDiaSrcId = MultiCriteriaDownselectVector( 

259 vectorKey="isAssocDiaSource" 

260 ) 

261 

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

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

264 ) 

265 

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

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

268 ) 

269 

270 self.process.calculateActions.numTotalFakeAssocSources = CountAction( 

271 vectorKey="fakeSourcesAssocDiaSrcId" 

272 ) 

273 

274 self.process.calculateActions.numFoundFakeAssocSources = CountAction( 

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

276 ) 

277 

278 self.process.calculateActions.fractionFoundFakesAssocDiaAll = FracThreshold( 

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

280 ) 

281 

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

283 self.produce.metric.units = { 

284 "numTotalFakeAssocSources": "ct", 

285 "numFoundFakeAssocSources": "ct", 

286 "fractionFoundFakesAssocDiaAll": "", 

287 }