jointcal.py

Go to the documentation of this file.

# This file is part of jointcal.
#
# Developed for the LSST Data Management System.
# This product includes software developed by the LSST Project
# (https://www.lsst.org).
# See the COPYRIGHT file at the top-level directory of this distribution
# for details of code ownership.
#
# 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 GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
import dataclasses
import collections
import os
 
import astropy.time
import numpy as np
import astropy.units as u
 
import lsst.geom
import lsst.utils
import lsst.pex.config as pexConfig
import lsst.pipe.base as pipeBase
from lsst.afw.image import fluxErrFromABMagErr
import lsst.pex.exceptions as pexExceptions
import lsst.afw.cameraGeom
import lsst.afw.table
import lsst.log
from lsst.pipe.tasks.colorterms import ColortermLibrary
from lsst.verify import Job, Measurement
 
from lsst.meas.algorithms import (LoadIndexedReferenceObjectsTask, ReferenceSourceSelectorTask,
                                  ReferenceObjectLoader)
from lsst.meas.algorithms.sourceSelector import sourceSelectorRegistry
 
from .dataIds import PerTractCcdDataIdContainer
 
import lsst.jointcal
from lsst.jointcal import MinimizeResult
 
__all__ = ["JointcalConfig", "JointcalRunner", "JointcalTask"]
 
Photometry = collections.namedtuple('Photometry', ('fit', 'model'))
Astrometry = collections.namedtuple('Astrometry', ('fit', 'model', 'sky_to_tan_projection'))
 
 
# TODO: move this to MeasurementSet in lsst.verify per DM-12655.
def add_measurement(job, name, value):
    meas = Measurement(job.metrics[name], value)
    job.measurements.insert(meas)
 
 
class JointcalRunner(pipeBase.ButlerInitializedTaskRunner):
    """Subclass of TaskRunner for jointcalTask (gen2)
 
    jointcalTask.runDataRef() takes a number of arguments, one of which is a list of dataRefs
    extracted from the command line (whereas most CmdLineTasks' runDataRef methods take
    single dataRef, are are called repeatedly). This class transforms the processed
    arguments generated by the ArgumentParser into the arguments expected by
    Jointcal.runDataRef().
 
    See pipeBase.TaskRunner for more information.
    """
 
    @staticmethod
    def getTargetList(parsedCmd, **kwargs):
        """
        Return a list of tuples per tract, each containing (dataRefs, kwargs).
 
        Jointcal operates on lists of dataRefs simultaneously.
        """
        kwargs['butler'] = parsedCmd.butler
 
        # organize data IDs by tract
        refListDict = {}
        for ref in parsedCmd.id.refList:
            refListDict.setdefault(ref.dataId["tract"], []).append(ref)
        # we call runDataRef() once with each tract
        result = [(refListDict[tract], kwargs) for tract in sorted(refListDict.keys())]
        return result
 
    def __call__(self, args):
        """
        Parameters
        ----------
        args
            Arguments for Task.runDataRef()
 
        Returns
        -------
        pipe.base.Struct
            if self.doReturnResults is False:
 
            - ``exitStatus``: 0 if the task completed successfully, 1 otherwise.
 
            if self.doReturnResults is True:
 
            - ``result``: the result of calling jointcal.runDataRef()
            - ``exitStatus``: 0 if the task completed successfully, 1 otherwise.
        """
        exitStatus = 0  # exit status for shell
 
        # NOTE: cannot call self.makeTask because that assumes args[0] is a single dataRef.
        dataRefList, kwargs = args
        butler = kwargs.pop('butler')
        task = self.TaskClass(config=self.config, log=self.log, butler=butler)
        result = None
        try:
            result = task.runDataRef(dataRefList, **kwargs)
            exitStatus = result.exitStatus
            job_path = butler.get('verify_job_filename')
            result.job.write(job_path[0])
        except Exception as e:  # catch everything, sort it out later.
            if self.doRaise:
                raise e
            else:
                exitStatus = 1
                eName = type(e).__name__
                tract = dataRefList[0].dataId['tract']
                task.log.fatal("Failed processing tract %s, %s: %s", tract, eName, e)
 
        # Put the butler back into kwargs for the other Tasks.
        kwargs['butler'] = butler
        if self.doReturnResults:
            return pipeBase.Struct(result=result, exitStatus=exitStatus)
        else:
            return pipeBase.Struct(exitStatus=exitStatus)
 
 
class JointcalTaskConnections(pipeBase.PipelineTaskConnections,
                              dimensions=("skymap", "tract", "instrument", "physical_filter")):
    """Middleware input/output connections for jointcal data."""
    inputCamera = pipeBase.connectionTypes.Input(
        doc="The camera instrument that took these observations.",
        name="camera",
        storageClass="Camera",
        dimensions=("instrument",),
        isCalibration=True
    )
    inputSourceTableVisit = pipeBase.connectionTypes.Input(
        doc="Source table in parquet format, per visit",
        name="sourceTable_visit",
        storageClass="DataFrame",
        dimensions=("instrument", "visit"),
        deferLoad=True,
        multiple=True,
    )
    inputVisitSummary = pipeBase.connectionTypes.Input(
        doc=("Per-visit consolidated exposure metadata built from calexps. "
             "These catalogs use detector id for the id and must be sorted for "
             "fast lookups of a detector."),
        name="visitSummary",
        storageClass="ExposureCatalog",
        dimensions=("instrument", "visit"),
        deferLoad=True,
        multiple=True,
    )
    # TODO DM-28991: This does not load enough refcat data: the graph only to gets
    # refcats that touch the tract, but we need to go larger because we're
    # loading visits that may extend further. How to test that?
    astrometryRefCat = pipeBase.connectionTypes.Input(
        doc="The astrometry reference catalog to match to loaded input catalog sources.",
        name="gaia_dr2_20200414",
        storageClass="SimpleCatalog",
        dimensions=("htm7",),
        deferLoad=True,
        multiple=True
    )
    photometryRefCat = pipeBase.connectionTypes.Input(
        doc="The photometry reference catalog to match to loaded input catalog sources.",
        name="ps1_pv3_3pi_20170110",
        storageClass="SimpleCatalog",
        dimensions=("htm7",),
        deferLoad=True,
        multiple=True
    )
 
    outputWcs = pipeBase.connectionTypes.Output(
        doc=("Per-tract, per-visit world coordinate systems derived from the fitted model."
             " These catalogs only contain entries for detectors with an output, and use"
             " the detector id for the catalog id, sorted on id for fast lookups of a detector."),
        name="jointcalSkyWcsCatalog",
        storageClass="ExposureCatalog",
        dimensions=("instrument", "visit", "skymap", "tract"),
        multiple=True
    )
    outputPhotoCalib = pipeBase.connectionTypes.Output(
        doc=("Per-tract, per-visit photometric calibrations derived from the fitted model."
             " These catalogs only contain entries for detectors with an output, and use"
             " the detector id for the catalog id, sorted on id for fast lookups of a detector."),
        name="jointcalPhotoCalibCatalog",
        storageClass="ExposureCatalog",
        dimensions=("instrument", "visit", "skymap", "tract"),
        multiple=True
    )
 
 
