Inheritance diagram for lsst.cp.pipe.measureCrosstalk.CrosstalkSolveTask:

Public Member Functions
def	runQuantum (self, butlerQC, inputRefs, outputRefs)

def	run (self, inputRatios, inputFluxes=None, camera=None, inputDims=None, outputDims=None)

def	measureCrosstalkCoefficients (self, ratios, rejIter, rejSigma)

def	debugRatios (self, stepname, ratios, i, j, coeff=0.0, valid=False)

Static Public Member Functions
def	filterCrosstalkCalib (inCalib)

Static Public Attributes
	ConfigClass = CrosstalkSolveConfig

Detailed Description

Task to solve crosstalk from pixel ratios.

Definition at line 387 of file measureCrosstalk.py.

Member Function Documentation

◆ debugRatios()

def lsst.cp.pipe.measureCrosstalk.CrosstalkSolveTask.debugRatios	(	self,
		stepname,
		ratios,
		i,
		j,
		coeff = `0.0`,
		valid = `False`
	)

Utility function to examine the final CT ratio set.

Parameters
----------
stepname : `str`
    State of processing to view.
ratios : `dict` of `dict` of `np.ndarray`
    Array of measured CT ratios, indexed by source/victim
    amplifier.
i : `str`
    Index of the source amplifier.
j : `str`
    Index of the target amplifier.
coeff : `float`, optional
    Coefficient calculated to plot along with the simple mean.
valid : `bool`, optional
    Validity to be added to the plot title.

Definition at line 657 of file measureCrosstalk.py.

     def debugRatios(self, stepname, ratios, i, j, coeff=0.0, valid=False):
         """Utility function to examine the final CT ratio set.
  
         Parameters
         ----------
         stepname : `str`
             State of processing to view.
         ratios : `dict` of `dict` of `np.ndarray`
             Array of measured CT ratios, indexed by source/victim
             amplifier.
         i : `str`
             Index of the source amplifier.
         j : `str`
             Index of the target amplifier.
         coeff : `float`, optional
             Coefficient calculated to plot along with the simple mean.
         valid : `bool`, optional
             Validity to be added to the plot title.
         """
         frame = getDebugFrame(self._display, stepname)
         if frame:
             if i == j or ratios is None or len(ratios) < 1:
                 pass
  
             ratioList = ratios[i][j]
             if ratioList is None or len(ratioList) < 1:
                 pass
  
             mean = np.mean(ratioList)
             std = np.std(ratioList)
             import matplotlib.pyplot as plt
             figure = plt.figure(1)
             figure.clear()
             plt.hist(x=ratioList, bins=len(ratioList),
                      cumulative=True, color='b', density=True, histtype='step')
             plt.xlabel("Measured pixel ratio")
             plt.ylabel(f"CDF: n={len(ratioList)}")
             plt.xlim(np.percentile(ratioList, [1.0, 99]))
             plt.axvline(x=mean, color="k")
             plt.axvline(x=coeff, color='g')
             plt.axvline(x=(std / np.sqrt(len(ratioList))), color='r')
             plt.axvline(x=-(std / np.sqrt(len(ratioList))), color='r')
             plt.title(f"(Source {i} -> Target {j}) mean: {mean:.2g} coeff: {coeff:.2g} valid: {valid}")
             figure.show()
  
             prompt = "Press Enter to continue: "
             while True:
                 ans = input(prompt).lower()
                 if ans in ("", "c",):
                     break
                 elif ans in ("pdb", "p",):
                     import pdb
                     pdb.set_trace()
             plt.close()
  
  

◆ filterCrosstalkCalib()

def lsst.cp.pipe.measureCrosstalk.CrosstalkSolveTask.filterCrosstalkCalib ( inCalib )

static

Apply valid constraints to the measured values.

Any measured coefficient that is determined to be invalid is
set to zero, and has the error set to nan.  The validation is
determined by checking that the measured coefficient is larger
than the calculated standard error of the mean.

Parameters
----------
inCalib : `lsst.ip.isr.CrosstalkCalib`
    Input calibration to filter.

Returns
-------
outCalib : `lsst.ip.isr.CrosstalkCalib`
     Filtered calibration.

