DM-20692: Convert to new PipelineTask API

python/lsst/meas/base/forcedPhotCcd.py

@@ -31,6 +31,8 @@
 import lsst.afw.table
 import lsst.sphgeom
 
+from lsst.pipe.base import PipelineTaskConnections
+import lsst.pipe.base.connectionTypes as cT
 from .forcedPhotImage import ForcedPhotImageTask, ForcedPhotImageConfig
 
 try:
@@ -138,32 +140,53 @@ def imageOverlapsTract(tract, imageWcs, imageBox):
     return tractPoly.intersects(imagePoly)  # "intersects" also covers "contains" or "is contained by"
 
 
-class ForcedPhotCcdConfig(ForcedPhotImageConfig):
+class ForcedPhotCcdConnections(PipelineTaskConnections,
+                               dimensions=("instrument", "visit", "detector", "skymap", "tract"),
+                               defaultTemplates={"inputCoaddName": "deep",
+                                                 "inputName": "calexp"}):
+    inputSchema = cT.InitInput(
+        doc="Schema for the input measurement catalogs.",
+        name="{inputCoaddName}Coadd_ref_schema",
+        storageClass="SourceCatalog",
+    )
+    outputSchema = cT.InitOutput(
+        doc="Schema for the output forced measurement catalogs.",
+        name="forced_src_schema",
+        storageClass="SourceCatalog",
+    )
+    exposure = cT.Input(
+        doc="Input exposure to perform photometry on.",
+        name="{inputName}",
+        storageClass="ExposureF",
+        dimensions=["instrument", "visit", "detector"],
+    )
+    refCat = cT.Input(
+        doc="Catalog of shapes and positions at which to force photometry.",
+        name="{inputCoaddName}Coadd_ref",
+        storageClass="SourceCatalog",
+        dimensions=["skymap", "tract", "patch"],
+    )
+    refWcs = cT.Input(
+        doc="Reference world coordinate system.",
+        name="{inputCoaddName}Coadd.wcs",
+        storageClass="Wcs",
+        dimensions=["abstract_filter", "skymap", "tract", "patch"],
+    )
+    measCat = cT.Output(
+        doc="Output forced photometry catalog.",
+        name="forced_src",
+        storageClass="SourceCatalog",
+        dimensions=["instrument", "visit", "detector", "skymap", "tract"],
+    )
+
+
+class ForcedPhotCcdConfig(ForcedPhotImageConfig, pipelineConnections=ForcedPhotCcdConnections):
     doApplyUberCal = lsst.pex.config.Field(
         dtype=bool,
         doc="Apply meas_mosaic ubercal results to input calexps?",
         default=False
     )
 
-    def setDefaults(self):
-        super().setDefaults()
-
-        # Override the Gen3 datasets from ForcedPhotImageTaskConfig.
-        # When these nameTemplate values are used in ForcedPhotCoadd,
-        # there is a coadd name that may change, depending on the
-        # coadd type.  For CCD forced photometry, there will likely
-        # only ever be a single calexp type, but for consistency, use
-        # the nameTemplate with nothing to substitute.
-        self.outputSchema.nameTemplate = "forced_src_schema"
-        self.exposure.nameTemplate = "{inputName}"
-        self.exposure.dimensions = ["instrument", "visit", "detector"]
-        self.measCat.nameTemplate = "forced_src"
-        self.measCat.dimensions = ["instrument", "visit", "detector", "skymap", "tract"]
-
-        self.formatTemplateNames({"inputName": "calexp",
-                                  "inputCoaddName": "deep"})
-        self.quantum.dimensions = ("instrument", "visit", "detector", "skymap", "tract")
-
 
 class ForcedPhotCcdTask(ForcedPhotImageTask):
     """A command-line driver for performing forced measurement on CCD images.
@@ -205,16 +228,15 @@ class ForcedPhotCcdTask(ForcedPhotImageTask):
     _DefaultName = "forcedPhotCcd"
     dataPrefix = ""
 
-    def adaptArgsAndRun(self, inputData, inputDataIds, outputDataIds, butler):
-        inputData['refWcs'] = butler.get(f"{self.config.refWcs.name}.wcs", inputDataIds["refWcs"])
-        inputData['refCat'] = self.filterReferences(inputData['exposure'],
-                                                    inputData['refCat'], inputData['refWcs'])
-        inputData['measCat'] = self.generateMeasCat(inputDataIds['exposure'],
-                                                    inputData['exposure'],
-                                                    inputData['refCat'], inputData['refWcs'],
-                                                    "visit_detector", butler)
-
-        return self.run(**inputData)
+    def runQuantum(self, butlerQC, inputRefs, outputRefs):
+        inputs = butlerQC.get(inputRefs)
+        inputs['refCat'] = self.filterReferences(inputs['exposure'], inputs['refCat'], inputs['refWcs'])
+        inputs['measCat'] = self.generateMeasCat(inputRefs.exposure.dataId,
+                                                 inputs['exposure'],
+                                                 inputs['refCat'], inputs['refWcs'],
+                                                 "visit_detector", butlerQC.registry)
+        outputs = self.run(**inputs)
+        butlerQC.put(outputs, outputRefs)
 
     def filterReferences(self, exposure, refCat, refWcs):
         """Filter reference catalog so that all sources are within the