class JointcalConfig(pipeBase.PipelineTaskConfig,
                     pipelineConnections=JointcalTaskConnections):
    """Configuration for JointcalTask"""
 
    doAstrometry = pexConfig.Field(
        doc="Fit astrometry and write the fitted result.",
        dtype=bool,
        default=True
    )
    doPhotometry = pexConfig.Field(
        doc="Fit photometry and write the fitted result.",
        dtype=bool,
        default=True
    )
    coaddName = pexConfig.Field(
        doc="Type of coadd, typically deep or goodSeeing",
        dtype=str,
        default="deep"
    )
    # TODO DM-29008: Change this to "ApFlux_12_0" before gen2 removal.
    sourceFluxType = pexConfig.Field(
        dtype=str,
        doc="Source flux field to use in source selection and to get fluxes from the catalog.",
        default='Calib'
    )
    positionErrorPedestal = pexConfig.Field(
        doc="Systematic term to apply to the measured position error (pixels)",
        dtype=float,
        default=0.02,
    )
    photometryErrorPedestal = pexConfig.Field(
        doc="Systematic term to apply to the measured error on flux or magnitude as a "
        "fraction of source flux or magnitude delta (e.g. 0.05 is 5% of flux or +50 millimag).",
        dtype=float,
        default=0.0,
    )
    # TODO: DM-6885 matchCut should be an geom.Angle
    matchCut = pexConfig.Field(
        doc="Matching radius between fitted and reference stars (arcseconds)",
        dtype=float,
        default=3.0,
    )
    minMeasurements = pexConfig.Field(
        doc="Minimum number of associated measured stars for a fitted star to be included in the fit",
        dtype=int,
        default=2,
    )
    minMeasuredStarsPerCcd = pexConfig.Field(
        doc="Minimum number of measuredStars per ccdImage before printing warnings",
        dtype=int,
        default=100,
    )
    minRefStarsPerCcd = pexConfig.Field(
        doc="Minimum number of measuredStars per ccdImage before printing warnings",
        dtype=int,
        default=30,
    )
    allowLineSearch = pexConfig.Field(
        doc="Allow a line search during minimization, if it is reasonable for the model"
        " (models with a significant non-linear component, e.g. constrainedPhotometry).",
        dtype=bool,
        default=False
    )
    astrometrySimpleOrder = pexConfig.Field(
        doc="Polynomial order for fitting the simple astrometry model.",
        dtype=int,
        default=3,
    )
    astrometryChipOrder = pexConfig.Field(
        doc="Order of the per-chip transform for the constrained astrometry model.",
        dtype=int,
        default=1,
    )
    astrometryVisitOrder = pexConfig.Field(
        doc="Order of the per-visit transform for the constrained astrometry model.",
        dtype=int,
        default=5,
    )
    useInputWcs = pexConfig.Field(
        doc="Use the input calexp WCSs to initialize a SimpleAstrometryModel.",
        dtype=bool,
        default=True,
    )
    astrometryModel = pexConfig.ChoiceField(
        doc="Type of model to fit to astrometry",
        dtype=str,
        default="constrained",
        allowed={"simple": "One polynomial per ccd",
                 "constrained": "One polynomial per ccd, and one polynomial per visit"}
    )
    photometryModel = pexConfig.ChoiceField(
        doc="Type of model to fit to photometry",
        dtype=str,
        default="constrainedMagnitude",
        allowed={"simpleFlux": "One constant zeropoint per ccd and visit, fitting in flux space.",
                 "constrainedFlux": "Constrained zeropoint per ccd, and one polynomial per visit,"
                 " fitting in flux space.",
                 "simpleMagnitude": "One constant zeropoint per ccd and visit,"
                 " fitting in magnitude space.",
                 "constrainedMagnitude": "Constrained zeropoint per ccd, and one polynomial per visit,"
                 " fitting in magnitude space.",
                 }
    )
    applyColorTerms = pexConfig.Field(
        doc="Apply photometric color terms to reference stars?"
            "Requires that colorterms be set to a ColortermLibrary",
        dtype=bool,
        default=False
    )
    colorterms = pexConfig.ConfigField(
        doc="Library of photometric reference catalog name to color term dict.",
        dtype=ColortermLibrary,
    )
    photometryVisitOrder = pexConfig.Field(
        doc="Order of the per-visit polynomial transform for the constrained photometry model.",
        dtype=int,
        default=7,
    )
    photometryDoRankUpdate = pexConfig.Field(
        doc=("Do the rank update step during minimization. "
             "Skipping this can help deal with models that are too non-linear."),
        dtype=bool,
        default=True,
    )
    astrometryDoRankUpdate = pexConfig.Field(
        doc=("Do the rank update step during minimization (should not change the astrometry fit). "
             "Skipping this can help deal with models that are too non-linear."),
        dtype=bool,
        default=True,
    )
    outlierRejectSigma = pexConfig.Field(
        doc="How many sigma to reject outliers at during minimization.",
        dtype=float,
        default=5.0,
    )
    maxPhotometrySteps = pexConfig.Field(
        doc="Maximum number of minimize iterations to take when fitting photometry.",
        dtype=int,
        default=20,
    )
    maxAstrometrySteps = pexConfig.Field(
        doc="Maximum number of minimize iterations to take when fitting photometry.",
        dtype=int,
        default=20,
    )
    astrometryRefObjLoader = pexConfig.ConfigurableField(
        target=LoadIndexedReferenceObjectsTask,
        doc="Reference object loader for astrometric fit",
    )
    photometryRefObjLoader = pexConfig.ConfigurableField(
        target=LoadIndexedReferenceObjectsTask,
        doc="Reference object loader for photometric fit",
    )
    sourceSelector = sourceSelectorRegistry.makeField(
        doc="How to select sources for cross-matching",
        default="astrometry"
    )
    astrometryReferenceSelector = pexConfig.ConfigurableField(
        target=ReferenceSourceSelectorTask,
        doc="How to down-select the loaded astrometry reference catalog.",
    )
    photometryReferenceSelector = pexConfig.ConfigurableField(
        target=ReferenceSourceSelectorTask,
        doc="How to down-select the loaded photometry reference catalog.",
    )
    astrometryReferenceErr = pexConfig.Field(
        doc=("Uncertainty on reference catalog coordinates [mas] to use in place of the `coord_*Err` fields. "
             "If None, then raise an exception if the reference catalog is missing coordinate errors. "
             "If specified, overrides any existing `coord_*Err` values."),
        dtype=float,
        default=None,
        optional=True
    )
 
    # configs for outputting debug information
    writeInitMatrix = pexConfig.Field(
        dtype=bool,
        doc=("Write the pre/post-initialization Hessian and gradient to text files, for debugging. "
             "The output files will be of the form 'astrometry_preinit-mat.txt', in the current directory. "
             "Note that these files are the dense versions of the matrix, and so may be very large."),
        default=False
    )
    writeChi2FilesInitialFinal = pexConfig.Field(
        dtype=bool,
        doc="Write .csv files containing the contributions to chi2 for the initialization and final fit.",
        default=False
    )
    writeChi2FilesOuterLoop = pexConfig.Field(
        dtype=bool,
        doc="Write .csv files containing the contributions to chi2 for the outer fit loop.",
        default=False
    )
    writeInitialModel = pexConfig.Field(
        dtype=bool,
        doc=("Write the pre-initialization model to text files, for debugging."
             " Output is written to `initial[Astro|Photo]metryModel.txt` in the current working directory."),
        default=False
    )
    debugOutputPath = pexConfig.Field(
        dtype=str,
        default=".",
        doc=("Path to write debug output files to. Used by "
             "`writeInitialModel`, `writeChi2Files*`, `writeInitMatrix`.")
    )
    detailedProfile = pexConfig.Field(
        dtype=bool,
        default=False,
        doc="Output separate profiling information for different parts of jointcal, e.g. data read, fitting"
    )
 
    def validate(self):
        super().validate()
        if self.doPhotometrydoPhotometry and self.applyColorTermsapplyColorTerms and len(self.colortermscolorterms.data) == 0:
            msg = "applyColorTerms=True requires the `colorterms` field be set to a ColortermLibrary."
            raise pexConfig.FieldValidationError(JointcalConfig.colorterms, self, msg)
        if self.doAstrometrydoAstrometry and not self.doPhotometrydoPhotometry and self.applyColorTermsapplyColorTerms:
            msg = ("Only doing astrometry, but Colorterms are not applied for astrometry;"
                   "applyColorTerms=True will be ignored.")
            lsst.log.warn(msg)
 
    def setDefaults(self):
        # Use science source selector which can filter on extendedness, SNR, and whether blended
        self.sourceSelectorsourceSelector.name = 'science'
        # Use only stars because aperture fluxes of galaxies are biased and depend on seeing
        self.sourceSelectorsourceSelector['science'].doUnresolved = True
        # with dependable signal to noise ratio.
        self.sourceSelectorsourceSelector['science'].doSignalToNoise = True
        # Min SNR must be > 0 because jointcal cannot handle negative fluxes,
        # and S/N > 10 to use sources that are not too faint, and thus better measured.
        self.sourceSelectorsourceSelector['science'].signalToNoise.minimum = 10.
        # Base SNR on CalibFlux because that is the flux jointcal that fits and must be positive
        fluxField = f"slot_{self.sourceFluxType}Flux_instFlux"
        self.sourceSelectorsourceSelector['science'].signalToNoise.fluxField = fluxField
        self.sourceSelectorsourceSelector['science'].signalToNoise.errField = fluxField + "Err"
        # Do not trust blended sources' aperture fluxes which also depend on seeing
        self.sourceSelectorsourceSelector['science'].doIsolated = True
        # Do not trust either flux or centroid measurements with flags,
        # chosen from the usual QA flags for stars)
        self.sourceSelectorsourceSelector['science'].doFlags = True
        badFlags = ['base_PixelFlags_flag_edge', 'base_PixelFlags_flag_saturated',
                    'base_PixelFlags_flag_interpolatedCenter', 'base_SdssCentroid_flag',
                    'base_PsfFlux_flag', 'base_PixelFlags_flag_suspectCenter']
        self.sourceSelectorsourceSelector['science'].flags.bad = badFlags
 
        # Default to Gaia-DR2 (with proper motions) for astrometry and
        # PS1-DR1 for photometry, with a reasonable initial filterMap.
        self.astrometryRefObjLoaderastrometryRefObjLoader.ref_dataset_name = "gaia_dr2_20200414"
        self.astrometryRefObjLoaderastrometryRefObjLoader.requireProperMotion = True
        self.astrometryRefObjLoaderastrometryRefObjLoader.anyFilterMapsToThis = 'phot_g_mean'
        self.photometryRefObjLoaderphotometryRefObjLoader.ref_dataset_name = "ps1_pv3_3pi_20170110"
 
 
