crosstalk.py

Go to the documentation of this file.

#
# LSST Data Management System
# Copyright 2008-2017 AURA/LSST.
#
# This product includes software developed by the
# LSST Project (http://www.lsst.org/).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the LSST License Statement and
# the GNU General Public License along with this program.  If not,
# see <https://www.lsstcorp.org/LegalNotices/>.
#
"""
Apply intra-detector crosstalk corrections
"""
import numpy as np
from astropy.table import Table
 
import lsst.afw.math
import lsst.afw.detection
import lsst.daf.butler
from lsst.pex.config import Config, Field, ChoiceField, ListField
from lsst.pipe.base import Task
 
from lsst.ip.isr import IsrCalib
 
 
__all__ = ["CrosstalkCalib", "CrosstalkConfig", "CrosstalkTask",
           "NullCrosstalkTask"]
 
 
class CrosstalkCalib(IsrCalib):
    """Calibration of amp-to-amp crosstalk coefficients.
 
    Parameters
    ----------
    detector : `lsst.afw.cameraGeom.Detector`, optional
        Detector to use to pull coefficients from.
    nAmp : `int`, optional
        Number of amplifiers to initialize.
    log : `logging.Logger`, optional
        Log to write messages to.
    **kwargs :
        Parameters to pass to parent constructor.
 
    Notes
    -----
    The crosstalk attributes stored are:
 
    hasCrosstalk : `bool`
        Whether there is crosstalk defined for this detector.
    nAmp : `int`
        Number of amplifiers in this detector.
    crosstalkShape : `tuple` [`int`, `int`]
        A tuple containing the shape of the ``coeffs`` matrix.  This
        should be equivalent to (``nAmp``, ``nAmp``).
    coeffs : `np.ndarray`
        A matrix containing the crosstalk coefficients.  coeff[i][j]
        contains the coefficients to calculate the contribution
        amplifier_j has on amplifier_i (each row[i] contains the
        corrections for detector_i).
    coeffErr : `np.ndarray`, optional
        A matrix (as defined by ``coeffs``) containing the standard
        distribution of the crosstalk measurements.
    coeffNum : `np.ndarray`, optional
        A matrix containing the number of pixel pairs used to measure
        the ``coeffs`` and ``coeffErr``.
    coeffValid : `np.ndarray`, optional
        A matrix of Boolean values indicating if the coefficient is
        valid, defined as abs(coeff) > coeffErr / sqrt(coeffNum).
    interChip : `dict` [`np.ndarray`]
        A dictionary keyed by detectorName containing ``coeffs``
        matrices used to correct for inter-chip crosstalk with a
        source on the detector indicated.
 
    """
    _OBSTYPE = 'CROSSTALK'
    _SCHEMA = 'Gen3 Crosstalk'
    _VERSION = 1.0
 
    def __init__(self, detector=None, nAmp=0, **kwargs):
        self.hasCrosstalkhasCrosstalk = False
        self.nAmpnAmp = nAmp if nAmp else 0
        self.crosstalkShapecrosstalkShape = (self.nAmpnAmp, self.nAmpnAmp)
 
        self.coeffscoeffs = np.zeros(self.crosstalkShapecrosstalkShape) if self.nAmpnAmp else None
        self.coeffErrcoeffErr = np.zeros(self.crosstalkShapecrosstalkShape) if self.nAmpnAmp else None
        self.coeffNumcoeffNum = np.zeros(self.crosstalkShapecrosstalkShape,
                                 dtype=int) if self.nAmpnAmp else None
        self.coeffValidcoeffValid = np.zeros(self.crosstalkShapecrosstalkShape,
                                   dtype=bool) if self.nAmpnAmp else None
        self.interChipinterChip = {}
 
        super().__init__(**kwargs)
        self.requiredAttributesrequiredAttributesrequiredAttributesrequiredAttributes.update(['hasCrosstalk', 'nAmp', 'coeffs',
                                        'coeffErr', 'coeffNum', 'coeffValid',
                                        'interChip'])
        if detector:
            self.fromDetectorfromDetectorfromDetector(detector)
 
    def updateMetadata(self, setDate=False, **kwargs):
        """Update calibration metadata.
 
        This calls the base class's method after ensuring the required
        calibration keywords will be saved.
 
        Parameters
        ----------
        setDate : `bool`, optional
            Update the CALIBDATE fields in the metadata to the current
            time. Defaults to False.
        kwargs :
            Other keyword parameters to set in the metadata.
        """
        kwargs['DETECTOR'] = self._detectorId_detectorId_detectorId
        kwargs['DETECTOR_NAME'] = self._detectorName_detectorName_detectorName
        kwargs['DETECTOR_SERIAL'] = self._detectorSerial_detectorSerial_detectorSerial
        kwargs['HAS_CROSSTALK'] = self.hasCrosstalkhasCrosstalk
        kwargs['NAMP'] = self.nAmpnAmp
        self.crosstalkShapecrosstalkShape = (self.nAmpnAmp, self.nAmpnAmp)
        kwargs['CROSSTALK_SHAPE'] = self.crosstalkShapecrosstalkShape
 
        super().updateMetadata(setDate=setDate, **kwargs)
 
    def fromDetector(self, detector, coeffVector=None):
        """Set calibration parameters from the detector.
 
        Parameters
        ----------
        detector : `lsst.afw.cameraGeom.Detector`
            Detector to use to set parameters from.
        coeffVector : `numpy.array`, optional
            Use the detector geometry (bounding boxes and flip
            information), but use ``coeffVector`` instead of the
            output of ``detector.getCrosstalk()``.
 
        Returns
        -------
        calib : `lsst.ip.isr.CrosstalkCalib`
            The calibration constructed from the detector.
 
        """
        if detector.hasCrosstalk() or coeffVector:
            self._detectorId_detectorId_detectorId = detector.getId()
            self._detectorName_detectorName_detectorName = detector.getName()
            self._detectorSerial_detectorSerial_detectorSerial = detector.getSerial()
 
            self.nAmpnAmp = len(detector)
            self.crosstalkShapecrosstalkShape = (self.nAmpnAmp, self.nAmpnAmp)
 
            if coeffVector is not None:
                crosstalkCoeffs = coeffVector
            else:
                crosstalkCoeffs = detector.getCrosstalk()
            if len(crosstalkCoeffs) == 1 and crosstalkCoeffs[0] == 0.0:
                return self
            self.coeffscoeffs = np.array(crosstalkCoeffs).reshape(self.crosstalkShapecrosstalkShape)
 
            if self.coeffscoeffs.shape != self.crosstalkShapecrosstalkShape:
                raise RuntimeError("Crosstalk coefficients do not match detector shape. "
                                   f"{self.crosstalkShape} {self.nAmp}")
 
            self.coeffErrcoeffErr = np.zeros(self.crosstalkShapecrosstalkShape)
            self.coeffNumcoeffNum = np.zeros(self.crosstalkShapecrosstalkShape, dtype=int)
            self.coeffValidcoeffValid = np.ones(self.crosstalkShapecrosstalkShape, dtype=bool)
            self.interChipinterChip = {}
 
            self.hasCrosstalkhasCrosstalk = True
            self.updateMetadataupdateMetadataupdateMetadata()
        return self
 
    @classmethod
    def fromDict(cls, dictionary):
        """Construct a calibration from a dictionary of properties.
 
        Must be implemented by the specific calibration subclasses.
 
        Parameters
        ----------
        dictionary : `dict`
            Dictionary of properties.
 
        Returns
        -------
        calib : `lsst.ip.isr.CalibType`
            Constructed calibration.
 
        Raises
        ------
        RuntimeError :
            Raised if the supplied dictionary is for a different
            calibration.
        """
        calib = cls()
 
        if calib._OBSTYPE != dictionary['metadata']['OBSTYPE']:
            raise RuntimeError(f"Incorrect crosstalk supplied.  Expected {calib._OBSTYPE}, "
                               f"found {dictionary['metadata']['OBSTYPE']}")
 
        calib.setMetadata(dictionary['metadata'])
 
        if 'detectorName' in dictionary:
            calib._detectorName = dictionary.get('detectorName')
        elif 'DETECTOR_NAME' in dictionary:
            calib._detectorName = dictionary.get('DETECTOR_NAME')
        elif 'DET_NAME' in dictionary['metadata']:
            calib._detectorName = dictionary['metadata']['DET_NAME']
        else:
            calib._detectorName = None
 
        if 'detectorSerial' in dictionary:
            calib._detectorSerial = dictionary.get('detectorSerial')
        elif 'DETECTOR_SERIAL' in dictionary:
            calib._detectorSerial = dictionary.get('DETECTOR_SERIAL')
        elif 'DET_SER' in dictionary['metadata']:
            calib._detectorSerial = dictionary['metadata']['DET_SER']
        else:
            calib._detectorSerial = None
 
        if 'detectorId' in dictionary:
            calib._detectorId = dictionary.get('detectorId')
        elif 'DETECTOR' in dictionary:
            calib._detectorId = dictionary.get('DETECTOR')
        elif 'DETECTOR' in dictionary['metadata']:
            calib._detectorId = dictionary['metadata']['DETECTOR']
        elif calib._detectorSerial:
            calib._detectorId = calib._detectorSerial
        else:
            calib._detectorId = None
 
        if 'instrument' in dictionary:
            calib._instrument = dictionary.get('instrument')
        elif 'INSTRUME' in dictionary['metadata']:
            calib._instrument = dictionary['metadata']['INSTRUME']
        else:
            calib._instrument = None
 
        calib.hasCrosstalk = dictionary.get('hasCrosstalk',
                                            dictionary['metadata'].get('HAS_CROSSTALK', False))
        if calib.hasCrosstalk:
            calib.nAmp = dictionary.get('nAmp', dictionary['metadata'].get('NAMP', 0))
            calib.crosstalkShape = (calib.nAmp, calib.nAmp)
            calib.coeffs = np.array(dictionary['coeffs']).reshape(calib.crosstalkShape)
            if 'coeffErr' in dictionary:
                calib.coeffErr = np.array(dictionary['coeffErr']).reshape(calib.crosstalkShape)
            else:
                calib.coeffErr = np.zeros_like(calib.coeffs)
            if 'coeffNum' in dictionary:
                calib.coeffNum = np.array(dictionary['coeffNum']).reshape(calib.crosstalkShape)
            else:
                calib.coeffNum = np.zeros_like(calib.coeffs, dtype=int)
            if 'coeffValid' in dictionary:
                calib.coeffValid = np.array(dictionary['coeffValid']).reshape(calib.crosstalkShape)
            else:
                calib.coeffValid = np.ones_like(calib.coeffs, dtype=bool)
 
            calib.interChip = dictionary.get('interChip', None)
            if calib.interChip:
                for detector in calib.interChip:
                    coeffVector = calib.interChip[detector]
                    calib.interChip[detector] = np.array(coeffVector).reshape(calib.crosstalkShape)
 
        calib.updateMetadata()
        return calib
 
    def toDict(self):
        """Return a dictionary containing the calibration properties.
 
        The dictionary should be able to be round-tripped through
        `fromDict`.
 
        Returns
        -------
        dictionary : `dict`
            Dictionary of properties.
        """
        self.updateMetadataupdateMetadataupdateMetadata()
 
        outDict = {}
        metadata = self.getMetadatagetMetadata()
        outDict['metadata'] = metadata
 
        outDict['hasCrosstalk'] = self.hasCrosstalkhasCrosstalk
        outDict['nAmp'] = self.nAmpnAmp
        outDict['crosstalkShape'] = self.crosstalkShapecrosstalkShape
 
        ctLength = self.nAmpnAmp*self.nAmpnAmp
        outDict['coeffs'] = self.coeffscoeffs.reshape(ctLength).tolist()
 
        if self.coeffErrcoeffErr is not None:
            outDict['coeffErr'] = self.coeffErrcoeffErr.reshape(ctLength).tolist()
        if self.coeffNumcoeffNum is not None:
            outDict['coeffNum'] = self.coeffNumcoeffNum.reshape(ctLength).tolist()
        if self.coeffValidcoeffValid is not None:
            outDict['coeffValid'] = self.coeffValidcoeffValid.reshape(ctLength).tolist()
 
        if self.interChipinterChip:
            outDict['interChip'] = dict()
            for detector in self.interChipinterChip:
                outDict['interChip'][detector] = self.interChipinterChip[detector].reshape(ctLength).tolist()
 
        return outDict
 
    @classmethod
    def fromTable(cls, tableList):
        """Construct calibration from a list of tables.
 
        This method uses the `fromDict` method to create the
        calibration, after constructing an appropriate dictionary from
        the input tables.
 
        Parameters
        ----------
        tableList : `list` [`lsst.afw.table.Table`]
            List of tables to use to construct the crosstalk
            calibration.
 
        Returns
        -------
        calib : `lsst.ip.isr.CrosstalkCalib`
            The calibration defined in the tables.
 
        """
        coeffTable = tableList[0]
 
        metadata = coeffTable.meta
        inDict = dict()
        inDict['metadata'] = metadata
        inDict['hasCrosstalk'] = metadata['HAS_CROSSTALK']
        inDict['nAmp'] = metadata['NAMP']
 
        inDict['coeffs'] = coeffTable['CT_COEFFS']
        if 'CT_ERRORS' in coeffTable.columns:
            inDict['coeffErr'] = coeffTable['CT_ERRORS']
        if 'CT_COUNTS' in coeffTable.columns:
            inDict['coeffNum'] = coeffTable['CT_COUNTS']
        if 'CT_VALID' in coeffTable.columns:
            inDict['coeffValid'] = coeffTable['CT_VALID']
 
        if len(tableList) > 1:
            inDict['interChip'] = dict()
            interChipTable = tableList[1]
            for record in interChipTable:
                inDict['interChip'][record['IC_SOURCE_DET']] = record['IC_COEFFS']
 
        return cls().fromDict(inDict)
 
    def toTable(self):
        """Construct a list of tables containing the information in this calibration.
 
        The list of tables should create an identical calibration
        after being passed to this class's fromTable method.
 
        Returns
        -------
        tableList : `list` [`lsst.afw.table.Table`]
            List of tables containing the crosstalk calibration
            information.
 
        """
        tableList = []
        self.updateMetadataupdateMetadataupdateMetadata()
        catalog = Table([{'CT_COEFFS': self.coeffscoeffs.reshape(self.nAmpnAmp*self.nAmpnAmp),
                          'CT_ERRORS': self.coeffErrcoeffErr.reshape(self.nAmpnAmp*self.nAmpnAmp),
                          'CT_COUNTS': self.coeffNumcoeffNum.reshape(self.nAmpnAmp*self.nAmpnAmp),
                          'CT_VALID': self.coeffValidcoeffValid.reshape(self.nAmpnAmp*self.nAmpnAmp),
                          }])
        # filter None, because astropy can't deal.
        inMeta = self.getMetadatagetMetadata().toDict()
        outMeta = {k: v for k, v in inMeta.items() if v is not None}
        outMeta.update({k: "" for k, v in inMeta.items() if v is None})
        catalog.meta = outMeta
        tableList.append(catalog)
 
        if self.interChipinterChip:
            interChipTable = Table([{'IC_SOURCE_DET': sourceDet,
                                     'IC_COEFFS': self.interChipinterChip[sourceDet].reshape(self.nAmpnAmp*self.nAmpnAmp)}
                                    for sourceDet in self.interChipinterChip.keys()])
            tableList.append(interChipTable)
        return tableList
 
    # Implementation methods.
    @staticmethod
    def extractAmp(image, amp, ampTarget, isTrimmed=False):
        """Extract the image data from an amp, flipped to match ampTarget.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image` or `lsst.afw.image.MaskedImage`
            Image containing the amplifier of interest.
        amp : `lsst.afw.cameraGeom.Amplifier`
            Amplifier on image to extract.
        ampTarget : `lsst.afw.cameraGeom.Amplifier`
            Target amplifier that the extracted image will be flipped
            to match.
        isTrimmed : `bool`
            The image is already trimmed.
            TODO : DM-15409 will resolve this.
 
        Returns
        -------
        output : `lsst.afw.image.Image`
            Image of the amplifier in the desired configuration.
        """
        X_FLIP = {lsst.afw.cameraGeom.ReadoutCorner.LL: False,
                  lsst.afw.cameraGeom.ReadoutCorner.LR: True,
                  lsst.afw.cameraGeom.ReadoutCorner.UL: False,
                  lsst.afw.cameraGeom.ReadoutCorner.UR: True}
        Y_FLIP = {lsst.afw.cameraGeom.ReadoutCorner.LL: False,
                  lsst.afw.cameraGeom.ReadoutCorner.LR: False,
                  lsst.afw.cameraGeom.ReadoutCorner.UL: True,
                  lsst.afw.cameraGeom.ReadoutCorner.UR: True}
 
        output = image[amp.getBBox() if isTrimmed else amp.getRawDataBBox()]
        thisAmpCorner = amp.getReadoutCorner()
        targetAmpCorner = ampTarget.getReadoutCorner()
 
        # Flipping is necessary only if the desired configuration doesn't match what we currently have
        xFlip = X_FLIP[targetAmpCorner] ^ X_FLIP[thisAmpCorner]
        yFlip = Y_FLIP[targetAmpCorner] ^ Y_FLIP[thisAmpCorner]
        return lsst.afw.math.flipImage(output, xFlip, yFlip)
 
    @staticmethod
    def calculateBackground(mi, badPixels=["BAD"]):
        """Estimate median background in image.
 
        Getting a great background model isn't important for crosstalk correction,
        since the crosstalk is at a low level. The median should be sufficient.
 
        Parameters
        ----------
        mi : `lsst.afw.image.MaskedImage`
            MaskedImage for which to measure background.
        badPixels : `list` of `str`
            Mask planes to ignore.
        Returns
        -------
        bg : `float`
            Median background level.
        """
        mask = mi.getMask()
        stats = lsst.afw.math.StatisticsControl()
        stats.setAndMask(mask.getPlaneBitMask(badPixels))
        return lsst.afw.math.makeStatistics(mi, lsst.afw.math.MEDIAN, stats).getValue()
 
    def subtractCrosstalk(self, thisExposure, sourceExposure=None, crosstalkCoeffs=None,
                          badPixels=["BAD"], minPixelToMask=45000,
                          crosstalkStr="CROSSTALK", isTrimmed=False,
                          backgroundMethod="None"):
        """Subtract the crosstalk from thisExposure, optionally using a different source.
 
        We set the mask plane indicated by ``crosstalkStr`` in a target amplifier
        for pixels in a source amplifier that exceed ``minPixelToMask``. Note that
        the correction is applied to all pixels in the amplifier, but only those
        that have a substantial crosstalk are masked with ``crosstalkStr``.
 
        The uncorrected image is used as a template for correction. This is good
        enough if the crosstalk is small (e.g., coefficients < ~ 1e-3), but if it's
        larger you may want to iterate.
 
        Parameters
        ----------
        thisExposure : `lsst.afw.image.Exposure`
            Exposure for which to subtract crosstalk.
        sourceExposure : `lsst.afw.image.Exposure`, optional
            Exposure to use as the source of the crosstalk.  If not set,
            thisExposure is used as the source (intra-detector crosstalk).
        crosstalkCoeffs : `numpy.ndarray`, optional.
            Coefficients to use to correct crosstalk.
        badPixels : `list` of `str`
            Mask planes to ignore.
        minPixelToMask : `float`
            Minimum pixel value (relative to the background level) in
            source amplifier for which to set ``crosstalkStr`` mask plane
            in target amplifier.
        crosstalkStr : `str`
            Mask plane name for pixels greatly modified by crosstalk
            (above minPixelToMask).
        isTrimmed : `bool`
            The image is already trimmed.
            This should no longer be needed once DM-15409 is resolved.
        backgroundMethod : `str`
            Method used to subtract the background.  "AMP" uses
            amplifier-by-amplifier background levels, "DETECTOR" uses full
            exposure/maskedImage levels.  Any other value results in no
            background subtraction.
        """
        mi = thisExposure.getMaskedImage()
        mask = mi.getMask()
        detector = thisExposure.getDetector()
        if self.hasCrosstalkhasCrosstalk is False:
            self.fromDetectorfromDetectorfromDetector(detector, coeffVector=crosstalkCoeffs)
 
        numAmps = len(detector)
        if numAmps != self.nAmpnAmp:
            raise RuntimeError(f"Crosstalk built for {self.nAmp} in {self._detectorName}, received "
                               f"{numAmps} in {detector.getName()}")
 
        if sourceExposure:
            source = sourceExposure.getMaskedImage()
            sourceDetector = sourceExposure.getDetector()
        else:
            source = mi
            sourceDetector = detector
 
        if crosstalkCoeffs is not None:
            coeffs = crosstalkCoeffs
        else:
            coeffs = self.coeffscoeffs
        self.loglog.debug("CT COEFF: %s", coeffs)
        # Set background level based on the requested method.  The
        # thresholdBackground holds the offset needed so that we only mask
        # pixels high relative to the background, not in an absolute
        # sense.
        thresholdBackground = self.calculateBackgroundcalculateBackground(source, badPixels)
 
        backgrounds = [0.0 for amp in sourceDetector]
        if backgroundMethod is None:
            pass
        elif backgroundMethod == "AMP":
            backgrounds = [self.calculateBackgroundcalculateBackground(source[amp.getBBox()], badPixels)
                           for amp in sourceDetector]
        elif backgroundMethod == "DETECTOR":
            backgrounds = [self.calculateBackgroundcalculateBackground(source, badPixels) for amp in sourceDetector]
 
        # Set the crosstalkStr bit for the bright pixels (those which will have
        # significant crosstalk correction)
        crosstalkPlane = mask.addMaskPlane(crosstalkStr)
        footprints = lsst.afw.detection.FootprintSet(source,
                                                     lsst.afw.detection.Threshold(minPixelToMask
                                                                                  + thresholdBackground))
        footprints.setMask(mask, crosstalkStr)
        crosstalk = mask.getPlaneBitMask(crosstalkStr)
 
        # Define a subtrahend image to contain all the scaled crosstalk signals
        subtrahend = source.Factory(source.getBBox())
        subtrahend.set((0, 0, 0))
 
        coeffs = coeffs.transpose()
        for ii, iAmp in enumerate(sourceDetector):
            iImage = subtrahend[iAmp.getBBox() if isTrimmed else iAmp.getRawDataBBox()]
            for jj, jAmp in enumerate(detector):
                if coeffs[ii, jj] == 0.0:
                    continue
                jImage = self.extractAmpextractAmp(mi, jAmp, iAmp, isTrimmed)
                jImage.getMask().getArray()[:] &= crosstalk  # Remove all other masks
                jImage -= backgrounds[jj]
                iImage.scaledPlus(coeffs[ii, jj], jImage)
 
        # Set crosstalkStr bit only for those pixels that have been significantly modified (i.e., those
        # masked as such in 'subtrahend'), not necessarily those that are bright originally.
        mask.clearMaskPlane(crosstalkPlane)
        mi -= subtrahend  # also sets crosstalkStr bit for bright pixels
 
 
