Generalize MetricAnalysisTask

Enable MetricAnalysisTask to work with any metric table. Inputs and dimensions specified at runtime via the pipeline yaml file.

python/lsst/analysis/tools/tasks/metricAnalysis.py

@@ -26,33 +26,76 @@
 )
 
 
+from lsst.pex.config import ListField
 from lsst.pipe.base import connectionTypes as ct
 
 from ..interfaces import AnalysisBaseConfig, AnalysisBaseConnections, AnalysisPipelineTask
 
 
 class MetricAnalysisConnections(
     AnalysisBaseConnections,
-    dimensions=("skymap",),
-    defaultTemplates={"metricBundleName": "objectTableCore_metrics"},
+    dimensions=(),
+    defaultTemplates={"metricBundleName": ""},
 ):
+
     data = ct.Input(
-        doc="A summary table of all metrics by tract.",
+        doc="A table containing metrics.",
         name="{metricBundleName}Table",
         storageClass="ArrowAstropy",
-        dimensions=("skymap",),
         deferLoad=True,
+        dimensions=(),
     )
 
+    def __init__(self, *, config=None):
+
+        self.dimensions.update(frozenset(sorted(config.outputDataDimensions)))
+        super().__init__(config=config)
+        self.data = ct.Input(
+            doc=self.data.doc,
+            name=self.data.name,
+            storageClass=self.data.storageClass,
+            deferLoad=self.data.deferLoad,
+            dimensions=frozenset(sorted(config.inputDataDimensions)),
+        )
+
 
-class MetricAnalysisConfig(AnalysisBaseConfig, pipelineConnections=MetricAnalysisConnections):
-    pass
+class MetricAnalysisConfig(
+    AnalysisBaseConfig,
+    pipelineConnections=MetricAnalysisConnections,
+):
+    inputDataDimensions = ListField[str](
+        doc="Dimensions of the input data.",
+        default=(),
+        optional=False,
+    )
+    outputDataDimensions = ListField[str](
+        doc="Dimensions of the output data.",
+        default=(),
+        optional=False,
+    )
 
 
 class MetricAnalysisTask(AnalysisPipelineTask):
-    """Turn metric bundles which are per tract into a
-    summary metric table.
-    """
+    """Perform an analysis of a metric table."""
 
     ConfigClass = MetricAnalysisConfig
     _DefaultName = "metricAnalysis"
+
+    def runQuantum(self, butlerQC, inputRefs, outputRefs):
+
+        inputs = butlerQC.get(inputRefs)
+        dataId = butlerQC.quantum.dataId
+        plotInfo = self.parsePlotInfo(inputs, dataId)
+
+        data = self.loadData(inputs.pop("data"))
+
+        # TODO: "bands" kwarg is a workaround for DM-47941.
+        outputs = self.run(
+            data=data,
+            plotInfo=plotInfo,
+            bands=dataId["band"],
+            band=dataId["band"],
+            **inputs,
+        )
+
+        butlerQC.put(outputs, outputRefs)