#! /usr/bin/env python
from stdatamodels.jwst import datamodels
from jwst.datamodels import ModelContainer
from ..stpipe import Step
import numpy as np
from . import spectral_leak
__all__ = ["SpectralLeakStep"]
[docs]
class SpectralLeakStep(Step):
"""
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']
[docs]
def process(self, input):
"""Execute the step.
Parameters
----------
input : container of models containing 1-D extracted spectra
Returns
-------
JWST DataModel
The corrected data model. This will be "input" 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:
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')
for i, x1d in enumerate(input_model): # input_model is a Model Container
# check that we have the correct type of data
if not isinstance(x1d, datamodels.MultiSpecModel):
self.log.warning("Data sent to spectral_leak step is not an extracted spectrum. It is {}." .format(type(x1d)))
for r in result:
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')
for r in result:
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')
ch1b = x1d
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')
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')
ch3a = x1d
ich3a = i # store the datamodel # to update later
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')
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 = spectral_leak.do_correction(sp_leak_ref, ch1b, ch3a)
result[ich3a].spec[0].spec_table.FLUX = corrected_3a
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')
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("Step is skipped")
return input