Coverage for python / lsst / cp / pipe / ptc / cpPtcFixupGainRatios.py: 48%

1# This file is part of cp_pipe. 

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 copy 

23import numpy as np 

24 

25from lsst.ip.isr import PhotonTransferCurveDataset 

26import lsst.pex.config as pexConfig 

27import lsst.pipe.base as pipeBase 

28import lsst.pipe.base.connectionTypes as cT 

29 

30from deprecated.sphinx import deprecated 

31 

32from ..utils import ampOffsetGainRatioFixup 

33 

34 

35__all__ = [ 

36 "PhotonTransferCurveFixupGainRatiosConfig", 

37 "PhotonTransferCurveFixupGainRatiosTask", 

38 "PhotonTransferCurveRenameConfig", 

39 "PhotonTransferCurveRenameTask", 

40] 

41 

42 

43# TODO DM-52883: Remove deprecated tasks. 

44@deprecated(reason="PhotonTransferCurveFixupGainRatiosTask is no longer used. " 

45 "This Task will be removed after v30.", 

46 version="v30.0", category=FutureWarning) 

47class PhotonTransferCurveFixupGainRatiosConnections( 

48 pipeBase.PipelineTaskConnections, 

49 dimensions=("instrument", "detector") 

50): 

51 exposureMetadata = cT.Input( 

52 name="cpPtcFixupGainRatiosIsrExp.metadata", 

53 doc="Input exposures for gain ratio fixup.", 

54 storageClass="PropertyList", 

55 dimensions=("instrument", "exposure", "detector"), 

56 multiple=True, 

57 ) 

58 inputPtc = cT.PrerequisiteInput( 

59 name="ptc", 

60 doc="Input PTC to modify.", 

61 storageClass="PhotonTransferCurveDataset", 

62 dimensions=("instrument", "detector"), 

63 isCalibration=True, 

64 ) 

65 outputPtc = cT.Output( 

66 name="ptcFixed", 

67 doc="Output modified PTC.", 

68 storageClass="PhotonTransferCurveDataset", 

69 dimensions=("instrument", "detector"), 

70 multiple=False, 

71 isCalibration=True, 

72 ) 

73 

74 

75# TODO DM-52883: Remove deprecated tasks. 

76@deprecated(reason="PhotonTransferCurveFixupGainRatiosTask is no longer used. " 

77 "This Task will be removed after v30.", 

78 version="v30.0", category=FutureWarning) 

79class PhotonTransferCurveFixupGainRatiosConfig( 

80 pipeBase.PipelineTaskConfig, 

81 pipelineConnections=PhotonTransferCurveFixupGainRatiosConnections, 

82): 

83 ampOffsetGainRatioMinAdu = pexConfig.Field( 

84 dtype=float, 

85 doc="Minimum number of adu to use for amp offset gain ratio fixup.", 

86 default=1000.0, 

87 ) 

88 ampOffsetGainRatioMaxAdu = pexConfig.Field( 

89 dtype=float, 

90 doc="Maximum number of adu to use for amp offset gain ratio fixup.", 

91 default=40000.0, 

92 ) 

93 

94 

95# TODO DM-52883: Remove deprecated tasks. 

96@deprecated(reason="PhotonTransferCurveFixupGainRatiosTask is no longer used. " 

97 "This Task will be removed after v30.", 

98 version="v30.0", category=FutureWarning) 

99class PhotonTransferCurveFixupGainRatiosTask(pipeBase.PipelineTask): 

100 """Task to use on-sky amp ratios to fix up gain ratios in a PTC. 

101 

102 This uses the ampOffsetGainRatioFixup with on-sky data (preferably 

103 twilight flats or similar) to update gain ratios. 

104 """ 

105 ConfigClass = PhotonTransferCurveFixupGainRatiosConfig 

106 _DefaultName = "cpPhotonTransferCurveFixupGainRatios" 

107 

108 def runQuantum(self, butlerQC, inputRefs, outputRefs): 

109 # docstring inherited. 

110 inputs = butlerQC.get(inputRefs) 

111 outputs = self.run(inputPtc=inputs["inputPtc"], exposureMetadata=inputs["exposureMetadata"]) 

112 butlerQC.put(outputs, outputRefs) 

113 

114 def run(self, *, inputPtc, exposureMetadata): 

115 """Run the gain ratio fixup task. 

116 

117 Parameters 

118 ---------- 

119 inputPtc : `lsst.ip.isr.PhotonTransferCurveDataset` 

120 Input PTC to modify. 

121 exposureMetadata: `list` [`lsst.daf.base.PropertyList`] 

122 Input exposure metadata. 

123 

124 Returns 

125 ------- 

126 results : `lsst.pipe.base.Struct` 

127 The output struct contains: 

128 

129 ``outputPtc`` 

130 The output modified ptc. 

131 """ 

132 ampNames = inputPtc.ampNames 

133 

134 # Create a set of fake partial PTC datasets. 

135 fakePtc = PhotonTransferCurveDataset( 

136 ampNames=ampNames, 

137 ptcFitType="FAKEPTC", 

138 covMatrixSide=1, 

139 covMatrixSideFullCovFit=1, 

140 ) 

