JP-3855 Update code style for spectral_leak step

.pre-commit-config.yaml

@@ -50,7 +50,6 @@ repos:
             jwst/photom/.* |
             jwst/saturation/.* |
             jwst/scripts/.* |
-            jwst/spectral_leak/.* |
             jwst/straylight/.* |
             jwst/tests/.* |
             docs/.* |

.ruff.toml

@@ -32,7 +32,6 @@ exclude = [
     "jwst/photom/**.py",
     "jwst/saturation/**.py",
     "jwst/scripts/**.py",
-    "jwst/spectral_leak/**.py",
     "jwst/straylight/**.py",
 ]
 
@@ -127,5 +126,4 @@ ignore-fully-untyped = true  # Turn off annotation checking for fully untyped co
 "jwst/photom/**.py" = ["D", "N", "A", "ARG", "B", "C4", "ICN", "INP", "ISC", "LOG", "NPY", "PGH", "PTH", "S", "SLF", "SLOT", "T20", "TRY", "UP", "YTT", "E501"]
 "jwst/saturation/**.py" = ["D", "N", "A", "ARG", "B", "C4", "ICN", "INP", "ISC", "LOG", "NPY", "PGH", "PTH", "S", "SLF", "SLOT", "T20", "TRY", "UP", "YTT", "E501"]
 "jwst/scripts/**.py" = ["D", "N", "A", "ARG", "B", "C4", "ICN", "INP", "ISC", "LOG", "NPY", "PGH", "PTH", "S", "SLF", "SLOT", "T20", "TRY", "UP", "YTT", "E501"]
-"jwst/spectral_leak/**.py" = ["D", "N", "A", "ARG", "B", "C4", "ICN", "INP", "ISC", "LOG", "NPY", "PGH", "PTH", "S", "SLF", "SLOT", "T20", "TRY", "UP", "YTT", "E501"]
 "jwst/straylight/**.py" = ["D", "N", "A", "ARG", "B", "C4", "ICN", "INP", "ISC", "LOG", "NPY", "PGH", "PTH", "S", "SLF", "SLOT", "T20", "TRY", "UP", "YTT", "E501"]

jwst/spectral_leak/__init__.py

@@ -1,3 +1,5 @@
+"""Apply a spectral leak correction to the Channel 3A of MIRI MRS data."""
+
 from .spectral_leak_step import SpectralLeakStep
 
 __all__ = ["SpectralLeakStep"]

jwst/spectral_leak/spectral_leak.py

@@ -8,15 +8,15 @@
 
 def do_correction(sp_leak_ref, ch1b, ch3a):
     """
-    Short Summary
-    -------------
+    Apply spectral leak correction to Channel 3A data using Channel 1B data.
+
     Using the spectral leak reference correction and spectrum containing CH1 B
     correct the CH 3 A spectrum.
 
     Parameters
     ----------
-    sp_leak_ref : string
-        name of the spectral leak reference file defined by datamodels.MirMrsPtCorrModel
+    sp_leak_ref : str
+        Name of the spectral leak reference file defined by datamodels.MirMrsPtCorrModel.
 
     ch1b : numpy array
         Input channel 1 B spectrum
@@ -28,7 +28,6 @@ def do_correction(sp_leak_ref, ch1b, ch3a):
     output_model : ~jwst.datamodels.RampModel
         Spectral leak corrected science data
     """
-
     wave1b = ch1b.spec[0].spec_table.WAVELENGTH
     spec1b = ch1b.spec[0].spec_table.FLUX
 
@@ -41,21 +40,27 @@ def do_correction(sp_leak_ref, ch1b, ch3a):
 
     # Spectral leak vector
 