class CrosstalkConfig(Config):
    """Configuration for intra-detector crosstalk removal."""
    minPixelToMask = Field(
        dtype=float,
        doc="Set crosstalk mask plane for pixels over this value.",
        default=45000
    )
    crosstalkMaskPlane = Field(
        dtype=str,
        doc="Name for crosstalk mask plane.",
        default="CROSSTALK"
    )
    crosstalkBackgroundMethod = ChoiceField(
        dtype=str,
        doc="Type of background subtraction to use when applying correction.",
        default="None",
        allowed={
            "None": "Do no background subtraction.",
            "AMP": "Subtract amplifier-by-amplifier background levels.",
            "DETECTOR": "Subtract detector level background."
        },
    )
    useConfigCoefficients = Field(
        dtype=bool,
        doc="Ignore the detector crosstalk information in favor of CrosstalkConfig values?",
        default=False,
    )
    crosstalkValues = ListField(
        dtype=float,
        doc=("Amplifier-indexed crosstalk coefficients to use.  This should be arranged as a 1 x nAmp**2 "
             "list of coefficients, such that when reshaped by crosstalkShape, the result is nAmp x nAmp. "
             "This matrix should be structured so CT * [amp0 amp1 amp2 ...]^T returns the column "
             "vector [corr0 corr1 corr2 ...]^T."),
        default=[0.0],
    )
    crosstalkShape = ListField(
        dtype=int,
        doc="Shape of the coefficient array.  This should be equal to [nAmp, nAmp].",
        default=[1],
    )
 
    def getCrosstalk(self, detector=None):
        """Return a 2-D numpy array of crosstalk coefficients in the proper shape.
 
        Parameters
        ----------
        detector : `lsst.afw.cameraGeom.detector`
            Detector that is to be crosstalk corrected.
 
        Returns
        -------
        coeffs : `numpy.ndarray`
            Crosstalk coefficients that can be used to correct the detector.
 
        Raises
        ------
        RuntimeError
            Raised if no coefficients could be generated from this detector/configuration.
        """
        if self.useConfigCoefficientsuseConfigCoefficients is True:
            coeffs = np.array(self.crosstalkValuescrosstalkValues).reshape(self.crosstalkShapecrosstalkShape)
            if detector is not None:
                nAmp = len(detector)
                if coeffs.shape != (nAmp, nAmp):
                    raise RuntimeError("Constructed crosstalk coeffients do not match detector shape. "
                                       f"{coeffs.shape} {nAmp}")
            return coeffs
        elif detector is not None and detector.hasCrosstalk() is True:
            # Assume the detector defines itself consistently.
            return detector.getCrosstalk()
        else:
            raise RuntimeError("Attempted to correct crosstalk without crosstalk coefficients")
 
    def hasCrosstalk(self, detector=None):
        """Return a boolean indicating if crosstalk coefficients exist.
 
        Parameters
        ----------
        detector : `lsst.afw.cameraGeom.detector`
            Detector that is to be crosstalk corrected.
 
        Returns
        -------
        hasCrosstalk : `bool`
            True if this detector/configuration has crosstalk coefficients defined.
        """
        if self.useConfigCoefficientsuseConfigCoefficients is True and self.crosstalkValuescrosstalkValues is not None:
            return True
        elif detector is not None and detector.hasCrosstalk() is True:
            return True
        else:
            return False
 
 
