"""
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
from importlib import import_module
from pathlib import Path
__all__ = ["remove_suffix"]
# Configure logging
logger = logging.getLogger(__name__)
# Suffixes that are hard-coded or otherwise
# have to exist. Used by `find_suffixes` to
# add to the result it produces.
SUFFIXES_TO_ADD = [
"ami",
"aminorm",
"ami-oi",
"aminorm-oi",
"blot",
"bsub",
"bsubints",
"c1d",
"cal",
"calints",
"cat",
"crf",
"crfints",
"dark",
"i2d",
"mbsub",
"median",
"phot",
"psfalign",
"psfstack",
"psfsub",
"ramp",
"rate",
"rateints",
"residual_fringe",
"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.
# This is produced by the `find_suffixes` function below
_calculated_suffixes = {
"lastframe",
"saturation",
"pixelreplacestep",
"reset",
"wavecorrstep",
"lastframestep",
"residual_fringe",
"cat",
"extract2dstep",
"guider_cds",
"gain_scale",
"bkg_subtract",
"resetstep",
"imprintstep",
"extract1dstep",
"pathlossstep",
"spec2pipeline",
"darkpipeline",
"jumpstep",
"masterbackgroundmosstep",
"s3d",
"saturationstep",
"jump",
"ipc",
"wfscmb",
"rampfitstep",
"sourcecatalogstep",
"outlier_detection",
"assignwcsstep",
"persistencestep",
"chargemigrationstep",
"assign_wcs",
"whtlt",
"assignmtwcsstep",
"engdblog",
"align_refs",
"master_background",
"tso3pipeline",
"backgroundstep",
"tweakregstep",
"i2d",
"whitelightstep",
"photom",
"guidercdsstep",
"imprint",
"resample_spec",
"white_light",
"resample",
"sourcetypestep",
"flat_field",
"extract_1d",
"skymatchstep",
"residualfringestep",
"superbias",
"fringestep",
"photomstep",
"klipstep",
"tweakreg",
"fringe",
"stackrefsstep",
"stackrefs",
"extract_2d",
"source_catalog",
"resamplestep",
"refpix",
"superbiasstep",
"dq_init",
"wfsscontamstep",
"straylightstep",
"spectral_leak",
"spectralleakstep",
"rscdstep",
"barshadowstep",
"msaflagopenstep",
"refpixstep",
"ami_normalize",
"wfscombinestep",
"aminormalizestep",
"rscd",
"firstframestep",
"ami_analyze",
"resamplespecstep",
"group_scale",
"linearitystep",
"combine1dstep",
"tsophotometrystep",
"spec3pipeline",
"clean_flicker_noise",
"cleanflickernoisestep",
"picture_frame",
"pictureframestep",
"outlierdetectionstep",
"groupscalestep",
"hlspstep",
"masterbackgroundstep",
"alignrefsstep",
"flatfieldstep",
"skymatch",
"cube_build",
"ramp_fit",
"firstframe",
"srctype",
"straylight",
"spec2nrslamp",
"pathloss",
"guiderpipeline",
"linearity",
"assign_mtwcs",
"hlsp",
"wfscombine",
"detector1pipeline",
"ami3pipeline",
"gainscalestep",
"coron3pipeline",
"image3pipeline",
"combine_1d",
"amianalyzestep",
"ipcstep",
"dark_current",
"dqinitstep",
"cubebuildstep",
"darkcurrentstep",
"persistence",
"image2pipeline",
"klip",
"emicorr",
"emicorrstep",
"badpixselfcalstep",
"targcentroidstep",
"targ_centroid",
"adaptive_trace_model",
"adaptivetracemodelstep",
}
# ##########
# 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 import Step
from jwst.stpipe.utilities import all_steps
jwst = import_module("jwst")
jwst_fpath = Path(jwst.__file__).parent
# First traverse the code base and find all
# `Step` classes. The default suffix is the
# class name.
suffixes = {klass_name.lower() for klass_name in all_steps()}
# Instantiate Steps/Pipelines from their configuration files.
# Different names and suffixes can be defined in this way.
# Note: Based on the `collect_pipeline_cfgs` script
config_path = jwst_fpath / "pipeline"
for config_file in config_path.iterdir():
if config_file.suffix == ".cfg":
try:
step = Step.from_config_file(str(config_path / config_file))
except Exception as err:
logger.debug("Configuration %s failed: %s", config_file, str(err))
else:
suffixes.add(step.name.lower())
if step.suffix is not None:
suffixes.add(step.suffix.lower())
# 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)}"
)