lsst.ip.isr.crosstalk.CrosstalkCalib Class Reference

Inheritance diagram for lsst.ip.isr.crosstalk.CrosstalkCalib:

Public Member Functions
	__init__ (self, detector=None, nAmp=0, **kwargs)
	updateMetadata (self, setDate=False, **kwargs)
	fromDetector (self, detector, coeffVector=None, coeffSqrVector=None)
	fromDict (cls, dictionary)
	toDict (self)
	fromTable (cls, tableList)
	toTable (self)
	subtractCrosstalk (self, thisExposure, sourceExposure=None, crosstalkCoeffs=None, crosstalkCoeffsSqr=None, crosstalkCoeffsValid=None, badPixels=["BAD"], minPixelToMask=45000, doSubtrahendMasking=False, crosstalkStr="CROSSTALK", isTrimmed=None, backgroundMethod="None", doSqrCrosstalk=False, fullAmplifier=False, parallelOverscan=None, detectorConfig=None, badAmpDict=None, ignoreVariance=False)
	requiredAttributes (self)
	requiredAttributes (self, value)
	__str__ (self)
	__eq__ (self, other)
	metadata (self)
	getMetadata (self)
	setMetadata (self, metadata)
	updateMetadataFromExposures (self, exposures)
	calibInfoFromDict (self, dictionary)
	determineCalibClass (cls, metadata, message)
	readText (cls, filename, **kwargs)
	writeText (self, filename, format="auto")
	readFits (cls, filename, **kwargs)
	writeFits (self, filename)
	validate (self, other=None)
	apply (self, target)

Static Public Member Functions
	extractAmp (image, ampToFlip, ampTarget, isTrimmed=False, fullAmplifier=False, parallelOverscan=None)
	calculateBackground (mi, badPixels=["BAD"])

Public Attributes
bool	hasCrosstalk = False
int	nAmp = nAmp if nAmp else 0
tuple	crosstalkShape = (self.nAmp, self.nAmp)
int	coeffs = np.zeros(self.crosstalkShape) if self.nAmp else None
int	coeffErr = np.zeros(self.crosstalkShape) if self.nAmp else None
int	coeffNum
int	coeffValid
int	coeffsSqr = np.zeros(self.crosstalkShape) if self.nAmp else None
int	coeffErrSqr = np.zeros(self.crosstalkShape) if self.nAmp else None
int	ampGainRatios = np.zeros(self.crosstalkShape) if self.nAmp else None
int	fitGains = np.zeros(self.nAmp) if self.nAmp else None
str	crosstalkRatiosUnits = 'adu' if self.nAmp else None
dict	interChip = {}
	requiredAttributes = set(["_OBSTYPE", "_SCHEMA", "_VERSION"])
	log = log if log else logging.getLogger(__name__)

Static Protected Member Functions
	_getAppropriateBBox (amp, isTrimmed, fullAmplifier)

Protected Attributes
	_instrument = None
	_raftName = None
	_slotName = None
	_detectorName = None
	_detectorSerial = None
	_detectorId = None
	_filter = None
str	_calibId = None
	_seqfile = None
	_seqname = None
	_seqcksum = None
	_metadata = PropertyList()
	_requiredAttributes

Static Protected Attributes
str	_OBSTYPE = "generic"
str	_SCHEMA = "NO SCHEMA"
int	_VERSION = 0