class CrosstalkTask(Task):
    """Apply intra-detector crosstalk correction."""
    ConfigClass = CrosstalkConfig
    _DefaultName = 'isrCrosstalk'
 
    def prepCrosstalk(self, dataRef, crosstalk=None):
        """Placeholder for crosstalk preparation method, e.g., for inter-detector crosstalk.
 
        Parameters
        ----------
        dataRef : `daf.persistence.butlerSubset.ButlerDataRef`
            Butler reference of the detector data to be processed.
        crosstalk : `~lsst.ip.isr.CrosstalkConfig`
            Crosstalk calibration that will be used.
 
        See also
        --------
        lsst.obs.decam.crosstalk.DecamCrosstalkTask.prepCrosstalk
        """
        return
 
    def run(self, exposure, crosstalk=None,
            crosstalkSources=None, isTrimmed=False, camera=None):
        """Apply intra-detector crosstalk correction
 
        Parameters
        ----------
        exposure : `lsst.afw.image.Exposure`
            Exposure for which to remove crosstalk.
        crosstalkCalib : `lsst.ip.isr.CrosstalkCalib`, optional
            External crosstalk calibration to apply.  Constructed from
            detector if not found.
        crosstalkSources : `defaultdict`, optional
            Image data for other detectors that are sources of
            crosstalk in exposure.  The keys are expected to be names
            of the other detectors, with the values containing
            `lsst.afw.image.Exposure` at the same level of processing
            as ``exposure``.
            The default for intra-detector crosstalk here is None.
        isTrimmed : `bool`, optional
            The image is already trimmed.
            This should no longer be needed once DM-15409 is resolved.
        camera : `lsst.afw.cameraGeom.Camera`, optional
            Camera associated with this exposure.  Only used for
            inter-chip matching.
 
        Raises
        ------
        RuntimeError
            Raised if called for a detector that does not have a
            crosstalk correction.  Also raised if the crosstalkSource
            is not an expected type.
        """
        if not crosstalk:
            crosstalk = CrosstalkCalib(log=self.log)
            crosstalk = crosstalk.fromDetector(exposure.getDetector(),
                                               coeffVector=self.config.crosstalkValues)
        if not crosstalk.log:
            crosstalk.log = self.log
        if not crosstalk.hasCrosstalk:
            raise RuntimeError("Attempted to correct crosstalk without crosstalk coefficients.")
 
        else:
            self.log.info("Applying crosstalk correction.")
            crosstalk.subtractCrosstalk(exposure, crosstalkCoeffs=crosstalk.coeffs,
                                        minPixelToMask=self.config.minPixelToMask,
                                        crosstalkStr=self.config.crosstalkMaskPlane, isTrimmed=isTrimmed,
                                        backgroundMethod=self.config.crosstalkBackgroundMethod)
 
            if crosstalk.interChip:
                if crosstalkSources:
                    # Parse crosstalkSources: Identify which detectors we have available
                    if isinstance(crosstalkSources[0], lsst.afw.image.Exposure):
                        # Received afwImage.Exposure
                        sourceNames = [exp.getDetector().getName() for exp in crosstalkSources]
                    elif isinstance(crosstalkSources[0], lsst.daf.butler.DeferredDatasetHandle):
                        # Received dafButler.DeferredDatasetHandle
                        detectorList = [source.dataId['detector'] for source in crosstalkSources]
                        sourceNames = [camera[detector].getName() for detector in detectorList]
                    else:
                        raise RuntimeError("Unknown object passed as crosstalk sources.",
                                           type(crosstalkSources[0]))
 
                    for detName in crosstalk.interChip:
                        if detName not in sourceNames:
                            self.log.warning("Crosstalk lists %s, not found in sources: %s",
                                             detName, sourceNames)
                            continue
                        # Get the coefficients.
                        interChipCoeffs = crosstalk.interChip[detName]
 
                        sourceExposure = crosstalkSources[sourceNames.index(detName)]
                        if isinstance(sourceExposure, lsst.daf.butler.DeferredDatasetHandle):
                            # Dereference the dafButler.DeferredDatasetHandle.
                            sourceExposure = sourceExposure.get()
                        if not isinstance(sourceExposure, lsst.afw.image.Exposure):
                            raise RuntimeError("Unknown object passed as crosstalk sources.",
                                               type(sourceExposure))
 
                        self.log.info("Correcting detector %s with ctSource %s",
                                      exposure.getDetector().getName(),
                                      sourceExposure.getDetector().getName())
                        crosstalk.subtractCrosstalk(exposure, sourceExposure=sourceExposure,
                                                    crosstalkCoeffs=interChipCoeffs,
                                                    minPixelToMask=self.config.minPixelToMask,
                                                    crosstalkStr=self.config.crosstalkMaskPlane,
                                                    isTrimmed=isTrimmed,
                                                    backgroundMethod=self.config.crosstalkBackgroundMethod)
                else:
                    self.log.warning("Crosstalk contains interChip coefficients, but no sources found!")
 
 
class NullCrosstalkTask(CrosstalkTask):
    def run(self, exposure, crosstalkSources=None):
        self.log.info("Not performing any crosstalk correction")