Definition at line 625 of file measureCrosstalk.py.

     def filterCrosstalkCalib(inCalib):
         """Apply valid constraints to the measured values.
  
         Any measured coefficient that is determined to be invalid is
         set to zero, and has the error set to nan.  The validation is
         determined by checking that the measured coefficient is larger
         than the calculated standard error of the mean.
  
         Parameters
         ----------
         inCalib : `lsst.ip.isr.CrosstalkCalib`
             Input calibration to filter.
  
         Returns
         -------
         outCalib : `lsst.ip.isr.CrosstalkCalib`
              Filtered calibration.
         """
         outCalib = CrosstalkCalib()
         outCalib.numAmps = inCalib.numAmps
  
         outCalib.coeffs = inCalib.coeffs
         outCalib.coeffs[~inCalib.coeffValid] = 0.0
  
         outCalib.coeffErr = inCalib.coeffErr
         outCalib.coeffErr[~inCalib.coeffValid] = np.nan
  
         outCalib.coeffNum = inCalib.coeffNum
         outCalib.coeffValid = inCalib.coeffValid
  
         return outCalib
  

◆ measureCrosstalkCoefficients()

def lsst.cp.pipe.measureCrosstalk.CrosstalkSolveTask.measureCrosstalkCoefficients	(	self,
		ratios,
		rejIter,
		rejSigma
	)

Measure crosstalk coefficients from the ratios.

Given a list of ratios for each target/source amp combination,
we measure a sigma clipped mean and error.

The coefficient errors returned are the standard deviation of
the final set of clipped input ratios.

Parameters
----------
ratios : `dict` of `dict` of `numpy.ndarray`
   Catalog of arrays of ratios.
rejIter : `int`
   Number of rejection iterations.
rejSigma : `float`
   Rejection threshold (sigma).

Returns
-------
calib : `lsst.ip.isr.CrosstalkCalib`
    The output crosstalk calibration.

Notes
-----
The lsstDebug.Info() method can be rewritten for __name__ =
`lsst.ip.isr.measureCrosstalk`, and supports the parameters:

debug.display['measure'] : `bool`
    Display the CDF of the combined ratio measurements for
    a pair of source/target amplifiers from the final set of
    clipped input ratios.

Definition at line 548 of file measureCrosstalk.py.

     def measureCrosstalkCoefficients(self, ratios, rejIter, rejSigma):
         """Measure crosstalk coefficients from the ratios.
  
         Given a list of ratios for each target/source amp combination,
         we measure a sigma clipped mean and error.
  
         The coefficient errors returned are the standard deviation of
         the final set of clipped input ratios.
  
         Parameters
         ----------
         ratios : `dict` of `dict` of `numpy.ndarray`
            Catalog of arrays of ratios.
         rejIter : `int`
            Number of rejection iterations.
         rejSigma : `float`
            Rejection threshold (sigma).
  
         Returns
         -------
         calib : `lsst.ip.isr.CrosstalkCalib`
             The output crosstalk calibration.
  
         Notes
         -----
         The lsstDebug.Info() method can be rewritten for __name__ =
         `lsst.ip.isr.measureCrosstalk`, and supports the parameters:
  
         debug.display['measure'] : `bool`
             Display the CDF of the combined ratio measurements for
             a pair of source/target amplifiers from the final set of
             clipped input ratios.
         """
         calib = CrosstalkCalib(nAmp=len(ratios))
  
         # Calibration stores coefficients as a numpy ndarray.
         ordering = list(ratios.keys())
         for ii, jj in itertools.product(range(calib.nAmp), range(calib.nAmp)):
             if ii == jj:
                 values = [0.0]
             else:
                 values = np.array(ratios[ordering[ii]][ordering[jj]])
                 values = values[np.abs(values) < 1.0]  # Discard unreasonable values
  
             calib.coeffNum[ii][jj] = len(values)
  
             if len(values) == 0:
                 self.log.warn("No values for matrix element %d,%d" % (ii, jj))
                 calib.coeffs[ii][jj] = np.nan
                 calib.coeffErr[ii][jj] = np.nan
                 calib.coeffValid[ii][jj] = False
             else:
                 if ii != jj:
                     for rej in range(rejIter):
                         lo, med, hi = np.percentile(values, [25.0, 50.0, 75.0])
                         sigma = 0.741*(hi - lo)
                         good = np.abs(values - med) < rejSigma*sigma
                         if good.sum() == len(good):
                             break
                         values = values[good]
  
                 calib.coeffs[ii][jj] = np.mean(values)
                 if calib.coeffNum[ii][jj] == 1:
                     calib.coeffErr[ii][jj] = np.nan
                 else:
                     correctionFactor = sigmaClipCorrection(rejSigma)
                     calib.coeffErr[ii][jj] = np.std(values) * correctionFactor
                 calib.coeffValid[ii][jj] = (np.abs(calib.coeffs[ii][jj])
                                             > calib.coeffErr[ii][jj] / np.sqrt(calib.coeffNum[ii][jj]))
  
             if calib.coeffNum[ii][jj] > 1:
                 self.debugRatios('measure', ratios, ordering[ii], ordering[jj],
                                  calib.coeffs[ii][jj], calib.coeffValid[ii][jj])
  
         return calib
  

