Source code for jwst.msaflagopen.msaflagopen_step

from stdatamodels.jwst import datamodels

from stpipe.crds_client import reference_uri_to_cache_path

from jwst.stpipe import Step
from jwst.assign_wcs import AssignWcsStep
from jwst.msaflagopen import msaflag_open

__all__ = ["MSAFlagOpenStep"]


[docs] class MSAFlagOpenStep(Step): """Flag pixels affected by MSA failed open shutters.""" class_alias = "msa_flagging" spec = """ """ # noqa: E501 reference_file_types = ["msaoper"]
[docs] def process(self, input_data): """ Flag data affected by open shutters. Parameters ---------- input_data : DataModel or str Science data to be corrected. Returns ------- DataModel Science data with DQ array modified. """ # Open the input data model with datamodels.open(input_data) as input_model: self.reference_name = self.get_reference_file(input_model, "msaoper") self.log.info("Using reference file %s", self.reference_name) # Check for a valid reference file if self.reference_name == "N/A": self.log.warning("No reference file found") self.log.warning("Step will be skipped") result = input_model.copy() result.meta.cal_step.msa_flagging = "SKIPPED" return result # Get the reference file names for constructing the WCS pipeline wcs_reffile_names = create_reference_filename_dictionary(input_model) # Do the DQ flagging result = msaflag_open.do_correction(input_model, self.reference_name, wcs_reffile_names) # set the step status to complete result.meta.cal_step.msa_flagging = "COMPLETE" return result
def create_reference_filename_dictionary(input_model): """ Get all relevant WCS reference files. Parameters ---------- input_model : DataModel Input data with WCS assigned. Returns ------- ref_files : dict Dictionary of reference files. Keys are CRDS reference types; values are file paths or 'N/A'. """ ref_files = {} for ref_type in AssignWcsStep.reference_file_types: ref_file = getattr(input_model.meta.ref_file, ref_type) ref_files[ref_type] = ref_file.name # Convert from crds protocol to absolute filenames for key in ref_files.keys(): if ref_files[key].startswith("crds://"): ref_files[key] = reference_uri_to_cache_path( ref_files[key], input_model.crds_observatory ) return ref_files