141 

142 for i, metadata in enumerate(exposureMetadata): 

143 fakePartialPtc = PhotonTransferCurveDataset(ampNames=ampNames, ptcFitType="PARTIAL") 

144 

145 for ampName in ampNames: 

146 fakePartialPtc.setAmpValuesPartialDataset( 

147 ampName, 

148 inputExpIdPair=(2*i, 2*i + 1), 

149 rawExpTime=float(i), 

150 rawMean=metadata[f"LSST ISR FINAL MEDIAN {ampName}"], 

151 rawVar=metadata[f"LSST ISR FINAL STDEV {ampName}"]**2., 

152 ampOffset=metadata[f"LSST ISR AMPOFFSET PEDESTAL {ampName}"], 

153 expIdMask=True, 

154 gain=inputPtc.gainUnadjusted[ampName], 

155 noise=metadata[f"LSST ISR READNOISE {ampName}"]*inputPtc.gain[ampName], 

156 covariance=np.zeros((1, 1)), 

157 covSqrtWeights=np.zeros((1, 1)), 

158 ) 

159 

160 fakePtc.appendPartialPtc(fakePartialPtc) 

161 

162 detectorMeans = np.zeros(len(exposureMetadata)) 

163 

164 for i in range(len(detectorMeans)): 

165 arr = np.asarray([fakePtc.rawMeans[ampName][i] for ampName in ampNames]) 

166 detectorMeans[i] = np.nanmean(arr) 

167 

168 index = np.argsort(detectorMeans) 

169 fakePtc.sort(index) 

170 

171 for ampName in ampNames: 

172 fakePtc.finalMeans[ampName][:] = fakePtc.rawMeans[ampName].copy() 

173 fakePtc.finalVars[ampName][:] = fakePtc.rawVars[ampName].copy() 

174 fakePtc.gainUnadjusted[ampName] = inputPtc.gainUnadjusted[ampName] 

175 fakePtc.gain[ampName] = inputPtc.gainUnadjusted[ampName] 

176 

177 ampOffsetGainRatioFixup( 

178 fakePtc, 

179 self.config.ampOffsetGainRatioMinAdu, 

180 self.config.ampOffsetGainRatioMaxAdu, 

181 log=self.log, 

182 ) 

183 

184 outputPtc = copy.copy(inputPtc) 

185 

186 # Replace the gain (leave gainUnadjusted alone). 

187 for ampName in ampNames: 

188 outputPtc.gain[ampName] = fakePtc.gain[ampName] 

189 

190 return pipeBase.Struct( 

191 outputPtc=outputPtc, 

192 ) 

193 

194 

195# TODO DM-52883: Remove deprecated tasks. 

196@deprecated(reason="PhotonTransferCurveRenameTask is no longer used. " 

197 "This Task will be removed after v30.", 

198 version="v30.0", category=FutureWarning) 

199class PhotonTransferCurveRenameConnections( 

200 pipeBase.PipelineTaskConnections, 

201 dimensions=("instrument", "detector") 

202): 

203 inputPtc = cT.PrerequisiteInput( 

204 name="ptcFixed", 

205 doc="Input PTC to rename.", 

206 storageClass="PhotonTransferCurveDataset", 

207 dimensions=("instrument", "detector"), 

208 isCalibration=True, 

209 ) 

210 outputPtc = cT.Output( 

211 name="ptc", 

212 doc="Output PTC that has been renamed.", 

213 storageClass="PhotonTransferCurveDataset", 

214 dimensions=("instrument", "detector"), 

215 multiple=False, 

216 isCalibration=True, 

217 ) 

218 

219 

220# TODO DM-52883: Remove deprecated tasks. 

221@deprecated(reason="PhotonTransferCurveRenameTask is no longer used. " 

222 "This Task will be removed after v30.", 

223 version="v30.0", category=FutureWarning) 

224class PhotonTransferCurveRenameConfig( 

225 pipeBase.PipelineTaskConfig, 

226 pipelineConnections=PhotonTransferCurveRenameConnections, 

227): 

228 pass 

229 

230 

231# TODO DM-52883: Remove deprecated tasks. 

232@deprecated(reason="PhotonTransferCurveRenameTask is no longer used. " 

233 "This Task will be removed after v30.", 

234 version="v30.0", category=FutureWarning) 

235class PhotonTransferCurveRenameTask(pipeBase.PipelineTask): 

236 """Task to rename a ptcFixed into a ptc.""" 

237 ConfigClass = PhotonTransferCurveRenameConfig 

238 _DefaultName = "cpPhotonTransferCurveRename" 

239 

240 def runQuantum(self, butlerQC, inputRefs, outputRefs): 

241 # docstring inherited. 

242 inputs = butlerQC.get(inputRefs) 

243 

244 outputs = pipeBase.Struct(outputPtc=inputs["inputPtc"]) 

245 butlerQC.put(outputs, outputRefs) 

246 

247 def run(self): 

248 pass