◆ run()

def lsst.cp.pipe.measureCrosstalk.CrosstalkSolveTask.run	(	self,
		inputRatios,
		inputFluxes = `None`,
		camera = `None`,
		inputDims = `None`,
		outputDims = `None`
	)

Combine ratios to produce crosstalk coefficients.

Parameters
----------
inputRatios : `list` [`dict` [`dict` [`dict` [`dict` [`list`]]]]]
    A list of nested dictionaries of ratios indexed by target
    and source chip, then by target and source amplifier.
inputFluxes : `list` [`dict` [`dict` [`list`]]]
    A list of nested dictionaries of source pixel fluxes, indexed
    by source chip and amplifier.
camera : `lsst.afw.cameraGeom.Camera`
    Input camera.
inputDims : `list` [`lsst.daf.butler.DataCoordinate`]
    DataIds to use to construct provenance.
outputDims : `list` [`lsst.daf.butler.DataCoordinate`]
    DataIds to use to populate the output calibration.

Returns
-------
results : `lsst.pipe.base.Struct`
    The results struct containing:

    ``outputCrosstalk`` : `lsst.ip.isr.CrosstalkCalib`
        Final crosstalk calibration.
    ``outputProvenance`` : `lsst.ip.isr.IsrProvenance`
        Provenance data for the new calibration.

Raises
------
RuntimeError
    Raised if the input data contains multiple target detectors.

Notes
-----
The lsstDebug.Info() method can be rewritten for __name__ =
`lsst.ip.isr.measureCrosstalk`, and supports the parameters:

debug.display['reduce'] : `bool`
    Display a histogram of the combined ratio measurements for
    a pair of source/target amplifiers from all input
    exposures/detectors.