python/lsst/meas/base/forcedPhotCoadd.py

@@ -24,12 +24,14 @@
 import lsst.coadd.utils
 import lsst.afw.table
 
-from .forcedPhotImage import ForcedPhotImageConfig, ForcedPhotImageTask
+
+from .forcedPhotImage import ForcedPhotImageConfig, ForcedPhotImageTask, ForcedPhotImageConnections
+
 
 __all__ = ("ForcedPhotCoaddConfig", "ForcedPhotCoaddTask")
 
 
-class ForcedPhotCoaddConfig(ForcedPhotImageConfig):
+class ForcedPhotCoaddConfig(ForcedPhotImageConfig, pipelineConnections=ForcedPhotImageConnections):
     footprintDatasetName = lsst.pex.config.Field(
         doc="Dataset (without coadd prefix) that should be used to obtain (Heavy)Footprints for sources. "
             "Must have IDs that match those of the reference catalog."

python/lsst/meas/base/forcedPhotImage.py

@@ -31,57 +31,59 @@
 import lsst.pipe.base
 import lsst.pex.config
 
+from lsst.pipe.base import PipelineTaskConnections, PipelineTaskConfig
+import lsst.pipe.base.connectionTypes as cT
+
 from .references import MultiBandReferencesTask
 from .forcedMeasurement import ForcedMeasurementTask
 from .applyApCorr import ApplyApCorrTask
 from .catalogCalculation import CatalogCalculationTask
 
-__all__ = ("ForcedPhotImageConfig", "ForcedPhotImageTask")
+__all__ = ("ForcedPhotImageConfig", "ForcedPhotImageTask", "ForcedPhotImageConnections")
 
 
-class ForcedPhotImageConfig(lsst.pipe.base.PipelineTaskConfig):
-    """Config class for forced measurement driver task."""
-    # Gen 3 options
-    inputSchema = lsst.pipe.base.InitInputDatasetField(
+class ForcedPhotImageConnections(PipelineTaskConnections,
+                                 dimensions=("abstract_filter", "skymap", "tract", "patch"),
+                                 defaultTemplates={"inputCoaddName": "deep",
+                                                   "outputCoaddName": "deep"}):
+    inputSchema = cT.InitInput(
         doc="Schema for the input measurement catalogs.",
-        nameTemplate="{inputCoaddName}Coadd_ref_schema",
+        name="{inputCoaddName}Coadd_ref_schema",
         storageClass="SourceCatalog",
     )
-    outputSchema = lsst.pipe.base.InitOutputDatasetField(
+    outputSchema = cT.InitOutput(
         doc="Schema for the output forced measurement catalogs.",
-        nameTemplate="{outputCoaddName}Coadd_forced_src_schema",
+        name="{outputCoaddName}Coadd_forced_src_schema",
         storageClass="SourceCatalog",
     )
-    exposure = lsst.pipe.base.InputDatasetField(
+    exposure = cT.Input(
         doc="Input exposure to perform photometry on.",
-        nameTemplate="{inputCoaddName}Coadd",
-        scalar=True,
+        name="{inputCoaddName}Coadd",
         storageClass="ExposureF",
         dimensions=["abstract_filter", "skymap", "tract", "patch"],
     )
-    refCat = lsst.pipe.base.InputDatasetField(
+    refCat = cT.Input(
         doc="Catalog of shapes and positions at which to force photometry.",
-        nameTemplate="{inputCoaddName}Coadd_ref",
-        scalar=True,
+        name="{inputCoaddName}Coadd_ref",
         storageClass="SourceCatalog",
         dimensions=["skymap", "tract", "patch"],
     )
-    refWcs = lsst.pipe.base.InputDatasetField(
+    refWcs = cT.Input(
         doc="Reference world coordinate system.",
-        nameTemplate="{inputCoaddName}Coadd",
-        scalar=True,
-        manualLoad=True,
-        storageClass="ExposureF",
+        name="{inputCoaddName}Coadd.wcs",
+        storageClass="Wcs",
         dimensions=["abstract_filter", "skymap", "tract", "patch"],
     )
-    measCat = lsst.pipe.base.OutputDatasetField(
+    measCat = cT.Output(
         doc="Output forced photometry catalog.",
-        nameTemplate="{outputCoaddName}Coadd_forced_src",
-        scalar=True,
+        name="{outputCoaddName}Coadd_forced_src",
         storageClass="SourceCatalog",
         dimensions=["abstract_filter", "skymap", "tract", "patch"],
     )
 
+
+class ForcedPhotImageConfig(PipelineTaskConfig, pipelineConnections=ForcedPhotImageConnections):
+    """Config class for forced measurement driver task."""
     # ForcedPhotImage options
     references = lsst.pex.config.ConfigurableField(
         target=MultiBandReferencesTask,
@@ -117,10 +119,6 @@ def setDefaults(self):
         super().setDefaults()
 
         self.catalogCalculation.plugins.names = []
-        self.formatTemplateNames({"inputCoaddName": "deep",
-                                  "outputCoaddName": "deep",
-                                  "inputName": None})
-        self.quantum.dimensions = ("abstract_filter", "skymap", "tract", "patch")
 
 
 class ForcedPhotImageTask(lsst.pipe.base.PipelineTask, lsst.pipe.base.CmdLineTask):
@@ -173,20 +171,18 @@ def __init__(self, butler=None, refSchema=None, initInputs=None, **kwds):
         if self.config.doApCorr:
             self.makeSubtask("applyApCorr", schema=self.measurement.schema)
         self.makeSubtask('catalogCalculation', schema=self.measurement.schema)
-
-    def getInitOutputDatasets(self):
-        return {"outputSchema": lsst.afw.table.SourceCatalog(self.measurement.schema)}
-
-    def adaptArgsAndRun(self, inputData, inputDataIds, outputDataIds, butler):
-        inputData['refWcs'] = butler.get(f"{self.config.refWcs.name}.wcs", inputDataIds["refWcs"])
-        inputData['measCat'] = self.generateMeasCat(inputDataIds['exposure'],
-                                                    inputData['exposure'],
-                                                    inputData['refCat'], inputData['refWcs'],
-                                                    "tract_patch", butler)
-
-        return self.run(**inputData)
-
-    def generateMeasCat(self, exposureDataId, exposure, refCat, refWcs, idPackerName, butler):
+        self.outputSchema = lsst.afw.table.SourceCatalog(self.measurement.schema)
+
+    def runQuantum(self, butlerQC, inputRefs, outputRefs):
+        inputs = butlerQC.get(inputRefs)
+        inputs['measCat'] = self.generateMeasCat(inputRefs.exposure.dataId,
+                                                 inputs['exposure'],
+                                                 inputs['refCat'], inputs['refWcs'],
+                                                 "tract_patch", butlerQC.registry)
+        outputs = self.run(**inputs)
+        butlerQC.put(outputs, outputRefs)
+
+    def generateMeasCat(self, exposureDataId, exposure, refCat, refWcs, idPackerName, registry):
         """Generate a measurement catalog for Gen3.
 
         Parameters
@@ -201,15 +197,15 @@ def generateMeasCat(self, exposureDataId, exposure, refCat, refWcs, idPackerName
             Reference world coordinate system.
         idPackerName : `str`
             Type of ID packer to construct from the registry.
-        butler : `lsst.daf.persistence.butler.Butler`
-            Butler to use to construct id packer.
+        registry : `lsst.daf.persistence.butler.Registry`
+            Registry to use to construct id packer.
 
         Returns
         -------
         measCat : `lsst.afw.table.SourceCatalog`
             Catalog of forced sources to measure.
         """
-        packer = butler.registry.makeDataIdPacker(idPackerName, exposureDataId)
+        packer = registry.makeDataIdPacker(idPackerName, exposureDataId)
         expId = packer.pack(exposureDataId)
         expBits = packer.maxBits
         idFactory = lsst.afw.table.IdFactory.makeSource(expId, 64 - expBits)