Detailed Description

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 : `numpy.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 : `numpy.ndarray`, optional
    A matrix (as defined by ``coeffs``) containing the standard
    distribution of the crosstalk measurements.
coeffNum : `numpy.ndarray`, optional
    A matrix containing the number of pixel pairs used to measure
    the ``coeffs`` and ``coeffErr``.
coeffValid : `numpy.ndarray`, optional
    A matrix of Boolean values indicating if the coefficient is
    valid, defined as abs(coeff) > coeffErr / sqrt(coeffNum).
coeffsSqr : `numpy.ndarray`, optional
    A matrix containing potential quadratic crosstalk coefficients
    (see e.g., Snyder+21, 2001.03223). coeffsSqr[i][j]
    contains the coefficients to calculate the contribution
    amplifier_j has on amplifier_i (each row[i] contains the
    corrections for detector_i).
coeffErrSqr : `numpy.ndarray`, optional
    A matrix (as defined by ``coeffsSqr``) containing the standard
    distribution of the quadratic term of the crosstalk measurements.
interChip : `dict` [`numpy.ndarray`]
    A dictionary keyed by detectorName containing ``coeffs``
    matrices used to correct for inter-chip crosstalk with a
    source on the detector indicated.

Version 1.1 adds quadratic coefficients, a matrix with the ratios
of amplifiers gains per detector, and a field to indicate the units
of the numerator and denominator of the source and target signals, with
"adu" meaning "ADU / ADU" and "electron" meaning "e- / e-".

Version 1.2 adds the original gains used in the crosstalk fit.

Definition at line 43 of file crosstalk.py.

Constructor & Destructor Documentation

__init__()

lsst.ip.isr.crosstalk.CrosstalkCalib.__init__	(		self,
			detector = None,
			nAmp = 0,
		**	kwargs )

Definition at line 107 of file crosstalk.py.

    def __init__(self, detector=None, nAmp=0, **kwargs):
        self.hasCrosstalk = False
        self.nAmp = nAmp if nAmp else 0
        self.crosstalkShape = (self.nAmp, self.nAmp)
 
        self.coeffs = np.zeros(self.crosstalkShape) if self.nAmp else None
        self.coeffErr = np.zeros(self.crosstalkShape) if self.nAmp else None
        self.coeffNum = np.zeros(self.crosstalkShape,
                                 dtype=int) if self.nAmp else None
        self.coeffValid = np.ones(self.crosstalkShape,
                                  dtype=bool) if self.nAmp else None
        # Quadratic terms, if any.
        self.coeffsSqr = np.zeros(self.crosstalkShape) if self.nAmp else None
        self.coeffErrSqr = np.zeros(self.crosstalkShape) if self.nAmp else None
 
        # Gain ratios
        self.ampGainRatios = np.zeros(self.crosstalkShape) if self.nAmp else None
 
        # Gains used for fit.
        self.fitGains = np.zeros(self.nAmp) if self.nAmp else None
 
        # Units
        self.crosstalkRatiosUnits = 'adu' if self.nAmp else None
 
        self.interChip = {}
 
        super().__init__(**kwargs)
        self.requiredAttributes.update(['hasCrosstalk', 'nAmp', 'coeffs',
                                        'coeffErr', 'coeffNum', 'coeffValid',
                                        'coeffsSqr', 'coeffErrSqr',
                                        'ampGainRatios', 'crosstalkRatiosUnits',
                                        'fitGains', 'interChip'])
        if detector:
            self.fromDetector(detector)
 

Member Function Documentation

__eq__()

lsst.ip.isr.calibType.IsrCalib.__eq__ ( self, other )

inherited

Calibration equivalence.

Running ``calib.log.setLevel(0)`` enables debug statements to
identify problematic fields.

Definition at line 105 of file calibType.py.

    def __eq__(self, other):
        """Calibration equivalence.
 
        Running ``calib.log.setLevel(0)`` enables debug statements to
        identify problematic fields.
        """
        if not isinstance(other, self.__class__):
            self.log.debug("Incorrect class type: %s %s", self.__class__, other.__class__)
            return False
 
        for attr in self._requiredAttributes:
            attrSelf = getattr(self, attr)
            attrOther = getattr(other, attr)
 
            if isinstance(attrSelf, dict):
                # Dictionary of arrays.
                if attrSelf.keys() != attrOther.keys():
                    self.log.debug("Dict Key Failure: %s %s %s", attr, attrSelf.keys(), attrOther.keys())
                    return False
                for key in attrSelf:
                    try:
                        if not np.allclose(attrSelf[key], attrOther[key], equal_nan=True):
                            self.log.debug("Array Failure: %s %s %s", key, attrSelf[key], attrOther[key])
                            return False
                    except TypeError:
                        # If it is not something numpy can handle
                        # (it's not a number or array of numbers),
                        # then it needs to have its own equivalence
                        # operator.
                        if np.all(attrSelf[key] != attrOther[key]):
                            return False
            elif (isinstance(attrSelf, np.ndarray) or isinstance(attrSelf, Column)
                  or isinstance(attrOther, np.ndarray) or isinstance(attrOther, Column)):
                # Bare array.
                if isinstance(attrSelf[0], (str, np.str_, np.bytes_)):
                    if not np.all(attrSelf == attrOther):
                        self.log.debug("Array Failure: %s %s %s", attr, attrSelf, attrOther)
                        return False
                else:
                    if not np.allclose(attrSelf, attrOther, equal_nan=True):
                        self.log.debug("Array Failure: %s %s %s", attr, attrSelf, attrOther)
                        return False
            elif type(attrSelf) is not type(attrOther):
                if set([attrSelf, attrOther]) == set([None, ""]):
                    # Fits converts None to "", but None is not "".
                    continue
                self.log.debug("Type Failure: %s %s %s %s %s", attr, type(attrSelf), type(attrOther),
                               attrSelf, attrOther)
                return False
            else:
                if attrSelf != attrOther:
                    self.log.debug("Value Failure: %s %s %s", attr, attrSelf, attrOther)
                    return False
 
        return True
 

__str__()

lsst.ip.isr.calibType.IsrCalib.__str__ ( self )

inherited

Definition at line 102 of file calibType.py.

    def __str__(self):
        return f"{self.__class__.__name__}(obstype={self._OBSTYPE}, detector={self._detectorName}, )"
 

_getAppropriateBBox()

lsst.ip.isr.crosstalk.CrosstalkCalib._getAppropriateBBox ( amp, isTrimmed, fullAmplifier )

staticprotected

Get the appropriate bounding box from an amplifier.

Parameters
----------
amp : `lsst.afw.cameraGeom.Amplifier`
    Amplifier to get bounding box.
isTrimmed : `bool`
    Is this a trimmed image?
fullAmplifier : `bool`
    Extract full amplifier for an untrimmed image?

Definition at line 544 of file crosstalk.py.

    def _getAppropriateBBox(amp, isTrimmed, fullAmplifier):
        """Get the appropriate bounding box from an amplifier.
 
        Parameters
        ----------
        amp : `lsst.afw.cameraGeom.Amplifier`
            Amplifier to get bounding box.
        isTrimmed : `bool`
            Is this a trimmed image?
        fullAmplifier : `bool`
            Extract full amplifier for an untrimmed image?
        """
        if isTrimmed:
            return amp.getBBox()
        elif fullAmplifier and not isTrimmed:
            return amp.getRawBBox()
        else:
            return amp.getRawDataBBox()
 

apply()

lsst.ip.isr.calibType.IsrCalib.apply ( self, target )

inherited

Method to apply the calibration to the target object.

Parameters
----------
target : `object`
    Thing to validate against.

Returns
-------
valid : `bool`
    Returns true if the calibration was applied correctly.

Raises
------
NotImplementedError
    Raised if not implemented.

Definition at line 706 of file calibType.py.

    def apply(self, target):
        """Method to apply the calibration to the target object.
 
        Parameters
        ----------
        target : `object`
            Thing to validate against.
 
        Returns
        -------
        valid : `bool`
            Returns true if the calibration was applied correctly.
 
        Raises
        ------
        NotImplementedError
            Raised if not implemented.
        """
        raise NotImplementedError("Must be implemented by subclass.")
 
 

calculateBackground()

lsst.ip.isr.crosstalk.CrosstalkCalib.calculateBackground ( mi, badPixels = ["BAD"] )

static

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.

Definition at line 564 of file crosstalk.py.

    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()
 

calibInfoFromDict()

lsst.ip.isr.calibType.IsrCalib.calibInfoFromDict ( self, dictionary )

inherited

Handle common keywords.

This isn't an ideal solution, but until all calibrations
expect to find everything in the metadata, they still need to
search through dictionaries.

Parameters
----------
dictionary : `dict` or `lsst.daf.base.PropertyList`
    Source for the common keywords.

Raises
------
RuntimeError
    Raised if the dictionary does not match the expected OBSTYPE.

Definition at line 325 of file calibType.py.

    def calibInfoFromDict(self, dictionary):
        """Handle common keywords.
 
        This isn't an ideal solution, but until all calibrations
        expect to find everything in the metadata, they still need to
        search through dictionaries.
 
        Parameters
        ----------
        dictionary : `dict` or `lsst.daf.base.PropertyList`
            Source for the common keywords.
 
        Raises
        ------
        RuntimeError
            Raised if the dictionary does not match the expected OBSTYPE.
        """
 
        def search(haystack, needles):
            """Search dictionary 'haystack' for an entry in 'needles'
            """
            test = [haystack.get(x) for x in needles]
            test = set([x for x in test if x is not None])
            if len(test) == 0:
                if "metadata" in haystack:
                    return search(haystack["metadata"], needles)
                else:
                    return None
            elif len(test) == 1:
                value = list(test)[0]
                if value == "":
                    return None
                else:
                    return value
            else:
                raise ValueError(f"Too many values found: {len(test)} {test} {needles}")
 
        if "metadata" in dictionary:
            metadata = dictionary["metadata"]
 
            if self._OBSTYPE != metadata["OBSTYPE"]:
                raise RuntimeError(f"Incorrect calibration supplied.  Expected {self._OBSTYPE}, "
                                   f"found {metadata['OBSTYPE']}")
 
        if (value := search(dictionary, ["INSTRUME", "instrument"])) is not None:
            self._instrument = value
        if (value := search(dictionary, ["RAFTNAME"])) is not None:
            self._slotName = value
        if (value := search(dictionary, ["DETECTOR", "detectorId"])) is not None:
            self._detectorId = value
        if (value := search(dictionary, ["DET_NAME", "DETECTOR_NAME", "detectorName"])) is not None:
            self._detectorName = value
        if (value := search(dictionary, ["DET_SER", "DETECTOR_SERIAL", "detectorSerial"])) is not None:
            self._detectorSerial = value
        if (value := search(dictionary, ["FILTER", "filterName"])) is not None:
            self._filter = value
        if (value := search(dictionary, ["CALIB_ID"])) is not None:
            self._calibId = value
        if (value := search(dictionary, ["SEQFILE"])) is not None:
            self._seqfile = value
        if (value := search(dictionary, ["SEQNAME"])) is not None:
            self._seqname = value
        if (value := search(dictionary, ["SEQCKSUM"])) is not None:
            self._seqcksum = value
 

determineCalibClass()

lsst.ip.isr.calibType.IsrCalib.determineCalibClass ( cls, metadata, message )

inherited

Attempt to find calibration class in metadata.

Parameters
----------
metadata : `dict` or `lsst.daf.base.PropertyList`
    Metadata possibly containing a calibration class entry.
message : `str`
    Message to include in any errors.

Returns
-------
calibClass : `object`
    The class to use to read the file contents.  Should be an
    `lsst.ip.isr.IsrCalib` subclass.

Raises
------
ValueError
    Raised if the resulting calibClass is the base
    `lsst.ip.isr.IsrClass` (which does not implement the
    content methods).

Definition at line 391 of file calibType.py.

    def determineCalibClass(cls, metadata, message):
        """Attempt to find calibration class in metadata.
 
        Parameters
        ----------
        metadata : `dict` or `lsst.daf.base.PropertyList`
            Metadata possibly containing a calibration class entry.
        message : `str`
            Message to include in any errors.
 
        Returns
        -------
        calibClass : `object`
            The class to use to read the file contents.  Should be an
            `lsst.ip.isr.IsrCalib` subclass.
 
        Raises
        ------
        ValueError
            Raised if the resulting calibClass is the base
            `lsst.ip.isr.IsrClass` (which does not implement the
            content methods).
        """
        calibClassName = metadata.get("CALIBCLS")
        if calibClassName is None:
            calibClassName = metadata.get("fileType")
            if calibClassName == "shutterMotionProfile":
                calibClassName = "lsst.ip.isr.ShutterMotionProfile"
        calibClass = doImport(calibClassName) if calibClassName is not None else cls
        if calibClass is IsrCalib:
            raise ValueError(f"Cannot use base class to read calibration data: {message}")
        return calibClass
 

extractAmp()

lsst.ip.isr.crosstalk.CrosstalkCalib.extractAmp ( image, ampToFlip, ampTarget, isTrimmed = False, fullAmplifier = False, parallelOverscan = None )

static

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`, optional
    The image is already trimmed.
