lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask Class Reference

Inheritance diagram for lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask:

Public Member Functions
def	runQuantum (self, butlerQC, inputRefs, outputRefs)
def	run (self, inputCovariances, camera=None, inputExpList=None)
def	fitCovariancesAstier (self, dataset)
def	getOutputPtcDataCovAstier (self, dataset, covFits, covFitsNoB)
def	fitPtc (self, dataset)
def	fillBadAmp (self, dataset, ptcFitType, ampName)

Static Public Attributes
	ConfigClass = PhotonTransferCurveSolveConfig

Detailed Description

Task to fit the PTC from flat covariances.
This task assembles the list of individual PTC datasets produced
by `PhotonTransferCurveSolveTask` into one single final PTC dataset.
The task fits the measured (co)variances to a polynomial model or to
the models described in equations 16 and 20 of Astier+19
(referred to as `POLYNOMIAL`, `EXPAPPROXIMATION`, and `FULLCOVARIANCE`
in the configuration options of the task, respectively). Parameters
of interest such as tghe gain and noise are derived from the fits.

Astier+19: "The Shape of the Photon Transfer Curve
of CCD sensors", arXiv:1905.08677

Definition at line 158 of file cpSolvePtcTask.py.

Member Function Documentation

fillBadAmp()

def lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask.fillBadAmp	(		self,
			dataset,
			ptcFitType,
			ampName
	)

Fill the dataset with NaNs if there are not enough good points.

Parameters
----------
dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
    The dataset containing the means, variances and exposure times.
ptcFitType : `str`
    Fit a 'POLYNOMIAL' (degree: 'polynomialFitDegree') or
    'EXPAPPROXIMATION' (Eq. 16 of Astier+19) to the PTC.
ampName : `str`
    Amplifier name.

Definition at line 678 of file cpSolvePtcTask.py.

    def fillBadAmp(self, dataset, ptcFitType, ampName):
        """Fill the dataset with NaNs if there are not enough good points.
 
        Parameters
        ----------
        dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
            The dataset containing the means, variances and exposure times.
        ptcFitType : `str`
            Fit a 'POLYNOMIAL' (degree: 'polynomialFitDegree') or
            'EXPAPPROXIMATION' (Eq. 16 of Astier+19) to the PTC.
        ampName : `str`
            Amplifier name.
        """
        dataset.badAmps.append(ampName)
        dataset.expIdMask[ampName] = np.repeat(False, len(dataset.rawExpTimes[ampName]))
        dataset.gain[ampName] = np.nan
        dataset.gainErr[ampName] = np.nan
        dataset.noise[ampName] = np.nan
        dataset.noiseErr[ampName] = np.nan
        dataset.ptcFitPars[ampName] = (np.repeat(np.nan, self.config.polynomialFitDegree + 1) if
                                       ptcFitType in ["POLYNOMIAL", ] else np.repeat(np.nan, 3))
        dataset.ptcFitParsError[ampName] = (np.repeat(np.nan, self.config.polynomialFitDegree + 1) if
                                            ptcFitType in ["POLYNOMIAL", ] else np.repeat(np.nan, 3))
        dataset.ptcFitChiSq[ampName] = np.nan
        dataset.finalVars[ampName] = np.repeat(np.nan, len(dataset.rawExpTimes[ampName]))
        dataset.finalModelVars[ampName] = np.repeat(np.nan, len(dataset.rawExpTimes[ampName]))
        dataset.finalMeans[ampName] = np.repeat(np.nan, len(dataset.rawExpTimes[ampName]))
 
        return

fitCovariancesAstier()

def lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask.fitCovariancesAstier	(		self,
			dataset
	)

Fit measured flat covariances to full model in Astier+19.

Parameters
----------
dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
    The dataset containing information such as the means, (co)variances,
    and exposure times.

