DM-31833: Move LoadDiaCatalogs to standalone pipelineTask

python/lsst/ap/verify/testPipeline.py

@@ -34,7 +34,7 @@
 import lsst.afw.image as afwImage
 import lsst.afw.math as afwMath
 import lsst.afw.table as afwTable
-from lsst.ap.association import (TransformDiaSourceCatalogConfig,
+from lsst.ap.association import (LoadDiaCatalogsConfig, TransformDiaSourceCatalogConfig,
                                  DiaPipelineConfig, FilterDiaSourceCatalogConfig)
 from lsst.pipe.base import PipelineTask, Struct
 from lsst.ip.isr import IsrTaskConfig
@@ -45,6 +45,41 @@
 from lsst.meas.transiNet import RBTransiNetConfig
 
 
+class MockLoadDiaCatalogsTask(PipelineTask):
+    """A do-nothing substitute for LoadDiaCatalogsTask.
+    """
+    ConfigClass = LoadDiaCatalogsConfig
+    _DefaultName = "notLoadDiaCatalogs"
+
+    def run(self, regionTime):
+        """Produce preloaded DiaSource and DiaObject outputs with no processing.
+
+        Parameters
+        ----------
+        regionTime : `lsst.pipe.base.utils.RegionTimeInfo`
+            A serializable container for a sky region and timespan.
+
+        Returns
+        -------
+        result : `lsst.pipe.base.Struct`
+            Results struct with components.
+
+            - ``diaObjects`` : Complete set of DiaObjects covering the input
+              exposure padded by ``pixelMargin``. DataFrame is indexed by
+              the ``diaObjectId`` column. (`pandas.DataFrame`)
+            - ``diaSources`` : Complete set of DiaSources covering the input
+              exposure padded by ``pixelMargin``. DataFrame is indexed by
+              ``diaObjectId``, ``band``, ``diaSourceId`` columns.
+              (`pandas.DataFrame`)
+            - ``diaForcedSources`` : Complete set of forced photometered fluxes
+            on the past 12 months of difference images at DiaObject locations.
+        """
+        return Struct(diaObjects=pandas.DataFrame(),
+                      diaSources=pandas.DataFrame(),
+                      diaForcedSources=pandas.DataFrame(),
+                      )
+
+
 class MockIsrTask(PipelineTask):
     """A do-nothing substitute for IsrTask.
     """
@@ -529,14 +564,12 @@ def __init__(self, initInputs, **kwargs):
     def runQuantum(self, butlerQC, inputRefs, outputRefs):
         inputs = butlerQC.get(inputRefs)
         inputs["band"] = butlerQC.quantum.dataId["band"]
-        inputs["visit"] = butlerQC.quantum.dataId["visit"]
-        inputs["detector"] = butlerQC.quantum.dataId["detector"]
 
         outputs = self.run(**inputs)
 
         butlerQC.put(outputs, outputRefs)
 
-    def run(self, diaSourceCat, diffIm, band, visit, detector, funcs=None):
+    def run(self, diaSourceCat, diffIm, band, reliability=None):
         """Produce transformation outputs with no processing.
 
         Parameters
@@ -545,12 +578,9 @@ def run(self, diaSourceCat, diffIm, band, visit, detector, funcs=None):
             The catalog to transform.
         diffIm : `lsst.afw.image.Exposure`
             An image, to provide supplementary information.
-        band : `str`
-            The band in which the sources were observed.
-        visit, detector: `int`
-            Visit and detector the sources were detected on.
-        funcs, optional
-            Unused.
+        reliability : `lsst.afw.table.SourceCatalog`, optional
+            Reliability (e.g. real/bogus) scores, row-matched to
+            ``diaSourceCat``.
 
         Returns
         -------
@@ -583,9 +613,6 @@ class MockDiaPipelineTask(PipelineTask):
     def runQuantum(self, butlerQC, inputRefs, outputRefs):
         inputs = butlerQC.get(inputRefs)
         inputs["idGenerator"] = self.config.idGenerator.apply(butlerQC.quantum.dataId)
-        # Need to set ccdExposureIdBits (now deprecated) to None and pass it,
-        # since there are non-optional positional arguments after it.
-        inputs["ccdExposureIdBits"] = None
         inputs["band"] = butlerQC.quantum.dataId["band"]
         if not self.config.doSolarSystemAssociation:
             inputs["solarSystemObjectTable"] = None
@@ -600,9 +627,11 @@ def run(self,
             diffIm,
             exposure,
             template,
-            ccdExposureIdBits,
+            preloadedDiaObjects,
+            preloadedDiaSources,
+            preloadedDiaForcedSources,
             band,
-            idGenerator=None):
+            idGenerator):
         """Produce DiaSource and DiaObject outputs with no processing.
 
         Parameters
@@ -619,17 +648,16 @@ def run(self,
             ``diffIm``.
         template : `lsst.afw.image.ExposureF`
             Template exposure used to create diffIm.
-        ccdExposureIdBits : `int`
-            Number of bits used for a unique ``ccdVisitId``.  Deprecated in
-            favor of ``idGenerator``, and ignored if that is present.  Pass
-            `None` explicitly to avoid a deprecation warning (a default is
-            impossible given that later positional arguments are not
-            defaulted).
+        preloadedDiaObjects : `pandas.DataFrame`
+            Previously detected DiaObjects, loaded from the APDB.
+        preloadedDiaSources : `pandas.DataFrame`
+            Previously detected DiaSources, loaded from the APDB.
+        preloadedDiaForcedSources : `pandas.DataFrame`
+            Catalog of previously detected forced DiaSources, from the APDB
         band : `str`
             The band in which the new DiaSources were detected.
-        idGenerator : `lsst.meas.base.IdGenerator`, optional
+        idGenerator : `lsst.meas.base.IdGenerator`
             Object that generates source IDs and random number generator seeds.
-            Will be required after ``ccdExposureIdBits`` is removed.
 
         Returns
         -------

tests/MockApPipe.yaml

@@ -7,6 +7,10 @@ parameters:
   refcat: gaia
   apdb_config: dummy_path.yaml
 tasks:
+  loadDiaCatalogs:
+    class: lsst.ap.verify.testPipeline.MockLoadDiaCatalogsTask
+    config:
+      apdb_config_url: parameters.apdb_config
   isr:
     class: lsst.ap.verify.testPipeline.MockIsrTask
     config:
@@ -51,6 +55,11 @@ tasks:
     config:
       doRemoveSkySources: True
       connections.coaddName: parameters.coaddName
+  getRegionTimeFromVisit:
+    # class: lsst.ap.verify.testPipeline.MockGetRegionTimeFromVisitTask
+    class: lsst.pipe.tasks.getRegionTimeFromVisit.GetRegionTimeFromVisitTask
+    config:
+      connections.coaddName: parameters.coaddName
   diaPipe:
     class: lsst.ap.verify.testPipeline.MockDiaPipelineTask
     config:

tests/test_testPipeline.py

@@ -33,13 +33,16 @@
 import lsst.afw.math as afwMath
 import lsst.afw.table as afwTable
 import lsst.skymap
+from lsst.daf.butler import Timespan
 import lsst.daf.butler.tests as butlerTests
+from lsst.meas.base import IdGenerator
 import lsst.pipe.base.testUtils as pipelineTests
+from lsst.pipe.base.utils import RegionTimeInfo
 from lsst.ap.verify.testPipeline import MockIsrTask, MockCharacterizeImageTask, \
     MockCalibrateTask, MockGetTemplateTask, \
     MockAlardLuptonSubtractTask, MockDetectAndMeasureTask, MockTransformDiaSourceCatalogTask, \
     MockRBTransiNetTask, MockDiaPipelineTask, MockFilterDiaSourceCatalogTask, \
-    MockSpatiallySampledMetricsTask
+    MockSpatiallySampledMetricsTask, MockLoadDiaCatalogsTask
 
 
 class MockTaskTestSuite(unittest.TestCase):
@@ -60,6 +63,7 @@ def setUpClass(cls):
 
         INSTRUMENT = "DummyCam"
         VISIT = 42
+        GROUP = '42'
         CCD = 101
         HTM = 42
         SKYMAP = "TreasureMap"
@@ -78,6 +82,7 @@ def setUpClass(cls):
         butlerTests.addDataIdValue(cls.repo, "subfilter", SUB_FILTER)
         butlerTests.addDataIdValue(cls.repo, "exposure", VISIT)
         butlerTests.addDataIdValue(cls.repo, "visit", VISIT)
+        butlerTests.addDataIdValue(cls.repo, "group", GROUP)
         butlerTests.addDataIdValue(cls.repo, "detector", CCD)
         butlerTests.addDataIdValue(cls.repo, "skymap", SKYMAP)
         butlerTests.addDataIdValue(cls.repo, "tract", TRACT)
@@ -88,6 +93,8 @@ def setUpClass(cls):
             {"instrument": INSTRUMENT, "exposure": VISIT, "detector": CCD})
         cls.visitId = cls.repo.registry.expandDataId(
             {"instrument": INSTRUMENT, "visit": VISIT, "detector": CCD})
+        cls.groupId = cls.repo.registry.expandDataId(
+            {"instrument": INSTRUMENT, "group": GROUP, "detector": CCD})
         cls.visitOnlyId = cls.repo.registry.expandDataId(
             {"instrument": INSTRUMENT, "visit": VISIT})
         cls.skymapId = cls.repo.registry.expandDataId({"skymap": SKYMAP})
@@ -99,6 +106,11 @@ def setUpClass(cls):
             {"skymap": SKYMAP, "tract": TRACT, "patch": PATCH, "band": BAND, "subfilter": SUB_FILTER})
         cls.htmId = cls.repo.registry.expandDataId({"htm7": HTM})
 
