Merge pull request #80 from lsst-ts/tickets/DM-29787

[DM-29787] Get instrument from pipeline in DonutCatalogOnlineTask

Jenkinsfile

@@ -57,6 +57,7 @@ pipeline {
                 withEnv(["HOME=${env.WORKSPACE}"]) {
                     sh """
                         source ${env.LSST_STACK}/loadLSST.bash
+                        setup ctrl_mpexec -t ${env.STACK_VERSION}
                         cd phosim_utils/
                         setup -k -r . -t ${env.STACK_VERSION}
                         scons
@@ -74,6 +75,7 @@ pipeline {
                 withEnv(["HOME=${env.WORKSPACE}"]) {
                     sh """
                         source ${env.LSST_STACK}/loadLSST.bash
+                        setup ctrl_mpexec -t ${env.STACK_VERSION}
                         cd phosim_utils/
                         setup -k -r . -t ${env.STACK_VERSION}
                         cd ..
@@ -128,7 +130,7 @@ pipeline {
                 }
               }
             }
-            
+
             // Change the ownership of workspace to Jenkins for the clean up
             // This is to work around the condition that the user ID of jenkins
             // is 1003 on TSSW Jenkins instance. In this post stage, it is the

doc/versionHistory.rst

@@ -6,6 +6,15 @@
 Version History
 ##################
 
+.. _lsst.ts.wep-1.6.1:
+
+-------------
+1.6.1
+-------------
+
+* Update GenerateDonutCatalogOnlineTask.py to get instrument directly from pipeline configuration.
+* Setup `ctrl_mpexec` package in Jenkinsfile so tests can run `pipetask` command.
+
 .. _lsst.ts.wep-1.6.0:
 
 -------------

python/lsst/ts/wep/task/GenerateDonutCatalogOnlineTask.py

@@ -99,10 +99,6 @@ def __init__(self, **kwargs):
         self.boresightDec = self.config.boresightDec
         self.boresightRotAng = self.config.boresightRotAng
 
-        # Need camera name to get the detectors to use to find X,Y pixel
-        # values for the sources
-        self.cameraName = self.config.cameraName
-
         # TODO: Temporary until DM-24162 is closed at which point we
         # can remove this
         os.environ["NUMEXPR_MAX_THREADS"] = "1"
@@ -131,9 +127,27 @@ def filterResults(self, resultsDataFrame):
 
         return resultsDataFrame
 