fullAmplifier : `bool`, optional
    Use full amplifier and not just imaging region.
parallelOverscan : `bool`, optional
    This has been deprecated and is unused, and will be removed
    after v29.

Returns
-------
output : `lsst.afw.image.Image`
    Amplifier from image, flipped to desired configuration.
    This will always return a copy of the original data.

Definition at line 489 of file crosstalk.py.

    def extractAmp(image, ampToFlip, ampTarget, isTrimmed=False, fullAmplifier=False, parallelOverscan=None):
        """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`, optional
            The image is already trimmed.
        fullAmplifier : `bool`, optional
            Use full amplifier and not just imaging region.
        parallelOverscan : `bool`, optional
            This has been deprecated and is unused, and will be removed
            after v29.
 
        Returns
        -------
        output : `lsst.afw.image.Image`
            Amplifier from image, flipped to desired configuration.
            This will always return a copy of the original data.
        """
        if parallelOverscan is not None:
            warnings.warn(
                "The parallelOverscan option has been deprecated and will be removed after v29.",
                FutureWarning,
            )
 
        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}
 
        bbox = CrosstalkCalib._getAppropriateBBox(ampToFlip, isTrimmed, fullAmplifier)
        output = image[bbox]
 
        sourceAmpCorner = ampToFlip.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[sourceAmpCorner]
        yFlip = Y_FLIP[targetAmpCorner] ^ Y_FLIP[sourceAmpCorner]
        # This always makes a copy of the image.
        return lsst.afw.math.flipImage(output, xFlip, yFlip)
 

fromDetector()

