cherry pick commit 523a8c7

CHANGES.rst

@@ -8,6 +8,8 @@ set_telescope_pointing
 
 - Add fgsid option to set_telescope_pointing [#6717]
 
+- Further restrict default models that can be updated. [#6767]
+
 - Update COARSE handling of FGS, pysiaf importing, model opening,
   and removal of stale code. [#6735]
 

jwst/datamodels/tests/test_open.py

@@ -25,6 +25,24 @@
 MEMORY = 100  # 100 bytes
 
 
+@pytest.mark.parametrize('guess', [True, False])
+def test_guess(guess):
+    """Test the guess parameter to the open func"""
+    path = Path(t_path('test.fits'))
+
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore", "model_type not found")
+        if guess is None or guess:
+            # Default is to guess at the model type.
+            with datamodels.open(path, guess=guess) as model:
+                assert isinstance(model, JwstDataModel)
+        else:
+            # Without guessing, the call should fail.
+            with pytest.raises(TypeError):
+                with datamodels.open(path, guess=guess) as model:
+                    pass
+
+
 def test_mirirampmodel_deprecation(tmp_path):
     """Test that a deprecated MIRIRampModel can be opened"""
     path = str(tmp_path / "ramp.fits")

jwst/datamodels/util.py

@@ -31,7 +31,7 @@ class NoTypeWarning(Warning):
     pass
 
 
-def open(init=None, memmap=False, **kwargs):
+def open(init=None, guess=True, memmap=False, **kwargs):
     """
     Creates a DataModel from a number of different types
 
@@ -56,6 +56,10 @@ def open(init=None, memmap=False, **kwargs):
 
         - dict: The object model tree for the data model
 
+    guess : bool
+        Guess as to the model type if the model type is not specifically known from the file.
+        If not guess and the model type is not explicit, raise a TypeError.
+
     memmap : bool
         Turn memmap of file on or off.  (default: False).
 
@@ -184,6 +188,8 @@ def open(init=None, memmap=False, **kwargs):
     # First try to get the class name from the primary header
     new_class = _class_from_model_type(hdulist)
     has_model_type = new_class is not None
+    if not guess and not has_model_type:
+        raise TypeError('Model type is not specifically defined and guessing has been disabled.')
 
     # Special handling for ramp files for backwards compatibility
     if new_class is None:

jwst/lib/set_telescope_pointing.py

@@ -77,8 +77,6 @@
 from ..lib.engdb_tools import ENGDB_Service
 from ..lib.pipe_utils import is_tso
 
-TYPES_TO_UPDATE = set(list(IMAGING_TYPES) + FGS_GUIDE_EXP_TYPES)
-
 __all__ = [
     'Methods',
     'TransformParameters',
@@ -98,6 +96,9 @@
 DEBUG_FULL = logging.DEBUG - 1
 LOGLEVELS = [logging.INFO, logging.DEBUG, DEBUG_FULL]
 
+EXPECTED_MODELS = (datamodels.Level1bModel, datamodels.ImageModel, datamodels.CubeModel)
+TYPES_TO_UPDATE = set(list(IMAGING_TYPES) + FGS_GUIDE_EXP_TYPES)
+
 
 # The available methods for transformation
 class Methods(Enum):
@@ -372,22 +373,27 @@ def as_reprdict(self):
         return r
 
 
-def add_wcs(filename, default_pa_v3=0., siaf_path=None, engdb_url=None,
+def add_wcs(filename, allow_any_file=False, default_pa_v3=0., siaf_path=None, engdb_url=None,
             fgsid=None, tolerance=60, allow_default=False, reduce_func=None,
             dry_run=False, save_transforms=None, **transform_kwargs):
-    """Add WCS information to a FITS file.
+    """Add WCS information to a JWST DataModel.
 
     Telescope orientation is attempted to be obtained from
     the engineering database. Failing that, a default pointing
     is used based on proposal target.
 
-    The FITS file is updated in-place.
+    The file is updated in-place.
 
     Parameters
     ----------
     filename : str
         The path to a data file.
 
+    allow_any_file : bool
+        Attempt to add the WCS information to any type of file.
+        The default, `False`, only allows modifications of files that contain
+        known datamodels of `Level1bmodel`, `ImageModel`, or `CubeModel`.
+
     default_pa_v3 : float
         The V3 position angle to use if the pointing information
         is not found.
@@ -425,7 +431,14 @@ def add_wcs(filename, default_pa_v3=0., siaf_path=None, engdb_url=None,
 
     Notes
     -----
-    This function adds absolute pointing information to the JWST datamodels provided.
+
+    This function adds absolute pointing information to the JWST
+    datamodels provided. By default, only Stage 1 and Stage 2a exposures are
+    allowed to be updated. These have the suffixes of "uncal", "rate", and
+    "rateints" representing datamodels Level1bModel, ImageModel, and CubeModel.
+    Any higher level product, from Stage 2b and beyond, that has had the
+    `assign_wcs` step applied, have improved WCS information. Running
+    this task on such files will potentially corrupt the WCS.
 
     It starts by populating the headers with values from the SIAF database.
     It adds the following keywords to all files:
@@ -456,13 +469,18 @@ def add_wcs(filename, default_pa_v3=0., siaf_path=None, engdb_url=None,
 
     It does not currently place the new keywords in any particular location
     in the header other than what is required by the standard.
+
     """
     logger.info('Updating WCS info for file %s', filename)
-    with datamodels.open(filename) as model:
-        if type(model) not in (datamodels.Level1bModel, datamodels.ImageModel, datamodels.CubeModel):
+    with datamodels.open(filename, guess=allow_any_file) as model:
+        if type(model) not in EXPECTED_MODELS:
             logger.warning(f'Input {model} is not of an expected type (uncal, rate, rateints)'
                            '\n    Updating pointing may have no effect or detrimental effects on the WCS information,'
                            '\n    especially if the input is the result of Level2b or higher calibration.')
+            if not allow_any_file:
+                raise TypeError(f'Input model {model} is not one of {EXPECTED_MODELS} and `allow_any_file` is `False`.'
+                                '\n\tFailing WCS processing.')
+
         t_pars, transforms = update_wcs(
             model,
             default_pa_v3=default_pa_v3,

jwst/lib/tests/test_set_telescope_pointing.py

@@ -7,9 +7,11 @@
 from pathlib import Path
 import sys
 from tempfile import TemporaryDirectory
+import warnings
 
 import pytest
 
+from astropy.io import fits
 from astropy.time import Time
 
 from jwst import datamodels
@@ -83,6 +85,59 @@
                  ]
 
 
+@pytest.fixture(params=[('good_model', True), ('bad_model', False), ('fits_nomodel', False)])
+def file_case(request, tmp_path):
+    """Generate files with different model states"""
+    case, allow = request.param
+
+    if case == 'good_model':
+        # Make a model that will always succeed
+        model = datamodels.Level1bModel((10, 10, 10, 10))
+        path = tmp_path / 'level1bmodel.fits'
+        model.save(path)
+    elif case == 'bad_model':
+        # Make a model that will fail if not allowed
+        model = datamodels.IFUCubeModel((10, 10, 10))
+        path = tmp_path / 'image.fits'
+        model.save(path)
+    elif case == 'fits_nomodel':
+        # Create just a plain anything FITS
+        hdu = fits.PrimaryHDU()
+        hdul = fits.HDUList([hdu])
+        path = tmp_path / 'empty.fits'
+        hdul.writeto(path)
+    else:
+        assert False, f'Cannot produce a file for {case}'
+
+    return path, allow
+
+
+@pytest.mark.parametrize('allow_any_file', [True, False])
+def test_allow_any_file(file_case, allow_any_file):
+    """Test various files against whether they should be allowed or not
+
+    Parameters
+    ----------
+    file_case : (Path-like, allow)
+        File to test and whether it should always be allowed.
+        If not `allow`, the file should be usable only when `allow_any_file`.
+
+    allow_any_file : bool
+        Value of `allow_any_file` to try
+    """
+    path, allow = file_case
+    with warnings.catch_warnings():
+        warnings.filterwarnings("ignore", "model_type not found")
+        if not allow and not allow_any_file:
+            with pytest.raises(TypeError):
+                stp.add_wcs(path, siaf_path=siaf_path, allow_any_file=allow_any_file, dry_run=True)
+        else:
+            # Expected error when trying to actually add the wcs.
+            # The provided files do not have sufficient info to do the calculations.
+            with pytest.raises(AttributeError):
+                stp.add_wcs(path, siaf_path=siaf_path, allow_any_file=allow_any_file, dry_run=True)
+
+
 @pytest.mark.parametrize(
     'method',
     [method for method in stp.Methods]

scripts/set_telescope_pointing.py

@@ -54,7 +54,9 @@
 if __name__ == '__main__':
 
     parser = argparse.ArgumentParser(
-        description='Update basic WCS information in JWST exposures from the engineering database.'
+        description=('Update basic WCS information in JWST exposures from the engineering database.'
+                     ' For detailed information, see'
+                     ' https://jwst-pipeline.readthedocs.io/en/latest/jwst/lib/set_telescope_pointing.html')
     )
     parser.add_argument(
         'exposure', type=str, nargs='+',
@@ -64,6 +66,18 @@
         '-v', '--verbose', action='count', default=0,
         help='Increase verbosity. Specifying multiple times adds more output.'
     )
+    parser.add_argument(
+        '--allow-any-file', action='store_true', default=False,
+        help='Attempt to update WCS for any file or model. Default: False'
+    )
+    parser.add_argument(
+        '--allow-default', action='store_true',
+        help='If pointing information cannot be determine, use header information.'
+    )
+    parser.add_argument(
+        '--dry-run', action='store_true',
+        help='Perform all actions but do not save the results'
+    )
     parser.add_argument(
         '--method',
         type=stp.Methods, choices=list(stp.Methods), default=stp.Methods.default,
@@ -85,14 +99,6 @@
         '--tolerance', type=int, default=60,
         help='Seconds beyond the observation time to search for telemetry. Default: %(default)s'
     )
-    parser.add_argument(
-        '--allow-default', action='store_true',
-        help='If pointing information cannot be determine, use header information.'
-    )
-    parser.add_argument(
-        '--dry-run', action='store_true',
-        help='Perform all actions but do not save the results'
-    )
     parser.add_argument(
         '--siaf', type=str, default=None,
         help='SIAF PRD sqlite database file. If not specified, default is to use `$XML_DATA/prd.db`'
@@ -137,6 +143,7 @@
         try:
             stp.add_wcs(
                 filename,
+                allow_any_file=args.allow_any_file,
                 siaf_path=args.siaf,
                 engdb_url=args.engdb_url,
                 fgsid=args.fgsid,
@@ -148,5 +155,6 @@
                 save_transforms=transform_path,
                 override_transforms=override_transforms
             )
-        except ValueError as exception:
-            logger.info('Cannot determine pointing information', exc_info=exception)
+        except (TypeError, ValueError) as exception:
+            logger.warning('Cannot determine pointing information: %s', str(exception))
+            logger.debug('Full exception:', exc_info=exception)