DM-50987: Add sattle to pipelines

python/lsst/pipe/tasks/calibrateImage.py

@@ -23,10 +23,12 @@
            "AllCentroidsFlaggedError"]
 
 import numpy as np
+import requests
+import os
 
 import lsst.afw.table as afwTable
 import lsst.afw.image as afwImage
-from lsst.ip.diffim.utils import evaluateMaskFraction
+from lsst.ip.diffim.utils import evaluateMaskFraction, populate_sattle_visit_cache
 import lsst.meas.algorithms
 import lsst.meas.algorithms.installGaussianPsf
 import lsst.meas.algorithms.measureApCorr
@@ -384,6 +386,21 @@ class CalibrateImageConfig(pipeBase.PipelineTaskConfig, pipelineConnections=Cali
         doc="If True, include astrometric errors in the output catalog.",
     )
 
+    run_sattle = pexConfig.Field(
+        dtype=bool,
+        default=False,
+        doc="If True, the sattle service will populate a cache for later use "
+            "in ip_diffim.detectAndMeasure alert verification."
+    )
+
+    sattle_historical = pexConfig.Field(
+        dtype=bool,
+        default=False,
+        doc="If re-running a pipeline that requires sattle, this should be set "
+            "to True. This will populate sattle's cache with the historic data "
+            "closest in time to the exposure.",
+    )
+
     def setDefaults(self):
         super().setDefaults()
 
@@ -565,6 +582,11 @@ def validate(self):
                         "doApplyFlatBackgroundRatio=True if do_illumination_correction=True."
                     )
 
+        if self.run_sattle:
+            if not os.getenv("SATTLE_URI_BASE"):
+                raise pexConfig.FieldValidationError(CalibrateImageConfig.run_sattle, self,
+                                                     "Sattle requested but URI environment variable not set.")
+
 
 class CalibrateImageTask(pipeBase.PipelineTask):
     """Compute the PSF, aperture corrections, astrometric and photometric
@@ -901,6 +923,17 @@ def run(
             result.applied_photo_calib = photo_calib
         else:
             result.applied_photo_calib = None
+
+        if self.config.run_sattle:
+            # send boresight and timing information to sattle so the cache
+            # is populated by the time we reach ip_diffim detectAndMeasure.
+            try:
+                populate_sattle_visit_cache(result.exposure.getInfo().getVisitInfo(),
+                                            historical=self.config.sattle_historical)
+                self.log.info('Successfully triggered load of sattle visit cache')
+            except requests.exceptions.HTTPError:
+                self.log.exception("Sattle visit cache update failed; continuing with image processing")
+
         return result
 
     def _apply_illumination_correction(self, exposure, background_flat, illumination_correction):

tests/test_calibrateImage.py

@@ -28,6 +28,8 @@
 import copy
 import numpy as np
 import esutil
+import os
+import requests
 
 import lsst.afw.image as afwImage
 import lsst.afw.math as afwMath
@@ -44,6 +46,7 @@
 import lsst.pipe.base.testUtils
 from lsst.pipe.tasks.calibrateImage import CalibrateImageTask, \
     NoPsfStarsToStarsMatchError, AllCentroidsFlaggedError
+import lsst.pex.config as pexConfig
 import lsst.utils.tests
 
 
@@ -120,6 +123,7 @@ def setUp(self):
         # We don't have many test points, so can't match on complicated shapes.
         self.config.astrometry.sourceSelector["science"].flags.good = []
         self.config.astrometry.matcher.numPointsForShape = 3
+        self.config.run_sattle = False
         # ApFlux has more noise than PsfFlux (the latter unrealistically small
         # in this test data), so we need to do magnitude rejection at higher
         # sigma, otherwise we can lose otherwise good sources.
@@ -584,6 +588,42 @@ def test_calibrate_image_illumcorr(self):
         self.assertIn(key, result.exposure.metadata)
         self.assertEqual(result.exposure.metadata[key], True)
 
+    @mock.patch.dict(os.environ, {"SATTLE_URI_BASE": ""})
+    def test_fail_on_sattle_miconfiguration(self):
+        """Test for failure if sattle is requested without appropriate configurations.
+        """
+        self.config.run_sattle = True
+        with self.assertRaises(pexConfig.FieldValidationError):
+            CalibrateImageTask(config=self.config)
+
+    @mock.patch.dict(os.environ, {"SATTLE_URI_BASE": "fake_host:1234"})
+    def test_continue_on_sattle_failure(self):
+        """Processing should continue when sattle returns status codes other than 200.
+        """
+        response = MockResponse({}, 500, "internal sattle error")
+
+        self.config.run_sattle = True
+        calibrate = CalibrateImageTask(config=self.config)
+        calibrate.astrometry.setRefObjLoader(self.ref_loader)
+        calibrate.photometry.match.setRefObjLoader(self.ref_loader)
+        with mock.patch('requests.put', return_value=response) as mock_put:
+            calibrate.run(exposures=self.exposure)
+            mock_put.assert_called_once()
+
+    @mock.patch.dict(os.environ, {"SATTLE_URI_BASE": "fake_host:1234"})
+    def test_sattle(self):
+        """Test for successful completion when sattle call returns successfully.
+        """
+        response = MockResponse({}, 200, "success")
+
+        self.config.run_sattle = True
+        calibrate = CalibrateImageTask(config=self.config)
+        calibrate.astrometry.setRefObjLoader(self.ref_loader)
+        calibrate.photometry.match.setRefObjLoader(self.ref_loader)
+        with mock.patch('requests.put', return_value=response) as mock_put:
+            calibrate.run(exposures=self.exposure)
+            mock_put.assert_called_once()
+
 
 class CalibrateImageTaskRunQuantumTests(lsst.utils.tests.TestCase):
     """Tests of ``CalibrateImageTask.runQuantum``, which need a test butler,
@@ -955,6 +995,21 @@ def mock_run(
             self.butler.get("initial_stars_footprints_detector", self.visit_id)
 
 
+class MockResponse:
+    """Provide a mock for requests.put calls"""
+    def __init__(self, json_data, status_code, text):
+        self.json_data = json_data
+        self.status_code = status_code
+        self.text = text
+
+    def json(self):
+        return self.json_data
+
+    def raise_for_status(self):
+        if self.status_code != 200:
+            raise requests.exceptions.HTTPError
+
+
 def setup_module(module):
     lsst.utils.tests.init()