lsst.ip.isr.crosstalk.CrosstalkCalib.fromDetector	(		self,
			detector,
			coeffVector = None,
			coeffSqrVector = 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()``.
coeffSqrVector : `numpy.array`, optional
    Quadratic crosstalk coefficients.

Returns
-------
calib : `lsst.ip.isr.CrosstalkCalib`
    The calibration constructed from the detector.

Reimplemented from lsst.ip.isr.calibType.IsrCalib.

Definition at line 167 of file crosstalk.py.

    def fromDetector(self, detector, coeffVector=None, coeffSqrVector=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()``.
        coeffSqrVector : `numpy.array`, optional
            Quadratic crosstalk coefficients.
 
        Returns
        -------
        calib : `lsst.ip.isr.CrosstalkCalib`
            The calibration constructed from the detector.
 
        """
        self._detectorId = detector.getId()
        self._detectorName = detector.getName()
        self._detectorSerial = detector.getSerial()
 
        self.nAmp = len(detector)
        self.crosstalkShape = (self.nAmp, self.nAmp)
 
        if coeffVector is not None:
            crosstalkCoeffs = coeffVector
        else:
            crosstalkCoeffs = detector.getCrosstalk()
        if coeffSqrVector is not None:
            self.coeffsSqr = coeffSqrVector
        else:
            self.coeffsSqr = np.zeros(self.crosstalkShape)
        if len(crosstalkCoeffs) == 1 and crosstalkCoeffs[0] == 0.0:
            return self
        self.coeffs = np.array(crosstalkCoeffs).reshape(self.crosstalkShape)
 
        if self.coeffs.shape != self.crosstalkShape:
            raise RuntimeError("Crosstalk coefficients do not match detector shape. "
                               f"{self.crosstalkShape} {self.nAmp}")
        # Set default as in __init__
        self.coeffErr = np.zeros(self.crosstalkShape)
        self.coeffNum = np.zeros(self.crosstalkShape, dtype=int)
        self.coeffValid = np.ones(self.crosstalkShape, dtype=bool)
        self.coeffErrSqr = np.zeros(self.crosstalkShape)
        self.ampGainRatios = np.zeros(self.crosstalkShape)
        self.fitGains = np.zeros(self.nAmp)
        self.crosstalkRatiosUnits = 'adu'
 
        self.interChip = {}
 
        self.hasCrosstalk = True
        self.updateMetadata()
 
        return self
 

fromDict()

lsst.ip.isr.crosstalk.CrosstalkCalib.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.

Reimplemented from lsst.ip.isr.calibType.IsrCalib.

Definition at line 226 of file crosstalk.py.

    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)
            calib.crosstalkRatiosUnits = dictionary.get(
                'crosstalkRatiosUnits',
                dictionary['metadata'].get('CROSSTALK_RATIOS_UNITS', 'adu'))
            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)
            if 'coeffsSqr' in dictionary:
                calib.coeffsSqr = np.array(dictionary['coeffsSqr']).reshape(calib.crosstalkShape)
            else:
                calib.coeffsSqr = np.zeros_like(calib.coeffs)
            if 'coeffErrSqr' in dictionary:
                calib.coeffErrSqr = np.array(dictionary['coeffErrSqr']).reshape(calib.crosstalkShape)
            else:
                calib.coeffErrSqr = np.zeros_like(calib.coeffs)
            if 'ampGainRatios' in dictionary:
                calib.ampGainRatios = np.array(dictionary['ampGainRatios']).reshape(calib.crosstalkShape)
            else:
                calib.ampGainRatios = np.zeros_like(calib.coeffs)
            if 'fitGains' in dictionary:
                # Compatibility for matrices that were stored with fitGains
                # of length 1.
                fitGains = np.array(dictionary['fitGains'])
                if len(fitGains) == 1:
                    # Expand to the correct number of amps, all zero (unknown).
                    calib.fitGains = np.zeros(calib.nAmp)
                else:
                    calib.fitGains = np.array(dictionary['fitGains']).reshape(calib.nAmp)
            else:
                calib.fitGains = np.zeros(calib.nAmp)
 
            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
 

fromTable()

lsst.ip.isr.crosstalk.CrosstalkCalib.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.

Reimplemented from lsst.ip.isr.calibType.IsrCalib.

Definition at line 393 of file crosstalk.py.

    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']
        calibVersion = metadata['CROSSTALK_VERSION']
        if calibVersion < 1.1:
            inDict['crosstalkRatiosUnits'] = 'adu'
        else:
            inDict['crosstalkRatiosUnits'] = metadata['CROSSTALK_RATIOS_UNITS']
        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 'CT_COEFFS_SQR' in coeffTable.columns:
            inDict['coeffsSqr'] = coeffTable['CT_COEFFS_SQR']
        if 'CT_ERRORS_SQR' in coeffTable.columns:
            inDict['coeffErrSqr'] = coeffTable['CT_ERRORS_SQR']
        if 'CT_AMP_GAIN_RATIOS' in coeffTable.columns:
            inDict['ampGainRatios'] = coeffTable['CT_AMP_GAIN_RATIOS']
        if 'CT_FIT_GAINS' in coeffTable.columns:
            inDict['fitGains'] = coeffTable['CT_FIT_GAINS']
 
        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)
 

getMetadata()

lsst.ip.isr.calibType.IsrCalib.getMetadata ( self )

inherited

Retrieve metadata associated with this calibration.

Returns
-------
meta : `lsst.daf.base.PropertyList`
    Metadata. The returned `~lsst.daf.base.PropertyList` can be
    modified by the caller and the changes will be written to
    external files.

Definition at line 174 of file calibType.py.

    def getMetadata(self):
        """Retrieve metadata associated with this calibration.
 
        Returns
        -------
        meta : `lsst.daf.base.PropertyList`
            Metadata. The returned `~lsst.daf.base.PropertyList` can be
            modified by the caller and the changes will be written to
            external files.
        """
        return self._metadata
 

metadata()

lsst.ip.isr.calibType.IsrCalib.metadata ( self )

inherited

Definition at line 171 of file calibType.py.

    def metadata(self):
        return self._metadata
 

readFits()

lsst.ip.isr.calibType.IsrCalib.readFits ( cls, filename, ** kwargs )

inherited

Read calibration data from a FITS file.

Parameters
----------
filename : `str`
    Filename to read data from.
kwargs : `dict` or collections.abc.Mapping`, optional
    Set of key=value pairs to pass to the ``fromTable``
    method.

Returns
-------
calib : `lsst.ip.isr.IsrCalib`
    Calibration contained within the file.

Definition at line 517 of file calibType.py.

    def readFits(cls, filename, **kwargs):
        """Read calibration data from a FITS file.
 
        Parameters
        ----------
        filename : `str`
            Filename to read data from.
        kwargs : `dict` or collections.abc.Mapping`, optional
            Set of key=value pairs to pass to the ``fromTable``
            method.
 
        Returns
        -------
        calib : `lsst.ip.isr.IsrCalib`
            Calibration contained within the file.
        """
        tableList = []
        tableList.append(Table.read(filename, hdu=1, mask_invalid=False))
        extNum = 2  # Fits indices start at 1, we've read one already.
        keepTrying = True
 
        while keepTrying:
            with warnings.catch_warnings():
                warnings.simplefilter("error")
                try:
                    newTable = Table.read(filename, hdu=extNum, mask_invalid=False)
                    tableList.append(newTable)
                    extNum += 1
                except Exception:
                    keepTrying = False
 
        calibClass = cls.determineCalibClass(tableList[0].meta, "readFits")
        for table in tableList:
            if calibClass._OBSTYPE == "BF_DISTORTION_MATRIX":
                table.convert_bytestring_to_unicode()
            for k, v in table.meta.items():
                if isinstance(v, fits.card.Undefined):
                    table.meta[k] = None
 
        if calibClass._OBSTYPE in ("PHOTODIODE", ):
            # Merge primary header, as these types store information
            # there.
            with fits.open(filename) as hdul:
                primaryHeader = hdul[0].header
            tableList[0].meta = merge_headers([tableList[0].meta, primaryHeader], mode="first")
 
        return calibClass.fromTable(tableList, **kwargs)
 

readText()

lsst.ip.isr.calibType.IsrCalib.readText ( cls, filename, ** kwargs )

inherited

Read calibration representation from a yaml/ecsv file.

Parameters
----------
filename : `str`
    Name of the file containing the calibration definition.
kwargs : `dict` or collections.abc.Mapping`, optional
    Set of key=value pairs to pass to the ``fromDict`` or
    ``fromTable`` methods.

Returns
-------
calib : `~lsst.ip.isr.IsrCalibType`
    Calibration class.

Raises
------
RuntimeError
    Raised if the filename does not end in ".ecsv" or ".yaml".

Definition at line 425 of file calibType.py.

    def readText(cls, filename, **kwargs):
        """Read calibration representation from a yaml/ecsv file.
 
        Parameters
        ----------
        filename : `str`
            Name of the file containing the calibration definition.
        kwargs : `dict` or collections.abc.Mapping`, optional
            Set of key=value pairs to pass to the ``fromDict`` or
            ``fromTable`` methods.
 
        Returns
        -------
        calib : `~lsst.ip.isr.IsrCalibType`
            Calibration class.
 
        Raises
        ------
        RuntimeError
            Raised if the filename does not end in ".ecsv" or ".yaml".
        """
        if filename.endswith((".ecsv", ".ECSV")):
            data = Table.read(filename, format="ascii.ecsv")
            calibClass = cls.determineCalibClass(data.meta, "readText/ECSV")
            return calibClass.fromTable([data], **kwargs)
        elif filename.endswith((".yaml", ".YAML")):
            with open(filename, "r") as f:
                data = yaml.load(f, Loader=yaml.CLoader)
            calibClass = cls.determineCalibClass(data["metadata"], "readText/YAML")
            return calibClass.fromDict(data, **kwargs)
        elif filename.endswith((".json", ".JSON")):
            with open(filename, "r") as f:
                data = json.load(f)
            calibClass = cls.determineCalibClass(data, "readText/JSON")
            return calibClass.fromDict(data, **kwargs)
        else:
            raise RuntimeError(f"Unknown filename extension: {filename}")
 

requiredAttributes() [1/2]

lsst.ip.isr.calibType.IsrCalib.requiredAttributes ( self )

inherited

Definition at line 162 of file calibType.py.

    def requiredAttributes(self):
        return self._requiredAttributes
 

requiredAttributes() [2/2]

lsst.ip.isr.calibType.IsrCalib.requiredAttributes ( self, value )

inherited

Definition at line 166 of file calibType.py.

    def requiredAttributes(self, value):
        self._requiredAttributes = value
 

setMetadata()

lsst.ip.isr.calibType.IsrCalib.setMetadata ( self, metadata )

inherited

Store a copy of the supplied metadata with this calibration.

Parameters
----------
metadata : `lsst.daf.base.PropertyList`
    Metadata to associate with the calibration.  Will be copied and
    overwrite existing metadata.

Reimplemented in lsst.ip.isr.transmissionCurve.IntermediateTransmissionCurve.

Definition at line 186 of file calibType.py.

    def setMetadata(self, metadata):
        """Store a copy of the supplied metadata with this calibration.
 
        Parameters
        ----------
        metadata : `lsst.daf.base.PropertyList`
            Metadata to associate with the calibration.  Will be copied and
            overwrite existing metadata.
        """
        if metadata is not None:
            self._metadata.update(metadata)
 
        # Ensure that we have the obs type required by calibration ingest
        self._metadata["OBSTYPE"] = self._OBSTYPE
        self._metadata[self._OBSTYPE + "_SCHEMA"] = self._SCHEMA
        self._metadata[self._OBSTYPE + "_VERSION"] = self._VERSION
 
        if isinstance(metadata, dict):
            self.calibInfoFromDict(metadata)
        elif isinstance(metadata, PropertyList):
            self.calibInfoFromDict(metadata.toDict())
 

subtractCrosstalk()

lsst.ip.isr.crosstalk.CrosstalkCalib.subtractCrosstalk	(		self,
			thisExposure,
			sourceExposure = None,
			crosstalkCoeffs = None,
			crosstalkCoeffsSqr = None,
			crosstalkCoeffsValid = None,
			badPixels = ["BAD"],
			minPixelToMask = 45000,
			doSubtrahendMasking = False,
			crosstalkStr = "CROSSTALK",
			isTrimmed = None,
			backgroundMethod = "None",
			doSqrCrosstalk = False,
			fullAmplifier = False,
			parallelOverscan = None,
			detectorConfig = None,
			badAmpDict = None,
			ignoreVariance = False )

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``, if ``doSubtrahendMasking`` is False.  With
that enabled, the mask is only set if the absolute value of
the correction applied exceeds ``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.
crosstalkCoeffsSqr : `numpy.ndarray`, optional.
    Quadratic coefficients to use to correct crosstalk.
crosstalkCoeffsValid : `numpy.ndarray`, optional
    Boolean array that is True where coefficients are valid.
badPixels : `list` of `str`, optional
    Mask planes to ignore.
minPixelToMask : `float`, optional
    Minimum pixel value to set the ``crosstalkStr`` mask
    plane.  If doSubtrahendMasking is True, this is calculated
    from the absolute magnitude of the subtrahend image.
    Otherwise, this sets the minimum source value to use to
    set that mask.
doSubtrahendMasking : `bool`, optional
    If true, the mask is calculated from the properties of the
    subtrahend image, not from the brightness of the source
    pixel.
crosstalkStr : `str`, optional
    Mask plane name for pixels greatly modified by crosstalk
    (above minPixelToMask).
isTrimmed : `bool`, optional
    This option has been deprecated and does not do anything.
    It will be removed after v29.
backgroundMethod : `str`, optional
    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.
doSqrCrosstalk: `bool`, optional
    Should the quadratic crosstalk coefficients be used for the
    crosstalk correction?
fullAmplifier : `bool`, optional
    Use full amplifier and not just imaging region.
parallelOverscan : `bool`, optional
    This option is deprecated and will be removed after v29.
detectorConfig : `lsst.ip.isr.overscanDetectorConfig`, optional
    Per-amplifier configs to use if parallelOverscan is True.
    This option is deprecated and will be removed after v29.
badAmpDict : `dict` [`str`, `bool`], optional
    Dictionary to identify bad amplifiers that should not be
    source or target for crosstalk correction.
ignoreVariance : `bool`, optional
    Ignore the variance plane when doing crosstalk correction?

Notes
-----

For a given image I, we want to find the crosstalk subtrahend
image CT, such that
                 I_corrected = I - CT
The subtrahend image is the sum of all crosstalk contributions
that appear in I, so we can build it up by amplifier. Each
amplifier A in image I sees the contributions from all other
amplifiers B_v != A.  For the current linear model, we set `sImage`
equal to the segment of the subtrahend image CT corresponding to
amplifier A, and then build it up as:
simage_linear = sum_v coeffsA_v * (B_v - bkg_v) where coeffsA_v
is the vector of crosstalk coefficients for sources that cause
images in amplifier A.  The bkg_v term in this equation is
identically 0.0 for all cameras except obs_subaru (and is only
non-zero there for historical reasons).
To include the non-linear term, we can again add to the subtrahend
image using the same loop, as:

simage_nonlinear = sum_v (coeffsA_v * B_v) + (NLcoeffsA_v * B_v * B_v)
                 = sum_v   linear_term_v   +    nonlinear_term_v

where coeffsA_v is the linear term, and NLcoeffsA_v are the quadratic
component. For LSSTCam, it has been observed that the linear_term_v >>
nonlinear_term_v.

Definition at line 587 of file crosstalk.py.

                          ignoreVariance=False):
        """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``, if ``doSubtrahendMasking`` is False.  With
        that enabled, the mask is only set if the absolute value of
        the correction applied exceeds ``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.
        crosstalkCoeffsSqr : `numpy.ndarray`, optional.
            Quadratic coefficients to use to correct crosstalk.
        crosstalkCoeffsValid : `numpy.ndarray`, optional
            Boolean array that is True where coefficients are valid.
        badPixels : `list` of `str`, optional
            Mask planes to ignore.
        minPixelToMask : `float`, optional
            Minimum pixel value to set the ``crosstalkStr`` mask
            plane.  If doSubtrahendMasking is True, this is calculated
            from the absolute magnitude of the subtrahend image.
            Otherwise, this sets the minimum source value to use to
            set that mask.
        doSubtrahendMasking : `bool`, optional
            If true, the mask is calculated from the properties of the
            subtrahend image, not from the brightness of the source
            pixel.
        crosstalkStr : `str`, optional
            Mask plane name for pixels greatly modified by crosstalk
            (above minPixelToMask).
        isTrimmed : `bool`, optional
            This option has been deprecated and does not do anything.
            It will be removed after v29.
        backgroundMethod : `str`, optional
            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.
        doSqrCrosstalk: `bool`, optional
            Should the quadratic crosstalk coefficients be used for the
            crosstalk correction?
        fullAmplifier : `bool`, optional
            Use full amplifier and not just imaging region.
        parallelOverscan : `bool`, optional
            This option is deprecated and will be removed after v29.
        detectorConfig : `lsst.ip.isr.overscanDetectorConfig`, optional
            Per-amplifier configs to use if parallelOverscan is True.
            This option is deprecated and will be removed after v29.
        badAmpDict : `dict` [`str`, `bool`], optional
            Dictionary to identify bad amplifiers that should not be
            source or target for crosstalk correction.
        ignoreVariance : `bool`, optional
            Ignore the variance plane when doing crosstalk correction?
 
        Notes
        -----
 
        For a given image I, we want to find the crosstalk subtrahend
        image CT, such that
                         I_corrected = I - CT
        The subtrahend image is the sum of all crosstalk contributions
        that appear in I, so we can build it up by amplifier. Each
        amplifier A in image I sees the contributions from all other
        amplifiers B_v != A.  For the current linear model, we set `sImage`
        equal to the segment of the subtrahend image CT corresponding to
        amplifier A, and then build it up as:
        simage_linear = sum_v coeffsA_v * (B_v - bkg_v) where coeffsA_v
        is the vector of crosstalk coefficients for sources that cause
        images in amplifier A.  The bkg_v term in this equation is
        identically 0.0 for all cameras except obs_subaru (and is only
        non-zero there for historical reasons).
        To include the non-linear term, we can again add to the subtrahend
        image using the same loop, as:
 
        simage_nonlinear = sum_v (coeffsA_v * B_v) + (NLcoeffsA_v * B_v * B_v)
                         = sum_v   linear_term_v   +    nonlinear_term_v
 
        where coeffsA_v is the linear term, and NLcoeffsA_v are the quadratic
        component. For LSSTCam, it has been observed that the linear_term_v >>
        nonlinear_term_v.
        """
        targetMaskedImage = thisExposure.maskedImage
        targetDetector = thisExposure.getDetector()
        if self.hasCrosstalk is False:
            self.fromDetector(targetDetector, coeffVector=crosstalkCoeffs)
 
        # TODO: Remove on DM-48394
        if isTrimmed is not None:
            warnings.warn(
                "The isTrimmed option has been deprecated and will be removed after v29.",
                FutureWarning,
            )
        isTrimmed = isTrimmedImage(targetMaskedImage, targetDetector)
 
        # TODO: Remove on DM-48394
        if parallelOverscan is not None:
            warnings.warn(
                "The parallelOverscan option has been deprecated and will be removed after v29.",
                FutureWarning,
            )
        if detectorConfig is not None:
            warnings.warn(
                "The detectorConfig option has been deprecated and will be removed after v29.",
                FutureWarning,
            )
 
        numAmps = len(targetDetector)
        if numAmps != self.nAmp:
            raise RuntimeError(f"Crosstalk built for {self.nAmp} in {self._detectorName}, received "
                               f"{numAmps} in {targetDetector.getName()}")
 
        if doSqrCrosstalk and crosstalkCoeffsSqr is None:
            raise RuntimeError("Attempted to perform NL crosstalk correction without NL "
                               "crosstalk coefficients.")
 
        if fullAmplifier and (backgroundMethod != "None"):
            raise RuntimeError("Cannot do full amplifier with background subtraction.")
 
        if sourceExposure:
            sourceMaskedImage = sourceExposure.maskedImage
            sourceDetector = sourceExposure.getDetector()
        else:
            sourceMaskedImage = targetMaskedImage
            sourceDetector = targetDetector
 
        if crosstalkCoeffs is not None:
            coeffs = np.asarray(crosstalkCoeffs).copy()
        else:
            coeffs = np.asarray(self.coeffs).copy()
        self.log.debug("CT COEFF: %s", coeffs)
 
        if doSqrCrosstalk:
            coeffsSqr = np.asarray(crosstalkCoeffsSqr).copy()
            self.log.debug("CT COEFF SQR: %s", coeffsSqr)
        else:
            coeffsSqr = np.zeros_like(coeffs)
 
        # Check for valid values; set to 0 otherwise.
        badCoeffs = ~np.isfinite(coeffs) | (coeffs == 0.0)
        coeffs[badCoeffs] = 0.0
        coeffsSqr[badCoeffs] = 0.0
        if crosstalkCoeffsValid is not None:
            coeffs[~crosstalkCoeffsValid] = 0.0
            coeffsSqr[~crosstalkCoeffsValid] = 0.0
 
        if badAmpDict:
            for index, amp in enumerate(sourceDetector):
                if badAmpDict[amp.getName()]:
                    coeffs[index, :] = 0.0
                    coeffs[:, index] = 0.0
 
        # Compute backgrounds if requested.
        backgrounds = {amp.getName(): 0.0 for amp in sourceDetector}
        if backgroundMethod == "AMP":
            backgrounds = {
                amp.getName(): self.calculateBackground(
                    sourceMaskedImage[self._getAppropriateBBox(
                        amp,
                        isTrimmed,
                        False,
                    )], badPixels)
                for amp in sourceDetector
            }
        elif backgroundMethod == "DETECTOR":
            background = self.calculateBackground(sourceMaskedImage, badPixels)
            backgrounds = {amp.getName(): background for amp in sourceDetector}
 
        # Add the crosstalk mask plane to the target mask.
        sourceCrosstalkPlane = sourceMaskedImage.mask.addMaskPlane(crosstalkStr)
        if sourceExposure is not None:
            targetCrosstalkPlane = targetMaskedImage.mask.addMaskPlane(crosstalkStr)
        else:
            targetCrosstalkPlane = sourceCrosstalkPlane
 
        # If we are not doing subtrahend masking, the CROSSTALK mask bit
        # is set according to the counts in the source pixel (above
        # background), regardless of the crosstalk coefficient value.
        if not doSubtrahendMasking:
            # When we are not using subtrahend masking, we set the mask.
            thresholdBackground = self.calculateBackground(sourceMaskedImage, badPixels)
            toMask = (sourceMaskedImage.image.array >= (minPixelToMask + thresholdBackground))
            sourceMaskedImage.mask.array[toMask] |= sourceMaskedImage.mask.getPlaneBitMask(crosstalkStr)
 
        crosstalk = sourceMaskedImage.mask.getPlaneBitMask(crosstalkStr)
 
        # Define a subtrahend image to contain all the scaled crosstalk
        # signals. These will be applied to the target image.
        subtrahend = targetMaskedImage.Factory(targetMaskedImage.getBBox())
        subtrahend.set((0, 0, 0))
 
        # If we are ignoring variance and doing subtrahend masking, then
        # we can work on only the image plane (3x speed increase).
        imageOnly = ignoreVariance and doSubtrahendMasking
 
        for sourceIndex, sourceAmp in enumerate(sourceDetector):
            for targetIndex, targetAmp in enumerate(targetDetector):
                coeff = coeffs[sourceIndex, targetIndex]
                coeffSqr = coeffsSqr[sourceIndex, targetIndex]
 
                if coeff == 0.0:
                    continue
 
                targetBBox = self._getAppropriateBBox(targetAmp, isTrimmed, fullAmplifier)
 
                # The extractAmp() method will always make a copy of the source
                # amplifier data.
                if imageOnly:
                    sourceImage = self.extractAmp(
                        sourceMaskedImage.image,
                        sourceAmp,
                        targetAmp,
                        isTrimmed=isTrimmed,
                        fullAmplifier=fullAmplifier,
                    )
                    targetImage = subtrahend[targetBBox].image
                else:
                    sourceImage = self.extractAmp(
                        sourceMaskedImage,
                        sourceAmp,
                        targetAmp,
                        isTrimmed=isTrimmed,
                        fullAmplifier=fullAmplifier,
                    )
                    targetImage = subtrahend[targetBBox]
 
                    # Remove all other masks from copied sourceImage.
                    sourceImage.mask.array[:] &= crosstalk
 
                if backgrounds[sourceAmp.getName()] != 0.0:
                    sourceImage -= backgrounds[sourceAmp.getName()]
 
                # This operation will also transfer the CROSSTALK mask bit from
                # above to the target (subtrahend) image if we are using a
                # masked image.
                targetImage.scaledPlus(coeff, sourceImage)
                if coeffSqr != 0.0:
                    sourceImage.scaledMultiplies(1.0, sourceImage)
                    targetImage.scaledPlus(coeffSqr, sourceImage)
 
        # Clear the mask in the target image, because the subtrahend image
        # contains the crosstalk mask.
        sourceMaskedImage.mask.clearMaskPlane(sourceCrosstalkPlane)
        if sourceExposure is not None:
            targetMaskedImage.mask.clearMaskPlane(targetCrosstalkPlane)
 
        if doSubtrahendMasking:
            # 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.
 
            # The existing mask in the subtrahend comes from the
            # threshold set above.  It should be cleared so we can
            # recalculate it.
            subtrahend.mask.clearMaskPlane(targetCrosstalkPlane)
 
            # For masking purposes, we only really care when the
            # correction is significantly different than the median
            # value on that amplifier (which includes the contribution
            # of every other amplifier background that crosstalks onto
            # that amplifier).  Subtract and save this "background"
            # level, so we can threshold to set the mask relative to
            # that background, but still include that contribution in
            # the correction we're applying.
            subtrahendBackgrounds = {}
            for amp in targetDetector:
                # Note that we never want the full amplifier for background
                # calculations.
                bbox = self._getAppropriateBBox(amp, isTrimmed, False)
                ampData = subtrahend[bbox]
                background = np.median(ampData.image.array)
                subtrahendBackgrounds[amp.getName()] = background
                ampData.image.array[:, :] -= background
                self.log.debug(f"Subtrahend background level: {amp.getName()} {background}")
 
            toMask = (subtrahend.image.array >= minPixelToMask) | (subtrahend.image.array <= -minPixelToMask)
            subtrahend.mask.array[toMask] |= subtrahend.mask.getPlaneBitMask(crosstalkStr)
 
            # Put the backgrounds back.
            for amp in targetDetector:
                bbox = self._getAppropriateBBox(amp, isTrimmed, False)
                ampData = subtrahend[bbox]
                background = subtrahendBackgrounds[amp.getName()]
                ampData.image.array[:, :] += background
 
        # Subtract subtrahend from input.  The mask plane is fully
        # populated, so this operation also sets the ``crosstalkStr``
        # bit where applicable.
        targetMaskedImage -= subtrahend
 
 

toDict()

lsst.ip.isr.crosstalk.CrosstalkCalib.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.

Reimplemented from lsst.ip.isr.calibType.IsrCalib.

Definition at line 345 of file crosstalk.py.

    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.updateMetadata()
 
        outDict = {}
        metadata = self.getMetadata()
        outDict['metadata'] = metadata
 
        outDict['hasCrosstalk'] = self.hasCrosstalk
        outDict['nAmp'] = self.nAmp
        outDict['crosstalkShape'] = self.crosstalkShape
        outDict['crosstalkRatiosUnits'] = self.crosstalkRatiosUnits
 
        ctLength = self.nAmp*self.nAmp
        outDict['coeffs'] = self.coeffs.reshape(ctLength).tolist()
 
        if self.coeffErr is not None:
            outDict['coeffErr'] = self.coeffErr.reshape(ctLength).tolist()
        if self.coeffNum is not None:
            outDict['coeffNum'] = self.coeffNum.reshape(ctLength).tolist()
        if self.coeffValid is not None:
            outDict['coeffValid'] = self.coeffValid.reshape(ctLength).tolist()
        if self.coeffsSqr is not None:
            outDict['coeffsSqr'] = self.coeffsSqr.reshape(ctLength).tolist()
        if self.coeffErrSqr is not None:
            outDict['coeffErrSqr'] = self.coeffErrSqr.reshape(ctLength).tolist()
        if self.ampGainRatios is not None:
            outDict['ampGainRatios'] = self.ampGainRatios.reshape(ctLength).tolist()
        if self.fitGains is not None:
            outDict['fitGains'] = self.fitGains.tolist()
 
        if self.interChip:
            outDict['interChip'] = dict()
            for detector in self.interChip:
                outDict['interChip'][detector] = self.interChip[detector].reshape(ctLength).tolist()
 
        return outDict
 

toTable()

lsst.ip.isr.crosstalk.CrosstalkCalib.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.

Reimplemented from lsst.ip.isr.calibType.IsrCalib.

Definition at line 448 of file crosstalk.py.

    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.updateMetadata()
        catalog = Table([{'CT_COEFFS': self.coeffs.reshape(self.nAmp*self.nAmp),
                          'CT_ERRORS': self.coeffErr.reshape(self.nAmp*self.nAmp),
                          'CT_COUNTS': self.coeffNum.reshape(self.nAmp*self.nAmp),
                          'CT_VALID': self.coeffValid.reshape(self.nAmp*self.nAmp),
                          'CT_COEFFS_SQR': self.coeffsSqr.reshape(self.nAmp*self.nAmp),
                          'CT_ERRORS_SQR': self.coeffErrSqr.reshape(self.nAmp*self.nAmp),
                          'CT_AMP_GAIN_RATIOS': self.ampGainRatios.reshape(self.nAmp*self.nAmp),
                          'CT_FIT_GAINS': self.fitGains,
                          }])
        # filter None, because astropy can't deal.
        inMeta = self.getMetadata().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.interChip:
            interChipTable = Table([{'IC_SOURCE_DET': sourceDet,
                                     'IC_COEFFS': self.interChip[sourceDet].reshape(self.nAmp*self.nAmp)}
                                   for sourceDet in self.interChip.keys()])
            tableList.append(interChipTable)
        return tableList
 

updateMetadata()

lsst.ip.isr.crosstalk.CrosstalkCalib.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.

Reimplemented from lsst.ip.isr.calibType.IsrCalib.

Definition at line 142 of file crosstalk.py.

    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
        kwargs['DETECTOR_NAME'] = self._detectorName
        kwargs['DETECTOR_SERIAL'] = self._detectorSerial
        kwargs['HAS_CROSSTALK'] = self.hasCrosstalk
        kwargs['NAMP'] = self.nAmp
        self.crosstalkShape = (self.nAmp, self.nAmp)
        kwargs['CROSSTALK_SHAPE'] = self.crosstalkShape
        kwargs['CROSSTALK_RATIOS_UNITS'] = self.crosstalkRatiosUnits
 
        super().updateMetadata(setDate=setDate, **kwargs)
 

updateMetadataFromExposures()

lsst.ip.isr.calibType.IsrCalib.updateMetadataFromExposures ( self, exposures )

inherited

Extract and unify metadata information.

Parameters
----------
exposures : `list`
    Exposures or other calibrations to scan.

Definition at line 294 of file calibType.py.

    def updateMetadataFromExposures(self, exposures):
        """Extract and unify metadata information.
 
        Parameters
        ----------
        exposures : `list`
            Exposures or other calibrations to scan.
        """
        # This list of keywords is the set of header entries that
        # should be checked and propagated.  Not having an entry is
        # not a failure, as they may not be defined for the exposures
        # being used.
        keywords = ["SEQNAME", "SEQFILE", "SEQCKSUM", "ODP", "AP0_RC"]
        metadata = {}
 
        for exp in exposures:
            try:
                expMeta = exp.getMetadata()
            except AttributeError:
                continue
            for key in keywords:
                if key in expMeta:
                    if key in metadata:
                        if metadata[key] != expMeta[key]:
                            self.log.warning("Metadata mismatch! Have: %s Found %s",
                                             metadata[key], expMeta[key])
                    else:
                        metadata[key] = expMeta[key]
 
        self.updateMetadata(**metadata, setCalibInfo=True)
 

validate()

lsst.ip.isr.calibType.IsrCalib.validate ( self, other = None )

inherited

Validate that this calibration is defined and can be used.

Parameters
----------
other : `object`, optional
    Thing to validate against.

Returns
-------
valid : `bool`
    Returns true if the calibration is valid and appropriate.

Reimplemented in lsst.ip.isr.linearize.Linearizer, and lsst.ip.isr.photodiodeCorrection.PhotodiodeCorrection.

Definition at line 691 of file calibType.py.

    def validate(self, other=None):
        """Validate that this calibration is defined and can be used.
 
        Parameters
        ----------
        other : `object`, optional
            Thing to validate against.
 
        Returns
        -------
        valid : `bool`
            Returns true if the calibration is valid and appropriate.
        """
        return False
 

writeFits()

lsst.ip.isr.calibType.IsrCalib.writeFits ( self, filename )

inherited

Write calibration data to a FITS file.

Parameters
----------
filename : `str`
    Filename to write data to.

Returns
-------
used : `str`
    The name of the file used to write the data.

Reimplemented in lsst.ip.isr.transmissionCurve.IntermediateTransmissionCurve.

Definition at line 565 of file calibType.py.

    def writeFits(self, filename):
        """Write calibration data to a FITS file.
 
        Parameters
        ----------
        filename : `str`
            Filename to write data to.
 
        Returns
        -------
        used : `str`
            The name of the file used to write the data.
        """
        tableList = self.toTable()
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=Warning, module="astropy.io")
            astropyList = [fits.table_to_hdu(table) for table in tableList]
            astropyList.insert(0, fits.PrimaryHDU())
 
            writer = fits.HDUList(astropyList)
            writer.writeto(filename, overwrite=True)
        return filename
 

