Coverage for python / lsst / analysis / tools / atools / photometricRepeatability.py: 19%

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 "StellarPhotometricRepeatability", 

25 "StellarPhotometricResidualsFocalPlane", 

26) 

27 

28from lsst.pex.config import Field 

29 

30from ..actions.plot import FocalPlanePlot, HistPanel, HistPlot, HistStatsPanel 

31from ..actions.scalar.scalarActions import ( 

32 CountAction, 

33 FracThreshold, 

34 MedianAction, 

35 SigmaMadAction, 

36 StdevAction, 

37) 

38from ..actions.vector import ( 

39 BandSelector, 

40 CalcSn, 

41 ConvertFluxToMag, 

42 LoadVector, 

43 MultiCriteriaDownselectVector, 

44 PerGroupStatistic, 

45 RangeSelector, 

46 ResidualWithPerGroupStatistic, 

47 SnSelector, 

48 ThresholdSelector, 

49) 

50from ..interfaces import AnalysisTool 

51 

52 

53class StellarPhotometricRepeatability(AnalysisTool): 

54 """Compute photometric repeatability from multiple measurements of a set of 

55 stars. First, a set of per-source quality criteria are applied. Second, 

56 the individual source measurements are grouped together by object index 

57 and per-group quantities are computed (e.g., a representative S/N for the 

58 group based on the median of associated per-source measurements). Third, 

59 additional per-group criteria are applied. Fourth, summary statistics are 

60 computed for the filtered groups. 

61 """ 

62 

63 fluxType = Field[str](doc="Flux type to calculate repeatability with", default="psfFlux") 

64 PA2Value = Field[float]( 

65 doc="Used to compute the percent of individual measurements that deviate by more than PA2Value" 

66 "from the mean of each measurement (PF1). Units of PA2Value are mmag.", 

67 default=15.0, 

68 ) 

69 

70 def setDefaults(self): 

71 super().setDefaults() 

72 

73 # Apply per-source selection criteria 

74 self.prep.selectors.bandSelector = BandSelector() 

75 self.prep.selectors.snSelector = SnSelector() 

76 self.prep.selectors.snSelector.fluxType = self.fluxType 

77 self.prep.selectors.snSelector.threshold = 200 

78 

79 # Compute per-group quantities 

80 self.process.buildActions.perGroupSn = PerGroupStatistic() 

81 self.process.buildActions.perGroupSn.buildAction = CalcSn() 

82 self.process.buildActions.perGroupSn.func = "median" 

83 self.process.buildActions.perGroupExtendedness = PerGroupStatistic() 

84 self.process.buildActions.perGroupExtendedness.buildAction.vectorKey = "extendedness" 

85 self.process.buildActions.perGroupExtendedness.func = "median" 

86 self.process.buildActions.perGroupCount = PerGroupStatistic() 

87 self.process.buildActions.perGroupCount.buildAction.vectorKey = f"{self.fluxType}" 

88 self.process.buildActions.perGroupCount.func = "count" 

89 # Use mmag units 

90 self.process.buildActions.perGroupStdev = PerGroupStatistic() 

91 self.process.buildActions.perGroupStdev.buildAction = ConvertFluxToMag( 

92 vectorKey=f"{self.fluxType}", 

93 returnMillimags=True, 

94 ) 

95 self.process.buildActions.perGroupStdev.func = "std" 

96 

97 # Filter on per-group quantities 

98 self.process.filterActions.perGroupStdevFiltered = MultiCriteriaDownselectVector( 

99 vectorKey="perGroupStdev" 

100 ) 

101 self.process.filterActions.perGroupStdevFiltered.selectors.count = ThresholdSelector( 

102 vectorKey="perGroupCount", 

103 op="ge", 

104 threshold=3, 

105 ) 

106 self.process.filterActions.perGroupStdevFiltered.selectors.sn = RangeSelector( 

107 vectorKey="perGroupSn", 

108 minimum=200, 

109 ) 

110 self.process.filterActions.perGroupStdevFiltered.selectors.extendedness = ThresholdSelector( 

111 vectorKey="perGroupExtendedness", 

112 op="le", 

113 threshold=0.5, 

114 ) 

115 

116 # Compute summary statistics on filtered groups 

117 self.process.calculateActions.photRepeatStdev = MedianAction(vectorKey="perGroupStdevFiltered") 

118 self.process.calculateActions.photRepeatNsources = CountAction(vectorKey="perGroupStdevFiltered") 

119 

120 self.produce.plot = HistPlot() 

121 

122 self.produce.plot.panels["panel_rms"] = HistPanel() 

123 

124 self.produce.plot.panels["panel_rms"].statsPanel = HistStatsPanel() 

125 self.produce.plot.panels["panel_rms"].statsPanel.statsLabels = ["N", "PA1", "PF1 %"] 

126 self.produce.plot.panels["panel_rms"].statsPanel.stat1 = ["photRepeatNsources"] 

