Respond to review.

Author: ebellm

python/lsst/ip/diffim/detectAndMeasure.py

@@ -21,9 +21,7 @@
 
 import numpy as np
 import requests
-import json
 import os
-from astropy.time import Time
 
 import lsst.afw.detection as afwDetection
 import lsst.afw.image as afwImage
@@ -32,6 +30,7 @@
 import lsst.daf.base as dafBase
 import lsst.geom
 from lsst.ip.diffim.utils import evaluateMaskFraction, computeDifferenceImageMetrics
+from lsst.ip.diffim.utils import populate_sattle_visit_cache
 from lsst.meas.algorithms import SkyObjectsTask, SourceDetectionTask, SetPrimaryFlagsTask, MaskStreaksTask
 from lsst.meas.base import ForcedMeasurementTask, ApplyApCorrTask, DetectorVisitIdGeneratorConfig
 import lsst.meas.deblender
@@ -313,18 +312,6 @@ class DetectAndMeasureConfig(pipeBase.PipelineTaskConfig,
         doc="If true, dia source bounding boxes will be sent for verification"
             "to the sattle service."
     )
-    sattle_host = pexConfig.Field(
-        dtype=str,
-        default=os.getenv("SATTLE_HOST"),
-        doc="Host address for sattle service.",
-        optional=True
-    )
-    sattle_port = pexConfig.Field(
-        dtype=str,
-        default=os.getenv("SATTLE_PORT"),
-        doc="Port to connect to sattle.",
-        optional=True
-    )
     sattle_historical = pexConfig.Field(
         dtype=bool,
         default=False,
@@ -403,6 +390,14 @@ def setDefaults(self):
             "STREAK", "INJECTED", "INJECTED_TEMPLATE"]
         self.skySources.avoidMask = ["DETECTED", "DETECTED_NEGATIVE", "BAD", "NO_DATA", "EDGE"]
 
+    def validate(self):
+        super().validate()
+
+        if self.run_sattle:
+            if not os.getenv("SATTLE_URI_BASE"):
+                raise pexConfig.FieldValidationError(DetectAndMeasureConfig.run_sattle, self,
+                                                     "Sattle requested but URI environment variable not set.")
+
 
 class DetectAndMeasureTask(lsst.pipe.base.PipelineTask):
     """Detect and measure sources on a difference image.
@@ -721,8 +716,6 @@ def processResults(self, science, matchedTemplate, difference, sources, idFactor
         diaSources = self._removeBadSources(initialDiaSources)
 
         if self.config.run_sattle:
-            if not self.config.sattle_host or not self.config.sattle_port:
-                raise RuntimeError("Sattle filtering is on but service endpoints not set.")
             diaSources = self.filterSatellites(diaSources, science)
 
         if self.config.doForcedMeasurement:
@@ -982,99 +975,80 @@ def calculateMetrics(self, science, difference, diaSources, kernelSources):
                 raise BadSubtractionError(ratio=metrics.differenceFootprintRatioStdev,
                                           threshold=self.config.badSubtractionVariationThreshold)
 
-    def filterSatellites(self, diaSources, science):
-
-        wcs = science.getWcs()
-        nbbox = []
+    def getSattleDiaSourceAllowlist(self, diaSources, science):
+        """Query the sattle service and determine which diaSources are allowed.
 
-        for source in diaSources:
-            fp = source.getFootprint()
-            source_bbox = fp.getBBox()
+        Parameters
+        ----------
+        diaSources : `lsst.afw.table.SourceCatalog`
+            The catalog of detected sources.
+        science : `lsst.afw.image.ExposureF`
+            Science exposure that was subtracted.
 
-            corners = [wcs.pixelToSky(source_bbox.beginX, source_bbox.beginY),
-                       wcs.pixelToSky(source_bbox.beginX, source_bbox.endY),
-                       wcs.pixelToSky(source_bbox.endX, source_bbox.endY),
-                       wcs.pixelToSky(source_bbox.endX, source_bbox.beginY)]
+        Returns
+        ----------
+        allow_list : `list` of `int`
+            diaSourceIds of diaSources that can be made public.
 