+        butlerTests.addDatasetType(cls.repo, "regionTimeInfo", cls.groupId.dimensions, "RegionTimeInfo")
+        butlerTests.addDatasetType(cls.repo, "preloaded_diaObjects", cls.groupId.dimensions, "DataFrame")
+        butlerTests.addDatasetType(cls.repo, "preloaded_diaSources", cls.groupId.dimensions, "DataFrame")
+        butlerTests.addDatasetType(cls.repo, "preloaded_diaForcedSources", cls.groupId.dimensions,
+                                   "DataFrame")
         butlerTests.addDatasetType(cls.repo, "postISRCCD", cls.exposureId.dimensions, "Exposure")
         butlerTests.addDatasetType(cls.repo, "icExp", cls.visitId.dimensions, "ExposureF")
         butlerTests.addDatasetType(cls.repo, "icSrc", cls.visitId.dimensions, "SourceCatalog")
@@ -141,6 +153,26 @@ def setUp(self):
         super().setUp()
         self.butler = butlerTests.makeTestCollection(self.repo, uniqueId=self.id())
 
+    def testMockLoadDiaCatalogsTask(self):
+        config = MockLoadDiaCatalogsTask.ConfigClass()
+        config.apdb_config_url = "testing_only"
+        task = MockLoadDiaCatalogsTask(config=config)
+        pipelineTests.assertValidInitOutput(task)
+        region = lsst.sphgeom.Circle()
+        timespan = Timespan.makeEmpty()
+        result = task.run(RegionTimeInfo(region=region, timespan=timespan))
+        pipelineTests.assertValidOutput(task, result)
+
+        self.butler.put(RegionTimeInfo(region=region, timespan=timespan), "regionTimeInfo", self.groupId)
+        quantum = pipelineTests.makeQuantum(
+            task, self.butler, self.groupId,
+            {"regionTime": self.groupId,
+             "diaObjects": self.groupId,
+             "diaSources": self.groupId,
+             "diaForcedSources": self.groupId,
+             })
+        pipelineTests.runTestQuantum(task, self.butler, quantum, mockRun=False)
+
     def testMockIsr(self):
         # Testing MockIsrTask is tricky because the real ISR has an unstable
         # interface with dozens of potential inputs, too many to pass through