def writeModel(model, filename, log):
    """Write model to outfile."""
    with open(filename, "w") as file:
        file.write(repr(model))
    log.info("Wrote %s to file: %s", model, filename)
 
 
@dataclasses.dataclass
class JointcalInputData:
    """The input data jointcal needs for each detector/visit."""
    visit: int
    """The visit identifier of this exposure."""
    catalog: lsst.afw.table.SourceCatalog
    """The catalog derived from this exposure."""
    visitInfo: lsst.afw.image.VisitInfo
    """The VisitInfo of this exposure."""
    detector: lsst.afw.cameraGeom.Detector
    """The detector of this exposure."""
    photoCalib: lsst.afw.image.PhotoCalib
    """The photometric calibration of this exposure."""
    wcs: lsst.afw.geom.skyWcs
    """The WCS of this exposure."""
    bbox: lsst.geom.Box2I
    """The bounding box of this exposure."""
    filter: lsst.afw.image.FilterLabel
    """The filter of this exposure."""
 
 
class JointcalTask(pipeBase.PipelineTask, pipeBase.CmdLineTask):
    """Astrometricly and photometricly calibrate across multiple visits of the
    same field.
 
    Parameters
    ----------
    butler : `lsst.daf.persistence.Butler`
        The butler is passed to the refObjLoader constructor in case it is
        needed. Ignored if the refObjLoader argument provides a loader directly.
        Used to initialize the astrometry and photometry refObjLoaders.
    initInputs : `dict`, optional
        Dictionary used to initialize PipelineTasks (empty for jointcal).
    """
 
    ConfigClass = JointcalConfig
    RunnerClass = JointcalRunner
    _DefaultName = "jointcal"
 
    def __init__(self, butler=None, initInputs=None, **kwargs):
        super().__init__(**kwargs)
        self.makeSubtask("sourceSelector")
        if self.config.doAstrometry:
            if initInputs is None:
                # gen3 middleware does refcat things internally (and will not have a butler here)
                self.makeSubtask('astrometryRefObjLoader', butler=butler)
            self.makeSubtask("astrometryReferenceSelector")
        else:
            self.astrometryRefObjLoaderastrometryRefObjLoader = None
        if self.config.doPhotometry:
            if initInputs is None:
                # gen3 middleware does refcat things internally (and will not have a butler here)
                self.makeSubtask('photometryRefObjLoader', butler=butler)
            self.makeSubtask("photometryReferenceSelector")
        else:
            self.photometryRefObjLoaderphotometryRefObjLoader = None
 
        # To hold various computed metrics for use by tests
        self.jobjob = Job.load_metrics_package(subset='jointcal')
 
    def runQuantum(self, butlerQC, inputRefs, outputRefs):
        # We override runQuantum to set up the refObjLoaders and write the
        # outputs to the correct refs.
        inputs = butlerQC.get(inputRefs)
        # We want the tract number for writing debug files
        tract = butlerQC.quantum.dataId['tract']
        if self.config.doAstrometry:
            self.astrometryRefObjLoaderastrometryRefObjLoader = ReferenceObjectLoader(
                dataIds=[ref.datasetRef.dataId
                         for ref in inputRefs.astrometryRefCat],
                refCats=inputs.pop('astrometryRefCat'),
                config=self.config.astrometryRefObjLoader,
                log=self.log)
        if self.config.doPhotometry:
            self.photometryRefObjLoaderphotometryRefObjLoader = ReferenceObjectLoader(
                dataIds=[ref.datasetRef.dataId
                         for ref in inputRefs.photometryRefCat],
                refCats=inputs.pop('photometryRefCat'),
                config=self.config.photometryRefObjLoader,
                log=self.log)
        outputs = self.runrun(**inputs, tract=tract)
        if self.config.doAstrometry:
            self._put_output_put_output(butlerQC, outputs.outputWcs, outputRefs.outputWcs,
                             inputs['inputCamera'], "setWcs")
        if self.config.doPhotometry:
            self._put_output_put_output(butlerQC, outputs.outputPhotoCalib, outputRefs.outputPhotoCalib,
                             inputs['inputCamera'], "setPhotoCalib")
 
    def _put_output(self, butlerQC, outputs, outputRefs, camera, setter):
        """Persist the output datasets to their appropriate datarefs.
 
        Parameters
        ----------
        butlerQC : `ButlerQuantumContext`
            A butler which is specialized to operate in the context of a
            `lsst.daf.butler.Quantum`; This is the input to `runQuantum`.
        outputs : `dict` [`tuple`, `lsst.afw.geom.SkyWcs`] or
                  `dict` [`tuple, `lsst.afw.image.PhotoCalib`]
            The fitted objects to persist.
        outputRefs : `list` [`OutputQuantizedConnection`]
            The DatasetRefs to persist the data to.
        camera : `lsst.afw.cameraGeom.Camera`
            The camera for this instrument, to get detector ids from.
        setter : `str`
            The method to call on the ExposureCatalog to set each output.
        """
        schema = lsst.afw.table.ExposureTable.makeMinimalSchema()
        schema.addField('visit', type='I', doc='Visit number')
 
        def new_catalog(visit, size):
            """Return an catalog ready to be filled with appropriate output."""
            catalog = lsst.afw.table.ExposureCatalog(schema)
            catalog.resize(size)
            catalog['visit'] = visit
            metadata = lsst.daf.base.PropertyList()
            metadata.add("COMMENT", "Catalog id is detector id, sorted.")
            metadata.add("COMMENT", "Only detectors with data have entries.")
            return catalog
 
        # count how many detectors have output for each visit
        detectors_per_visit = collections.defaultdict(int)
        for key in outputs:
            # key is (visit, detector_id), and we only need visit here
            detectors_per_visit[key[0]] += 1
 
        for ref in outputRefs:
            visit = ref.dataId['visit']
            catalog = new_catalog(visit, detectors_per_visit[visit])
            # Iterate over every detector and skip the ones we don't have output for.
            i = 0
            for detector in camera:
                detectorId = detector.getId()
                key = (ref.dataId['visit'], detectorId)
                if key not in outputs:
                    # skip detectors we don't have output for
                    self.log.debug("No %s output for detector %s in visit %s",
                                   setter[3:], detectorId, visit)
                    continue
 
                catalog[i].setId(detectorId)
                getattr(catalog[i], setter)(outputs[key])
                i += 1
 
            catalog.sort()  # ensure that the detectors are in sorted order, for fast lookups
            butlerQC.put(catalog, ref)
            self.log.info("Wrote %s detectors to %s", i, ref)
 
    def run(self, inputSourceTableVisit, inputVisitSummary, inputCamera, tract=None):
        # Docstring inherited.
 
        # We take values out of the Parquet table, and put them in "flux_",
        # and the config.sourceFluxType field is used during that extraction,
        # so just use "flux" here.
        sourceFluxField = "flux"
        jointcalControl = lsst.jointcal.JointcalControl(sourceFluxField)
        associations = lsst.jointcal.Associations()
        self.focalPlaneBBoxfocalPlaneBBox = inputCamera.getFpBBox()
        oldWcsList, bands = self._load_data_load_data(inputSourceTableVisit,
                                            inputVisitSummary,
                                            associations,
                                            jointcalControl,
                                            inputCamera)
 
        boundingCircle, center, radius, defaultFilter = self._prep_sky_prep_sky(associations, bands)
        epoch = self._compute_proper_motion_epoch_compute_proper_motion_epoch(associations.getCcdImageList())
 
        if self.config.doAstrometry:
            astrometry = self._do_load_refcat_and_fit_do_load_refcat_and_fit(associations, defaultFilter, center, radius,
                                                      name="astrometry",
                                                      refObjLoader=self.astrometryRefObjLoaderastrometryRefObjLoader,
                                                      referenceSelector=self.astrometryReferenceSelector,
                                                      fit_function=self._fit_astrometry_fit_astrometry,
                                                      tract=tract,
                                                      epoch=epoch)
            astrometry_output = self._make_output_make_output(associations.getCcdImageList(),
                                                  astrometry.model,
                                                  "makeSkyWcs")
        else:
            astrometry_output = None
 
        if self.config.doPhotometry:
            photometry = self._do_load_refcat_and_fit_do_load_refcat_and_fit(associations, defaultFilter, center, radius,
                                                      name="photometry",
                                                      refObjLoader=self.photometryRefObjLoaderphotometryRefObjLoader,
                                                      referenceSelector=self.photometryReferenceSelector,
                                                      fit_function=self._fit_photometry_fit_photometry,
                                                      tract=tract,
                                                      epoch=epoch,
                                                      reject_bad_fluxes=True)
            photometry_output = self._make_output_make_output(associations.getCcdImageList(),
                                                  photometry.model,
                                                  "toPhotoCalib")
        else:
            photometry_output = None
 
        return pipeBase.Struct(outputWcs=astrometry_output,
                               outputPhotoCalib=photometry_output,
                               job=self.jobjob,
                               astrometryRefObjLoader=self.astrometryRefObjLoaderastrometryRefObjLoader,
                               photometryRefObjLoader=self.photometryRefObjLoaderphotometryRefObjLoader)
 
    def _make_schema_table(self):
        """Return an afw SourceTable to use as a base for creating the
        SourceCatalog to insert values from the dataFrame into.
 
        Returns
        -------
        table : `lsst.afw.table.SourceTable`
            Table with schema and slots to use to make SourceCatalogs.
        """
        schema = lsst.afw.table.SourceTable.makeMinimalSchema()
        schema.addField("centroid_x", "D")
        schema.addField("centroid_y", "D")
        schema.addField("centroid_xErr", "F")
        schema.addField("centroid_yErr", "F")
        schema.addField("shape_xx", "D")
        schema.addField("shape_yy", "D")
        schema.addField("shape_xy", "D")
        schema.addField("flux_instFlux", "D")
        schema.addField("flux_instFluxErr", "D")
        table = lsst.afw.table.SourceTable.make(schema)
        table.defineCentroid("centroid")
        table.defineShape("shape")
        return table
 
    def _extract_detector_catalog_from_visit_catalog(self, table, visitCatalog, detectorId):
        """Return an afw SourceCatalog extracted from a visit-level dataframe,
        limited to just one detector.
 
        Parameters
        ----------
        table : `lsst.afw.table.SourceTable`
            Table factory to use to make the SourceCatalog that will be
            populated with data from ``visitCatalog``.
        visitCatalog : `pandas.DataFrame`
            DataFrame to extract a detector catalog from.
        detectorId : `int`
            Numeric id of the detector to extract from ``visitCatalog``.
 
        Returns
        -------
        catalog : `lsst.afw.table.SourceCatalog`
            Detector-level catalog extracted from ``visitCatalog``.
        """
        # map from dataFrame column to afw table column
        mapping = {'sourceId': 'id',
                   'x': 'centroid_x',
                   'y': 'centroid_y',
                   'xErr': 'centroid_xErr',
                   'yErr': 'centroid_yErr',
                   'Ixx': 'shape_xx',
                   'Iyy': 'shape_yy',
                   'Ixy': 'shape_xy',
                   f'{self.config.sourceFluxType}_instFlux': 'flux_instFlux',
                   f'{self.config.sourceFluxType}_instFluxErr': 'flux_instFluxErr',
                   }
        catalog = lsst.afw.table.SourceCatalog(table)
        matched = visitCatalog['ccd'] == detectorId
        catalog.resize(sum(matched))
        view = visitCatalog.loc[matched]
        for dfCol, afwCol in mapping.items():
            catalog[afwCol] = view[dfCol]
 
        self.log.debug("%d sources selected in visit %d detector %d",
                       len(catalog),
                       view['visit'].iloc[0],  # all visits in this catalog are the same, so take the first
                       detectorId)
        return catalog
 
    def _load_data(self, inputSourceTableVisit, inputVisitSummary, associations,
                   jointcalControl, camera):
        """Read the data that jointcal needs to run. (Gen3 version)
 
        Modifies ``associations`` in-place with the loaded data.
 
        Parameters
        ----------
        inputSourceTableVisit : `list` [`lsst.daf.butler.DeferredDatasetHandle`]
            References to visit-level DataFrames to load the catalog data from.
        inputVisitSummary : `list` [`lsst.daf.butler.DeferredDatasetHandle`]
            Visit-level exposure summary catalog with metadata.
        associations : `lsst.jointcal.Associations`
            Object to add the loaded data to by constructing new CcdImages.
        jointcalControl : `jointcal.JointcalControl`
            Control object for C++ associations management.
        camera : `lsst.afw.cameraGeom.Camera`
            Camera object for detector geometry.
 
        Returns
        -------
        oldWcsList: `list` [`lsst.afw.geom.SkyWcs`]
            The original WCS of the input data, to aid in writing tests.
        bands : `list` [`str`]
            The filter bands of each input dataset.
        """
        oldWcsList = []
        filters = []
        load_cat_prof_file = 'jointcal_load_data.prof' if self.config.detailedProfile else ''
        with pipeBase.cmdLineTask.profile(load_cat_prof_file):
            table = self._make_schema_table_make_schema_table()  # every detector catalog has the same layout
            # No guarantee that the input is in the same order of visits, so we have to map one of them.
            catalogMap = {ref.dataId['visit']: i for i, ref in enumerate(inputSourceTableVisit)}
            detectorDict = {detector.getId(): detector for detector in camera}
 
            for visitSummaryRef in inputVisitSummary:
                visitSummary = visitSummaryRef.get()
                visitCatalog = inputSourceTableVisit[catalogMap[visitSummaryRef.dataId['visit']]].get()
                selected = self.sourceSelector.run(visitCatalog)
 
                # Build a CcdImage for each detector in this visit.
                detectors = {id: index for index, id in enumerate(visitSummary['id'])}
                for id, index in detectors.items():
                    catalog = self._extract_detector_catalog_from_visit_catalog_extract_detector_catalog_from_visit_catalog(table, selected.sourceCat, id)
                    data = self._make_one_input_data_make_one_input_data(visitSummary[index], catalog, detectorDict)
                    result = self._build_ccdImage_build_ccdImage(data, associations, jointcalControl)
                    if result is not None:
                        oldWcsList.append(result.wcs)
                        # A visit has only one band, so we can just use the first.
                        filters.append(data.filter)
        if len(filters) == 0:
            raise RuntimeError("No data to process: did source selector remove all sources?")
        filters = collections.Counter(filters)
 
        return oldWcsList, filters
 
    def _make_one_input_data(self, visitRecord, catalog, detectorDict):
        """Return a data structure for this detector+visit."""
        return JointcalInputData(visit=visitRecord['visit'],
                                 catalog=catalog,
                                 visitInfo=visitRecord.getVisitInfo(),
                                 detector=detectorDict[visitRecord.getId()],
                                 photoCalib=visitRecord.getPhotoCalib(),
                                 wcs=visitRecord.getWcs(),
                                 bbox=visitRecord.getBBox(),
                                 # ExposureRecord doesn't have a FilterLabel yet,
                                 # so we have to make one.
                                 filter=lsst.afw.image.FilterLabel(band=visitRecord['band'],
                                                                   physical=visitRecord['physical_filter']))
 
    # We don't currently need to persist the metadata.
    # If we do in the future, we will have to add appropriate dataset templates
    # to each obs package (the metadata template should look like `jointcal_wcs`).
    def _getMetadataName(self):
        return None
 
    @classmethod
    def _makeArgumentParser(cls):
        """Create an argument parser"""
        parser = pipeBase.ArgumentParser(name=cls._DefaultName_DefaultName)
        parser.add_id_argument("--id", "calexp", help="data ID, e.g. --id visit=6789 ccd=0..9",
                               ContainerClass=PerTractCcdDataIdContainer)
        return parser
 
    def _build_ccdImage(self, data, associations, jointcalControl):
        """
        Extract the necessary things from this catalog+metadata to add a new
        ccdImage.
 
        Parameters
        ----------
        data : `JointcalInputData`
            The loaded input data.
        associations : `lsst.jointcal.Associations`
            Object to add the info to, to construct a new CcdImage
        jointcalControl : `jointcal.JointcalControl`
            Control object for associations management
 
        Returns
        ------
        namedtuple or `None`
            ``wcs``
                The TAN WCS of this image, read from the calexp
                (`lsst.afw.geom.SkyWcs`).
            ``key``
                A key to identify this dataRef by its visit and ccd ids
                (`namedtuple`).
            `None`
            if there are no sources in the loaded catalog.
        """
        if len(data.catalog) == 0:
            self.log.warn("No sources selected in visit %s ccd %s", data.visit, data.detector.getId())
            return None
 
        associations.createCcdImage(data.catalog,
                                    data.wcs,
                                    data.visitInfo,
                                    data.bbox,
                                    data.filter.physicalLabel,
                                    data.photoCalib,
                                    data.detector,
                                    data.visit,
                                    data.detector.getId(),
                                    jointcalControl)
 
        Result = collections.namedtuple('Result_from_build_CcdImage', ('wcs', 'key'))
        Key = collections.namedtuple('Key', ('visit', 'ccd'))
        return Result(data.wcs, Key(data.visit, data.detector.getId()))
 
    def _readDataId(self, butler, dataId):
        """Read all of the data for one dataId from the butler. (gen2 version)"""
        # Not all instruments have `visit` in their dataIds.
        if "visit" in dataId.keys():
            visit = dataId["visit"]
        else:
            visit = butler.getButler().queryMetadata("calexp", ("visit"), butler.dataId)[0]
        detector = butler.get('calexp_detector', dataId=dataId)
 
        catalog = butler.get('src',
                             flags=lsst.afw.table.SOURCE_IO_NO_FOOTPRINTS,
                             dataId=dataId)
        goodSrc = self.sourceSelector.run(catalog)
        self.log.debug("%d sources selected in visit %d detector %d",
                       len(goodSrc.sourceCat),
                       visit,
                       detector.getId())
        return JointcalInputData(visit=visit,
                                 catalog=goodSrc.sourceCat,
                                 visitInfo=butler.get('calexp_visitInfo', dataId=dataId),
                                 detector=detector,
                                 photoCalib=butler.get('calexp_photoCalib', dataId=dataId),
                                 wcs=butler.get('calexp_wcs', dataId=dataId),
                                 bbox=butler.get('calexp_bbox', dataId=dataId),
                                 filter=butler.get('calexp_filterLabel', dataId=dataId))
 
    def loadData(self, dataRefs, associations, jointcalControl):
        """Read the data that jointcal needs to run. (Gen2 version)"""
        visit_ccd_to_dataRef = {}
        oldWcsList = []
        filters = []
        load_cat_prof_file = 'jointcal_loadData.prof' if self.config.detailedProfile else ''
        with pipeBase.cmdLineTask.profile(load_cat_prof_file):
            # Need the bounding-box of the focal plane (the same for all visits) for photometry visit models
            camera = dataRefs[0].get('camera', immediate=True)
            self.focalPlaneBBoxfocalPlaneBBox = camera.getFpBBox()
            for dataRef in dataRefs:
                data = self._readDataId_readDataId(dataRef.getButler(), dataRef.dataId)
                result = self._build_ccdImage_build_ccdImage(data, associations, jointcalControl)
                if result is None:
                    continue
                oldWcsList.append(result.wcs)
                visit_ccd_to_dataRef[result.key] = dataRef
                filters.append(data.filter)
        if len(filters) == 0:
            raise RuntimeError("No data to process: did source selector remove all sources?")
        filters = collections.Counter(filters)
 
        return oldWcsList, filters, visit_ccd_to_dataRef
 
    def _getDebugPath(self, filename):
        """Constructs a path to filename using the configured debug path.
        """
        return os.path.join(self.config.debugOutputPath, filename)
 
    def _prep_sky(self, associations, filters):
        """Prepare on-sky and other data that must be computed after data has
        been read.
        """
        associations.computeCommonTangentPoint()
 
        boundingCircle = associations.computeBoundingCircle()
        center = lsst.geom.SpherePoint(boundingCircle.getCenter())
        radius = lsst.geom.Angle(boundingCircle.getOpeningAngle().asRadians(), lsst.geom.radians)
 
        self.log.info(f"Data has center={center} with radius={radius.asDegrees()} degrees.")
 
        # Determine a default filter band associated with the catalog. See DM-9093
        defaultFilter = filters.most_common(1)[0][0]
        self.log.debug("Using '%s' filter for reference flux", defaultFilter.physicalLabel)
 
        return boundingCircle, center, radius, defaultFilter
 
    @pipeBase.timeMethod
    def runDataRef(self, dataRefs):
        """
        Jointly calibrate the astrometry and photometry across a set of images.
 
        NOTE: this is for gen2 middleware only.
 
        Parameters
        ----------
        dataRefs : `list` of `lsst.daf.persistence.ButlerDataRef`
            List of data references to the exposures to be fit.
 
        Returns
        -------
        result : `lsst.pipe.base.Struct`
            Struct of metadata from the fit, containing:
 
            ``dataRefs``
                The provided data references that were fit (with updated WCSs)
            ``oldWcsList``
                The original WCS from each dataRef
            ``metrics``
                Dictionary of internally-computed metrics for testing/validation.
        """
        if len(dataRefs) == 0:
            raise ValueError('Need a non-empty list of data references!')
 
        exitStatus = 0  # exit status for shell
 
        sourceFluxField = "slot_%sFlux" % (self.config.sourceFluxType,)
        jointcalControl = lsst.jointcal.JointcalControl(sourceFluxField)
        associations = lsst.jointcal.Associations()
 
        oldWcsList, filters, visit_ccd_to_dataRef = self.loadDataloadData(dataRefs,
                                                                  associations,
                                                                  jointcalControl)
 
        boundingCircle, center, radius, defaultFilter = self._prep_sky_prep_sky(associations, filters)
        epoch = self._compute_proper_motion_epoch_compute_proper_motion_epoch(associations.getCcdImageList())
 
        tract = dataRefs[0].dataId['tract']
 
        if self.config.doAstrometry:
            astrometry = self._do_load_refcat_and_fit_do_load_refcat_and_fit(associations, defaultFilter, center, radius,
                                                      name="astrometry",
                                                      refObjLoader=self.astrometryRefObjLoaderastrometryRefObjLoader,
                                                      referenceSelector=self.astrometryReferenceSelector,
                                                      fit_function=self._fit_astrometry_fit_astrometry,
                                                      tract=tract,
                                                      epoch=epoch)
            self._write_astrometry_results_write_astrometry_results(associations, astrometry.model, visit_ccd_to_dataRef)
        else:
            astrometry = Astrometry(None, None, None)
 
        if self.config.doPhotometry:
            photometry = self._do_load_refcat_and_fit_do_load_refcat_and_fit(associations, defaultFilter, center, radius,
                                                      name="photometry",
                                                      refObjLoader=self.photometryRefObjLoaderphotometryRefObjLoader,
                                                      referenceSelector=self.photometryReferenceSelector,
                                                      fit_function=self._fit_photometry_fit_photometry,
                                                      tract=tract,
                                                      epoch=epoch,
                                                      reject_bad_fluxes=True)
            self._write_photometry_results_write_photometry_results(associations, photometry.model, visit_ccd_to_dataRef)
        else:
            photometry = Photometry(None, None)
 
        return pipeBase.Struct(dataRefs=dataRefs,
                               oldWcsList=oldWcsList,
                               job=self.jobjob,
                               astrometryRefObjLoader=self.astrometryRefObjLoaderastrometryRefObjLoader,
                               photometryRefObjLoader=self.photometryRefObjLoaderphotometryRefObjLoader,
                               defaultFilter=defaultFilter,
                               epoch=epoch,
                               exitStatus=exitStatus)
 
    def _get_refcat_coordinate_error_override(self, refCat, name):
        """Check whether we should override the refcat coordinate errors, and
        return the overridden error if necessary.
 
        Parameters
        ----------
        refCat : `lsst.afw.table.SimpleCatalog`
            The reference catalog to check for a ``coord_raErr`` field.
        name : `str`
            Whether we are doing "astrometry" or "photometry".
 
        Returns
        -------
        refCoordErr : `float`
            The refcat coordinate error to use, or NaN if we are not overriding
            those fields.
 
        Raises
        ------
        lsst.pex.config.FieldValidationError
            Raised if the refcat does not contain coordinate errors and
            ``config.astrometryReferenceErr`` is not set.
        """
        # This value doesn't matter for photometry, so just set something to
        # keep old refcats from causing problems.
        if name.lower() == "photometry":
            if 'coord_raErr' not in refCat.schema:
                return 100
            else:
                return float('nan')
 
        if self.config.astrometryReferenceErr is None and 'coord_raErr' not in refCat.schema:
            msg = ("Reference catalog does not contain coordinate errors, "
                   "and config.astrometryReferenceErr not supplied.")
            raise pexConfig.FieldValidationError(JointcalConfig.astrometryReferenceErr,
                                                 self.config,
                                                 msg)
 
        if self.config.astrometryReferenceErr is not None and 'coord_raErr' in refCat.schema:
            self.log.warn("Overriding reference catalog coordinate errors with %f/coordinate [mas]",
                          self.config.astrometryReferenceErr)
 
        if self.config.astrometryReferenceErr is None:
            return float('nan')
        else:
            return self.config.astrometryReferenceErr
 
    def _compute_proper_motion_epoch(self, ccdImageList):
        """Return the proper motion correction epoch of the provided images.
 
        Parameters
        ----------
        ccdImageList : `list` [`lsst.jointcal.CcdImage`]
            The images to compute the appropriate epoch for.
 
        Returns
        -------
        epoch : `astropy.time.Time`
            The date to use for proper motion corrections.
        """
        mjds = [ccdImage.getMjd() for ccdImage in ccdImageList]
        return astropy.time.Time(np.mean(mjds), format='mjd', scale="tai")
 
    def _do_load_refcat_and_fit(self, associations, defaultFilter, center, radius,
                                tract="", match_cut=3.0,
                                reject_bad_fluxes=False, *,
                                name="", refObjLoader=None, referenceSelector=None,
                                fit_function=None, epoch=None):
        """Load reference catalog, perform the fit, and return the result.
 
        Parameters
        ----------
        associations : `lsst.jointcal.Associations`
            The star/reference star associations to fit.
        defaultFilter : `lsst.afw.image.FilterLabel`
            filter to load from reference catalog.
        center : `lsst.geom.SpherePoint`
            ICRS center of field to load from reference catalog.
        radius : `lsst.geom.Angle`
            On-sky radius to load from reference catalog.
        name : `str`
            Name of thing being fit: "astrometry" or "photometry".
        refObjLoader : `lsst.meas.algorithms.LoadReferenceObjectsTask`
            Reference object loader to use to load a reference catalog.
        referenceSelector : `lsst.meas.algorithms.ReferenceSourceSelectorTask`
            Selector to use to pick objects from the loaded reference catalog.
        fit_function : callable
            Function to call to perform fit (takes Associations object).
        tract : `str`, optional
            Name of tract currently being fit.
        match_cut : `float`, optional
            Radius in arcseconds to find cross-catalog matches to during
            associations.associateCatalogs.
        reject_bad_fluxes : `bool`, optional
            Reject refCat sources with NaN/inf flux or NaN/0 fluxErr.
        epoch : `astropy.time.Time`, optional
            Epoch to which to correct refcat proper motion and parallax,
            or `None` to not apply such corrections.
 
        Returns
        -------
        result : `Photometry` or `Astrometry`
            Result of `fit_function()`
        """
        self.log.info("====== Now processing %s...", name)
        # TODO: this should not print "trying to invert a singular transformation:"
        # if it does that, something's not right about the WCS...
        associations.associateCatalogs(match_cut)
        add_measurement(self.jobjob, 'jointcal.associated_%s_fittedStars' % name,
                        associations.fittedStarListSize())
 
        applyColorterms = False if name.lower() == "astrometry" else self.config.applyColorTerms
        refCat, fluxField = self._load_reference_catalog_load_reference_catalog(refObjLoader, referenceSelector,
                                                         center, radius, defaultFilter,
                                                         applyColorterms=applyColorterms,
                                                         epoch=epoch)
        refCoordErr = self._get_refcat_coordinate_error_override_get_refcat_coordinate_error_override(refCat, name)
 
        associations.collectRefStars(refCat,
                                     self.config.matchCut*lsst.geom.arcseconds,
                                     fluxField,
                                     refCoordinateErr=refCoordErr,
                                     rejectBadFluxes=reject_bad_fluxes)
        add_measurement(self.jobjob, 'jointcal.collected_%s_refStars' % name,
                        associations.refStarListSize())
 
        associations.prepareFittedStars(self.config.minMeasurements)
 
        self._check_star_lists_check_star_lists(associations, name)
        add_measurement(self.jobjob, 'jointcal.selected_%s_refStars' % name,
                        associations.nFittedStarsWithAssociatedRefStar())
        add_measurement(self.jobjob, 'jointcal.selected_%s_fittedStars' % name,
                        associations.fittedStarListSize())
        add_measurement(self.jobjob, 'jointcal.selected_%s_ccdImages' % name,
                        associations.nCcdImagesValidForFit())
 
        load_cat_prof_file = 'jointcal_fit_%s.prof'%name if self.config.detailedProfile else ''
        dataName = "{}_{}".format(tract, defaultFilter.bandLabel)
        with pipeBase.cmdLineTask.profile(load_cat_prof_file):
            result = fit_function(associations, dataName)
        # TODO DM-12446: turn this into a "butler save" somehow.
        # Save reference and measurement chi2 contributions for this data
        if self.config.writeChi2FilesInitialFinal:
            baseName = self._getDebugPath_getDebugPath(f"{name}_final_chi2-{dataName}")
            result.fit.saveChi2Contributions(baseName+"{type}")
            self.log.info("Wrote chi2 contributions files: %s", baseName)
 
        return result
 
    def _load_reference_catalog(self, refObjLoader, referenceSelector, center, radius, filterLabel,
                                applyColorterms=False, epoch=None):
        """Load the necessary reference catalog sources, convert fluxes to
        correct units, and apply color term corrections if requested.
 
        Parameters
        ----------
        refObjLoader : `lsst.meas.algorithms.LoadReferenceObjectsTask`
            The reference catalog loader to use to get the data.
        referenceSelector : `lsst.meas.algorithms.ReferenceSourceSelectorTask`
            Source selector to apply to loaded reference catalog.
        center : `lsst.geom.SpherePoint`
            The center around which to load sources.
        radius : `lsst.geom.Angle`
            The radius around ``center`` to load sources in.
        filterLabel : `lsst.afw.image.FilterLabel`
            The camera filter to load fluxes for.
        applyColorterms : `bool`
            Apply colorterm corrections to the refcat for ``filterName``?
        epoch : `astropy.time.Time`, optional
            Epoch to which to correct refcat proper motion and parallax,
            or `None` to not apply such corrections.
 
        Returns
        -------
        refCat : `lsst.afw.table.SimpleCatalog`
            The loaded reference catalog.
        fluxField : `str`
            The name of the reference catalog flux field appropriate for ``filterName``.
        """
        skyCircle = refObjLoader.loadSkyCircle(center,
                                               radius,
                                               filterLabel.bandLabel,
                                               epoch=epoch)
 
        selected = referenceSelector.run(skyCircle.refCat)
        # Need memory contiguity to get reference filters as a vector.
        if not selected.sourceCat.isContiguous():
            refCat = selected.sourceCat.copy(deep=True)
        else:
            refCat = selected.sourceCat
 
        if applyColorterms:
            refCatName = refObjLoader.ref_dataset_name
            self.log.info("Applying color terms for physical filter=%r reference catalog=%s",
                          filterLabel.physicalLabel, refCatName)
            colorterm = self.config.colorterms.getColorterm(filterLabel.physicalLabel,
                                                            refCatName,
                                                            doRaise=True)
 
            refMag, refMagErr = colorterm.getCorrectedMagnitudes(refCat)
            refCat[skyCircle.fluxField] = u.Magnitude(refMag, u.ABmag).to_value(u.nJy)
            # TODO: I didn't want to use this, but I'll deal with it in DM-16903
            refCat[skyCircle.fluxField+'Err'] = fluxErrFromABMagErr(refMagErr, refMag) * 1e9
 
        return refCat, skyCircle.fluxField
 
    def _check_star_lists(self, associations, name):
        # TODO: these should be len(blah), but we need this properly wrapped first.
        if associations.nCcdImagesValidForFit() == 0:
            raise RuntimeError('No images in the ccdImageList!')
        if associations.fittedStarListSize() == 0:
            raise RuntimeError('No stars in the {} fittedStarList!'.format(name))
        if associations.refStarListSize() == 0:
            raise RuntimeError('No stars in the {} reference star list!'.format(name))
 
    def _logChi2AndValidate(self, associations, fit, model, chi2Label, writeChi2Name=None):
        """Compute chi2, log it, validate the model, and return chi2.
 
        Parameters
        ----------
        associations : `lsst.jointcal.Associations`
            The star/reference star associations to fit.
        fit : `lsst.jointcal.FitterBase`
            The fitter to use for minimization.
        model : `lsst.jointcal.Model`
            The model being fit.
        chi2Label : `str`
            Label to describe the chi2 (e.g. "Initialized", "Final").
        writeChi2Name : `str`, optional
            Filename prefix to write the chi2 contributions to.
            Do not supply an extension: an appropriate one will be added.
 
        Returns
        -------
        chi2: `lsst.jointcal.Chi2Accumulator`
            The chi2 object for the current fitter and model.
 
        Raises
        ------
        FloatingPointError
            Raised if chi2 is infinite or NaN.
        ValueError
            Raised if the model is not valid.
        """
        if writeChi2Name is not None:
            fullpath = self._getDebugPath_getDebugPath(writeChi2Name)
            fit.saveChi2Contributions(fullpath+"{type}")
            self.log.info("Wrote chi2 contributions files: %s", fullpath)
 
        chi2 = fit.computeChi2()
        self.log.info("%s %s", chi2Label, chi2)
        self._check_stars_check_stars(associations)
        if not np.isfinite(chi2.chi2):
            raise FloatingPointError(f'{chi2Label} chi2 is invalid: {chi2}')
        if not model.validate(associations.getCcdImageList(), chi2.ndof):
            raise ValueError("Model is not valid: check log messages for warnings.")
        return chi2
 
    def _fit_photometry(self, associations, dataName=None):
        """
        Fit the photometric data.
 
        Parameters
        ----------
        associations : `lsst.jointcal.Associations`
            The star/reference star associations to fit.
        dataName : `str`
            Name of the data being processed (e.g. "1234_HSC-Y"), for
            identifying debugging files.
 
        Returns
        -------
        fit_result : `namedtuple`
            fit : `lsst.jointcal.PhotometryFit`
                The photometric fitter used to perform the fit.
            model : `lsst.jointcal.PhotometryModel`
                The photometric model that was fit.
        """
        self.log.info("=== Starting photometric fitting...")
 
        # TODO: should use pex.config.RegistryField here (see DM-9195)
        if self.config.photometryModel == "constrainedFlux":
            model = lsst.jointcal.ConstrainedFluxModel(associations.getCcdImageList(),
                                                       self.focalPlaneBBoxfocalPlaneBBox,
                                                       visitOrder=self.config.photometryVisitOrder,
                                                       errorPedestal=self.config.photometryErrorPedestal)
            # potentially nonlinear problem, so we may need a line search to converge.
            doLineSearch = self.config.allowLineSearch
        elif self.config.photometryModel == "constrainedMagnitude":
            model = lsst.jointcal.ConstrainedMagnitudeModel(associations.getCcdImageList(),
                                                            self.focalPlaneBBoxfocalPlaneBBox,
                                                            visitOrder=self.config.photometryVisitOrder,
                                                            errorPedestal=self.config.photometryErrorPedestal)
            # potentially nonlinear problem, so we may need a line search to converge.
            doLineSearch = self.config.allowLineSearch
        elif self.config.photometryModel == "simpleFlux":
            model = lsst.jointcal.SimpleFluxModel(associations.getCcdImageList(),
                                                  errorPedestal=self.config.photometryErrorPedestal)
            doLineSearch = False  # purely linear in model parameters, so no line search needed
        elif self.config.photometryModel == "simpleMagnitude":
            model = lsst.jointcal.SimpleMagnitudeModel(associations.getCcdImageList(),
                                                       errorPedestal=self.config.photometryErrorPedestal)
            doLineSearch = False  # purely linear in model parameters, so no line search needed
 
        fit = lsst.jointcal.PhotometryFit(associations, model)
        # TODO DM-12446: turn this into a "butler save" somehow.
        # Save reference and measurement chi2 contributions for this data
        if self.config.writeChi2FilesInitialFinal:
            baseName = f"photometry_initial_chi2-{dataName}"
        else:
            baseName = None
        if self.config.writeInitialModel:
            fullpath = self._getDebugPath_getDebugPath("initialPhotometryModel.txt")
            writeModel(model, fullpath, self.log)
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialized", writeChi2Name=baseName)
 
        def getChi2Name(whatToFit):
            if self.config.writeChi2FilesOuterLoop:
                return f"photometry_init-%s_chi2-{dataName}" % whatToFit
            else:
                return None
 
        # The constrained model needs the visit transform fit first; the chip
        # transform is initialized from the singleFrame PhotoCalib, so it's close.
        dumpMatrixFile = self._getDebugPath_getDebugPath("photometry_preinit") if self.config.writeInitMatrix else ""
        if self.config.photometryModel.startswith("constrained"):
            # no line search: should be purely (or nearly) linear,
            # and we want a large step size to initialize with.
            fit.minimize("ModelVisit", dumpMatrixFile=dumpMatrixFile)
            self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize ModelVisit",
                                     writeChi2Name=getChi2Name("ModelVisit"))
            dumpMatrixFile = ""  # so we don't redo the output on the next step
 
        fit.minimize("Model", doLineSearch=doLineSearch, dumpMatrixFile=dumpMatrixFile)
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize Model",
                                 writeChi2Name=getChi2Name("Model"))
 
        fit.minimize("Fluxes")  # no line search: always purely linear.
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize Fluxes",
                                 writeChi2Name=getChi2Name("Fluxes"))
 
        fit.minimize("Model Fluxes", doLineSearch=doLineSearch)
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize ModelFluxes",
                                 writeChi2Name=getChi2Name("ModelFluxes"))
 
        model.freezeErrorTransform()
        self.log.debug("Photometry error scales are frozen.")
 
        chi2 = self._iterate_fit_iterate_fit(associations,
                                 fit,
                                 self.config.maxPhotometrySteps,
                                 "photometry",
                                 "Model Fluxes",
                                 doRankUpdate=self.config.photometryDoRankUpdate,
                                 doLineSearch=doLineSearch,
                                 dataName=dataName)
 
        add_measurement(self.jobjob, 'jointcal.photometry_final_chi2', chi2.chi2)
        add_measurement(self.jobjob, 'jointcal.photometry_final_ndof', chi2.ndof)
        return Photometry(fit, model)
 
    def _fit_astrometry(self, associations, dataName=None):
        """
        Fit the astrometric data.
 
        Parameters
        ----------
        associations : `lsst.jointcal.Associations`
            The star/reference star associations to fit.
        dataName : `str`
            Name of the data being processed (e.g. "1234_HSC-Y"), for
            identifying debugging files.
 
        Returns
        -------
        fit_result : `namedtuple`
            fit : `lsst.jointcal.AstrometryFit`
                The astrometric fitter used to perform the fit.
            model : `lsst.jointcal.AstrometryModel`
                The astrometric model that was fit.
            sky_to_tan_projection : `lsst.jointcal.ProjectionHandler`
                The model for the sky to tangent plane projection that was used in the fit.
        """
 
        self.log.info("=== Starting astrometric fitting...")
 
        associations.deprojectFittedStars()
 
        # NOTE: need to return sky_to_tan_projection so that it doesn't get garbage collected.
        # TODO: could we package sky_to_tan_projection and model together so we don't have to manage
        # them so carefully?
        sky_to_tan_projection = lsst.jointcal.OneTPPerVisitHandler(associations.getCcdImageList())
 
        if self.config.astrometryModel == "constrained":
            model = lsst.jointcal.ConstrainedAstrometryModel(associations.getCcdImageList(),
                                                             sky_to_tan_projection,
                                                             chipOrder=self.config.astrometryChipOrder,
                                                             visitOrder=self.config.astrometryVisitOrder)
        elif self.config.astrometryModel == "simple":
            model = lsst.jointcal.SimpleAstrometryModel(associations.getCcdImageList(),
                                                        sky_to_tan_projection,
                                                        self.config.useInputWcs,
                                                        nNotFit=0,
                                                        order=self.config.astrometrySimpleOrder)
 
        fit = lsst.jointcal.AstrometryFit(associations, model, self.config.positionErrorPedestal)
        # TODO DM-12446: turn this into a "butler save" somehow.
        # Save reference and measurement chi2 contributions for this data
        if self.config.writeChi2FilesInitialFinal:
            baseName = f"astrometry_initial_chi2-{dataName}"
        else:
            baseName = None
        if self.config.writeInitialModel:
            fullpath = self._getDebugPath_getDebugPath("initialAstrometryModel.txt")
            writeModel(model, fullpath, self.log)
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initial", writeChi2Name=baseName)
 
        def getChi2Name(whatToFit):
            if self.config.writeChi2FilesOuterLoop:
                return f"astrometry_init-%s_chi2-{dataName}" % whatToFit
            else:
                return None
 
        dumpMatrixFile = self._getDebugPath_getDebugPath("astrometry_preinit") if self.config.writeInitMatrix else ""
        # The constrained model needs the visit transform fit first; the chip
        # transform is initialized from the detector's cameraGeom, so it's close.
        if self.config.astrometryModel == "constrained":
            fit.minimize("DistortionsVisit", dumpMatrixFile=dumpMatrixFile)
            self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize DistortionsVisit",
                                     writeChi2Name=getChi2Name("DistortionsVisit"))
            dumpMatrixFile = ""  # so we don't redo the output on the next step
 
        fit.minimize("Distortions", dumpMatrixFile=dumpMatrixFile)
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize Distortions",
                                 writeChi2Name=getChi2Name("Distortions"))
 
        fit.minimize("Positions")
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize Positions",
                                 writeChi2Name=getChi2Name("Positions"))
 
        fit.minimize("Distortions Positions")
        self._logChi2AndValidate_logChi2AndValidate(associations, fit, model, "Initialize DistortionsPositions",
                                 writeChi2Name=getChi2Name("DistortionsPositions"))
 
        chi2 = self._iterate_fit_iterate_fit(associations,
                                 fit,
                                 self.config.maxAstrometrySteps,
                                 "astrometry",
                                 "Distortions Positions",
                                 doRankUpdate=self.config.astrometryDoRankUpdate,
                                 dataName=dataName)
 
        add_measurement(self.jobjob, 'jointcal.astrometry_final_chi2', chi2.chi2)
        add_measurement(self.jobjob, 'jointcal.astrometry_final_ndof', chi2.ndof)
 
        return Astrometry(fit, model, sky_to_tan_projection)
 
    def _check_stars(self, associations):
        """Count measured and reference stars per ccd and warn/log them."""
        for ccdImage in associations.getCcdImageList():
            nMeasuredStars, nRefStars = ccdImage.countStars()
            self.log.debug("ccdImage %s has %s measured and %s reference stars",
                           ccdImage.getName(), nMeasuredStars, nRefStars)
            if nMeasuredStars < self.config.minMeasuredStarsPerCcd:
                self.log.warn("ccdImage %s has only %s measuredStars (desired %s)",
                              ccdImage.getName(), nMeasuredStars, self.config.minMeasuredStarsPerCcd)
            if nRefStars < self.config.minRefStarsPerCcd:
                self.log.warn("ccdImage %s has only %s RefStars (desired %s)",
                              ccdImage.getName(), nRefStars, self.config.minRefStarsPerCcd)
 
    def _iterate_fit(self, associations, fitter, max_steps, name, whatToFit,
                     dataName="",
                     doRankUpdate=True,
                     doLineSearch=False):
        """Run fitter.minimize up to max_steps times, returning the final chi2.
 
        Parameters
        ----------
        associations : `lsst.jointcal.Associations`
            The star/reference star associations to fit.
        fitter : `lsst.jointcal.FitterBase`
            The fitter to use for minimization.
        max_steps : `int`
            Maximum number of steps to run outlier rejection before declaring
            convergence failure.
        name : {'photometry' or 'astrometry'}
            What type of data are we fitting (for logs and debugging files).
        whatToFit : `str`
            Passed to ``fitter.minimize()`` to define the parameters to fit.
        dataName : `str`, optional
            Descriptive name for this dataset (e.g. tract and filter),
            for debugging.
        doRankUpdate : `bool`, optional
            Do an Eigen rank update during minimization, or recompute the full
            matrix and gradient?
        doLineSearch : `bool`, optional
            Do a line search for the optimum step during minimization?
 
        Returns
        -------
        chi2: `lsst.jointcal.Chi2Statistic`
            The final chi2 after the fit converges, or is forced to end.
 
        Raises
        ------
        FloatingPointError
            Raised if the fitter fails with a non-finite value.
        RuntimeError
            Raised if the fitter fails for some other reason;
            log messages will provide further details.
        """
        dumpMatrixFile = self._getDebugPath_getDebugPath(f"{name}_postinit") if self.config.writeInitMatrix else ""
        oldChi2 = lsst.jointcal.Chi2Statistic()
        oldChi2.chi2 = float("inf")
        for i in range(max_steps):
            if self.config.writeChi2FilesOuterLoop:
                writeChi2Name = f"{name}_iterate_{i}_chi2-{dataName}"
            else:
                writeChi2Name = None
            result = fitter.minimize(whatToFit,
                                     self.config.outlierRejectSigma,
                                     doRankUpdate=doRankUpdate,
                                     doLineSearch=doLineSearch,
                                     dumpMatrixFile=dumpMatrixFile)
            dumpMatrixFile = ""  # clear it so we don't write the matrix again.
            chi2 = self._logChi2AndValidate_logChi2AndValidate(associations, fitter, fitter.getModel(),
                                            f"Fit iteration {i}", writeChi2Name=writeChi2Name)
 
            if result == MinimizeResult.Converged:
                if doRankUpdate:
                    self.log.debug("fit has converged - no more outliers - redo minimization "
                                   "one more time in case we have lost accuracy in rank update.")
                    # Redo minimization one more time in case we have lost accuracy in rank update
                    result = fitter.minimize(whatToFit, self.config.outlierRejectSigma)
                    chi2 = self._logChi2AndValidate_logChi2AndValidate(associations, fitter, fitter.getModel(), "Fit completed")
 
                # log a message for a large final chi2, TODO: DM-15247 for something better
                if chi2.chi2/chi2.ndof >= 4.0:
                    self.log.error("Potentially bad fit: High chi-squared/ndof.")
 
                break
            elif result == MinimizeResult.Chi2Increased:
                self.log.warn("Still some outliers remaining but chi2 increased - retry")
                # Check whether the increase was large enough to cause trouble.
                chi2Ratio = chi2.chi2 / oldChi2.chi2
                if chi2Ratio > 1.5:
                    self.log.warn('Significant chi2 increase by a factor of %.4g / %.4g = %.4g',
                                  chi2.chi2, oldChi2.chi2, chi2Ratio)
                # Based on a variety of HSC jointcal logs (see DM-25779), it
                # appears that chi2 increases more than a factor of ~2 always
                # result in the fit diverging rapidly and ending at chi2 > 1e10.
                # Using 10 as the "failure" threshold gives some room between
                # leaving a warning and bailing early.
                if chi2Ratio > 10:
                    msg = ("Large chi2 increase between steps: fit likely cannot converge."
                           " Try setting one or more of the `writeChi2*` config fields and looking"
                           " at how individual star chi2-values evolve during the fit.")
                    raise RuntimeError(msg)
                oldChi2 = chi2
            elif result == MinimizeResult.NonFinite:
                filename = self._getDebugPath_getDebugPath("{}_failure-nonfinite_chi2-{}.csv".format(name, dataName))
                # TODO DM-12446: turn this into a "butler save" somehow.
                fitter.saveChi2Contributions(filename+"{type}")
                msg = "Nonfinite value in chi2 minimization, cannot complete fit. Dumped star tables to: {}"
                raise FloatingPointError(msg.format(filename))
            elif result == MinimizeResult.Failed:
                raise RuntimeError("Chi2 minimization failure, cannot complete fit.")
            else:
                raise RuntimeError("Unxepected return code from minimize().")
        else:
            self.log.error("%s failed to converge after %d steps"%(name, max_steps))
 
        return chi2
 
    def _make_output(self, ccdImageList, model, func):
        """Return the internal jointcal models converted to the afw
        structures that will be saved to disk.
 
        Parameters
        ----------
        ccdImageList : `lsst.jointcal.CcdImageList`
            The list of CcdImages to get the output for.
        model : `lsst.jointcal.AstrometryModel` or `lsst.jointcal.PhotometryModel`
            The internal jointcal model to convert for each `lsst.jointcal.CcdImage`.
        func : `str`
            The name of the function to call on ``model`` to get the converted
            structure. Must accept an `lsst.jointcal.CcdImage`.
 
        Returns
        -------
        output : `dict` [`tuple`, `lsst.jointcal.AstrometryModel`] or
                 `dict` [`tuple`, `lsst.jointcal.PhotometryModel`]
            The data to be saved, keyed on (visit, detector).
        """
        output = {}
        for ccdImage in ccdImageList:
            ccd = ccdImage.ccdId
            visit = ccdImage.visit
            self.log.debug("%s for visit: %d, ccd: %d", func, visit, ccd)
            output[(visit, ccd)] = getattr(model, func)(ccdImage)
        return output
 
    def _write_astrometry_results(self, associations, model, visit_ccd_to_dataRef):
        """
        Write the fitted astrometric results to a new 'jointcal_wcs' dataRef.
 
        Parameters
        ----------
        associations : `lsst.jointcal.Associations`
            The star/reference star associations to fit.
        model : `lsst.jointcal.AstrometryModel`
            The astrometric model that was fit.
        visit_ccd_to_dataRef : `dict` of Key: `lsst.daf.persistence.ButlerDataRef`
            Dict of ccdImage identifiers to dataRefs that were fit.
        """
        ccdImageList = associations.getCcdImageList()
        output = self._make_output_make_output(ccdImageList, model, "makeSkyWcs")
        for key, skyWcs in output.items():
            dataRef = visit_ccd_to_dataRef[key]
            try:
                dataRef.put(skyWcs, 'jointcal_wcs')
            except pexExceptions.Exception as e:
                self.log.fatal('Failed to write updated Wcs: %s', str(e))
                raise e
 
    def _write_photometry_results(self, associations, model, visit_ccd_to_dataRef):
        """
        Write the fitted photometric results to a new 'jointcal_photoCalib' dataRef.
 
        Parameters
        ----------
        associations : `lsst.jointcal.Associations`
            The star/reference star associations to fit.
        model : `lsst.jointcal.PhotometryModel`
            The photoometric model that was fit.
        visit_ccd_to_dataRef : `dict` of Key: `lsst.daf.persistence.ButlerDataRef`
            Dict of ccdImage identifiers to dataRefs that were fit.
        """
 
        ccdImageList = associations.getCcdImageList()
        output = self._make_output_make_output(ccdImageList, model, "toPhotoCalib")
        for key, photoCalib in output.items():
            dataRef = visit_ccd_to_dataRef[key]
            try:
                dataRef.put(photoCalib, 'jointcal_photoCalib')
            except pexExceptions.Exception as e:
                self.log.fatal('Failed to write updated PhotoCalib: %s', str(e))
                raise e