"""
Suffix manipulation.
Notes
-----
``KNOW_SUFFIXES`` is the list used by `remove_suffix`. This
list is generated by the function ``combine_suffixes``. The function uses
``SUFFIXES_TO_ADD`` to add other suffixes that it would otherwise not
discover or should absolutely be in the list. The function uses
'SUFFIXES_TO_DISCARD` for strings found that are not to be considered
suffixes.
Hence, to update ``KNOW_SUFFIXES``, update both ``SUFFIXES_TO_ADD`` and
``SUFFIXES_TO_DISCARD`` as necessary, then use the output of
``find_suffixes``.
"""
import itertools
import logging
import re
__all__ = ["remove_suffix"]
# Configure logging
logger = logging.getLogger(__name__)
# Suffixes that are hard-coded or otherwise have to exist.
# Used by `combine_suffixes` to add to the automatically derived results
# from `find_suffixes`.
SUFFIXES_TO_ADD = [
"ami",
"ami-oi",
"aminorm",
"aminorm-oi",
"blot",
"bsub",
"bsubints",
"c1d",
"cal",
"calints",
"cat",
"crf",
"crfints",
"dark",
"i2d",
"mbsub",
"median",
"phot",
"psfalign",
"psfstack",
"psfsub",
"ramp",
"rate",
"rateints",
"s2d",
"s3d",
"snr",
"uncal",
"wfscmb",
"whtlt",
"x1d",
"x1dints",
]
# Suffixes that are discovered but should not be considered.
# Used by `find_suffixes` to remove undesired values it has found.
SUFFIXES_TO_DISCARD = [
"engdblogstep",
"functionwrapper",
"pipeline",
"rscd_step",
"step",
"systemcall",
]
# Calculated suffixes, from the step name and alias.
# This is produced by the `find_suffixes` function below
_calculated_suffixes = {
"adaptive_trace_model",
"adaptivetracemodelstep",
"align_refs",
"alignrefsstep",
"ami3pipeline",
"ami_analyze",
"ami_normalize",
"amianalyzestep",
"aminormalizestep",
"assign_mtwcs",
"assign_wcs",
"assignmtwcsstep",
"assignwcsstep",
"backgroundstep",
"badpix_selfcal",
"badpixselfcalstep",
"barshadow",
"barshadowstep",
"bkg_subtract",
"calwebb_ami3",
"calwebb_coron3",
"calwebb_dark",
"calwebb_detector1",
"calwebb_guider",
"calwebb_image2",
"calwebb_image3",
"calwebb_spec2",
"calwebb_spec3",
"calwebb_tso3",
"calwebb_wfs-image3",
"charge_migration",
"chargemigrationstep",
"clean_flicker_noise",
"cleanflickernoisestep",
"combine1dstep",
"combine_1d",
"coron3pipeline",
"cube_build",
"cubebuildstep",
"dark_current",
"darkcurrentstep",
"darkpipeline",
"detector1pipeline",
"dq_init",
"dqinitstep",
"emicorr",
"emicorrstep",
"extract1dstep",
"extract2dstep",
"extract_1d",
"extract_2d",
"firstframe",
"firstframestep",
"flat_field",
"flatfieldstep",
"fringe",
"fringestep",
"gain_scale",
"gainscalestep",
"group_scale",
"groupscalestep",
"guider_cds",
"guidercdsstep",
"guiderpipeline",
"hlsp",
"hlspstep",
"image2pipeline",
"image3pipeline",
"imprint_subtract",
"imprintstep",
"ipc",
"ipcstep",
"jump",
"jumpstep",
"klip",
"klipstep",
"lastframe",
"lastframestep",
"linearity",
"linearitystep",
"master_background",
"master_background_mos",
"masterbackgroundmosstep",
"masterbackgroundstep",
"msa_flagging",
"msaflagopenstep",
"outlier_detection",
"outlierdetectionstep",
"pathloss",
"pathlossstep",
"persistence",
"persistencestep",
"photom",
"photomstep",
"picture_frame",
"pictureframestep",
"pixel_replace",
"pixelreplacestep",
"ramp_fit",
"rampfitstep",
"refpix",
"refpixstep",
"resample",
"resample_spec",
"resamplespecstep",
"resamplestep",
"reset",
"resetstep",
"residual_fringe",
"residualfringestep",
"rscd",
"rscdstep",
"saturation",
"saturationstep",
"skymatch",
"skymatchstep",
"source_catalog",
"sourcecatalogstep",
"sourcetypestep",
"spec2pipeline",
"spec3pipeline",
"spectral_leak",
"spectralleakstep",
"srctype",
"stack_refs",
"stackrefsstep",
"straylight",
"straylightstep",
"superbias",
"superbiasstep",
"targ_centroid",
"targcentroidstep",
"tso3pipeline",
"tso_photometry",
"tsophotometrystep",
"tweakreg",
"tweakregstep",
"wavecorr",
"wavecorrstep",
"wfscombinestep",
"wfss_contam",
"wfsscontamstep",
"white_light",
"whitelightstep",
}
# ##########
# Suffix API
# ##########
[docs]
def remove_suffix(name):
"""
Remove the suffix if a known suffix is already in name.
Parameters
----------
name : str
The name to remove the suffix from.
Returns
-------
name, separator : str
Matched name and separator, respectively.
"""
separator = None
match = REMOVE_SUFFIX_REGEX.match(name)
try:
name = match.group("root")
separator = match.group("separator")
except AttributeError:
pass
if separator is None:
separator = "_"
return name, separator
def replace_suffix(name, new_suffix):
"""
Replace suffix on name.
Parameters
----------
name : str
The name to replace the suffix of.
Expected to be only the basename; no extensions.
new_suffix : str
The new suffix to use.
Returns
-------
new_name : str
Name with new suffix.
"""
no_suffix, separator = remove_suffix(name)
return no_suffix + separator + new_suffix
# #####################################
# Functions to generate `KNOW_SUFFIXES`
# #####################################
def combine_suffixes(
to_add=(_calculated_suffixes, SUFFIXES_TO_ADD), to_remove=(SUFFIXES_TO_DISCARD,)
):
"""
Combine the suffix lists into a single list.
Parameters
----------
to_add : [iterable[, ...]]
List of iterables to add to the combined list.
to_remove : [iterable[, ...]]
List of iterables to remove from the combined list.
Returns
-------
suffixes : list
The list of suffixes.
"""
combined = set(itertools.chain.from_iterable(to_add))
combined.difference_update(itertools.chain.from_iterable(to_remove))
combined = list(combined)
combined.sort()
return combined
def find_suffixes():
"""
Find all possible suffixes from the ``jwst`` package.
Returns
-------
suffixes : set
The set of all programmatically findable suffixes.
Notes
-----
This will load all of the ``jwst`` package. Consider if this
is worth doing dynamically or only as a utility to update
a static list.
"""
from jwst.stpipe.utilities import all_steps
# Traverse the code base and find all `Step` classes.
# The suffix may be the class name or alias.
all_step_classes = all_steps()
suffixes = set()
for class_name, class_type in all_step_classes.items():
suffixes.add(class_name.lower())
suffixes.add(class_type.class_alias)
# That's all folks
return list(suffixes)
# --------------------------------------------------
# The set of suffixes used by the pipeline.
# This set is generated by `combine_suffixes`.
# Only update this list by `combine_suffixes`.
# Modify `SUFFIXES_TO_ADD` and `SUFFIXES_TO_DISCARD`
# to change the results.
# --------------------------------------------------
KNOW_SUFFIXES = combine_suffixes()
# Regex for removal
REMOVE_SUFFIX_REGEX = re.compile(
"^(?P<root>.+?)((?P<separator>_|-)(" + "|".join(KNOW_SUFFIXES) + "))?$"
)
# ############################################
# Main
# Find and report differences from known list.
# ############################################
if __name__ == "__main__":
print("Searching code base for calibration suffixes...") # noqa: T201
calculated_suffixes = find_suffixes()
found_suffixes = combine_suffixes(
to_add=(calculated_suffixes, SUFFIXES_TO_ADD), to_remove=(SUFFIXES_TO_DISCARD,)
)
print(f"Known list has {len(KNOW_SUFFIXES)} suffixes. Found {len(found_suffixes)} suffixes.") # noqa: T201
print( # noqa: T201
f"Suffixes that have changed are {set(found_suffixes).symmetric_difference(KNOW_SUFFIXES)}"
)