Definition at line 415 of file measureCrosstalk.py.

     def run(self, inputRatios, inputFluxes=None, camera=None, inputDims=None, outputDims=None):
         """Combine ratios to produce crosstalk coefficients.
  
         Parameters
         ----------
         inputRatios : `list` [`dict` [`dict` [`dict` [`dict` [`list`]]]]]
             A list of nested dictionaries of ratios indexed by target
             and source chip, then by target and source amplifier.
         inputFluxes : `list` [`dict` [`dict` [`list`]]]
             A list of nested dictionaries of source pixel fluxes, indexed
             by source chip and amplifier.
         camera : `lsst.afw.cameraGeom.Camera`
             Input camera.
         inputDims : `list` [`lsst.daf.butler.DataCoordinate`]
             DataIds to use to construct provenance.
         outputDims : `list` [`lsst.daf.butler.DataCoordinate`]
             DataIds to use to populate the output calibration.
  
         Returns
         -------
         results : `lsst.pipe.base.Struct`
             The results struct containing:
  
             ``outputCrosstalk`` : `lsst.ip.isr.CrosstalkCalib`
                 Final crosstalk calibration.
             ``outputProvenance`` : `lsst.ip.isr.IsrProvenance`
                 Provenance data for the new calibration.
  
         Raises
         ------
         RuntimeError
             Raised if the input data contains multiple target detectors.
  
         Notes
         -----
         The lsstDebug.Info() method can be rewritten for __name__ =
         `lsst.ip.isr.measureCrosstalk`, and supports the parameters:
  
         debug.display['reduce'] : `bool`
             Display a histogram of the combined ratio measurements for
             a pair of source/target amplifiers from all input
             exposures/detectors.
  
         """
         if outputDims:
             calibChip = outputDims['detector']
             instrument = outputDims['instrument']
         else:
             # calibChip needs to be set manually in Gen2.
             calibChip = None
             instrument = None
  
         if camera and calibChip:
             calibDetector = camera[calibChip]
         else:
             calibDetector = None
  
         self.log.info("Combining measurements from %d ratios and %d fluxes",
                       len(inputRatios), len(inputFluxes) if inputFluxes else 0)
  
         if inputFluxes is None:
             inputFluxes = [None for exp in inputRatios]
  
         combinedRatios = defaultdict(lambda: defaultdict(list))
         combinedFluxes = defaultdict(lambda: defaultdict(list))
         for ratioDict, fluxDict in zip(inputRatios, inputFluxes):
             for targetChip in ratioDict:
                 if calibChip and targetChip != calibChip and targetChip != calibDetector.getName():
                     raise RuntimeError(f"Target chip: {targetChip} does not match calibration dimension: "
                                        f"{calibChip}, {calibDetector.getName()}!")
  
                 sourceChip = targetChip
                 if sourceChip in ratioDict[targetChip]:
                     ratios = ratioDict[targetChip][sourceChip]
  
                     for targetAmp in ratios:
                         for sourceAmp in ratios[targetAmp]:
                             combinedRatios[targetAmp][sourceAmp].extend(ratios[targetAmp][sourceAmp])
                             if fluxDict:
                                 combinedFluxes[targetAmp][sourceAmp].extend(fluxDict[sourceChip][sourceAmp])
                 # TODO: DM-21904
                 # Iterating over all other entries in ratioDict[targetChip] will yield
                 # inter-chip terms.
  
         for targetAmp in combinedRatios:
             for sourceAmp in combinedRatios[targetAmp]:
                 self.log.info("Read %d pixels for %s -> %s",
                               len(combinedRatios[targetAmp][sourceAmp]),
                               targetAmp, sourceAmp)
                 if len(combinedRatios[targetAmp][sourceAmp]) > 1:
                     self.debugRatios('reduce', combinedRatios, targetAmp, sourceAmp)
  
         if self.config.fluxOrder == 0:
             self.log.info("Fitting crosstalk coefficients.")
             calib = self.measureCrosstalkCoefficients(combinedRatios,
                                                       self.config.rejIter, self.config.rejSigma)
         else:
             raise NotImplementedError("Non-linear crosstalk terms are not yet supported.")
  
         self.log.info("Number of valid coefficients: %d", np.sum(calib.coeffValid))
  
         if self.config.doFiltering:
             # This step will apply the calculated validity values to
             # censor poorly measured coefficients.
             self.log.info("Filtering measured crosstalk to remove invalid solutions.")
             calib = self.filterCrosstalkCalib(calib)
  
         # Populate the remainder of the calibration information.
         calib.hasCrosstalk = True
         calib.interChip = {}
  
         # calibChip is the detector dimension, which is the detector Id
         calib._detectorId = calibChip
         if calibDetector:
             calib._detectorName = calibDetector.getName()
             calib._detectorSerial = calibDetector.getSerial()
  
         calib._instrument = instrument
         calib.updateMetadata(setCalibId=True, setDate=True)
  
         # Make an IsrProvenance().
         provenance = IsrProvenance(calibType="CROSSTALK")
         provenance._detectorName = calibChip
         if inputDims:
             provenance.fromDataIds(inputDims)
             provenance._instrument = instrument
         provenance.updateMetadata()
  
         return pipeBase.Struct(
             outputCrosstalk=calib,
             outputProvenance=provenance,
         )
  

◆ runQuantum()

def lsst.cp.pipe.measureCrosstalk.CrosstalkSolveTask.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 394 of file measureCrosstalk.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)
  
         # Use the dimensions to set calib/provenance information.
         inputs['inputDims'] = [exp.dataId.byName() for exp in inputRefs.inputRatios]
         inputs['outputDims'] = outputRefs.outputCrosstalk.dataId.byName()
  
         outputs = self.run(**inputs)
         butlerQC.put(outputs, outputRefs)
  

Member Data Documentation

◆ ConfigClass

lsst.cp.pipe.measureCrosstalk.CrosstalkSolveTask.ConfigClass = CrosstalkSolveConfig

static

Definition at line 391 of file measureCrosstalk.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/measureCrosstalk.py

Public Member Functions

Static Public Member Functions

Static Public Attributes