-    leak = np.interp(wave3a, 2*wave1b, spec1b) * np.interp(wave3a, leak_wave, leak_percent)
-    # Factor of 2 in 2*wavelb is  necessary because we're dealing with second order light.  Without this the interpolation simply
-    # uses the last point in the spectrum, which is close enough to correct as won't be noticeable for BAND spectra,
-    # but for CH spectra can be quite different.
+    # Factor of 2 in 2*wavelb is  necessary because we're dealing with second order light.
+    # Without this the interpolation simply uses the last point in the spectrum, which is close
+    # enough to correct as won't be noticeable for BAND spectra, but for CH spectra can be
+    # quite different.
+
+    leak = np.interp(wave3a, 2 * wave1b, spec1b) * np.interp(wave3a, leak_wave, leak_percent)
 
     # Correct the data
     spec3a_corr = spec3a - leak
 
     # now check if there exists residual fringe corrected data
     spec3a_corr_rf = None
 
-    if isinstance(ch1b, datamodels.MRSMultiSpecModel) and isinstance(ch3a, datamodels.MRSMultiSpecModel):
+    if isinstance(ch1b, datamodels.MRSMultiSpecModel) and isinstance(
+        ch3a, datamodels.MRSMultiSpecModel
+    ):
         spec1b_rf = ch1b.spec[0].spec_table.RF_FLUX
         spec3a_rf = ch3a.spec[0].spec_table.RF_FLUX
-        leak_rf = np.interp(wave3a, 2*wave1b, spec1b_rf) * np.interp(wave3a, leak_wave, leak_percent)
+        leak_rf = np.interp(wave3a, 2 * wave1b, spec1b_rf) * np.interp(
+            wave3a, leak_wave, leak_percent
+        )
         # Correct the data
         spec3a_corr_rf = spec3a_rf - leak_rf
     return spec3a_corr, spec3a_corr_rf

jwst/spectral_leak/spectral_leak_step.py

@@ -12,102 +12,112 @@
 
 class SpectralLeakStep(Step):
     """
+    Apply a spectral leak correction to the Channel 3A of MIRI MRS data.
+
     The MIRI MRS has a spectral leak in which 6 micron light leaks into the
     12 micron channel.  This step applies a correction to the 12 micron channel.
     """
 
     class_alias = "spectral_leak"
 
-    reference_file_types = ['mrsptcorr']
+    reference_file_types = ["mrsptcorr"]
 
-    def process(self, input):
-        """Execute the step.
+    def process(self, input_data):
+        """
+        Apply a spectral leak  correction to MIRI MRS Channel 3A data.
 
         Parameters
         ----------
-        input : container of models containing 1-D extracted spectra
+        input_data : ~jwst.datamodels.ModelContainer
+            Container of models containing 1-D extracted spectra
 
         Returns
         -------
         JWST DataModel
