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

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 ..utils import ampOffsetGainRatioFixup 

31 

32 

33__all__ = [ 

34 "PhotonTransferCurveFixupGainRatiosConfig", 

35 "PhotonTransferCurveFixupGainRatiosTask", 

36 "PhotonTransferCurveRenameConfig", 

37 "PhotonTransferCurveRenameTask", 

38] 

39 

40 

41class PhotonTransferCurveFixupGainRatiosConnections( 

42 pipeBase.PipelineTaskConnections, 

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

44): 

45 exposureMetadata = cT.Input( 

46 name="cpPtcFixupGainRatiosIsrExp.metadata", 

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

48 storageClass="PropertyList", 

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

50 multiple=True, 

51 ) 

52 inputPtc = cT.PrerequisiteInput( 

53 name="ptc", 

54 doc="Input PTC to modify.", 

55 storageClass="PhotonTransferCurveDataset", 

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

57 isCalibration=True, 

58 ) 

59 outputPtc = cT.Output( 

60 name="ptcFixed", 

61 doc="Output modified PTC.", 

62 storageClass="PhotonTransferCurveDataset", 

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

64 multiple=False, 

65 isCalibration=True, 

66 ) 

67 

68 

69class PhotonTransferCurveFixupGainRatiosConfig( 

70 pipeBase.PipelineTaskConfig, 

71 pipelineConnections=PhotonTransferCurveFixupGainRatiosConnections, 

72): 

73 ampOffsetGainRatioMinAdu = pexConfig.Field( 

74 dtype=float, 

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

76 default=1000.0, 

77 ) 

78 ampOffsetGainRatioMaxAdu = pexConfig.Field( 

79 dtype=float, 

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

81 default=40000.0, 

82 ) 

83 

84 

85class PhotonTransferCurveFixupGainRatiosTask(pipeBase.PipelineTask): 

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

87 

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

89 twilight flats or similar) to update gain ratios. 

90 """ 

91 ConfigClass = PhotonTransferCurveFixupGainRatiosConfig 

92 _DefaultName = "cpPhotonTransferCurveFixupGainRatios" 

93 

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

95 # docstring inherited. 

96 inputs = butlerQC.get(inputRefs) 

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

98 butlerQC.put(outputs, outputRefs) 

99 

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

101 """Run the gain ratio fixup task. 

102 

103 Parameters 

104 ---------- 

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

106 Input PTC to modify. 

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

108 Input exposure metadata. 

109 

110 Returns 

111 ------- 

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

113 The output struct contains: 

114 

115 ``outputPtc`` 

116 The output modified ptc. 

117 """ 

118 ampNames = inputPtc.ampNames 

119 

120 # Create a set of fake partial PTC datasets. 

121 fakePtc = PhotonTransferCurveDataset( 

122 ampNames=ampNames, 

123 ptcFitType="FAKEPTC", 

124 covMatrixSide=1, 

125 covMatrixSideFullCovFit=1, 

126 ) 

127 

128 for i, metadata in enumerate(exposureMetadata): 

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

130 

131 for ampName in ampNames: 

132 fakePartialPtc.setAmpValuesPartialDataset( 

133 ampName, 

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

135 rawExpTime=float(i), 

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

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

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

139 expIdMask=True, 

140 gain=inputPtc.gainUnadjusted[ampName], 

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

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

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

144 ) 

145 

146 fakePtc.appendPartialPtc(fakePartialPtc) 

147 

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

149 

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

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

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

153 

154 index = np.argsort(detectorMeans) 

155 fakePtc.sort(index) 

156 

157 for ampName in ampNames: 

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

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

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

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

162 

163 ampOffsetGainRatioFixup( 

164 fakePtc, 

165 self.config.ampOffsetGainRatioMinAdu, 

166 self.config.ampOffsetGainRatioMaxAdu, 

167 log=self.log, 

168 ) 

169 

170 outputPtc = copy.copy(inputPtc) 

171 

172 # Replace the gain (leave gainUnadjusted alone). 

173 for ampName in ampNames: 

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

175 

176 return pipeBase.Struct( 

177 outputPtc=outputPtc, 

178 ) 

179 

180 

181class PhotonTransferCurveRenameConnections( 

182 pipeBase.PipelineTaskConnections, 

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

184): 

185 inputPtc = cT.PrerequisiteInput( 

186 name="ptcFixed", 

187 doc="Input PTC to rename.", 

188 storageClass="PhotonTransferCurveDataset", 

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

190 isCalibration=True, 

191 ) 

192 outputPtc = cT.Output( 

193 name="ptc", 

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

195 storageClass="PhotonTransferCurveDataset", 

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

197 multiple=False, 

198 isCalibration=True, 

199 ) 

200 

201 

202class PhotonTransferCurveRenameConfig( 

203 pipeBase.PipelineTaskConfig, 

204 pipelineConnections=PhotonTransferCurveRenameConnections, 

205): 

206 pass 

207 

208 

209class PhotonTransferCurveRenameTask(pipeBase.PipelineTask): 

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

211 ConfigClass = PhotonTransferCurveRenameConfig 

212 _DefaultName = "cpPhotonTransferCurveRename" 

213 

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

215 # docstring inherited. 

216 inputs = butlerQC.get(inputRefs) 

217 

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

219 butlerQC.put(outputs, outputRefs) 

220 

221 def run(self): 

222 pass