writeText()

lsst.ip.isr.calibType.IsrCalib.writeText ( self, filename, format = "auto" )

inherited

Write the calibration data to a text file.

Parameters
----------
filename : `str`
    Name of the file to write.
format : `str`
    Format to write the file as.  Supported values are:
        ``"auto"`` : Determine filetype from filename.
        ``"yaml"`` : Write as yaml.
        ``"ecsv"`` : Write as ecsv.

Returns
-------
used : `str`
    The name of the file used to write the data.  This may
    differ from the input if the format is explicitly chosen.

Raises
------
RuntimeError
    Raised if filename does not end in a known extension, or
    if all information cannot be written.

Notes
-----
The file is written to YAML/ECSV format and will include any
associated metadata.

Definition at line 463 of file calibType.py.

    def writeText(self, filename, format="auto"):
        """Write the calibration data to a text file.
 
        Parameters
        ----------
        filename : `str`
            Name of the file to write.
        format : `str`
            Format to write the file as.  Supported values are:
                ``"auto"`` : Determine filetype from filename.
                ``"yaml"`` : Write as yaml.
                ``"ecsv"`` : Write as ecsv.
 
        Returns
        -------
        used : `str`
            The name of the file used to write the data.  This may
            differ from the input if the format is explicitly chosen.
 
        Raises
        ------
        RuntimeError
            Raised if filename does not end in a known extension, or
            if all information cannot be written.
 
        Notes
        -----
        The file is written to YAML/ECSV format and will include any
        associated metadata.
        """
        if format == "yaml" or (format == "auto" and filename.lower().endswith((".yaml", ".YAML"))):
            outDict = self.toDict()
            path, ext = os.path.splitext(filename)
            filename = path + ".yaml"
            with open(filename, "w") as f:
                yaml.dump(outDict, f)
        elif format == "ecsv" or (format == "auto" and filename.lower().endswith((".ecsv", ".ECSV"))):
            tableList = self.toTable()
            if len(tableList) > 1:
                # ECSV doesn't support multiple tables per file, so we
                # can only write the first table.
                raise RuntimeError(f"Unable to persist {len(tableList)}tables in ECSV format.")
 
            table = tableList[0]
            path, ext = os.path.splitext(filename)
            filename = path + ".ecsv"
            table.write(filename, format="ascii.ecsv")
        else:
            raise RuntimeError(f"Attempt to write to a file {filename} "
                               "that does not end in '.yaml' or '.ecsv'")
 
        return filename
 