-            The corrected data model. This will be "input" if the step is skipped,
+            The corrected data model. This will be the input model if the step is skipped,
             otherwise it will be a corrected 1D extracted spectrum that contains
             the MRS channel 3B range.
         """
-
         ch1b = None
         ch3a = None
         ich3a = None
         ch1b_wave = 6.0
         ch3a_wave = 12.0
 
-        with datamodels.open(input) as input_model:
+        with datamodels.open(input_data) as input_model:
             if isinstance(input_model, ModelContainer):
                 result = input_model.copy()  # copy input to return
                 # Retrieve the reference parameters for this type of data
-                sp_leak_ref = self.get_reference_file(input_model[0], 'mrsptcorr')
+                sp_leak_ref = self.get_reference_file(input_model[0], "mrsptcorr")
 
                 for i, x1d in enumerate(input_model):  # input_model is a Model Container
                     # check that we have the correct type of data
-                    if  isinstance(x1d, datamodels.MultiSpecModel):
-                        self.log.debug(' Data is MIRI MRS MultiSpecModel data')
+                    if isinstance(x1d, datamodels.MultiSpecModel):
+                        self.log.debug(" Data is MIRI MRS MultiSpecModel data")
                     elif isinstance(x1d, datamodels.MRSMultiSpecModel):
-                        self.log.debug(' Data is  MIRI MRS MRSMultiSpecModel data')
+                        self.log.debug(" Data is  MIRI MRS MRSMultiSpecModel data")
                     else:
-                        self.log.warning("Data sent to spectral_leak step is not an extracted spectrum. "
-                                         " It is  {}." .format(type(x1d)))
+                        self.log.warning(
+                            "Data sent to spectral_leak step is not an extracted spectrum. "
+                            f" It is  {type(x1d)}."
+                        )
                         for r in result:
-                            r.meta.cal_step.spectral_leak = 'SKIPPED'
+                            r.meta.cal_step.spectral_leak = "SKIPPED"
                         return result
                     channel = x1d.meta.instrument.channel
                     band = x1d.meta.instrument.band
                     srctype = x1d.spec[0].source_type
-                    if srctype == 'EXTENDED':
-                        self.log.warning('No spectral leak correction for extended source data')
+                    if srctype == "EXTENDED":
+                        self.log.warning("No spectral leak correction for extended source data")
                         for r in result:
-                            r.meta.cal_step.spectral_leak = 'SKIPPED'
+                            r.meta.cal_step.spectral_leak = "SKIPPED"
                         return result
                     # search x1d containing CH 1 B
-                    if '1' in channel and 'MEDIUM' in band:
-                        self.log.info('Found CH 1B in input data')
+                    if "1" in channel and "MEDIUM" in band:
+                        self.log.info("Found CH 1B in input data")
                         ch1b = x1d
-                    elif '1' in channel and 'MULTIPLE' in band:
+                    elif "1" in channel and "MULTIPLE" in band:
                         # read in the wavelength array and see
                         # if it covers ch1b_wave
                         wave = x1d.spec[0].spec_table.WAVELENGTH
                         if np.min(wave) < ch1b_wave and np.max(wave) > ch1b_wave:
-                            self.log.info('Found CH 1B in the input data')
+                            self.log.info("Found CH 1B in the input data")
                             ch1b = x1d
                     # search x1d containing CH 3 A
-                    if '3' in channel and 'SHORT' in band:
-                        self.log.info('Found CH 3A in the input data')
+                    if "3" in channel and "SHORT" in band:
+                        self.log.info("Found CH 3A in the input data")
                         ch3a = x1d
                         ich3a = i  # store the datamodel # to update later
-                    elif '3' in channel and 'MULTIPLE' in band:
+                    elif "3" in channel and "MULTIPLE" in band:
                         # read in the wavelength array and see
                         # if it covers ch3a_wave
                         wave = x1d.spec[0].spec_table.WAVELENGTH
                         if np.min(wave) < ch3a_wave and np.max(wave) > ch3a_wave:
-                            self.log.info('Found CH 3A in the input data')
+                            self.log.info("Found CH 3A in the input data")
                             ch3a = x1d
                             ich3a = i  # store the datamodel to update later
 
                 # done looping over data now if 1B and 3A data exists make a correction
                 # update result and return
                 if ch1b is not None and ch3a is not None:
-                    corrected_3a, corrected_3a_rf = spectral_leak.do_correction(sp_leak_ref, ch1b, ch3a)
+                    corrected_3a, corrected_3a_rf = spectral_leak.do_correction(
+                        sp_leak_ref, ch1b, ch3a
+                    )
                     result[ich3a].spec[0].spec_table.FLUX = corrected_3a
                     if corrected_3a_rf is not None:
                         result[ich3a].spec[0].spec_table.RF_FLUX = corrected_3a_rf
-                    result[ich3a].meta.cal_step.spectral_leak = 'COMPLETE'
+                    result[ich3a].meta.cal_step.spectral_leak = "COMPLETE"
                     return result
                 else:
                     for r in result:
-                        r.meta.cal_step.spectral_leak = 'SKIPPED'
-                    self.log.warning('CH1B and CH3A were not found. No spectral leak correction')
+                        r.meta.cal_step.spectral_leak = "SKIPPED"
+                    self.log.warning("CH1B and CH3A were not found. No spectral leak correction")
                     return result
 
             else:
-                self.log.warning("Data sent to spectral_leak step is not a ModelContainer. It is  {}." .format(type(input_model)))
+                self.log.warning(
+                    "Data sent to spectral_leak step is not a ModelContainer."
+                    "It is {type(input_model)}."
+                )
                 self.log.warning("Step is skipped")
-                return input
+                return input_data