Returns
-------
dataset: `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
    This is the same dataset as the input paramter, however, it has been modified
    to include information such as the fit vectors and the fit parameters. See
    the class `PhotonTransferCurveDatase`.

Definition at line 280 of file cpSolvePtcTask.py.

    def fitCovariancesAstier(self, dataset):
        """Fit measured flat covariances to full model in Astier+19.
 
        Parameters
        ----------
        dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
            The dataset containing information such as the means, (co)variances,
            and exposure times.
 
        Returns
        -------
        dataset: `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
            This is the same dataset as the input paramter, however, it has been modified
            to include information such as the fit vectors and the fit parameters. See
            the class `PhotonTransferCurveDatase`.
        """
 
        covFits, covFitsNoB = fitDataFullCovariance(dataset)
        dataset = self.getOutputPtcDataCovAstier(dataset, covFits, covFitsNoB)
 
        return dataset
 

fitPtc()

def lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask.fitPtc	(		self,
			dataset
	)

Fit the photon transfer curve to a polynomial or to Astier+19 approximation.

Fit the photon transfer curve with either a polynomial of the order
specified in the task config, or using the exponential approximation
in Astier+19 (Eq. 16).

Sigma clipping is performed iteratively for the fit, as well as an
initial clipping of data points that are more than
config.initialNonLinearityExclusionThreshold away from lying on a
straight line. This other step is necessary because the photon transfer
curve turns over catastrophically at very high flux (because saturation
drops the variance to ~0) and these far outliers cause the initial fit
to fail, meaning the sigma cannot be calculated to perform the
sigma-clipping.

Parameters
----------
dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
    The dataset containing the means, variances and exposure times.

Returns
-------
dataset: `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
    This is the same dataset as the input parameter, however, it has been modified
    to include information such as the fit vectors and the fit parameters. See
    the class `PhotonTransferCurveDatase`.

Raises
------
RuntimeError:
    Raises if dataset.ptcFitType is None or empty.

Definition at line 493 of file cpSolvePtcTask.py.

    def fitPtc(self, dataset):
        """Fit the photon transfer curve to a polynomial or to Astier+19 approximation.
 
        Fit the photon transfer curve with either a polynomial of the order
        specified in the task config, or using the exponential approximation
        in Astier+19 (Eq. 16).
 
        Sigma clipping is performed iteratively for the fit, as well as an
        initial clipping of data points that are more than
        config.initialNonLinearityExclusionThreshold away from lying on a
        straight line. This other step is necessary because the photon transfer
        curve turns over catastrophically at very high flux (because saturation
        drops the variance to ~0) and these far outliers cause the initial fit
        to fail, meaning the sigma cannot be calculated to perform the
        sigma-clipping.
 
        Parameters
        ----------
        dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
            The dataset containing the means, variances and exposure times.
 
        Returns
        -------
        dataset: `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
            This is the same dataset as the input parameter, however, it has been modified
            to include information such as the fit vectors and the fit parameters. See
            the class `PhotonTransferCurveDatase`.
 
        Raises
        ------
        RuntimeError:
            Raises if dataset.ptcFitType is None or empty.
        """
        if dataset.ptcFitType:
            ptcFitType = dataset.ptcFitType
        else:
            raise RuntimeError("ptcFitType is None of empty in PTC dataset.")
        matrixSide = self.config.maximumRangeCovariancesAstier
        nanMatrix = np.empty((matrixSide, matrixSide))
        nanMatrix[:] = np.nan
 
        for amp in dataset.ampNames:
            lenInputTimes = len(dataset.rawExpTimes[amp])
            listNanMatrix = np.empty((lenInputTimes, matrixSide, matrixSide))
            listNanMatrix[:] = np.nan
 
            dataset.covariancesModel[amp] = listNanMatrix
            dataset.aMatrix[amp] = nanMatrix
            dataset.bMatrix[amp] = nanMatrix
            dataset.covariancesModelNoB[amp] = listNanMatrix
            dataset.aMatrixNoB[amp] = nanMatrix
 
        def errFunc(p, x, y):
            return ptcFunc(p, x) - y
 
        sigmaCutPtcOutliers = self.config.sigmaCutPtcOutliers
        maxIterationsPtcOutliers = self.config.maxIterationsPtcOutliers
 
        for i, ampName in enumerate(dataset.ampNames):
            timeVecOriginal = np.ravel(np.array(dataset.rawExpTimes[ampName]))
            meanVecOriginal = np.ravel(np.array(dataset.rawMeans[ampName]))
            varVecOriginal = np.ravel(np.array(dataset.rawVars[ampName]))
            varVecOriginal = self._makeZeroSafe(varVecOriginal)
 
            goodPoints = self._getInitialGoodPoints(meanVecOriginal, varVecOriginal,
                                                    self.config.initialNonLinearityExclusionThresholdPositive,
                                                    self.config.initialNonLinearityExclusionThresholdNegative,
                                                    self.config.minMeanRatioTest,
                                                    self.config.minVarPivotSearch)
            if not (goodPoints.any()):
                msg = (f"SERIOUS: All points in goodPoints: {goodPoints} are bad."
                       f"Setting {ampName} to BAD.")
                self.log.warn(msg)
                # Fill entries with NaNs
                self.fillBadAmp(dataset, ptcFitType, ampName)
                continue
 
            mask = goodPoints
 
            if ptcFitType == 'EXPAPPROXIMATION':
                ptcFunc = funcAstier
                parsIniPtc = [-1e-9, 1.0, 10.]  # a00, gain, noisei^2
                # lowers and uppers obtained from BOT data studies by C. Lage (UC Davis, 11/2020).
                bounds = self._boundsForAstier(parsIniPtc, lowers=[-1e-4, 0.5, -2000],
                                               uppers=[1e-4, 2.5, 2000])
            if ptcFitType == 'POLYNOMIAL':
                ptcFunc = funcPolynomial
                parsIniPtc = self._initialParsForPolynomial(self.config.polynomialFitDegree + 1)
                bounds = self._boundsForPolynomial(parsIniPtc)
 
            # Before bootstrap fit, do an iterative fit to get rid of outliers
            count = 1
            while count <= maxIterationsPtcOutliers:
                # Note that application of the mask actually shrinks the array
                # to size rather than setting elements to zero (as we want) so
                # always update mask itself and re-apply to the original data
                meanTempVec = meanVecOriginal[mask]
                varTempVec = varVecOriginal[mask]
                res = least_squares(errFunc, parsIniPtc, bounds=bounds, args=(meanTempVec, varTempVec))
                pars = res.x
 
                # change this to the original from the temp because the masks are ANDed
                # meaning once a point is masked it's always masked, and the masks must
                # always be the same length for broadcasting
                sigResids = (varVecOriginal - ptcFunc(pars, meanVecOriginal))/np.sqrt(varVecOriginal)
                newMask = np.array([True if np.abs(r) < sigmaCutPtcOutliers else False for r in sigResids])
                mask = mask & newMask
                if not (mask.any() and newMask.any()):
                    msg = (f"SERIOUS: All points in either mask: {mask} or newMask: {newMask} are bad. "
                           f"Setting {ampName} to BAD.")
                    self.log.warn(msg)
                    # Fill entries with NaNs
                    self.fillBadAmp(dataset, ptcFitType, ampName)
                    break
                nDroppedTotal = Counter(mask)[False]
                self.log.debug(f"Iteration {count}: discarded {nDroppedTotal} points in total for {ampName}")
                count += 1
                # objects should never shrink
                assert (len(mask) == len(timeVecOriginal) == len(meanVecOriginal) == len(varVecOriginal))
            if not (mask.any() and newMask.any()):
                continue
            dataset.expIdMask[ampName] = np.array(dataset.expIdMask[ampName])
            # store the final mask
            if len(dataset.expIdMask[ampName]):
                dataset.expIdMask[ampName] &= mask  # bitwise_and if there is already a  mask
            else:
                dataset.expIdMask[ampName] = mask
            parsIniPtc = pars
            meanVecFinal = meanVecOriginal[mask]
            varVecFinal = varVecOriginal[mask]
 
            if Counter(mask)[False] > 0:
                self.log.info((f"Number of points discarded in PTC of amplifier {ampName}:"
                               f" {Counter(mask)[False]} out of {len(meanVecOriginal)}"))
 
            if (len(meanVecFinal) < len(parsIniPtc)):
                msg = (f"SERIOUS: Not enough data points ({len(meanVecFinal)}) compared to the number of "
                       f"parameters of the PTC model({len(parsIniPtc)}). Setting {ampName} to BAD.")
                self.log.warn(msg)
                # Fill entries with NaNs
                self.fillBadAmp(dataset, ptcFitType, ampName)
                continue
            # Fit the PTC
            if self.config.doFitBootstrap:
                parsFit, parsFitErr, reducedChiSqPtc = fitBootstrap(parsIniPtc, meanVecFinal,
                                                                    varVecFinal, ptcFunc,
                                                                    weightsY=1./np.sqrt(varVecFinal))
            else:
                parsFit, parsFitErr, reducedChiSqPtc = fitLeastSq(parsIniPtc, meanVecFinal,
                                                                  varVecFinal, ptcFunc,
                                                                  weightsY=1./np.sqrt(varVecFinal))
            dataset.ptcFitPars[ampName] = parsFit
            dataset.ptcFitParsError[ampName] = parsFitErr
            dataset.ptcFitChiSq[ampName] = reducedChiSqPtc
            # Masked variances (measured and modeled) and means. Need to pad the array so astropy.Table does
            # not crash (the mask may vary per amp).
            padLength = len(dataset.rawExpTimes[ampName]) - len(varVecFinal)
            dataset.finalVars[ampName] = np.pad(varVecFinal, (0, padLength), 'constant',
                                                constant_values=np.nan)
            dataset.finalModelVars[ampName] = np.pad(ptcFunc(parsFit, meanVecFinal), (0, padLength),
                                                     'constant', constant_values=np.nan)
            dataset.finalMeans[ampName] = np.pad(meanVecFinal, (0, padLength), 'constant',
                                                 constant_values=np.nan)
            if ptcFitType == 'EXPAPPROXIMATION':
                ptcGain = parsFit[1]
                ptcGainErr = parsFitErr[1]
                ptcNoise = np.sqrt(np.fabs(parsFit[2]))
                ptcNoiseErr = 0.5*(parsFitErr[2]/np.fabs(parsFit[2]))*np.sqrt(np.fabs(parsFit[2]))
            if ptcFitType == 'POLYNOMIAL':
                ptcGain = 1./parsFit[1]
                ptcGainErr = np.fabs(1./parsFit[1])*(parsFitErr[1]/parsFit[1])
                ptcNoise = np.sqrt(np.fabs(parsFit[0]))*ptcGain
                ptcNoiseErr = (0.5*(parsFitErr[0]/np.fabs(parsFit[0]))*(np.sqrt(np.fabs(parsFit[0]))))*ptcGain
            dataset.gain[ampName] = ptcGain
            dataset.gainErr[ampName] = ptcGainErr
            dataset.noise[ampName] = ptcNoise
            dataset.noiseErr[ampName] = ptcNoiseErr
 
        if not len(dataset.ptcFitType) == 0:
            dataset.ptcFitType = ptcFitType
        if len(dataset.badAmps) == 0:
            dataset.badAmps = np.repeat(np.nan, len(list(dataset.rawExpTimes.values())[0]))
 
        return dataset
 

getOutputPtcDataCovAstier()

def lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask.getOutputPtcDataCovAstier	(		self,
			dataset,
			covFits,
			covFitsNoB
	)

Get output data for PhotonTransferCurveCovAstierDataset from CovFit objects.

Parameters
----------
dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
    The dataset containing information such as the means, variances and exposure times.
covFits: `dict`
    Dictionary of CovFit objects, with amp names as keys.
covFitsNoB : `dict`
     Dictionary of CovFit objects, with amp names as keys, and 'b=0' in Eq. 20 of Astier+19.

Returns
-------
dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
    This is the same dataset as the input paramter, however, it has been modified
    to include extra information such as the mask 1D array, gains, reoudout noise, measured signal,
    measured variance, modeled variance, a, and b coefficient matrices (see Astier+19) per amplifier.
    See the class `PhotonTransferCurveDatase`.

Definition at line 302 of file cpSolvePtcTask.py.

    def getOutputPtcDataCovAstier(self, dataset, covFits, covFitsNoB):
        """Get output data for PhotonTransferCurveCovAstierDataset from CovFit objects.
 
        Parameters
        ----------
        dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
            The dataset containing information such as the means, variances and exposure times.
        covFits: `dict`
            Dictionary of CovFit objects, with amp names as keys.
        covFitsNoB : `dict`
             Dictionary of CovFit objects, with amp names as keys, and 'b=0' in Eq. 20 of Astier+19.
 
        Returns
        -------
        dataset : `lsst.ip.isr.ptcDataset.PhotonTransferCurveDataset`
            This is the same dataset as the input paramter, however, it has been modified
            to include extra information such as the mask 1D array, gains, reoudout noise, measured signal,
            measured variance, modeled variance, a, and b coefficient matrices (see Astier+19) per amplifier.
            See the class `PhotonTransferCurveDatase`.
        """
        assert(len(covFits) == len(covFitsNoB))
 
        for i, amp in enumerate(dataset.ampNames):
            lenInputTimes = len(dataset.rawExpTimes[amp])
            # Not used when ptcFitType is 'FULLCOVARIANCE'
            dataset.ptcFitPars[amp] = [np.nan]
            dataset.ptcFitParsError[amp] = [np.nan]
            dataset.ptcFitChiSq[amp] = np.nan
            if amp in covFits:
                fit = covFits[amp]
                fitNoB = covFitsNoB[amp]
                # Save full covariances, covariances models, and their weights
                # dataset.expIdMask is already full
                dataset.covariances[amp] = fit.cov
                dataset.covariancesModel[amp] = fit.evalCovModel()
                dataset.covariancesSqrtWeights[amp] = fit.sqrtW
                dataset.aMatrix[amp] = fit.getA()
                dataset.bMatrix[amp] = fit.getB()
                dataset.covariancesModelNoB[amp] = fitNoB.evalCovModel()
                dataset.aMatrixNoB[amp] = fitNoB.getA()
 
                (meanVecFinal, varVecFinal, varVecModel,
                    wc, varMask) = fit.getFitData(0, 0, divideByMu=False)
                gain = fit.getGain()
 
                dataset.gain[amp] = gain
                dataset.gainErr[amp] = fit.getGainErr()
                dataset.noise[amp] = np.sqrt(fit.getRon())
                dataset.noiseErr[amp] = fit.getRonErr()
                dataset.finalVars[amp] = varVecFinal
                dataset.finalModelVars[amp] = varVecModel
                dataset.finalMeans[amp] = meanVecFinal
 
            else:
                # Bad amp
                # Entries need to have proper dimensions so read/write with astropy.Table works.
                matrixSide = self.config.maximumRangeCovariancesAstier
                nanMatrix = np.full((matrixSide, matrixSide), np.nan)
                listNanMatrix = np.full((lenInputTimes, matrixSide, matrixSide), np.nan)
 
                dataset.covariances[amp] = listNanMatrix
                dataset.covariancesModel[amp] = listNanMatrix
                dataset.covariancesSqrtWeights[amp] = listNanMatrix
                dataset.aMatrix[amp] = nanMatrix
                dataset.bMatrix[amp] = nanMatrix
                dataset.covariancesModelNoB[amp] = listNanMatrix
                dataset.aMatrixNoB[amp] = nanMatrix
 
                dataset.expIdMask[amp] = np.repeat(np.nan, lenInputTimes)
                dataset.gain[amp] = np.nan
                dataset.gainErr[amp] = np.nan
                dataset.noise[amp] = np.nan
                dataset.noiseErr[amp] = np.nan
                dataset.finalVars[amp] = np.repeat(np.nan, lenInputTimes)
                dataset.finalModelVars[amp] = np.repeat(np.nan, lenInputTimes)
                dataset.finalMeans[amp] = np.repeat(np.nan, lenInputTimes)
 
        return dataset
 

run()

def lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask.run	(		self,
			inputCovariances,
			camera = None,
			inputExpList = None
	)

Fit measure covariances to different models.

Parameters
----------
inputCovariances : `list` [`lsst.ip.isr.PhotonTransferCurveDataset`]
    List of lsst.ip.isr.PhotonTransferCurveDataset datasets.

camera : `lsst.afw.cameraGeom.Camera`, optional
    Input camera.

inputExpList : `list` [`~lsst.afw.image.exposure.exposure.ExposureF`], optional
    List of exposures.

Returns
-------
results : `lsst.pipe.base.Struct`
    The results struct containing:
    ``outputPtcDatset`` : `lsst.ip.isr.PhotonTransferCurveDataset`
        Final PTC dataset, containing information such as the means, variances,
        and exposure times.

Definition at line 191 of file cpSolvePtcTask.py.

    def run(self, inputCovariances, camera=None, inputExpList=None):
        """Fit measure covariances to different models.
 
        Parameters
        ----------
        inputCovariances : `list` [`lsst.ip.isr.PhotonTransferCurveDataset`]
            List of lsst.ip.isr.PhotonTransferCurveDataset datasets.
 
        camera : `lsst.afw.cameraGeom.Camera`, optional
            Input camera.
 
        inputExpList : `list` [`~lsst.afw.image.exposure.exposure.ExposureF`], optional
            List of exposures.
 
        Returns
        -------
        results : `lsst.pipe.base.Struct`
            The results struct containing:
            ``outputPtcDatset`` : `lsst.ip.isr.PhotonTransferCurveDataset`
                Final PTC dataset, containing information such as the means, variances,
                and exposure times.
        """
        # Assemble partial PTC datasets into a single dataset.
        ampNames = np.unique(inputCovariances[0].ampNames)
        datasetPtc = PhotonTransferCurveDataset(ampNames, self.config.ptcFitType,
                                                self.config.maximumRangeCovariancesAstier)
        for partialPtcDataset in inputCovariances:
            if partialPtcDataset.ptcFitType == 'DUMMY':
                continue
            for ampName in ampNames:
                datasetPtc.inputExpIdPairs[ampName].append(partialPtcDataset.inputExpIdPairs[ampName])
                if type(partialPtcDataset.rawExpTimes[ampName]) is list:
                    datasetPtc.rawExpTimes[ampName].append(partialPtcDataset.rawExpTimes[ampName][0])
                else:
                    datasetPtc.rawExpTimes[ampName].append(partialPtcDataset.rawExpTimes[ampName])
                if type(partialPtcDataset.rawMeans[ampName]) is list:
                    datasetPtc.rawMeans[ampName].append(partialPtcDataset.rawMeans[ampName][0])
                else:
                    datasetPtc.rawMeans[ampName].append(partialPtcDataset.rawMeans[ampName])
                if type(partialPtcDataset.rawVars[ampName]) is list:
                    datasetPtc.rawVars[ampName].append(partialPtcDataset.rawVars[ampName][0])
                else:
                    datasetPtc.rawVars[ampName].append(partialPtcDataset.rawVars[ampName])
                if type(partialPtcDataset.expIdMask[ampName]) is list:
                    datasetPtc.expIdMask[ampName].append(partialPtcDataset.expIdMask[ampName][0])
                else:
                    datasetPtc.expIdMask[ampName].append(partialPtcDataset.expIdMask[ampName])
                datasetPtc.covariances[ampName].append(np.array(partialPtcDataset.covariances[ampName][0]))
                datasetPtc.covariancesSqrtWeights[ampName].append(
                    np.array(partialPtcDataset.covariancesSqrtWeights[ampName][0]))
        # Sort arrays that are filled so far in the final dataset by rawMeans index
        for ampName in ampNames:
            index = np.argsort(np.ravel(np.array(datasetPtc.rawMeans[ampName])))
            datasetPtc.inputExpIdPairs[ampName] = np.array(datasetPtc.inputExpIdPairs[ampName])[index]
            datasetPtc.rawExpTimes[ampName] = np.array(datasetPtc.rawExpTimes[ampName])[index]
            datasetPtc.rawMeans[ampName] = np.array(datasetPtc.rawMeans[ampName])[index]
            datasetPtc.rawVars[ampName] = np.array(datasetPtc.rawVars[ampName])[index]
            datasetPtc.expIdMask[ampName] = np.array(datasetPtc.expIdMask[ampName])[index]
            datasetPtc.covariances[ampName] = np.array(datasetPtc.covariances[ampName])[index]
            datasetPtc.covariancesSqrtWeights[ampName] = np.array(
                datasetPtc.covariancesSqrtWeights[ampName])[index]
        if self.config.ptcFitType == "FULLCOVARIANCE":
            # Calculate covariances and fit them, including the PTC, to Astier+19 full model (Eq. 20)
            # First, fit get the flat pairs that are masked, fitting C_00 vs mu to
            # the EXPAPPROXIMATION model (Eq. 16 in Astier+19).
            # The points at these fluxes will also be masked when calculating the other covariances, C_ij)
            tempDatasetPtc = copy.copy(datasetPtc)
            tempDatasetPtc.ptcFitType = "EXPAPPROXIMATION"
            tempDatasetPtc = self.fitPtc(tempDatasetPtc)
            for ampName in datasetPtc.ampNames:
                datasetPtc.expIdMask[ampName] = tempDatasetPtc.expIdMask[ampName]
            datasetPtc.fitType = "FULLCOVARIANCE"
            datasetPtc = self.fitCovariancesAstier(datasetPtc)
        # The other options are: self.config.ptcFitType in ("EXPAPPROXIMATION", "POLYNOMIAL")
        else:
            # Fit the PTC to a polynomial or to Astier+19 exponential approximation (Eq. 16).
            # Fill up PhotonTransferCurveDataset object.
            datasetPtc = self.fitPtc(datasetPtc)
        if inputExpList is not None:
            # It should be a list of exposures, to get the detector.
            detector = inputExpList[0].getDetector()
        else:
            detector = None
        datasetPtc.updateMetadata(setDate=True, camera=camera, detector=detector)
 
        return pipeBase.Struct(
            outputPtcDataset=datasetPtc,
        )
 

runQuantum()

def lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask.runQuantum	(		self,
			butlerQC,
			inputRefs,
			outputRefs
	)

Ensure that the input and output dimensions are passed along.

Parameters
----------
butlerQC : `~lsst.daf.butler.butlerQuantumContext.ButlerQuantumContext`
    Butler to operate on.
inputRefs : `~lsst.pipe.base.connections.InputQuantizedConnection`
    Input data refs to load.
ouptutRefs : `~lsst.pipe.base.connections.OutputQuantizedConnection`
    Output data refs to persist.

Definition at line 175 of file cpSolvePtcTask.py.

    def runQuantum(self, butlerQC, inputRefs, outputRefs):
        """Ensure that the input and output dimensions are passed along.
 
        Parameters
        ----------
        butlerQC : `~lsst.daf.butler.butlerQuantumContext.ButlerQuantumContext`
            Butler to operate on.
        inputRefs : `~lsst.pipe.base.connections.InputQuantizedConnection`
            Input data refs to load.
        ouptutRefs : `~lsst.pipe.base.connections.OutputQuantizedConnection`
            Output data refs to persist.
        """
        inputs = butlerQC.get(inputRefs)
        outputs = self.run(inputCovariances=inputs['inputCovariances'], camera=inputs['camera'])
        butlerQC.put(outputs, outputRefs)
 

Member Data Documentation

ConfigClass

lsst.cp.pipe.ptc.cpSolvePtcTask.PhotonTransferCurveSolveTask.ConfigClass = PhotonTransferCurveSolveConfig

static

Definition at line 172 of file cpSolvePtcTask.py.

---

The documentation for this class was generated from the following file:

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-0.6.0/Linux64/cp_pipe/22.0.1-18-gb17765a+2264247a6b/python/lsst/cp/pipe/ptc/cpSolvePtcTask.py