DM-30386: Make DRP wrapper task for DiaCalculation task

pipelines/DRP.yaml

@@ -76,6 +76,13 @@ tasks:
       connections.diaSourceTables: goodSeeingDiff_diaSrcTable
       connections.assocDiaSourceTable: goodSeeingDiff_assocDiaSrcTable
       connections.diaObjectTable: goodSeeingDiff_diaObjTable
+  drpDiaCalculation:
+    class: lsst.pipe.tasks.drpDiaCalculationPipe.DrpDiaCalculationPipeTask
+    config:
+      connections.coaddName: goodSeeing
+      connections.assocDiaSourceTable: goodSeeingDiff_assocDiaSrcTable
+      connections.diaObjectTable: goodSeeingDiff_diaObjTable
+      connections.fullDiaObjectTable: goodSeeingDiff_fullDiaObjTable
   forcedPhotDiffim:
     class: lsst.meas.base.ForcedPhotCcdTask
     config:
@@ -172,6 +179,7 @@ subsets:
       - transformDiaSourceCat
       - consolidateDiaSourceTable
       - drpAssociation
+      - drpDiaCalculation
       - forcedPhotDiffim
     description: Subset for running image differencing branch of the DRP pipeline
 
@@ -182,3 +190,5 @@ contracts:
   - imageDifference.connections.subtractedExposure == forcedPhotDiffim.connections.exposure
   - transformDiaSourceCat.connections.diaSourceTable == consolidateDiaSourceTable.connections.inputCatalogs
   - transformDiaSourceCat.connections.diaSourceTable == drpAssociation.connections.diaSourceTables
+  - drpAssociation.connections.assocDiaSourceTable == drpDiaCalculation.connections.assocDiaSourceTable
+  - drpAssociation.connections.diaObjectTable == drpDiaCalculation.connections.diaObjectTable

python/lsst/pipe/tasks/drpDiaCalculationPipe.py

@@ -0,0 +1,139 @@
+# This file is part of pipe_tasks.
+
+# 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/>.
+#
+
+"""Pipeline for computing DiaObject summary/light curve values.
+"""
+
+__all__ = ["DrpDiaCalculationPipeTask",
+           "DrpDiaCalculationPipeConfig",
+           "DrpDiaCalculationPipeConnections"]
+
+import pandas as pd
+
+from lsst.meas.base import DiaObjectCalculationTask
+import lsst.pex.config as pexConfig
+import lsst.pipe.base as pipeBase
+
+
+class DrpDiaCalculationPipeConnections(pipeBase.PipelineTaskConnections,
+                                       dimensions=("tract", "patch", "skymap"),
+                                       defaultTemplates={"coaddName": "deep",
+                                                         "fakesType": ""}):
+    assocDiaSourceTable = pipeBase.connectionTypes.Input(
+        doc="Catalog of DiaSources covering the patch and associated with a "
+            "DiaObject.",
+        name="{fakesType}{coaddName}Diff_assocDiaSrcTable",
+        storageClass="DataFrame",
+        dimensions=("tract", "patch"),
+    )
+    diaObjectTable = pipeBase.connectionTypes.Input(
+        doc="Catalog of DiaObjects created from spatially associating "
+            "DiaSources.",
+        name="{fakesType}{coaddName}Diff_diaObjTable",
+        storageClass="DataFrame",
+        dimensions=("tract", "patch"),
+    )
+    fullDiaObjectTable = pipeBase.connectionTypes.Output(
+        doc="Catalog of DiaObjects created from spatially associating "
+            "DiaSources.",
+        name="{fakesType}{coaddName}Diff_fullDiaObjTable",
+        storageClass="DataFrame",
+        dimensions=("tract", "patch"),
+    )
+
+
+class DrpDiaCalculationPipeConfig(
+        pipeBase.PipelineTaskConfig,
+        pipelineConnections=DrpDiaCalculationPipeConnections):
+    filterNames = pexConfig.ListField(
+        dtype=str,
+        default=['u', 'g', 'r', 'i', 'z', 'y'],
+        doc="List of filters to attempt to calculate DiaObject summary "
+            "values."
+    )
+    diaCalculation = pexConfig.ConfigurableField(
+        target=DiaObjectCalculationTask,
+        doc="Task to compute summary statistics for DiaObjects.",
+    )
+
+    def setDefaults(self):
+        self.diaCalculation.plugins = ["ap_meanPosition",
+                                       "ap_HTMIndex",
+                                       "ap_diaObjectFlag",
+                                       "ap_meanFlux",
+                                       "ap_percentileFlux",
+                                       "ap_sigmaFlux",
+                                       "ap_chi2Flux",
+                                       "ap_madFlux",
+                                       "ap_skewFlux",
+                                       "ap_minMaxFlux",
+                                       "ap_maxSlopeFlux",
+                                       "ap_meanErrFlux",
+                                       "ap_linearFit",
+                                       "ap_stetsonJ",
+                                       "ap_meanTotFlux",
+                                       "ap_sigmaTotFlux"]
+
+
+class DrpDiaCalculationPipeTask(pipeBase.PipelineTask):
+    """Driver pipeline for loading DiaSource catalogs in a patch/tract
+    region and associating them.
+    """
+    ConfigClass = DrpDiaCalculationPipeConfig
+    _DefaultName = "drpDiaCalculation"
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.makeSubtask("diaCalculation")
+
+    def run(self, assocDiaSourceTable, diaObjectTable):
+        """Compute summary statistics over the input set of DiaSources and
+        store summary statistics into the associated DiaObjects.
+
+        Parameters
+        ----------
+        assocDiaSourceTable : `pandas.DataFrame`
+            Set of DiaSources spatially associated into the DiaObjects in
+            ``diaObjectTable``.
+        diaObjectTable : `pandas.DataFrame`
+            DiaObjects created from associating the sources in
+            ``assocDiaSourceTable``. All ids in the catalog must have a
+            corresponding DiaSource in the input catalog.
+
+        Returns
+        -------
+        results : `lsst.pipe.base.Struct`
+            Struct containing
+
+            ``fullDiaObjectTable``
+                DiaObjects with computed summary statistics based on their
+                associated DiaSource light curves. (`pandas.DataFrame`).
+        """
+        # Return empty dataFrame if no DiaObjects in this patch.
+        if len(diaObjectTable) <= 0 or len(assocDiaSourceTable) <= 0:
+            return pipeBase.Struct(fullDiaObjectTable=pd.DataFrame())
+        result = self.diaCalculation.run(
+            diaObjectTable,
+            assocDiaSourceTable,
+            diaObjectTable["diaObjectId"].to_numpy(),
+            self.config.filterNames)
+        return pipeBase.Struct(fullDiaObjectTable=result.updatedDiaObjects)