-            tmp = []
-            for c, corner in enumerate(corners):
-                tmp.append([corner.getRa().asDegrees(), corner.getDec().asDegrees()])
-            nbbox.append(tmp)
+        Raises
+        ------
+        requests.HTTPError
+            Raised if sattle call does not return success.
+        """
 
-        detector_id = science.getDetector().getId()
-        visit_id = science.getInfo().getVisitInfo().getId()
+        wcs = science.getWcs()
+        visit_info = science.getInfo().getVisitInfo()
+        visit_id = visit_info.getId()
+        sattle_uri_base = os.getenv('SATTLE_URI_BASE')
 
         dia_sources_json = []
-        for i, source in enumerate(diaSources):
-            dia_sources_json.append(
-                {"diasource_id": source['id'], "bbox": nbbox[i]})
-
-        sattle_output = requests.put(
-            f'{self.config.sattle_host}:{self.config.sattle_port}/diasource_allow_list',
-            json={"visit_id": visit_id, "detector_id": detector_id, "diasources": dia_sources_json})
-
-        if sattle_output.status_code != 200:
-            # Check if the cache is missing the visit and retry once
-            if sattle_output.status_code == 404:
-                try:
-                    self.log.info('Visit not found in sattle cache, re-sending')
-
-                    visit_id = science.getInfo().getVisitInfo().id
-
-                    visit_mjd = Time(
-                        science.getInfo().getVisitInfo().getDate().toPython()).mjd
-
-                    exposure_time_days = science.getInfo().getVisitInfo().getExposureTime() / 86400.0
-                    exposure_end_mjd = visit_mjd + exposure_time_days / 2.0
-                    exposure_start_mjd = visit_mjd - exposure_time_days / 2.0
-
-                    boresight_ra = science.getInfo().getVisitInfo().boresightRaDec[0].asDegrees()
-                    boresight_dec = science.getInfo().getVisitInfo().boresightRaDec[1].asDegrees()
+        for source in diaSources:
+            source_bbox = source.getFootprint().getBBox()
+            corners = [wcs.pixelToSky(lsst.geom.Point2D(c)) for c in source_bbox.getCorners()]
+            bbox_radec = [[pt.getRa().asDegrees(), pt.getDec().asDegrees()] for pt in corners]
+            dia_sources_json.append({"diasource_id": source["id"], "bbox": bbox_radec})
 
-                    r = requests.put(
-                        f'{self.config.sattle_host}:{self.config.sattle_port}/visit_cache',
-                        json={"visit_id": visit_id,
-                              "exposure_start_mjd": exposure_start_mjd,
-                              "exposure_end_mjd": exposure_end_mjd,
-                              "boresight_ra": boresight_ra,
-                              "boresight_dec": boresight_dec,
-                              "historical": self.config.sattle_historical})
+        payload = {"visit_id": visit_id, "detector_id": science.getDetector(), "diasources": dia_sources_json,
+                   "historical": self.config.sattle_historical}
 
-                    if r.status_code != 200:
-                        raise RuntimeError(r.text)
+        sattle_output = requests.put(f'{sattle_uri_base}/diasource_allow_list',
+                                     json=payload)
 
-                    sattle_output = requests.put(
-                        f'{self.config.sattle_host}:{self.config.sattle_port}/'
-                        'diasource_allow_list',
-                        json={"visit_id": visit_id, "detector_id": detector_id,
-                              "diasources": dia_sources_json})
+        # retry once if visit cache is not populated
+        if sattle_output.status_code == 404:
+            self.log.warning(f'Visit {visit_id} not found in sattle cache, re-sending')
+            populate_sattle_visit_cache(visit_info, historical=self.config.sattle_historical)
+            sattle_output = requests.put(f'{sattle_uri_base}/diasource_allow_list', json=payload)
 
-                    # fail on any non-success
-                    if sattle_output.status_code != 200:
-                        raise RuntimeError(sattle_output.text)
+        sattle_output.raise_for_status()
 
-                except (requests.RequestException, ConnectionError) as e:
-                    raise RuntimeError(sattle_output.text) from e
+        return sattle_output.json()['allow_list']
 
-            else:
-                raise RuntimeError(sattle_output.text)
+    def filterSatellites(self, diaSources, science):
+        """Remove diaSources overlapping predicted satellite positions.
 