-    def run(
+    def runQuantum(
         self,
-        refCatalogs: typing.List[afwTable.SimpleCatalog],
+        butlerQC: pipeBase.ButlerQuantumContext,
+        inputRefs: pipeBase.InputQuantizedConnection,
+        outputRefs: pipeBase.OutputQuantizedConnection,
+    ):
+
+        # Get the instrument we are running the pipeline with
+        cameraName = outputRefs.donutCatalog.dataId["instrument"]
+
+        # Get the input reference catalogs for the task
+        inputs = butlerQC.get(inputRefs)
+
+        # Run task on specified instrument
+        outputs = self.run(cameraName, **inputs)
+
+        # Use butler to store output in repository
+        butlerQC.put(outputs, outputRefs)
+
+    def run(
+        self, cameraName: str, refCatalogs: typing.List[afwTable.SimpleCatalog]
     ) -> pipeBase.Struct:
 
         refObjLoader = ReferenceObjectLoader(
@@ -154,11 +168,11 @@ def run(
         centroidY = []
         det_names = []
 
-        # Get camera. Only 'lsstCam' for now.
-        if self.cameraName == "lsstCam":
+        # Get camera. Only 'LSSTCam' for now.
+        if cameraName == "LSSTCam":
             camera = obs_lsst.LsstCam.getCamera()
         else:
-            raise ValueError(f"{self.cameraName} is not a valid camera name.")
+            raise ValueError(f"{cameraName} is not a valid camera name.")
 
         # Create WCS holder
         detWcs = WcsSol(camera=camera)

tests/task/test_generateDonutCatalogOnlineTask.py

@@ -29,6 +29,7 @@
     GenerateDonutCatalogOnlineTask,
     GenerateDonutCatalogOnlineTaskConfig,
 )
+from lsst.ts.wep.Utility import runProgram
 
 
 class TestGenerateDonutCatalogOnlineTask(unittest.TestCase):
@@ -44,20 +45,30 @@ def setUp(self):
         self.butler = dafButler.Butler(self.repoDir)
         self.registry = self.butler.registry
 
+    def _writePipetaskCmd(self, runName, taskName):
+
+        pipetaskCmd = "pipetask run "
+        pipetaskCmd += f"-b {self.repoDir} "  # Specify repo
+        pipetaskCmd += "-i refcats "  # Specify collections with data to use
+        # Specify task
+        pipetaskCmd += f"-t lsst.ts.wep.task.{taskName} "
+        pipetaskCmd += "--instrument lsst.obs.lsst.LsstCam "
+        pipetaskCmd += f"--register-dataset-types --output-run {runName}"
+
+        return pipetaskCmd
+
     def validateConfigs(self):
 
         self.config.boresightRa = 0.03
         self.config.boresightDec = -0.02
         self.config.boresightRotAng = 90.0
         self.config.filterName = "r"
-        self.cameraName = "lsstFamCam"
         self.task = GenerateDonutCatalogOnlineTask(config=self.config)
 
         self.assertEqual(self.task.boresightRa, 0.03)
         self.assertEqual(self.task.boresightDec, -0.02)
         self.assertEqual(self.task.boresightRotAng, 90.0)
         self.assertEqual(self.task.filterName, "r")
-        self.assertEqual(self.task.cameraName, "lsstFamCam")
 
     def testFilterResults(self):
 
@@ -68,6 +79,58 @@ def testFilterResults(self):
         filteredDataFrame = self.task.filterResults(testDataFrame)
         np.testing.assert_array_equal(filteredDataFrame, testDataFrame)
 
+    def testPipeline(self):
+        """
+        Test that the task runs in a pipeline. Also functions as a test of
+        runQuantum function.
+        """
+
+        # Run pipeline command
+        runName = "run1"
+        taskName = "GenerateDonutCatalogOnlineTask.GenerateDonutCatalogOnlineTask"
+        pipetaskCmd = self._writePipetaskCmd(runName, taskName)
+        runProgram(pipetaskCmd)
+
+        # Test instrument matches
+        pipelineButler = dafButler.Butler(self.repoDir)
+        outputDf = pipelineButler.get(
+            "donutCatalog", dataId={"instrument": "LSSTCam"}, collections=[f"{runName}"]
+        )
+        self.assertEqual(len(outputDf), 8)
+        self.assertEqual(len(outputDf.query('detector == "R22_S11"')), 4)
+        self.assertEqual(len(outputDf.query('detector == "R22_S10"')), 4)
+        self.assertCountEqual(
+            [
+                3806.7636478057957,
+                2806.982895217227,
+                607.3861483168994,
+                707.3972344551466,
+                614.607342274194,
+                714.6336433247832,
+                3815.2649173460436,
+                2815.0561553920156,
+            ],
+            outputDf["centroid_x"],
+        )
+        self.assertCountEqual(
+            [
+                3196.070534224157,
+                2195.666002294077,
+                394.8907003737886,
+                394.9087004171349,
+                396.2407036464963,
+                396.22270360324296,
+                3196.1965343932648,
+                2196.188002312585,
+            ],
+            outputDf["centroid_y"],
+        )
+
+        # Clean up
+        cleanUpCmd = "butler prune-collection "
+        cleanUpCmd += f"{self.repoDir} {runName} --purge --unstore"
+        runProgram(cleanUpCmd)
+
     def testDonutCatalogGeneration(self):
         """
         Test that task creates a dataframe with detector information.
@@ -80,7 +143,7 @@ def testDonutCatalogGeneration(self):
         ).expanded()
         for ref in datasetGenerator:
             deferredList.append(self.butler.getDeferred(ref, collections=["refcats"]))
-        taskOutput = self.task.run(deferredList)
+        taskOutput = self.task.run("LSSTCam", deferredList)
         outputDf = taskOutput.donutCatalog
 
         # Compare ra, dec info to original input catalog