linearize.py

Go to the documentation of this file.

#
# LSST Data Management System
# Copyright 2016 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 <http://www.lsstcorp.org/LegalNotices/>.
#
import abc
import numpy as np
 
from astropy.table import Table
 
import lsst.afw.math as afwMath
from lsst.pipe.base import Struct
from lsst.geom import Box2I, Point2I, Extent2I
from .applyLookupTable import applyLookupTable
from .calibType import IsrCalib
 
__all__ = ["Linearizer",
           "LinearizeBase", "LinearizeLookupTable", "LinearizeSquared",
           "LinearizeProportional", "LinearizePolynomial", "LinearizeSpline", "LinearizeNone"]
 
 
class Linearizer(IsrCalib):
    """Parameter set for linearization.
 
    These parameters are included in cameraGeom.Amplifier, but
    should be accessible externally to allow for testing.
 
    Parameters
    ----------
    table : `numpy.array`, optional
        Lookup table; a 2-dimensional array of floats:
            - one row for each row index (value of coef[0] in the amplifier)
            - one column for each image value
        To avoid copying the table the last index should vary fastest
        (numpy default "C" order)
    detector : `lsst.afw.cameraGeom.Detector`, optional
        Detector object.  Passed to self.fromDetectorfromDetectorfromDetector() on init.
    log : `logging.Logger`, optional
        Logger to handle messages.
    kwargs : `dict`, optional
        Other keyword arguments to pass to the parent init.
 
    Raises
    ------
    RuntimeError :
        Raised if the supplied table is not 2D, or if the table has fewer
        columns than rows (indicating that the indices are swapped).
 
    Notes
    -----
    The linearizer attributes stored are:
 
    hasLinearity : `bool`
        Whether a linearity correction is defined for this detector.
    override : `bool`
        Whether the detector parameters should be overridden.
    ampNames : `list` [`str`]
        List of amplifier names to correct.
    linearityCoeffs : `dict` [`str`, `numpy.array`]
        Coefficients to use in correction.  Indexed by amplifier
        names.  The format of the array depends on the type of
        correction to apply.
    linearityType : `dict` [`str`, `str`]
        Type of correction to use, indexed by amplifier names.
    linearityBBox : `dict` [`str`, `lsst.geom.Box2I`]
        Bounding box the correction is valid over, indexed by
        amplifier names.
    fitParams : `dict` [`str`, `numpy.array`], optional
        Linearity fit parameters used to construct the correction
        coefficients, indexed as above.
    fitParamsErr : `dict` [`str`, `numpy.array`], optional
        Uncertainty values of the linearity fit parameters used to
        construct the correction coefficients, indexed as above.
    fitChiSq : `dict` [`str`, `float`], optional
        Chi-squared value of the linearity fit, indexed as above.
    fitResiduals : `dict` [`str`, `numpy.array`], optional
        Residuals of the fit, indexed as above. Used for
        calculating photdiode corrections
    linearFit : The linear fit to the low flux region of the curve.
        [intercept, slope].
    tableData : `numpy.array`, optional
        Lookup table data for the linearity correction.
    """
    _OBSTYPE = "LINEARIZER"
    _SCHEMA = 'Gen3 Linearizer'
    _VERSION = 1.1
 
    def __init__(self, table=None, **kwargs):
        self.hasLinearityhasLinearity = False
        self.overrideoverride = False
 
        self.ampNamesampNames = list()
        self.linearityCoeffslinearityCoeffs = dict()
        self.linearityTypelinearityType = dict()
        self.linearityBBoxlinearityBBox = dict()
        self.fitParamsfitParams = dict()
        self.fitParamsErrfitParamsErr = dict()
        self.fitChiSqfitChiSq = dict()
        self.fitResidualsfitResiduals = dict()
        self.linearFitlinearFit = dict()
        self.tableDatatableData = None
        if table is not None:
            if len(table.shape) != 2:
                raise RuntimeError("table shape = %s; must have two dimensions" % (table.shape,))
            if table.shape[1] < table.shape[0]:
                raise RuntimeError("table shape = %s; indices are switched" % (table.shape,))
            self.tableDatatableData = np.array(table, order="C")
 
        super().__init__(**kwargs)
        self.requiredAttributesrequiredAttributesrequiredAttributesrequiredAttributes.update(['hasLinearity', 'override',
                                        'ampNames',
                                        'linearityCoeffs', 'linearityType', 'linearityBBox',
                                        'fitParams', 'fitParamsErr', 'fitChiSq',
                                        'fitResiduals', 'linearFit', 'tableData'])
 
    def updateMetadata(self, setDate=False, **kwargs):
        """Update metadata keywords with new values.
 
        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['HAS_LINEARITY'] = self.hasLinearityhasLinearity
        kwargs['OVERRIDE'] = self.overrideoverride
        kwargs['HAS_TABLE'] = self.tableDatatableData is not None
 
        super().updateMetadata(setDate=setDate, **kwargs)
 
    def fromDetector(self, detector):
        """Read linearity parameters from a detector.
 
        Parameters
        ----------
        detector : `lsst.afw.cameraGeom.detector`
            Input detector with parameters to use.
 
        Returns
        -------
        calib : `lsst.ip.isr.Linearizer`
            The calibration constructed from the detector.
        """
        self._detectorName_detectorName_detectorName = detector.getName()
        self._detectorSerial_detectorSerial_detectorSerial = detector.getSerial()
        self._detectorId_detectorId_detectorId = detector.getId()
        self.hasLinearityhasLinearity = True
 
        # Do not translate Threshold, Maximum, Units.
        for amp in detector.getAmplifiers():
            ampName = amp.getName()
            self.ampNamesampNames.append(ampName)
            self.linearityTypelinearityType[ampName] = amp.getLinearityType()
            self.linearityCoeffslinearityCoeffs[ampName] = amp.getLinearityCoeffs()
            self.linearityBBoxlinearityBBox[ampName] = amp.getBBox()
 
        return self
 
    @classmethod
    def fromDict(cls, dictionary):
        """Construct a calibration from a dictionary of properties
 
        Parameters
        ----------
        dictionary : `dict`
            Dictionary of properties
 
        Returns
        -------
        calib : `lsst.ip.isr.Linearity`
            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 linearity supplied.  Expected {calib._OBSTYPE}, "
                               f"found {dictionary['metadata']['OBSTYPE']}")
 
        calib.setMetadata(dictionary['metadata'])
 
        calib.hasLinearity = dictionary.get('hasLinearity',
                                            dictionary['metadata'].get('HAS_LINEARITY', False))
        calib.override = dictionary.get('override', True)
 
        if calib.hasLinearity:
            for ampName in dictionary['amplifiers']:
                amp = dictionary['amplifiers'][ampName]
                calib.ampNames.append(ampName)
                calib.linearityCoeffs[ampName] = np.array(amp.get('linearityCoeffs', [0.0]))
                calib.linearityType[ampName] = amp.get('linearityType', 'None')
                calib.linearityBBox[ampName] = amp.get('linearityBBox', None)
 
                calib.fitParams[ampName] = np.array(amp.get('fitParams', [0.0]))
                calib.fitParamsErr[ampName] = np.array(amp.get('fitParamsErr', [0.0]))
                calib.fitChiSq[ampName] = amp.get('fitChiSq', np.nan)
                calib.fitResiduals[ampName] = np.array(amp.get('fitResiduals', [0.0]))
                calib.linearFit[ampName] = np.array(amp.get('linearFit', [0.0]))
 
            calib.tableData = dictionary.get('tableData', None)
            if calib.tableData:
                calib.tableData = np.array(calib.tableData)
 
        return calib
 
    def toDict(self):
        """Return linearity parameters as a dict.
 
        Returns
        -------
        outDict : `dict`:
        """
        self.updateMetadataupdateMetadataupdateMetadata()
 
        outDict = {'metadata': self.getMetadatagetMetadata(),
                   'detectorName': self._detectorName_detectorName_detectorName,
                   'detectorSerial': self._detectorSerial_detectorSerial_detectorSerial,
                   'detectorId': self._detectorId_detectorId_detectorId,
                   'hasTable': self.tableDatatableData is not None,
                   'amplifiers': dict(),
                   }
        for ampName in self.linearityTypelinearityType:
            outDict['amplifiers'][ampName] = {'linearityType': self.linearityTypelinearityType[ampName],
                                              'linearityCoeffs': self.linearityCoeffslinearityCoeffs[ampName].tolist(),
                                              'linearityBBox': self.linearityBBoxlinearityBBox[ampName],
                                              'fitParams': self.fitParamsfitParams[ampName].tolist(),
                                              'fitParamsErr': self.fitParamsErrfitParamsErr[ampName].tolist(),
                                              'fitChiSq': self.fitChiSqfitChiSq[ampName],
                                              'fitResiduals': self.fitResidualsfitResiduals[ampName].tolist(),
                                              'linearFit': self.linearFitlinearFit[ampName].tolist()}
        if self.tableDatatableData is not None:
            outDict['tableData'] = self.tableDatatableData.tolist()
 
        return outDict
 
    @classmethod
    def fromTable(cls, tableList):
        """Read linearity from a FITS file.
 
        This method uses the `fromDict` method to create the
        calibration, after constructing an appropriate dictionary from
        the input tables.
 
        Parameters
        ----------
        tableList : `list` [`astropy.table.Table`]
            afwTable read from input file name.
 
        Returns
        -------
        linearity : `~lsst.ip.isr.linearize.Linearizer``
            Linearity parameters.
 
        Notes
        -----
        The method reads a FITS file with 1 or 2 extensions. The metadata is
        read from the header of extension 1, which must exist.  Then the table
        is loaded, and  the ['AMPLIFIER_NAME', 'TYPE', 'COEFFS', 'BBOX_X0',
        'BBOX_Y0', 'BBOX_DX', 'BBOX_DY'] columns are read and used to set each
        dictionary by looping over rows.
        Extension 2 is then attempted to read in the try block (which only
        exists for lookup tables). It has a column named 'LOOKUP_VALUES' that
        contains a vector of the lookup entries in each row.
        """
        coeffTable = tableList[0]
 
        metadata = coeffTable.meta
        inDict = dict()
        inDict['metadata'] = metadata
        inDict['hasLinearity'] = metadata.get('HAS_LINEARITY', False)
        inDict['amplifiers'] = dict()
 
        for record in coeffTable:
            ampName = record['AMPLIFIER_NAME']
 
            fitParams = record['FIT_PARAMS'] if 'FIT_PARAMS' in record.columns else np.array([0.0])
            fitParamsErr = record['FIT_PARAMS_ERR'] if 'FIT_PARAMS_ERR' in record.columns else np.array([0.0])
            fitChiSq = record['RED_CHI_SQ'] if 'RED_CHI_SQ' in record.columns else np.nan
            fitResiduals = record['FIT_RES'] if 'FIT_RES' in record.columns else np.array([0.0])
            linearFit = record['LIN_FIT'] if 'LIN_FIT' in record.columns else np.array([0.0])
 
            inDict['amplifiers'][ampName] = {
                'linearityType': record['TYPE'],
                'linearityCoeffs': record['COEFFS'],
                'linearityBBox': Box2I(Point2I(record['BBOX_X0'], record['BBOX_Y0']),
                                       Extent2I(record['BBOX_DX'], record['BBOX_DY'])),
                'fitParams': fitParams,
                'fitParamsErr': fitParamsErr,
                'fitChiSq': fitChiSq,
                'fitResiduals': fitResiduals,
                'linearFit': linearFit,
            }
 
        if len(tableList) > 1:
            tableData = tableList[1]
            inDict['tableData'] = [record['LOOKUP_VALUES'] for record in tableData]
 
        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` [`astropy.table.Table`]
            List of tables containing the linearity calibration
            information.
        """
 
        tableList = []
        self.updateMetadataupdateMetadataupdateMetadata()
        catalog = Table([{'AMPLIFIER_NAME': ampName,
                          'TYPE': self.linearityTypelinearityType[ampName],
                          'COEFFS': self.linearityCoeffslinearityCoeffs[ampName],
                          'BBOX_X0': self.linearityBBoxlinearityBBox[ampName].getMinX(),
                          'BBOX_Y0': self.linearityBBoxlinearityBBox[ampName].getMinY(),
                          'BBOX_DX': self.linearityBBoxlinearityBBox[ampName].getWidth(),
                          'BBOX_DY': self.linearityBBoxlinearityBBox[ampName].getHeight(),
                          'FIT_PARAMS': self.fitParamsfitParams[ampName],
                          'FIT_PARAMS_ERR': self.fitParamsErrfitParamsErr[ampName],
                          'RED_CHI_SQ': self.fitChiSqfitChiSq[ampName],
                          'FIT_RES': self.fitResidualsfitResiduals[ampName],
                          'LIN_FIT': self.linearFitlinearFit[ampName],
                          } for ampName in self.ampNamesampNames])
        catalog.meta = self.getMetadatagetMetadata().toDict()
        tableList.append(catalog)
 
        if self.tableDatatableData is not None:
            catalog = Table([{'LOOKUP_VALUES': value} for value in self.tableDatatableData])
            tableList.append(catalog)
        return(tableList)
 
    def getLinearityTypeByName(self, linearityTypeName):
        """Determine the linearity class to use from the type name.
 
        Parameters
        ----------
        linearityTypeName : str
            String name of the linearity type that is needed.
 
        Returns
        -------
        linearityType : `~lsst.ip.isr.linearize.LinearizeBase`
            The appropriate linearity class to use.  If no matching class
            is found, `None` is returned.
        """
        for t in [LinearizeLookupTable,
                  LinearizeSquared,
                  LinearizePolynomial,
                  LinearizeProportional,
                  LinearizeSpline,
                  LinearizeNone]:
            if t.LinearityType == linearityTypeName:
                return t
        return None
 
    def validate(self, detector=None, amplifier=None):
        """Validate linearity for a detector/amplifier.
 
        Parameters
        ----------
        detector : `lsst.afw.cameraGeom.Detector`, optional
            Detector to validate, along with its amplifiers.
        amplifier : `lsst.afw.cameraGeom.Amplifier`, optional
            Single amplifier to validate.
 
        Raises
        ------
        RuntimeError :
            Raised if there is a mismatch in linearity parameters, and
            the cameraGeom parameters are not being overridden.
        """
        amplifiersToCheck = []
        if detector:
            if self._detectorName_detectorName_detectorName != detector.getName():
                raise RuntimeError("Detector names don't match: %s != %s" %
                                   (self._detectorName_detectorName_detectorName, detector.getName()))
            if int(self._detectorId_detectorId_detectorId) != int(detector.getId()):
                raise RuntimeError("Detector IDs don't match: %s != %s" %
                                   (int(self._detectorId_detectorId_detectorId), int(detector.getId())))
            if self._detectorSerial_detectorSerial_detectorSerial != detector.getSerial():
                raise RuntimeError("Detector serial numbers don't match: %s != %s" %
                                   (self._detectorSerial_detectorSerial_detectorSerial, detector.getSerial()))
            if len(detector.getAmplifiers()) != len(self.linearityCoeffslinearityCoeffs.keys()):
                raise RuntimeError("Detector number of amps = %s does not match saved value %s" %
                                   (len(detector.getAmplifiers()),
                                    len(self.linearityCoeffslinearityCoeffs.keys())))
            amplifiersToCheck.extend(detector.getAmplifiers())
 
        if amplifier:
            amplifiersToCheck.extend(amplifier)
 
        for amp in amplifiersToCheck:
            ampName = amp.getName()
            if ampName not in self.linearityCoeffslinearityCoeffs.keys():
                raise RuntimeError("Amplifier %s is not in linearity data" %
                                   (ampName, ))
            if amp.getLinearityType() != self.linearityTypelinearityType[ampName]:
                if self.overrideoverride:
                    self.loglog.warning("Overriding amplifier defined linearityType (%s) for %s",
                                     self.linearityTypelinearityType[ampName], ampName)
                else:
                    raise RuntimeError("Amplifier %s type %s does not match saved value %s" %
                                       (ampName, amp.getLinearityType(), self.linearityTypelinearityType[ampName]))
            if (amp.getLinearityCoeffs().shape != self.linearityCoeffslinearityCoeffs[ampName].shape or not
                    np.allclose(amp.getLinearityCoeffs(), self.linearityCoeffslinearityCoeffs[ampName], equal_nan=True)):
                if self.overrideoverride:
                    self.loglog.warning("Overriding amplifier defined linearityCoeffs (%s) for %s",
                                     self.linearityCoeffslinearityCoeffs[ampName], ampName)
                else:
                    raise RuntimeError("Amplifier %s coeffs %s does not match saved value %s" %
                                       (ampName, amp.getLinearityCoeffs(), self.linearityCoeffslinearityCoeffs[ampName]))
 
    def applyLinearity(self, image, detector=None, log=None):
        """Apply the linearity to an image.
 
        If the linearity parameters are populated, use those,
        otherwise use the values from the detector.
 
        Parameters
        ----------
        image : `~lsst.afw.image.image`
            Image to correct.
        detector : `~lsst.afw.cameraGeom.detector`
            Detector to use for linearity parameters if not already
            populated.
        log : `~logging.Logger`, optional
            Log object to use for logging.
        """
        if log is None:
            log = self.loglog
        if detector and not self.hasLinearityhasLinearity:
            self.fromDetectorfromDetectorfromDetector(detector)
 
        self.validatevalidatevalidate(detector)
 
        numAmps = 0
        numLinearized = 0
        numOutOfRange = 0
        for ampName in self.linearityTypelinearityType.keys():
            linearizer = self.getLinearityTypeByNamegetLinearityTypeByName(self.linearityTypelinearityType[ampName])
            numAmps += 1
            if linearizer is not None:
                ampView = image.Factory(image, self.linearityBBoxlinearityBBox[ampName])
                success, outOfRange = linearizer()(ampView, **{'coeffs': self.linearityCoeffslinearityCoeffs[ampName],
                                                               'table': self.tableDatatableData,
                                                               'log': self.loglog})
                numOutOfRange += outOfRange
                if success:
                    numLinearized += 1
                elif log is not None:
                    log.warning("Amplifier %s did not linearize.",
                                ampName)
        return Struct(
            numAmps=numAmps,
            numLinearized=numLinearized,
            numOutOfRange=numOutOfRange
        )
 
 
class LinearizeBase(metaclass=abc.ABCMeta):
    """Abstract base class functor for correcting non-linearity.
 
    Subclasses must define __call__ and set class variable
    LinearityType to a string that will be used for linearity type in
    the cameraGeom.Amplifier.linearityType field.
 
    All linearity corrections should be defined in terms of an
    additive correction, such that:
 
    corrected_value = uncorrected_value + f(uncorrected_value)
    """
    LinearityType = None  # linearity type, a string used for AmpInfoCatalogs
 
    @abc.abstractmethod
    def __call__(self, image, **kwargs):
        """Correct non-linearity.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image`
            Image to be corrected
        kwargs : `dict`
            Dictionary of parameter keywords:
            ``"coeffs"``
                Coefficient vector (`list` or `numpy.array`).
            ``"table"``
                Lookup table data (`numpy.array`).
            ``"log"``
                Logger to handle messages (`logging.Logger`).
 
        Returns
        -------
        output : `bool`
            If true, a correction was applied successfully.
 
        Raises
        ------
        RuntimeError:
            Raised if the linearity type listed in the
            detector does not match the class type.
        """
        pass
 
 
class LinearizeLookupTable(LinearizeBase):
    """Correct non-linearity with a persisted lookup table.
 
    The lookup table consists of entries such that given
    "coefficients" c0, c1:
 
    for each i,j of image:
        rowInd = int(c0)
        colInd = int(c1 + uncorrImage[i,j])
        corrImage[i,j] = uncorrImage[i,j] + table[rowInd, colInd]
 
    - c0: row index; used to identify which row of the table to use
            (typically one per amplifier, though one can have multiple
            amplifiers use the same table)
    - c1: column index offset; added to the uncorrected image value
            before truncation; this supports tables that can handle
            negative image values; also, if the c1 ends with .5 then
            the nearest index is used instead of truncating to the
            next smaller index
    """
    LinearityType = "LookupTable"
 
    def __call__(self, image, **kwargs):
        """Correct for non-linearity.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image`
            Image to be corrected
        kwargs : `dict`
            Dictionary of parameter keywords:
            ``"coeffs"``
                Columnation vector (`list` or `numpy.array`).
            ``"table"``
                Lookup table data (`numpy.array`).
            ``"log"``
                Logger to handle messages (`logging.Logger`).
 
        Returns
        -------
        output : `tuple` [`bool`, `int`]
            If true, a correction was applied successfully.  The
            integer indicates the number of pixels that were
            uncorrectable by being out of range.
 
        Raises
        ------
        RuntimeError:
            Raised if the requested row index is out of the table
            bounds.
        """
        numOutOfRange = 0
 
        rowInd, colIndOffset = kwargs['coeffs'][0:2]
        table = kwargs['table']
        log = kwargs['log']
 
        numTableRows = table.shape[0]
        rowInd = int(rowInd)
        if rowInd < 0 or rowInd > numTableRows:
            raise RuntimeError("LinearizeLookupTable rowInd=%s not in range[0, %s)" %
                               (rowInd, numTableRows))
        tableRow = np.array(table[rowInd, :], dtype=image.getArray().dtype)
 
        numOutOfRange += applyLookupTable(image, tableRow, colIndOffset)
 
        if numOutOfRange > 0 and log is not None:
            log.warning("%s pixels were out of range of the linearization table",
                        numOutOfRange)
        if numOutOfRange < image.getArray().size:
            return True, numOutOfRange
        else:
            return False, numOutOfRange
 
 
class LinearizePolynomial(LinearizeBase):
    """Correct non-linearity with a polynomial mode.
 
    corrImage = uncorrImage + sum_i c_i uncorrImage^(2 + i)
 
    where c_i are the linearity coefficients for each amplifier.
    Lower order coefficients are not included as they duplicate other
    calibration parameters:
        ``"k0"``
            A coefficient multiplied by uncorrImage**0 is equivalent to
            bias level.  Irrelevant for correcting non-linearity.
        ``"k1"``
            A coefficient multiplied by uncorrImage**1 is proportional
            to the gain.  Not necessary for correcting non-linearity.
    """
    LinearityType = "Polynomial"
 
    def __call__(self, image, **kwargs):
        """Correct non-linearity.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image`
            Image to be corrected
        kwargs : `dict`
            Dictionary of parameter keywords:
            ``"coeffs"``
                Coefficient vector (`list` or `numpy.array`).
                If the order of the polynomial is n, this list
                should have a length of n-1 ("k0" and "k1" are
                not needed for the correction).
            ``"log"``
                Logger to handle messages (`logging.Logger`).
 
        Returns
        -------
        output : `tuple` [`bool`, `int`]
            If true, a correction was applied successfully.  The
            integer indicates the number of pixels that were
            uncorrectable by being out of range.
        """
        if not np.any(np.isfinite(kwargs['coeffs'])):
            return False, 0
        if not np.any(kwargs['coeffs']):
            return False, 0
 
        ampArray = image.getArray()
        correction = np.zeros_like(ampArray)
        for order, coeff in enumerate(kwargs['coeffs'], start=2):
            correction += coeff * np.power(ampArray, order)
        ampArray += correction
 
        return True, 0
 
 
class LinearizeSquared(LinearizeBase):
    """Correct non-linearity with a squared model.
 
    corrImage = uncorrImage + c0*uncorrImage^2
 
    where c0 is linearity coefficient 0 for each amplifier.
    """
    LinearityType = "Squared"
 
    def __call__(self, image, **kwargs):
        """Correct for non-linearity.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image`
            Image to be corrected
        kwargs : `dict`
            Dictionary of parameter keywords:
            ``"coeffs"``
                Coefficient vector (`list` or `numpy.array`).
            ``"log"``
                Logger to handle messages (`logging.Logger`).
 
        Returns
        -------
        output : `tuple` [`bool`, `int`]
            If true, a correction was applied successfully.  The
            integer indicates the number of pixels that were
            uncorrectable by being out of range.
        """
 
        sqCoeff = kwargs['coeffs'][0]
        if sqCoeff != 0:
            ampArr = image.getArray()
            ampArr *= (1 + sqCoeff*ampArr)
            return True, 0
        else:
            return False, 0
 
 
class LinearizeSpline(LinearizeBase):
    """Correct non-linearity with a spline model.
 
    corrImage = uncorrImage - Spline(coeffs, uncorrImage)
 
    Notes
    -----
 
    The spline fit calculates a correction as a function of the
    expected linear flux term.  Because of this, the correction needs
    to be subtracted from the observed flux.
 
    """
    LinearityType = "Spline"
 
    def __call__(self, image, **kwargs):
        """Correct for non-linearity.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image`
            Image to be corrected
        kwargs : `dict`
            Dictionary of parameter keywords:
            ``"coeffs"``
                Coefficient vector (`list` or `numpy.array`).
            ``"log"``
                Logger to handle messages (`logging.Logger`).
 
        Returns
        -------
        output : `tuple` [`bool`, `int`]
            If true, a correction was applied successfully.  The
            integer indicates the number of pixels that were
            uncorrectable by being out of range.
        """
        splineCoeff = kwargs['coeffs']
        centers, values = np.split(splineCoeff, 2)
        interp = afwMath.makeInterpolate(centers.tolist(), values.tolist(),
                                         afwMath.stringToInterpStyle("AKIMA_SPLINE"))
 
        ampArr = image.getArray()
        delta = interp.interpolate(ampArr.flatten())
        ampArr -= np.array(delta).reshape(ampArr.shape)
 
        return True, 0
 
 
class LinearizeProportional(LinearizeBase):
    """Do not correct non-linearity.
    """
    LinearityType = "Proportional"
 
    def __call__(self, image, **kwargs):
        """Do not correct for non-linearity.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image`
            Image to be corrected
        kwargs : `dict`
            Dictionary of parameter keywords:
            ``"coeffs"``
                Coefficient vector (`list` or `numpy.array`).
            ``"log"``
                Logger to handle messages (`logging.Logger`).
 
        Returns
        -------
        output : `tuple` [`bool`, `int`]
            If true, a correction was applied successfully.  The
            integer indicates the number of pixels that were
            uncorrectable by being out of range.
        """
        return True, 0
 
 
class LinearizeNone(LinearizeBase):
    """Do not correct non-linearity.
    """
    LinearityType = "None"
 
    def __call__(self, image, **kwargs):
        """Do not correct for non-linearity.
 
        Parameters
        ----------
        image : `lsst.afw.image.Image`
            Image to be corrected
        kwargs : `dict`
            Dictionary of parameter keywords:
            ``"coeffs"``
                Coefficient vector (`list` or `numpy.array`).
            ``"log"``
                Logger to handle messages (`logging.Logger`).
 
        Returns
        -------
        output : `tuple` [`bool`, `int`]
            If true, a correction was applied successfully.  The
            integer indicates the number of pixels that were
            uncorrectable by being out of range.
        """
        return True, 0