Member Data Documentation

_calibId

str lsst.ip.isr.calibType.IsrCalib._calibId = None

protectedinherited

Definition at line 76 of file calibType.py.

_detectorId

lsst.ip.isr.calibType.IsrCalib._detectorId = None

protectedinherited

Definition at line 74 of file calibType.py.

_detectorName

lsst.ip.isr.calibType.IsrCalib._detectorName = None

protectedinherited

Definition at line 72 of file calibType.py.

_detectorSerial

lsst.ip.isr.calibType.IsrCalib._detectorSerial = None

protectedinherited

Definition at line 73 of file calibType.py.

_filter

lsst.ip.isr.calibType.IsrCalib._filter = None

protectedinherited

Definition at line 75 of file calibType.py.

_instrument

lsst.ip.isr.calibType.IsrCalib._instrument = None

protectedinherited

Definition at line 69 of file calibType.py.

_metadata

lsst.ip.isr.calibType.IsrCalib._metadata = PropertyList()

protectedinherited

Definition at line 80 of file calibType.py.

_OBSTYPE

str lsst.ip.isr.calibType.IsrCalib._OBSTYPE = "generic"

staticprotectedinherited

Definition at line 64 of file calibType.py.

_raftName

lsst.ip.isr.calibType.IsrCalib._raftName = None