127 self.produce.plot.panels["panel_rms"].statsPanel.stat2 = ["photRepeatStdev"] 

128 self.produce.plot.panels["panel_rms"].statsPanel.stat3 = ["photRepeatOutlier"] 

129 

130 self.produce.plot.panels["panel_rms"].refRelativeToMedian = True 

131 

132 self.produce.plot.panels["panel_rms"].label = "rms (mmag)" 

133 self.produce.plot.panels["panel_rms"].hists = dict(perGroupStdevFiltered="Filtered per group rms") 

134 

135 self.produce.metric.units = { # type: ignore 

136 "photRepeatStdev": "mmag", 

137 "photRepeatOutlier": "percent", 

138 "photRepeatNsources": "ct", 

139 } 

140 

141 def finalize(self): 

142 super().finalize() 

143 self.process.buildActions.perGroupSn.buildAction.fluxType = f"{self.fluxType}" 

144 self.process.buildActions.perGroupCount.buildAction.vectorKey = f"{self.fluxType}" 

145 self.process.buildActions.perGroupStdev.buildAction = ConvertFluxToMag( 

146 vectorKey=f"{self.fluxType}", 

147 returnMillimags=True, 

148 ) 

149 self.process.calculateActions.photRepeatOutlier = FracThreshold( 

150 vectorKey="perGroupStdevFiltered", 

151 op="ge", 

152 threshold=self.PA2Value, 

153 percent=True, 

154 relative_to_median=True, 

155 ) 

156 

157 if isinstance(self.produce.plot, HistPlot): 

158 self.produce.plot.panels["panel_rms"].referenceValue = self.PA2Value 

159 

160 self.produce.metric.newNames = { 

161 "photRepeatStdev": "{band}_stellarPhotRepeatStdev", 

162 "photRepeatOutlier": "{band}_stellarPhotRepeatOutlierFraction", 

163 "photRepeatNsources": "{band}_ct", 

164 } 

165 

166 

167class StellarPhotometricResidualsFocalPlane(AnalysisTool): 

168 """Plot mean photometric residuals as a function of the position on the 

169 focal plane. 

170 

171 First, a set of per-source quality criteria are applied. Second, the 

172 individual source measurements are grouped together by object index 

173 and the per-group magnitude is computed. The residuals between the 

174 individual sources and these magnitudes are then used to construct a plot 

175 showing the mean residual as a function of the focal-plane position. 

176 """ 

177 

178 fluxType = Field[str](doc="Flux type to calculate repeatability with", default="psfFlux") 

179 

180 def setDefaults(self): 

181 super().setDefaults() 

182 

183 # Apply per-source selection criteria 

184 self.prep.selectors.bandSelector = BandSelector() 

185 self.prep.selectors.snSelector = SnSelector() 

186 self.prep.selectors.snSelector.fluxType = "psfFlux" 

187 self.prep.selectors.snSelector.threshold = 200 

188 

189 self.process.buildActions.z = ResidualWithPerGroupStatistic() 

190 self.process.buildActions.z.buildAction = ConvertFluxToMag( 

191 vectorKey=f"{self.fluxType}", 

192 returnMillimags=True, 

193 ) 

194 self.process.buildActions.z.func = "median" 

195 

196 self.process.buildActions.x = LoadVector(vectorKey="x") 

197 self.process.buildActions.y = LoadVector(vectorKey="y") 

198 

199 self.process.buildActions.detector = LoadVector(vectorKey="detector") 

200 

201 self.process.buildActions.statMask = SnSelector() 

202 self.process.buildActions.statMask.threshold = 200 

203 self.process.buildActions.statMask.fluxType = f"{self.fluxType}" 

204 

205 self.process.calculateActions.photResidTractMedian = MedianAction(vectorKey="z") 

206 self.process.calculateActions.photResidTractStdev = StdevAction(vectorKey="z") 

207 self.process.calculateActions.photResidTractSigmaMad = SigmaMadAction(vectorKey="z") 

208 

209 self.produce.plot = FocalPlanePlot() 

210 self.produce.plot.nBins = 100 

211 self.produce.plot.zAxisLabel = "Mag - Mag$_{median}$ (mmag)" 

212 

213 self.produce.metric.units = { # type: ignore 

214 "photResidTractSigmaMad": "mmag", 

215 "photResidTractStdev": "mmag", 

216 "photResidTractMedian": "mmag", 

217 } 

218 

219 def finalize(self): 

220 super().finalize() 

221 self.process.buildActions.z.buildAction = ConvertFluxToMag( 

222 vectorKey=f"{self.fluxType}", 

223 returnMillimags=True, 

224 ) 

225 self.process.buildActions.statMask.fluxType = f"{self.fluxType}" 

226 

227 self.produce.metric.newNames = { 

228 "photResidTractSigmaMad": "{band}_photResidTractSigmaMad", 

229 "photResidTractStdev": "{band}_photResidTractStdev", 

230 "photResidTractMedian": "{band}_photResidTractMedian", 

231 }