JP-4151: Give TSO3 x1dints files an S_REGION

changes/9927.pipeline.rst

@@ -0,0 +1 @@
+Add S_REGION to x1dints files generated by calwebb_tso3

jwst/pipeline/calwebb_tso3.py

@@ -172,6 +172,8 @@ def process(self, input_data):
                 log.warning("extract_1d step could not be completed for any integrations")
                 log.warning("x1dints products will not be created.")
             else:
+                # Set S_REGION to allow the x1dints file to show up in MAST spatial queries
+                self._populate_tso_spectral_sregion(x1d_result, input_models)
                 self.save_model(x1d_result, suffix="x1dints")
 
         # Done with all the inputs
@@ -192,3 +194,43 @@ def process(self, input_data):
         # All done. Nothing to return, because all products have
         # been created here.
         return
+
+    def _populate_tso_spectral_sregion(self, model, cal_model_list):
+        """
+        Generate cumulative S_REGION footprint from input images.
+
+        Take the input S_REGION values from all input models, and combine
+        them using a polygon union to create a cumulative footprint for the output
+        TSO spectral product.
+        The union is performed in pixel coordinates to avoid distortion,
+        so the WCS of the first input model is used to convert to and from sky coordinates.
+
+        Parameters
+        ----------
+        model : `~stdatamodels.jwst.datamodels.TSOMultiSpecModel`
+            The newly generated TSOMultiSpecModel made as part of
+            the save operation for spec3 processing of TSO data.
+
+        cal_model_list : `~jwst.datamodels.ModelContainer`
+            The input models provided to Tso3Pipeline by the
+            input association.
+        """
+        if (len(cal_model_list) == 0) or (len(model.spec) == 0):
+            log.warning("No input or output models provided; cannot set S_REGION.")
+            return
+        has_sregion = [w.meta.wcsinfo.hasattr("s_region") for w in cal_model_list]
+        if not all(has_sregion):
+            log.warning(
+                "One or more input model(s) are missing an `s_region` attribute; "
+                "output S_REGION will not be set."
+            )
+            return
+
+        input_sregions = [w.meta.wcsinfo.s_region for w in cal_model_list]
+        if not all(s == input_sregions[0] for s in input_sregions):
+            log.warning(
+                "Input models have different S_REGION values; this is unexpected for tso3 data. "
+                "Setting output S_REGION to the first input model's value."
+            )
+        model.spec[0].s_region = input_sregions[0]
+        return

jwst/pipeline/tests/test_calwebb_tso3.py

@@ -1,4 +1,7 @@
+import stdatamodels.jwst.datamodels as dm
+
 from jwst.associations.asn_from_list import asn_from_list
+from jwst.datamodels import ModelContainer
 from jwst.extract_1d.tests.helpers import mock_niriss_soss_96_func, mock_niriss_soss_full_func
 from jwst.pipeline.calwebb_tso3 import Tso3Pipeline
 
@@ -23,6 +26,7 @@ def niriss_soss_tso(subarray="SUBSTRIP96"):
     else:
         input_model = mock_niriss_soss_96_func()
     input_model.meta.wcs = None
+    input_model.meta.wcsinfo.s_region = "POLYGON ICRS 0 0 0 1 1 1 1 0"
     input_model.meta.visit.tsovisit = True
     input_model.int_times = input_model.int_times
     return input_model
@@ -59,6 +63,9 @@ def test_niriss_soss(tmp_path):
     for filename in expected:
         assert (tmp_path / filename).exists()
 
+    with dm.open(tmp_path / "test_tso3_x1dints.fits") as x1d:
+        assert x1d.spec[0].s_region == "POLYGON ICRS 0 0 0 1 1 1 1 0"
+
 
 def test_niriss_soss_full(tmp_path):
     """Smoke test for tso3 for an invalid NIRISS SOSS TSO mode."""
@@ -74,3 +81,72 @@ def test_niriss_soss_full(tmp_path):
         assert (tmp_path / filename).exists()
     for filename in not_expected:
         assert not (tmp_path / filename).exists()
+
+
+def test_populate_tso_spectral_sregion(log_watcher):
+    model = dm.TSOMultiSpecModel()
+    model.spec.append(dm.SpecModel())
+    model.spec.append(dm.SpecModel())
+    cal_model = dm.CubeModel((10, 10, 10))
+    cal_model_list = ModelContainer([cal_model, cal_model.copy()])
+
+    # no s_region attributes present
+    watcher = log_watcher(
+        "jwst.pipeline.calwebb_tso3",
+        message="One or more input model(s) are missing an `s_region` attribute;",
+        level="warning",
+    )
+    result = Tso3Pipeline()._populate_tso_spectral_sregion(model, cal_model_list)
+    watcher.assert_seen()
+    assert result is None
+
+    # s_regions are all the same, should run without issues
+    for m in cal_model_list:
+        m.meta.wcsinfo.s_region = "POLYGON ICRS 0 0 0 1 1 1 1 0"
+    result = Tso3Pipeline()._populate_tso_spectral_sregion(model, cal_model_list)
+    assert model.spec[0].s_region == "POLYGON ICRS 0 0 0 1 1 1 1 0"
+    assert not model.spec[1].hasattr("s_region")
+
+    # s_regions differ, should warn and set to first
+    cal_model_list[1].meta.wcsinfo.s_region = "POLYGON ICRS 1 1 1 2 2 2 2 1"
+    watcher = log_watcher(
+        "jwst.pipeline.calwebb_tso3",
+        message="Input models have different S_REGION values;",
+        level="warning",
+    )
+    result = Tso3Pipeline()._populate_tso_spectral_sregion(model, cal_model_list)
+    watcher.assert_seen()
+    assert model.spec[0].s_region == "POLYGON ICRS 0 0 0 1 1 1 1 0"
+    assert not model.spec[1].hasattr("s_region")
+
+
+def test_populate_tso_spectral_sregion_empty_container(log_watcher):
+    """Test with an empty ModelContainer."""
+    model = dm.TSOMultiSpecModel()
+    model.spec.append(dm.SpecModel())
+    cal_model_list = ModelContainer()
+
+    # Should handle empty container gracefully
+    watcher = log_watcher(
+        "jwst.pipeline.calwebb_tso3", message="No input or output models provided;", level="warning"
+    )
+    result = Tso3Pipeline()._populate_tso_spectral_sregion(model, cal_model_list)
+    watcher.assert_seen()
+    assert result is None
+    assert not model.spec[0].hasattr("s_region")
+
+
+def test_populate_tso_spectral_sregion_no_spec(log_watcher):
+    """Test with a model that has no spec entries."""
+    model = dm.TSOMultiSpecModel()
+    cal_model = dm.CubeModel((10, 10, 10))
+    cal_model.meta.wcsinfo.s_region = "POLYGON ICRS 0 0 0 1 1 1 1 0"
+    cal_model_list = ModelContainer([cal_model])
+
+    # Should handle gracefully when model.spec is empty
+    watcher = log_watcher(
+        "jwst.pipeline.calwebb_tso3", message="No input or output models provided;", level="warning"
+    )
+    result = Tso3Pipeline()._populate_tso_spectral_sregion(model, cal_model_list)
+    watcher.assert_seen()
+    assert result is None