@@ -327,7 +359,7 @@ def testMockRBTransiNetTask(self):
     def testMockTransformDiaSourceCatalogTask(self):
         task = MockTransformDiaSourceCatalogTask(initInputs=afwTable.SourceCatalog())
         pipelineTests.assertValidInitOutput(task)
-        result = task.run(afwTable.SourceCatalog(), afwImage.ExposureF(), 'k', 42, 2)
+        result = task.run(afwTable.SourceCatalog(), afwImage.ExposureF(), 'k')
         pipelineTests.assertValidOutput(task, result)
 
         self.butler.put(afwTable.SourceCatalog(), "deepDiff_candidateDiaSrc", self.visitId)
@@ -347,9 +379,14 @@ def testMockDiaPipelineTask(self):
         task = MockDiaPipelineTask(config=config)
         pipelineTests.assertValidInitOutput(task)
         result = task.run(pandas.DataFrame(), pandas.DataFrame(), afwImage.ExposureF(),
-                          afwImage.ExposureF(), afwImage.ExposureF(), 42, 'k')
+                          afwImage.ExposureF(), afwImage.ExposureF(),
+                          pandas.DataFrame(), pandas.DataFrame(), pandas.DataFrame(),
+                          'k', IdGenerator())
         pipelineTests.assertValidOutput(task, result)
 
+        self.butler.put(pandas.DataFrame(), "preloaded_diaObjects", self.groupId)
+        self.butler.put(pandas.DataFrame(), "preloaded_diaSources", self.groupId)
+        self.butler.put(pandas.DataFrame(), "preloaded_diaForcedSources", self.groupId)
         self.butler.put(pandas.DataFrame(), "deepDiff_diaSrcTable", self.visitId)
         self.butler.put(pandas.DataFrame(), "visitSsObjects", self.visitId)
         self.butler.put(afwImage.ExposureF(), "deepDiff_differenceExp", self.visitId)
@@ -362,6 +399,9 @@ def testMockDiaPipelineTask(self):
              "diffIm": self.visitId,
              "exposure": self.visitId,
              "template": self.visitId,
+             "preloadedDiaObjects": self.groupId,
+             "preloadedDiaSources": self.groupId,
+             "preloadedDiaForcedSources": self.groupId,
              "apdbMarker": self.visitId,
              "associatedDiaSources": self.visitId,
              "diaForcedSources": self.visitId,