protectedinherited

Definition at line 70 of file calibType.py.

_requiredAttributes

lsst.ip.isr.calibType.IsrCalib._requiredAttributes

protectedinherited

Definition at line 115 of file calibType.py.

_SCHEMA

str lsst.ip.isr.calibType.IsrCalib._SCHEMA = "NO SCHEMA"

staticprotectedinherited

Definition at line 65 of file calibType.py.

_seqcksum

lsst.ip.isr.calibType.IsrCalib._seqcksum = None

protectedinherited

Definition at line 79 of file calibType.py.

_seqfile

lsst.ip.isr.calibType.IsrCalib._seqfile = None

protectedinherited

Definition at line 77 of file calibType.py.

_seqname

lsst.ip.isr.calibType.IsrCalib._seqname = None

protectedinherited

Definition at line 78 of file calibType.py.

_slotName

lsst.ip.isr.calibType.IsrCalib._slotName = None

protectedinherited

Definition at line 71 of file calibType.py.

_VERSION

int lsst.ip.isr.calibType.IsrCalib._VERSION = 0

staticprotectedinherited

Definition at line 66 of file calibType.py.

ampGainRatios

int lsst.ip.isr.crosstalk.CrosstalkCalib.ampGainRatios = np.zeros(self.crosstalkShape) if self.nAmp else None