-        sattle_output_array = json.loads(sattle_output.content)
+        Parameters
+        ----------
+        diaSources : `lsst.afw.table.SourceCatalog`
+            The catalog of detected sources.
+        science : `lsst.afw.image.ExposureF`
+            Science exposure that was subtracted.
 
-        if sattle_output_array['allow_list'] == []:
-            self.log.warning('Sattle output array is empty, all sources removed')
-            diaSources = diaSources[0:0].copy(deep=True)
-        else:
+        Returns
+        ----------
+        filterdDiaSources : `lsst.afw.table.SourceCatalog`
+            Filtered catalog of diaSources
+        """
 
-            allowed_ids = []
-            for source in diaSources:
-                if source['id'] in sattle_output_array['allow_list']:
-                    allowed_ids.append(True)
-                else:
-                    allowed_ids.append(False)
+        allow_list = self.getSattleDiaSourceAllowlist(diaSources, science)
 
+        if len(allow_list):
+            allow_set = set(allow_list)
+            allowed_ids = [source['id'] in allow_set for source in diaSources]
             diaSources = diaSources[np.array(allowed_ids)].copy(deep=True)
-
+        else:
+            self.log.warning('Sattle allowlist is empty, all diaSources removed')
+            diaSources = diaSources[0:0].copy(deep=True)
         return diaSources
 
     def _runStreakMasking(self, difference):

python/lsst/ip/diffim/utils.py

@@ -28,6 +28,8 @@
 
 import itertools
 import numpy as np
+import os
+import requests
 import lsst.geom as geom
 import lsst.afw.detection as afwDetection
 import lsst.afw.image as afwImage
@@ -415,3 +417,40 @@ def footprint_mean(sources, sky=0):
                                  differenceFootprintSkyRatioMean=sky_mean,
                                  differenceFootprintSkyRatioStdev=sky_std,
                                  )
+
+
+def populate_sattle_visit_cache(visit_info, historical=False):
+    """Populate a cache of predicted satellite positions in the sattle service.
+
+    Parameters
+    ----------
+    visit_info: `lsst.afw.table.ExposureRecord.visitInfo`
+        Visit info for the science exposure being processed.
+    historical: `bool`
+        Set to True if observations are older than the current day.
+
+    Raises
+    ------
+    requests.HTTPError
+        Raised if sattle call does not return success.
+    """
+
+    visit_mjd = visit_info.getDate().toAstropy().mjd
+
+    exposure_time_days = visit_info.getExposureTime() / 86400.0
+    exposure_end_mjd = visit_mjd + exposure_time_days / 2.0
+    exposure_start_mjd = visit_mjd - exposure_time_days / 2.0
+
+    boresight_ra = visit_info.boresightRaDec.getRa().asDegrees()
+    boresight_dec = visit_info.boresightRaDec.getDec().asDegrees()
+
+    r = requests.put(
+        f'{os.getenv("SATTLE_URI_BASE")}/visit_cache',
+        json={"visit_id": visit_info.getId(),
+              "exposure_start_mjd": exposure_start_mjd,
+              "exposure_end_mjd": exposure_end_mjd,
+              "boresight_ra": boresight_ra,
+              "boresight_dec": boresight_dec,
+              "historical": historical})
+
+    r.raise_for_status()