Definition at line 123 of file crosstalk.py.

coeffErr

int lsst.ip.isr.crosstalk.CrosstalkCalib.coeffErr = np.zeros(self.crosstalkShape) if self.nAmp else None

Definition at line 113 of file crosstalk.py.

coeffErrSqr

int lsst.ip.isr.crosstalk.CrosstalkCalib.coeffErrSqr = np.zeros(self.crosstalkShape) if self.nAmp else None

Definition at line 120 of file crosstalk.py.

coeffNum

int lsst.ip.isr.crosstalk.CrosstalkCalib.coeffNum

Initial value:

=  np.zeros(self.crosstalkShape,
                                 dtype=int) if self.nAmp else None

Definition at line 114 of file crosstalk.py.

coeffs

int lsst.ip.isr.crosstalk.CrosstalkCalib.coeffs = np.zeros(self.crosstalkShape) if self.nAmp else None

Definition at line 112 of file crosstalk.py.

coeffsSqr

int lsst.ip.isr.crosstalk.CrosstalkCalib.coeffsSqr = np.zeros(self.crosstalkShape) if self.nAmp else None

Definition at line 119 of file crosstalk.py.

coeffValid

int lsst.ip.isr.crosstalk.CrosstalkCalib.coeffValid

Initial value:

=  np.ones(self.crosstalkShape,
                                  dtype=bool) if self.nAmp else None

Definition at line 116 of file crosstalk.py.

crosstalkRatiosUnits

str lsst.ip.isr.crosstalk.CrosstalkCalib.crosstalkRatiosUnits = 'adu' if self.nAmp else None

Definition at line 129 of file crosstalk.py.

crosstalkShape

tuple lsst.ip.isr.crosstalk.CrosstalkCalib.crosstalkShape = (self.nAmp, self.nAmp)

Definition at line 110 of file crosstalk.py.

fitGains

int lsst.ip.isr.crosstalk.CrosstalkCalib.fitGains = np.zeros(self.nAmp) if self.nAmp else None

Definition at line 126 of file crosstalk.py.

hasCrosstalk

bool lsst.ip.isr.crosstalk.CrosstalkCalib.hasCrosstalk = False

Definition at line 108 of file crosstalk.py.

interChip

dict lsst.ip.isr.crosstalk.CrosstalkCalib.interChip = {}

Definition at line 131 of file crosstalk.py.

log

lsst.ip.isr.calibType.IsrCalib.log = log if log else logging.getLogger(__name__)

inherited

Definition at line 96 of file calibType.py.

nAmp

int lsst.ip.isr.crosstalk.CrosstalkCalib.nAmp = nAmp if nAmp else 0

Definition at line 109 of file crosstalk.py.

requiredAttributes

lsst.ip.isr.calibType.IsrCalib.requiredAttributes = set(["_OBSTYPE", "_SCHEMA", "_VERSION"])

inherited

Definition at line 90 of file calibType.py.

---

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

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-12.1.0/Linux/ip_isr/gc24b5d6ed1+9f17e571f4/python/lsst/ip/isr/crosstalk.py