"""
Class for guiding calibration object generation in PypeIt.
.. include common links, assuming primary doc root is up one directory
.. include:: ../include/links.rst
"""
from abc import ABCMeta
from collections import Counter
import copy
import datetime
import os
from pathlib import Path
import numpy as np
import yaml
from pypeit import __version__
from pypeit import log
from pypeit import PypeItError
from pypeit import alignframe
from pypeit import flatfield
from pypeit import edgetrace
from pypeit import scattlight
from pypeit import slittrace
from pypeit import state
from pypeit import wavecalib
from pypeit import wavetilts
from pypeit.calibframe import CalibFrame
from pypeit.images import buildimage
from pypeit.metadata import PypeItMetaData
from pypeit.core import framematch
from pypeit.core import parse
from pypeit.core import scattlight as core_scattlight
from pypeit.par import pypeitpar
from pypeit.spectrographs.spectrograph import Spectrograph
from pypeit import utils
from IPython import embed
[docs]
class Calibrations:
"""
Class designed to guide the generation of calibration images and objects in
PypeIt.
Args:
fitstbl (:class:`~pypeit.metadata.PypeItMetaData`):
The class holding the metadata for all the frames in this PypeIt run.
If None, we are using this class as a glorified dict to hold the objects.
par (:class:`~pypeit.par.pypeitpar.CalibrationsPar`):
Parameter set defining optional parameters of PypeIt's algorithms
for Calibrations
spectrograph (:class:`~pypeit.spectrographs.spectrograph.Spectrograph`):
Spectrograph object
caldir (:obj:`str`, `Path`_):
Path for the processed calibration files.
qadir (:obj:`str`, optional):
Path for quality assessment output. If not provided, no QA
plots are saved.
reuse_calibs (:obj:`bool`, optional):
Instead of reprocessing them, load existing calibration files from
disk if they exist.
show (:obj:`bool`, optional):
Show plots of PypeIt's results as the code progresses. Requires
interaction from the user.
user_slits (:obj:`dict`, optional):
A limited set of slits selected by the user for analysis. See
:func:`~pypeit.slittrace.SlitTraceSet.user_mask`.
chk_version (:obj:`bool`, optional):
When reading in existing files written by PypeIt, perform strict
version checking to ensure a valid file. If False, the code will
try to keep going, but this may lead to faults and quiet failures.
User beware!
Attributes:
fitstbl (:class:`~pypeit.metadata.PypeItMetaData`):
See instantiation arguments.
par (:class:`~pypeit.par.pypeitpar.CalibrationsPar`):
See instantiation arguments.
spectrograph (:class:`~pypeit.spectrographs.spectrograph.Spectrograph`):
See instantiation arguments.
calib_dir (`Path`_):
Path for the processed calibration files.
qa_path (`Path`_):
Path for the QA diagnostics.
reuse_calibs (:obj:`bool`):
See instantiation arguments.
show (:obj:`bool`):
See instantiation arguments.
user_slits (:obj:`dict`):
See instantiation arguments.
det (:obj:`int`, :obj:`tuple`):
The single detector or set of detectors in a mosaic to process.
frame (:obj:`int`):
The index of a raw file in :attr:`fitstbl` used to set the
calibration group.
calib_ID (:obj:`int`):
The calibration group associated with :attr:`frame`.
msarc (:class:`~pypeit.images.buildimage.ArcImage`):
Arc calibration frame
mstilt (:class:`~pypeit.images.buildimage.TiltImage`):
Tilt calibration frame
alignments (:class:`~pypeit.alignframe.Alignments`):
Alignment calibration frame
msbias (:class:`~pypeit.images.buildimage.BiasImage`):
Bias calibration frame
msdark (:class:`~pypeit.images.buildimage.DarkImage`):
Dark calibration frame
msbpm (`numpy.ndarray`_):
Boolean array with the bad-pixel mask (pixels that should masked are
set to True).
wv_calib (:class:`~pypeit.wavecalib.WaveCalib`):
Wavelength calibration frame
slits (:class:`~pypeit.slittrace.SlitTraceSet`):
Slit tracing calibration frame
wavetilts (:class:`~pypeit.wavetilts.WaveTilts`):
Tilts calibration frame
flatimages (:class:`~pypeit.flatfield.FlatImages`):
Flat-field calibration frame
steps (:obj:`list`):
A list of strings setting the set of processing steps to be
completed (not necessarily those that were successful). See the
``default_steps`` functions of each subclass.
success (:obj:`bool`):
Flag that the calibrations were all generated successfully.
failed_step (:obj:`str`):
If the calibrations were unsuccessful, this is the step that
led to the fault.
"""
__metaclass__ = ABCMeta
# Mapping from calibration step name to (frametype, frameclass) for file lookup.
# bpm is None because it produces no CalibFrame output file.
step_frame_map = {
'bias': ('bias', buildimage.BiasImage),
'dark': ('dark', buildimage.DarkImage),
'bpm': None,
'arc': ('arc', buildimage.ArcImage),
'tiltimg': ('tilt', buildimage.TiltImage),
'slits': ('trace', slittrace.SlitTraceSet),
'wv_calib': ('arc', wavecalib.WaveCalib),
'tilts': ('tilt', wavetilts.WaveTilts),
'scattlight': ('scattlight', scattlight.ScatteredLight),
'flats': ('illumflat', flatfield.FlatImages),
'align': ('align', alignframe.Alignments),
}
[docs]
@staticmethod
def get_instance(fitstbl, par, spectrograph, caldir, calib_ID:str,
frame:int, det:int, **kwargs):
"""
Get the instance of the appropriate subclass of :class:`Calibrations` to
use for reducing data from the provided ``spectrograph``. For argument
descriptions, see :class:`Calibrations`.
"""
calibclass = MultiSlitCalibrations if spectrograph.pypeline in ['MultiSlit', 'Echelle'] \
else IFUCalibrations
return calibclass(fitstbl, par, spectrograph, caldir, calib_ID, frame, det,
**kwargs)
def __init__(self, fitstbl, par, spectrograph, caldir, calib_ID:str,
frame:int, det:int, qadir=None,
reuse_calibs=False, show=False, user_slits=None, chk_version=True,
state=None):
# Check the types
# TODO -- Remove this None option once we have data models for all the Calibrations
# outputs and use them to feed Reduce instead of the Calibrations object
if not isinstance(fitstbl, PypeItMetaData) and fitstbl is not None:
raise PypeItError('fitstbl must be an PypeItMetaData object')
if not isinstance(par, pypeitpar.CalibrationsPar):
raise PypeItError('Input parameters must be a CalibrationsPar instance.')
if not isinstance(spectrograph, Spectrograph):
raise PypeItError('Must provide Spectrograph instance to Calibrations.')
# Required inputs
self.fitstbl = fitstbl
self.par = par
self.spectrograph = spectrograph
self.calib_ID = calib_ID
self.det = det
self.frame = frame
# Calibrations
self.reuse_calibs = reuse_calibs
self.chk_version = chk_version
self.calib_dir = Path(caldir).absolute()
if not self.calib_dir.exists():
self.calib_dir.mkdir(parents=True)
# QA
self.qa_path = None if qadir is None else Path(qadir).absolute()
if self.qa_path is not None:
# TODO: This should only be defined in one place! Where?...
qa_png_path = self.qa_path / 'PNGs'
self.write_qa = self.qa_path is not None
if self.write_qa and not qa_png_path.exists():
qa_png_path.mkdir(parents=True)
# Debugging
self.show = show
# State
self.state = state
# Restrict on slits?
self.user_slits = user_slits
# Attributes
self.msarc = None
self.mstilt = None
self.alignments = None
self.msbias = None
self.msdark = None
self.msbpm = None
self.wv_calib = None
self.slits = None
self.msscattlight = None
self.wavetilts = None
self.flatimages = None
self.buildwaveTilts = None
# Steps
self.steps = self.__class__.default_steps()
self.success = False
self.failed_step = None
[docs]
def check_calibrations(self, file_list, check_lamps=True):
"""
Check if the input calibration files are consistent with each other.
This step is usually needed when combining calibration frames of a given type.
This routine currently only prints out warning messages if the calibration files are not consistent.
Note: The exposure times are currently checked in the combine step, so they are not checked here.
Parameters
----------
file_list : list
List of calibration files to check
check_lamps : bool, optional
Check if the lamp status is the same for all the files. Default is True.
"""
lampstat = [None] * len(file_list)
# Loop on the files
for ii, ifile in enumerate(file_list):
# Save the lamp status
headarr = [h.copy() for h in self.spectrograph.get_headarr(ifile)]
lampstat[ii] = self.spectrograph.get_lamps_status(headarr)
# Check that the lamps being combined are all the same
if check_lamps:
if not lampstat[1:] == lampstat[:-1]:
log.warning("The following files contain different lamp status")
# Get the longest strings
maxlen = max([len("Filename")] + [len(os.path.split(x)[1]) for x in file_list])
maxlmp = max([len("Lamp status")] + [len(x) for x in lampstat])
strout = "{0:" + str(maxlen) + "} {1:s}"
# Print the messages
print(' ' + '-' * maxlen + " " + '-' * maxlmp)
print(' ' + strout.format("Filename", "Lamp status"))
print(' ' + '-' * maxlen + " " + '-' * maxlmp)
for ff, file in enumerate(file_list):
print(' '
+ strout.format(os.path.split(file)[1], " ".join(lampstat[ff].split("_"))))
print(' ' + '-' * maxlen + " " + '-' * maxlmp)
[docs]
def find_calibrations(self, frametype, frameclass):
"""
Find calibration files and identifiers.
Parameters
----------
frametype : :obj:`str`
Calibration frame type. Must be a valid frame type; see
:func:`~pypeit.core.framematch.valid_frametype`.
frameclass : :class:`~pypeit.calibframe.CalibFrame`
The subclass used to store the processed calibration data.
Returns
-------
raw_files : :obj:`list`
The list of raw files in :attr:`fitstbl` with the provided
frametype.
cal_file : `Path`_
The path with/for the processed calibration frame
calib_key : :obj:`str`
The calibration identifier
setup : :obj:`str`
The setup/configuration identifier
calib_id : :obj:`list`
The calibration groups
detname : :obj:`str`
The detector/mosaic identifier
"""
# NOTE: This will raise an exception if the frametype is not valid!
framematch.valid_frametype(frametype, raise_error=True)
if not issubclass(frameclass, CalibFrame):
raise PypeItError(f'CODING ERROR: {frameclass} is not a subclass of CalibFrame.')
# Grab rows with relevant frames
detname = self.spectrograph.get_det_name(self.det)
rows = self.fitstbl.find_frames(frametype, calib_ID=self.calib_ID, index=True)
if len(rows) == 0:
# No raw files are available. Attempt to find an existing and
# relevant calibration frame based on the setup/configuration and
# calibration group of the (science) frame to be calibrated.
setup = self.fitstbl['setup'][self.frame]
cal_file = frameclass.glob(self.calib_dir, setup, self.calib_ID, detname=detname)
if cal_file is None or len(cal_file) > 1:
return [], None, None, setup, None, detname
cal_file = cal_file[0]
calib_key = frameclass.parse_key_dir(str(cal_file), from_filename=True)[0]
calib_id = frameclass.parse_calib_key(calib_key)[1]
return [], cal_file, calib_key, setup, frameclass.ingest_calib_id(calib_id), detname
# Otherwise, use the metadata for the raw frames to set the name of
# the processed calibration frame.
setup = self.fitstbl['setup'][rows[0]]
calib_id = self.fitstbl['calib'][rows[0]]
calib_key = frameclass.construct_calib_key(setup, calib_id, detname)
# Construct the expected calibration frame file name
cal_file = Path(frameclass.construct_file_name(calib_key, calib_dir=self.calib_dir))
return self.fitstbl.frame_paths(rows), cal_file, calib_key, setup, \
frameclass.ingest_calib_id(calib_id), detname
# def set_config(self, frame, det, par=None):
# """
# Specify the critical attributes of the class to perform a set of calibrations.
#
# Operations are:
#
# - Set the frame
# - Use the frame to find the calibration group
# - Set the detector/mosaic
# - Set the parameters
#
# Args:
# frame (:obj:`int`):
# The row index in :attr:`fitstbl` with the frame to calibrate.
# det (:obj:`int`):
# Detector number.
# par (:class:`~pypeit.par.pypeitpar.CalibrationsPar`, optional):
# Parameters used by the calibration procedures. If None, use
# :attr:`par`.
# """
# # Initialize for this setup
# self.frame = frame
# # Find the calibration groups associated with this frame. Note
# # find_frame_calib_groups *always* returns a list. Science frames only
# # have one calibration group, but calibration frames can have many. So
# # for both science and calibration frames, we just set the calibration
# # group to the first group in the returned list.
# self.calib_ID = self.fitstbl.find_frame_calib_groups(self.frame)[0]
# self.det = det
# if par is not None:
# self.par = par
[docs]
def base_state(self, step:str, object):
if self.state is None:
return
# Created/loaded?
if object is None:
self.state.update_calib(step, self.calib_ID, self.det, 'status', 'undone')
return
# Success
self.state.update_calib(step, self.calib_ID, self.det, 'status', 'success')
self.state.update_calib(step, self.calib_ID, self.det,
'output_file', object.get_path())
# Input files
self.state.update_calib(step, self.calib_ID, self.det,
'input_files', self.raw_files)
return
[docs]
def arc_state(self):
self.base_state('arc', self.msarc)
[docs]
def get_arc(self, force:str=None):
"""
Load or generate the arc calibration frame.
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.images.buildimage.ArcImage`: The processed
calibration image.
"""
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Find the calibrations
frame = {'type': 'arc', 'class': buildimage.ArcImage}
self.raw_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
if len(self.raw_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
self.msarc = None
return self.msarc
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.msarc = self.process_load_selection(frame, cal_file, force)
if not self.success or self.msarc is not None:
return self.msarc
# Reset the BPM
self.get_bpm(frame=self.raw_files[0])
# Perform a check on the files
self.check_calibrations(self.raw_files)
# Otherwise, create the processed file.
log.info(f'Preparing a {frame["class"].calib_type} calibration frame.')
self.msarc = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['arcframe'], self.raw_files,
bias=self.msbias, bpm=self.msbpm,
dark=self.msdark, calib_dir=self.calib_dir,
setup=setup, calib_id=calib_id)
# Save the result
self.msarc.to_file()
# Return it
return self.msarc
[docs]
def tiltimg_state(self):
self.base_state('tiltimg', self.mstilt)
[docs]
def get_tiltimg(self, force:str=None):
"""
Load or generate the tilt calibration frame.
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.images.buildimage.TiltImage`: The processed
calibration image.
"""
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Find the calibrations
frame = {'type': 'tilt', 'class':buildimage.TiltImage}
self.raw_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
if len(self.raw_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
self.mstilt = None
return self.mstilt
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.mstilt = self.process_load_selection(frame, cal_file, force)
if not self.success or self.mstilt is not None:
return self.mstilt
# Reset the BPM
self.get_bpm(frame=self.raw_files[0])
# Perform a check on the files
self.check_calibrations(self.raw_files)
# Otherwise, create the processed file.
log.info(f'Preparing a {frame["class"].calib_type} calibration frame.')
self.mstilt = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['tiltframe'], self.raw_files,
bias=self.msbias, bpm=self.msbpm,
dark=self.msdark, slits=self.slits,
calib_dir=self.calib_dir, setup=setup,
calib_id=calib_id)
# Save the result
self.mstilt.to_file()
# Return it
return self.mstilt
[docs]
def get_align(self, force:str=None):
"""
Load or generate the alignment calibration frame.
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.alignframe.Alignments`: The processed alignment
image.
"""
# Check for existing data
if not self._chk_objs(['msbpm', 'slits']):
raise PypeItError('Must have the bpm and slits to make the alignments!')
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Find the calibrations
frame = {'type': 'align', 'class': alignframe.Alignments}
self.raw_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
if len(raw_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
self.alignments = None
return self.alignments
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.alignments = self.process_load_selection(frame, cal_file, force)
if not self.success:
return None
elif self.alignments is not None:
self.alignments.is_synced(self.slits)
return self.alignments
# Reset the BPM
self.get_bpm(frame=raw_files[0])
# Perform a check on the files
self.check_calibrations(raw_files)
# Otherwise, create the processed file.
log.info(f'Preparing a {frame["class"].calib_type} calibration frame.')
msalign = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['alignframe'], raw_files,
bias=self.msbias, bpm=self.msbpm,
dark=self.msdark, calib_dir=self.calib_dir,
setup=setup, calib_id=calib_id)
# Instantiate
# TODO: From JFH: Do we need the bpm here? Check that this was in the previous code.
alignment = alignframe.TraceAlignment(msalign, self.slits, self.spectrograph,
self.par['alignment'], det=self.det,
qa_path=self.qa_path, msbpm=self.msbpm)
self.alignments = alignment.run(show=self.show)
# NOTE: The alignment object inherets the calibration frame naming from
# the msalign image.
self.alignments.to_file()
return self.alignments
[docs]
def bias_state(self):#, outfile:str):
# Base
self.base_state('bias', self.msbias)
# Metrics
if self.msbias is None:
return
self.state.update_calib('bias', self.calib_ID, self.det, 'mean', self.msbias.image.mean())
self.state.update_calib('bias', self.calib_ID, self.det, 'std', self.msbias.image.std())
[docs]
def get_bias(self, force:str=None):
"""
Load or generate the bias calibration frame.
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.images.buildimage.BiasImage`: The processed
calibration image.
"""
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Find the calibrations
frame = {'type': 'bias', 'class': buildimage.BiasImage}
self.raw_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
# If no raw files are available and no processed calibration frame
if len(self.raw_files) == 0 and cal_file is None:
log.warning(
f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing without a bias...'
)
self.msbias = None
return self.msbias
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.msbias = self.process_load_selection(frame, cal_file, force)
if not self.success or self.msbias is not None:
return self.msbias
# Perform a check on the files
self.check_calibrations(self.raw_files)
# Otherwise, create the processed file.
log.info(f'Preparing a {frame["class"].calib_type} calibration frame.')
self.msbias = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['biasframe'], self.raw_files,
calib_dir=self.calib_dir, setup=setup,
calib_id=calib_id)
# Save the result
self.msbias.to_file()
# State
#self.bias_state(self.msbias.get_path())
# Return it
return self.msbias
[docs]
def dark_state(self):#, outfile:str):
# Base
self.base_state('dark', self.msdark)
# Metrics
if self.msdark is None:
return
self.state.update_calib('dark', self.calib_ID, self.det, 'mean', self.msdark.image.mean())
self.state.update_calib('dark', self.calib_ID, self.det, 'std', self.msdark.image.std())
[docs]
def get_dark(self, force:str=None):
"""
Load or generate the dark calibration frame.
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.images.buildimage.DarkImage`: The processed
calibration image.
"""
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Find the calibrations
frame = {'type': 'dark', 'class': buildimage.DarkImage}
self.raw_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
if len(self.raw_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
self.msdark = None
return self.msdark
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.msdark = self.process_load_selection(frame, cal_file, force)
if not self.success or self.msdark is not None:
return self.msdark
# TODO: If a bias has been constructed and it will be subtracted from
# the science images, it should also be subtracted from this image. If
# it isn't, subtracting the dark will effectively lead to subtracting
# the bias twice.
# TODO: The order is such that the bpm doesn't exist yet. But calling
# buildimage_fromlist will create the bpm if it isn't passed. So
# calling get_dark then get_bpm unnecessarily creates the bpm twice. Is
# there any reason why creation of the bpm should come after the dark,
# or can we change the order?
# Perform a check on the files
self.check_calibrations(raw_files)
# Otherwise, create the processed file.
self.msdark = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['darkframe'], raw_files,
bias=self.msbias, calib_dir=self.calib_dir,
setup=setup, calib_id=calib_id)
# Save the result
self.msdark.to_file()
# Return it
return self.msdark
[docs]
def bpm_state(self):
pass
[docs]
def get_bpm(self, frame=None, force:str=None):
"""
Load or generate the bad pixel mask.
This is primarily a wrapper for
:func:`~pypeit.spectrographs.spectrograph.Spectrograph.bpm`.
Args:
force (:obj:`str`, optional):
Currently ignored
frame (:obj:`int`, optional):
The row index in :attr:`fitstbl`
Returns:
`numpy.ndarray`_: The bad pixel mask, which should match the shape
and orientation of a *trimmed* and PypeIt-oriented science image!
"""
# Check internals
self._chk_set(['par', 'det'])
# Set the frame to use for the BPM
if frame is None:
frame = self.fitstbl.frame_paths(self.frame)
# Build it
self.msbpm = self.spectrograph.bpm(frame, self.det,
msbias=self.msbias if self.par['bpm_usebias'] else None)
# Return
return self.msbpm
[docs]
def scattlight_state(self):
self.base_state('scattlight', self.msscattlight)
[docs]
def get_scattlight(self, force:str=None):
"""
Load or generate the scattered light model.
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.scattlight.ScatteredLight`: The processed calibration image including the model.
"""
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Prep
frame = {'type': 'scattlight', 'class': scattlight.ScatteredLight}
self.raw_files, cal_file, calib_key, setup, calib_id, detname = \
self.find_calibrations(frame['type'], frame['class'])
scatt_idx = self.fitstbl.find_frames(frame['type'], calib_ID=self.calib_ID, index=True)
if len(self.raw_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
return self.msscattlight
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.msscattlight = self.process_load_selection(frame, cal_file, force)
if not self.success or self.msscattlight is not None:
return self.msscattlight
# Check for existing data to generate the scattered light image
if not self._chk_objs(['msbpm', 'slits']):
log.warning('Must have the bpm and the slits defined to make a scattered light image! '
'Skipping and may crash down the line')
return self.msscattlight
# Scattered light model does not exist or we're not reusing it.
# Need to build everything from scratch. Start with the trace image.
log.info('Creating scattered light calibration frame using files: ')
for f in self.raw_files:
log.info(f' {Path(f).name}')
# Reset the BPM
self.get_bpm(frame=self.raw_files[0])
# Perform a check on the files
self.check_calibrations(raw_scattlight_files)
binning = self.fitstbl[scatt_idx[0]]['binning']
dispname = self.fitstbl[scatt_idx[0]]['dispname']
scattlightImage = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['scattlightframe'], self.raw_files,
bias=self.msbias, bpm=self.msbpm,
dark=self.msdark, calib_dir=self.calib_dir,
setup=setup, calib_id=calib_id)
spatbin = parse.parse_binning(binning)[1]
pad = self.par['scattlight_pad'] // spatbin
offslitmask = self.slits.slit_img(pad=pad, flexure=None) == -1
# Get starting parameters for the scattered light model
x0, bounds = self.spectrograph.scattered_light_archive(binning, dispname)
# Perform a fit to the scattered light
model, modelpar, success = core_scattlight.scattered_light(scattlightImage.image, self.msbpm, offslitmask,
x0, bounds)
if not success:
# Something went awry
log.warning('Scattered light modelling failed. Continuing, but likely to fail soon...')
self.success = False
return self.msscattlight
# Now generate the DataModel
self.msscattlight = scattlight.ScatteredLight(PYP_SPEC=self.spectrograph.name,
pypeline=self.spectrograph.pypeline,
detname=scattlightImage.detector.name,
nspec=scattlightImage.shape[0], nspat=scattlightImage.shape[1],
binning=scattlightImage.detector.binning,
pad=self.par['scattlight_pad'],
scattlight_raw=scattlightImage.image,
scattlight_model=model,
scattlight_param=modelpar)
# TODO :: Should we go back and recalculate the slit edges once the scattered light is known?
if self.msscattlight is not None:
# Show the result if requested
if self.show:
self.msscattlight.show()
# Save the master scattered light model
self.msscattlight.set_paths(self.calib_dir, setup, calib_id, detname)
self.msscattlight.to_file()
return self.msscattlight
[docs]
def _flat_correction_images(self):
"""
Reconstruct the applied flat-field correction images present in the
current ``FlatImages`` product.
The set mirrors what ``FlatImages.show()`` displays: pixel-to-pixel
response, slit (spatial) illumination, and spectral illumination.
Each image hovers about 1.0, so its mean/scatter characterize the
correction.
Generated by JXP and Claude.
Returns:
:obj:`dict`: Mapping of correction name (``'pixelflat'``,
``'spat_illum'``, ``'spec_illum'``) to its full-detector image
(`numpy.ndarray`_). Empty if there are no flats or no slits.
"""
fi = self.flatimages
if fi is None or self.slits is None:
return {}
imgs = {}
# Pixel-to-pixel response
if fi.pixelflat_norm is not None:
imgs['pixelflat'] = fi.pixelflat_norm
# Slit (spatial) illumination: prefer the dedicated illumflat
# bsplines, else use the pixelflat's (illumination can come from the
# pixelflat frames), matching fit2illumflat / merge precedence.
if fi.illumflat_spat_bsplines is not None \
or fi.pixelflat_spat_bsplines is not None:
ft = 'illum' if fi.illumflat_spat_bsplines is not None else 'pixel'
imgs['spat_illum'] = fi.fit2illumflat(self.slits, frametype=ft)
# Spectral illumination
if fi.pixelflat_spec_illum is not None:
imgs['spec_illum'] = fi.pixelflat_spec_illum
return imgs
[docs]
def _flat_qa_files(self):
"""
Find the flat-field QA PNGs associated with this calibration.
Generated by JXP and Claude.
Returns:
:obj:`list`: Sorted list of QA PNG paths (as strings) for this
flat's calibration key. Empty if QA is disabled or none found.
"""
if self.qa_path is None or self.flatimages is None:
return []
key = getattr(self.flatimages, 'calib_key', None)
png_dir = Path(self.qa_path) / 'PNGs'
if key is None or not png_dir.exists():
return []
# Flat QA file prefixes (per-slit spatial illumination + detector
# structure); match those that carry this calibration's key.
prefixes = ('Spatillum_FineCorr', 'DetectorStructure')
return sorted(str(p) for p in png_dir.glob('*.png')
if key in p.name and p.name.startswith(prefixes))
[docs]
def flats_state(self):
"""
Record the flat-field state: status, output/input files, pixel-flat
provenance, which corrections are present, the flat QA files, and a
per-slit status with per-correction mean/RMS metrics.
Generated by JXP and Claude.
"""
if self.state is None:
return
cid, det = self.calib_ID, self.det
if self.flatimages is None:
self.state.update_calib('flats', cid, det, 'status', 'undone')
return
# Status + output file
self.state.update_calib('flats', cid, det, 'status', 'success')
# Get the output path defensively: a FlatImages merged from a
# user/archived pixelflat may lack calib_key, so get_path() can raise;
# never let that abort the rest of the state recording.
try:
out_path = self.flatimages.get_path()
except Exception as e:
log.warning(f"Could not determine flat output path for state: {e}")
out_path = None
if out_path is not None:
self.state.update_calib('flats', cid, det, 'output_file', out_path)
# Input files: grouped by role plus a de-duplicated union for the
# generic input list. (Stashed by get_flats.)
inputs = getattr(self, '_flat_input_files', None) or {}
pix = inputs.get('pixelflat') or []
ill = inputs.get('illumflat') or []
lamp = inputs.get('lampoff') or []
if pix:
self.state.update_calib('flats', cid, det, 'pixelflat_files', pix)
if ill:
self.state.update_calib('flats', cid, det, 'illumflat_files', ill)
if lamp:
self.state.update_calib('flats', cid, det, 'lampoff_files', lamp)
union = list(dict.fromkeys([*pix, *ill, *lamp]))
if union:
self.state.update_calib('flats', cid, det, 'input_files', union)
# Pixel-flat provenance
source = getattr(self, '_pixelflat_source', None)
if source is not None:
self.state.update_calib('flats', cid, det, 'pixelflat_source',
source)
# Which corrections are present + per-slit metrics
corr_imgs = self._flat_correction_images()
if corr_imgs:
self.state.update_calib('flats', cid, det, 'corrections',
list(corr_imgs.keys()))
# QA files
qa_files = self._flat_qa_files()
if qa_files:
self.state.update_calib('flats', cid, det, 'qa_files', qa_files)
# Per-slit status + per-correction mean/RMS
if self.slits is None:
return
slitid_img = self.slits.slit_img()
for islit, spat in enumerate(self.slits.spat_id):
spat = int(spat)
# Status from the slit bitmask (mirrors wv_calib/tilts)
if self.slits.bitmask.flagged(self.slits.mask[islit],
flag='BADFLATCALIB'):
slit_status = 'fail'
elif self.slits.bitmask.flagged(self.slits.mask[islit],
flag='SKIPFLATCALIB'):
slit_status = 'skip'
else:
slit_status = 'success'
self.state.update_calib('flats', cid, det, 'status', slit_status,
slit=spat)
# Per-correction mean/RMS over this slit's good, on-slit pixels
onslit = slitid_img == spat
metrics = {}
for name, img in corr_imgs.items():
good = onslit & np.isfinite(img)
if not np.any(good):
continue
metrics[name] = state.FlatCorrectionMetric(
mean=float(np.mean(img[good])),
rms=float(np.std(img[good])))
if metrics:
self.state.update_calib('flats', cid, det, 'corrections',
metrics, slit=spat)
[docs]
def get_flats(self, force:str=None):
"""
Load or generate the flat-field calibration images.
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.flatfield.FlatImages`: The processed calibration
image.
"""
# Check for existing data
if not self._chk_objs(['msarc', 'msbpm', 'slits', 'wv_calib']):
log.warning('Must have the arc, bpm, slits, and wv_calib defined to make flats! '
'Skipping and may crash down the line')
# TODO: Why was this an empty object and not None?
self.flatimages = None #flatfield.FlatImages()
return self.flatimages
# Slit and tilt traces are required to flat-field the data
if not self._chk_objs(['slits', 'wavetilts']):
# TODO: Why doesn't this fault?
log.warning('Flats were requested, but there are quantities missing necessary to '
'create flats. Proceeding without flat fielding....')
# TODO: Why was this an empty object and not None?
self.flatimages = None #flatfield.FlatImages()
return self.flatimages
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# State: capture whether the user supplied a pixel-flat file *before*
# the slitless step (below) overwrites the parameter, so flats_state
# can record the pixel-flat provenance. (Generated by JXP and Claude)
user_pixelflat = self.par['flatfield']['pixelflat_file'] is not None
# generate the slitless pixel flat (if frames available).
slitless_rows = self.fitstbl.find_frames('slitless_pixflat', calib_ID=self.calib_ID, index=True)
if len(slitless_rows) > 0:
sflat = flatfield.SlitlessFlat(self.fitstbl, slitless_rows, self.spectrograph,
self.par, qa_path=self.qa_path)
# A pixel flat will be saved to disc and self.par['flatfield']['pixelflat_file'] will be updated
self.par['flatfield']['pixelflat_file'] = \
sflat.make_slitless_pixflat(msbias=self.msbias, msdark=self.msdark, calib_dir=self.calib_dir,
write_qa=self.write_qa, show=self.show)
# get illumination flat frames
illum_frame = {'type': 'illumflat', 'class': flatfield.FlatImages}
raw_illum_files, illum_cal_file, illum_calib_key, illum_setup, illum_calib_id, detname \
= self.find_calibrations(illum_frame['type'], illum_frame['class'])
# get pixel flat frames
pixel_frame = {'type': 'pixelflat', 'class': flatfield.FlatImages}
raw_pixel_files, pixel_cal_file, pixel_calib_key, pixel_setup, pixel_calib_id, detname \
= [], None, None, illum_setup, None, detname
# read in the raw pixelflat frames only if the user has not provided a pixelflat_file
if self.par['flatfield']['pixelflat_file'] is None:
raw_pixel_files, pixel_cal_file, pixel_calib_key, pixel_setup, pixel_calib_id, detname \
= self.find_calibrations(pixel_frame['type'], pixel_frame['class'])
# get lamp off flat frames
raw_lampoff_files = self.fitstbl.find_frame_files('lampoffflats', calib_ID=self.calib_ID)
# State: stash the raw input groups and the pixel-flat provenance for
# flats_state to record. Done here (before the reuse/no-frames
# branches below) so the inputs are captured on every path.
# (Generated by JXP and Claude)
self._flat_input_files = {'pixelflat': list(raw_pixel_files),
'illumflat': list(raw_illum_files),
'lampoff': list(raw_lampoff_files)}
if len(slitless_rows) > 0:
self._pixelflat_source = 'slitless'
elif user_pixelflat:
self._pixelflat_source = 'user_file'
elif len(raw_pixel_files) > 0:
self._pixelflat_source = 'raw'
else:
self._pixelflat_source = None
# Check if we have any calibration frames to work with
if len(raw_pixel_files) == 0 and pixel_cal_file is None \
and len(raw_illum_files) == 0 and illum_cal_file is None:
# if no calibration frames are found, check if the user has provided a pixel flat file
if self.par['flatfield']['pixelflat_file'] is not None:
log.warning(f'No raw {pixel_frame["type"]} or {illum_frame["type"]} frames found but a '
'user-defined pixel flat file was provided. Using that file.')
self.flatimages = flatfield.FlatImages(PYP_SPEC=self.spectrograph.name, spat_id=self.slits.spat_id)
self.flatimages.calib_key = flatfield.FlatImages.construct_calib_key(self.fitstbl['setup'][self.frame],
self.calib_ID, detname)
self.flatimages = flatfield.load_pixflat(self.par['flatfield']['pixelflat_file'], self.spectrograph,
self.det, self.flatimages, calib_dir=self.calib_dir,
chk_version=self.chk_version)
else:
log.warning(f'No raw {pixel_frame["type"]} or {illum_frame["type"]} frames found and '
'unable to identify a relevant processed calibration frame. Continuing...')
self.flatimages = None
return self.flatimages
# If a processed calibration frame exists and we want to reuse it, do
# so. The processed pixel_flat takes precedence, and a warning is
# issued if both are present and not the same.
if illum_cal_file is not None and pixel_cal_file is not None \
and pixel_cal_file != illum_cal_file:
log.warning('Processed calibration frames were found for both pixel and '
'slit-illumination flats, and the files are not the same. Ignoring the '
'slit-illumination flat.')
cal_file = illum_cal_file if pixel_cal_file is None else pixel_cal_file
calib_key = illum_calib_key if pixel_calib_key is None else pixel_calib_key
setup = illum_setup if pixel_setup is None else pixel_setup
calib_id = illum_calib_id if pixel_calib_id is None else pixel_calib_id
if cal_file.exists() and self.reuse_calibs and not force == 'remake':
self.flatimages = flatfield.FlatImages.from_file(cal_file,
chk_version=self.chk_version)
self.flatimages.is_synced(self.slits)
# Load user defined files
if self.par['flatfield']['pixelflat_file'] is not None:
# Load
self.flatimages = flatfield.load_pixflat(self.par['flatfield']['pixelflat_file'], self.spectrograph,
self.det, self.flatimages, calib_dir=self.calib_dir,
chk_version=self.chk_version)
# update slits
self.slits.mask_flats(self.flatimages)
return self.flatimages
# Generate the image(s) from scratch
pixelflatImages, illumflatImages = None, None
lampoff_flat = None
# Check if the image files are the same
pix_is_illum = Counter(raw_illum_files) == Counter(raw_pixel_files)
if len(raw_pixel_files) > 0:
# Reset the BPM
self.get_bpm(frame=raw_pixel_files[0])
# Perform a check on the files
self.check_calibrations(raw_pixel_files)
log.info('Creating pixel-flat calibration frame using files: ')
for f in raw_pixel_files:
log.info(f' {Path(f).name}')
pixel_flat = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['pixelflatframe'],
raw_pixel_files, dark=self.msdark,
slits=self.slits,
bias=self.msbias, bpm=self.msbpm,
scattlight=self.msscattlight)
if len(raw_lampoff_files) > 0:
# Reset the BPM
self.get_bpm(frame=raw_lampoff_files[0])
# Perform a check on the files
self.check_calibrations(raw_lampoff_files)
log.info('Subtracting lamp off flats using files: ')
for f in raw_lampoff_files:
log.info(f' {Path(f).name}')
lampoff_flat = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['lampoffflatsframe'],
raw_lampoff_files,
slits=self.slits,
dark=self.msdark, bias=self.msbias,
bpm=self.msbpm, scattlight=self.msscattlight)
pixel_flat = pixel_flat.sub(lampoff_flat)
# Initialise the pixel flat
pixelFlatField = flatfield.FlatField(pixel_flat, self.spectrograph,
self.par['flatfield'], self.slits, wavetilts=self.wavetilts,
wv_calib=self.wv_calib, qa_path=self.qa_path,
calib_key=calib_key)
# Generate
pixelflatImages = pixelFlatField.run(doqa=self.write_qa, show=self.show)
# Set flatimages in case we want to apply the pixel-to-pixel
# sensitivity corrections to the illumflat
self.flatimages = pixelflatImages
# State is recorded later, in flats_state(), from the merged
# FlatImages product and the slit bitmask.
# Only build illum_flat if the input files are different from the pixel flat
if not pix_is_illum and len(raw_illum_files) > 0:
# Reset the BPM
self.get_bpm(frame=raw_illum_files[0])
# Perform a check on the files
self.check_calibrations(raw_illum_files)
log.info('Creating slit-illumination flat calibration frame using files: ')
for f in raw_illum_files:
log.info(f' {Path(f).name}')
illum_flat = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['illumflatframe'], raw_illum_files,
dark=self.msdark, bias=self.msbias, scattlight=self.msscattlight,
slits=self.slits, flatimages=self.flatimages, bpm=self.msbpm)
if len(raw_lampoff_files) > 0:
log.info('Subtracting lamp off flats using files: ')
for f in raw_lampoff_files:
log.info(f' {Path(f).name}')
if lampoff_flat is None:
# Perform a check on the files
self.check_calibrations(raw_lampoff_files)
# Build the image
lampoff_flat = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['lampoffflatsframe'],
raw_lampoff_files,
dark=self.msdark,
bias=self.msbias,
slits=self.slits,
scattlight=self.msscattlight,
bpm=self.msbpm)
illum_flat = illum_flat.sub(lampoff_flat)
# Initialise the illum flat
illumFlatField = flatfield.FlatField(illum_flat, self.spectrograph,
self.par['flatfield'], self.slits, wavetilts=self.wavetilts,
wv_calib=self.wv_calib, spat_illum_only=True,
qa_path=self.qa_path, calib_key=calib_key)
# Generate
illumflatImages = illumFlatField.run(doqa=self.write_qa, show=self.show)
# Merge the illum flat with the pixel flat
if pixelflatImages is not None:
# Combine the pixelflat and illumflat parameters into flatimages.
# This will merge the attributes of pixelflatImages that are not None
# with the attributes of illflatImages that are not None. Default is
# to take pixelflatImages.
self.flatimages = flatfield.merge(pixelflatImages, illumflatImages)
else:
# No pixel flat, but there might be an illumflat. This will mean that
# the attributes prefixed with 'pixelflat_' will all be None.
self.flatimages = illumflatImages
if self.flatimages is not None:
self.flatimages.set_paths(self.calib_dir, setup, calib_id, detname)
# Save flat images
self.flatimages.to_file()
# Save slits too, in case they were tweaked
self.slits.to_file()
# State (output_file etc.) is recorded later in flats_state().
# Apply user-supplied images
# NOTE: These are the *final* images, not just a stack, and it will
# over-ride what is generated below (if generated).
# TODO: Why is this done after writing the image above? If we instead
# wrote the file after applying this user-defined pixelflat, we wouldn't
# need to re-read the user-provided file when ingesting the existing
# flat file. Is this to allow the user to change the pixel flat file?
# Should we allow that?
if self.par['flatfield']['pixelflat_file'] is not None:
# Load
self.flatimages = flatfield.load_pixflat(self.par['flatfield']['pixelflat_file'], self.spectrograph,
self.det, self.flatimages, calib_dir=self.calib_dir,
chk_version=self.chk_version)
return self.flatimages
[docs]
def _slits_qa_files(self):
"""
Find the slit/order QA PNGs for this calibration — chiefly the
echelle order-prediction figure (``Edges_*_orders_qa``).
Generated by JXP and Claude.
Returns:
:obj:`list`: Sorted list of QA PNG paths (as strings) starting
with ``Edges`` and carrying this calibration's key; empty if QA
is disabled or none found.
"""
if self.qa_path is None or self.slits is None:
return []
key = getattr(self.slits, 'calib_key', None)
png_dir = Path(self.qa_path) / 'PNGs'
if key is None or not png_dir.exists():
return []
return sorted(str(p) for p in png_dir.glob('*.png')
if key in p.name and p.name.startswith('Edges'))
[docs]
def slits_state(self):
# Base
self.base_state('slits', self.slits)
if self.slits is None:
return
self.state.update_calib('slits', self.calib_ID, self.det,
'nslits', self.slits.nslits)
# Slit/order QA (e.g. the echelle order-prediction figure; Stage 3b).
qa_files = self._slits_qa_files()
if qa_files:
self.state.update_calib('slits', self.calib_ID, self.det,
'qa_files', qa_files)
for islit in range(self.slits.nslits):
slit_ID = int(self.slits.slitord_id[islit])
self.state.update_calib('slits', self.calib_ID, self.det,
'center',
np.mean(self.slits.center[:,islit]),
slit=slit_ID)
self.state.update_calib('slits', self.calib_ID, self.det,
'status', 'success', slit=slit_ID)
[docs]
def get_slits(self, force:str=None):
"""
Load or generate the definition of the slit boundaries.
Returns:
:class:`~pypeit.slittrace.SlitTraceSet`: Traces of the
slit edges; also kept internally as :attr:`slits`.
"""
# Check for existing data
if not self._chk_objs(['msbpm']):
return None
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Prep
frame = {'type': 'trace', 'class': slittrace.SlitTraceSet}
raw_trace_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
raw_lampoff_files = self.fitstbl.find_frame_files('lampoffflats', calib_ID=self.calib_ID)
if len(raw_trace_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
self.slits = None
return self.slits
# Record the raw trace frames now, so slits_state() reports the input
# files even when the slits are reused/loaded below (the from-scratch
# build path further down also sets this). Generated by JXP and Claude.
self.raw_files = raw_trace_files
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.slits = self.process_load_selection(frame, cal_file, force)
if not self.success:
return None
elif self.slits is not None:
self.slits.mask = self.slits.mask_init.copy()
if self.user_slits is not None:
self.slits.user_mask(detname, self.user_slits)
return self.slits
# Slits don't exist or we're not resusing them. See if the Edges
# calibration frame exists.
edges_file = Path(edgetrace.EdgeTraceSet.construct_file_name(calib_key,
calib_dir=self.calib_dir)).absolute()
# If so, reuse it?
if edges_file.exists() and self.reuse_calibs and force != 'remake':
# Yep! Load it and parse it into slits.
self.slits = edgetrace.EdgeTraceSet.from_file(edges_file,
chk_version=self.chk_version).get_slits()
# Write the slits calibration file
self.slits.to_file()
if self.user_slits is not None:
self.slits.user_mask(detname, self.user_slits)
return self.slits
# Need to build everything from scratch. Start with the trace image.
log.info('Creating edge tracing calibration frame using files: ')
for f in raw_trace_files:
log.info(f' {Path(f).name}')
self.raw_files = raw_trace_files
# Reset the BPM
self.get_bpm(frame=raw_trace_files[0])
# Perform a check on the files
self.check_calibrations(raw_trace_files)
# NOTE: self.msscattlight is *always* created after identifying the
# slits, meaning that it is redundant to pass the scattlight argument
# here.
traceImage = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['traceframe'], raw_trace_files,
bias=self.msbias, bpm=self.msbpm,
dark=self.msdark, calib_dir=self.calib_dir,
setup=setup, calib_id=calib_id)
if len(raw_lampoff_files) > 0:
log.info('Subtracting lamp off flats using files: ')
for f in raw_lampoff_files:
log.info(f' {Path(f).name}')
# Reset the BPM
self.get_bpm(frame=raw_trace_files[0])
# Perform a check on the files
self.check_calibrations(raw_lampoff_files)
lampoff_flat = buildimage.buildimage_fromlist(self.spectrograph, self.det,
self.par['lampoffflatsframe'],
raw_lampoff_files, dark=self.msdark,
bias=self.msbias, bpm=self.msbpm)
traceImage = traceImage.sub(lampoff_flat)
edges = edgetrace.EdgeTraceSet(traceImage, self.spectrograph, self.par['slitedges'],
qa_path=self.qa_path, auto=True)
if not edges.success:
# Something went amiss
log.warning('Edge tracing failed. Continuing, but likely to fail soon...')
traceImage = None
edges = None
self.success = False
self.slits = None
return self.slits
# Save the result
edges.to_file()
# Show the result if requested
if self.show:
edges.show(in_ginga=True)
# Get the slits from the result of the edge tracing, delete
# the edges object, and save the slits, if requested
self.slits = edges.get_slits()
traceImage = None
edges = None
self.slits.to_file()
# State
#self.slits_state(self.slits.get_path())
if self.user_slits is not None:
self.slits.user_mask(detname, self.user_slits)
return self.slits
[docs]
def _wv_calib_qa_files(self):
"""
Find the wavelength-calibration QA PNGs for this calibration.
Generated by JXP and Claude.
Returns:
:obj:`list`: Sorted list of QA PNG paths (as strings) — the
per-slit ``Arc_1dfit_*`` / ``Arc_FWHMfit_*`` figures carrying this
calibration's key. Empty if QA is disabled or none found.
"""
if self.qa_path is None or self.wv_calib is None:
return []
key = getattr(self.wv_calib, 'calib_key', None)
png_dir = Path(self.qa_path) / 'PNGs'
if key is None or not png_dir.exists():
return []
# The wavelength 1D fit + FWHM figures, the echelle 2D-fit figures
# (Arc_2dfit_global / Arc_2dfit_orders), and the arc-line tilt figures
# (the user groups Arc_tilts_* under wv_calib too; Round-2 #7, 3b).
prefixes = ('Arc_1dfit', 'Arc_FWHMfit', 'Arc_2dfit', 'Arc_tilts')
return sorted(str(p) for p in png_dir.glob('*.png')
if key in p.name and p.name.startswith(prefixes))
[docs]
def wv_calib_state(self):
self.base_state('wv_calib', self.wv_calib)
if self.wv_calib is None or self.slits is None:
return
# Record the wavelength QA PNGs for the dashboard (C9).
qa_files = self._wv_calib_qa_files()
if qa_files:
self.state.update_calib('wv_calib', self.calib_ID, self.det,
'qa_files', qa_files)
for islit in range(self.slits.nslits):
slit_ID = int(self.slits.slitord_id[islit])
# Status
if self.slits.bitmask.flagged(
self.slits.mask[islit], flag='BADWVCALIB'):
status = 'fail'
else:
status = 'success'
self.state.update_calib('wv_calib', self.calib_ID, self.det,
'status', status, slit=slit_ID)
# Metrics
if status == 'success':
self.state.update_calib('wv_calib', self.calib_ID, self.det,
'rms', self.wv_calib.wv_fits[islit].rms,
slit=slit_ID)
[docs]
def get_wv_calib(self, force:str=None):
"""
Load or generate the 1D wavelength calibrations
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.wavecalib.WaveCalib`: Object containing wavelength
calibrations and the updated slit mask array.
"""
# No wavelength calibration requested
if self.par['wavelengths']['reference'] == 'pixel':
log.info('Wavelength "reference" parameter set to "pixel"; no wavelength '
'calibration will be performed.')
self.wv_calib = None
return self.wv_calib
# Check for existing data
req_objs = ['msarc', 'msbpm', 'slits']
if not self._chk_objs(req_objs):
log.warning('Not enough information to load/generate the wavelength calibration. '
'Skipping and may crash down the line')
return None
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Find the calibrations
frame = {'type': 'arc', 'class': wavecalib.WaveCalib}
self.raw_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
if len(self.raw_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
self.wv_calib = None
return self.wv_calib
# If a processed calibration frame exists and
# we want to reuse it, do so (or just load it):
self.wv_calib = self.process_load_selection(frame, cal_file, force)
if not self.success:
return None
elif self.wv_calib is not None:
self.wv_calib.chk_synced(self.slits)
self.slits.mask_wvcalib(self.wv_calib)
if self.par['wavelengths']['method'] == 'echelle':
log.info('Method set to Echelle -- checking wv_calib for 2dfits')
if not hasattr(self.wv_calib, 'wv_fit2d'):
raise PypeItError('There is no 2d fit in this Echelle wavelength '
'calibration! Please generate a new one with a 2d fit.')
# Return
if self.par['wavelengths']['redo_slits'] is None:
#self.wv_calib_state()
return self.wv_calib
# Determine lamp list to use for wavecalib
# Find all the arc frames in this calibration group
is_arc = self.fitstbl.find_frames('arc', calib_ID=self.calib_ID)
lamps = self.spectrograph.get_lamps(self.fitstbl[is_arc]) \
if self.par['wavelengths']['lamps'] == ['use_header'] \
else self.par['wavelengths']['lamps']
meta_dict = dict(self.fitstbl[is_arc][0]) \
if self.spectrograph.pypeline == 'Echelle' \
and not self.spectrograph.ech_fixed_format else None
# Instantiate
# TODO: Pull out and pass only the necessary parts of meta_dict to
# this, or include the relevant parts as parameters. See comments
# in PRs #1454 and #1476 on this.
# TODO: (Added 30 Mar 2023) The need for the meta_dict is for echelle
# wavelength calibration. Create EchelleCalibrations and
# EchelleBuildWaveCalib subclasses instead..
log.info(f'Preparing a {wavecalib.WaveCalib.calib_type} calibration frame.')
waveCalib = wavecalib.BuildWaveCalib(self.msarc, self.slits, self.spectrograph,
self.par['wavelengths'], lamps, meta_dict=meta_dict,
det=self.det, qa_path=self.qa_path)
self.wv_calib = waveCalib.run(skip_QA=(not self.write_qa),
prev_wvcalib=self.wv_calib)
# If orders were found, save slits to disk
# or if redo_slits
if (self.par['wavelengths']['redo_slits'] is not None) or (
self.spectrograph.pypeline == 'Echelle' and not self.spectrograph.ech_fixed_format):
self.slits.to_file()
# Save calibration frame
self.wv_calib.to_file()
# State
#self.wvcalib_state(self.wv_calib.get_path())
# Return
return self.wv_calib
[docs]
def tilts_state(self):#, buildTilts, outfile:str):
self.base_state('tilts', self.wavetilts)
# Update
if self.wavetilts is None or self.slits is None:
return
for islit in range(self.slits.nslits):
slit_ID = int(self.slits.slitord_id[islit])
# Status
if self.slits.bitmask.flagged(
self.slits.mask[islit], flag='BADTILTCALIB'):
status = 'fail'
else:
status = 'success'
self.state.update_calib('tilts', self.calib_ID, self.det,
'status', status, slit=slit_ID)
# Metrics
if self.buildwaveTilts is not None and status == 'success':
rms = self.buildwaveTilts.all_fit_dict[islit]['pypeitFit'].calc_fit_rms(
x2=self.buildwaveTilts.all_fit_dict[islit]['pypeitFit'].x2)
self.state.update_calib(
'tilts', self.calib_ID, self.det,
'rms', rms, slit=slit_ID)
[docs]
def get_tilts(self, force:str=None):
"""
Load or generate the wavelength tilts calibration frame
Args:
force (:obj:`str`, optional):
'remake' -- Force the frame to be remade.
'reload' -- Reload the frame if it exists.
None -- Load the existing frame if it exists and reuse_calibs=True
Returns:
:class:`~pypeit.wavetilts.WaveTilts`: Object containing the
wavelength tilt calibration.
"""
# Check for existing data
# TODO: add mstilt_inmask to this list when it gets implemented.
if not self._chk_objs(['mstilt', 'msbpm', 'slits', 'wv_calib']):
log.warning('Do not have all the necessary objects for tilts. Skipping and may crash '
'down the line.')
return None
# Check internals
self._chk_set(['det', 'calib_ID', 'par'])
# Find the calibrations
frame = {'type': 'tilt', 'class': wavetilts.WaveTilts}
self.raw_files, cal_file, calib_key, setup, calib_id, detname \
= self.find_calibrations(frame['type'], frame['class'])
if len(self.raw_files) == 0 and cal_file is None:
log.warning(f'No raw {frame["type"]} frames found and unable to identify a relevant '
'processed calibration frame. Continuing...')
self.wavetilts = None
return self.wavetilts
# If a processed calibration frame exists and we want to reuse it, do
# so:
self.wavetilts = self.process_load_selection(frame, cal_file, force)
if not self.success:
return None
elif self.wavetilts is not None:
self.wavetilts.is_synced(self.slits)
self.slits.mask_wavetilts(self.wavetilts)
return self.wavetilts
# Get flexure
_spat_flexure = self.mstilt.spat_flexure \
if self.par['tiltframe']['process']['spat_flexure_correct'] else None
# get measured fwhm from wv_calib
measured_fwhms = [wvfit.fwhm for wvfit in self.wv_calib.wv_fits]
# Build
self.buildwaveTilts = wavetilts.BuildWaveTilts(
self.mstilt, self.slits, self.spectrograph, self.par['tilts'],
self.par['wavelengths'], det=self.det, qa_path=self.qa_path,
spat_flexure=_spat_flexure, measured_fwhms=measured_fwhms)
# Write
self.wavetilts = self.buildwaveTilts.run(doqa=self.write_qa, show=self.show)
self.wavetilts.to_file()
# State
#self.tilts_state(buildwaveTilts, self.wavetilts.get_path())
return self.wavetilts
[docs]
def process_load_selection(self, frame, cal_file, force):
"""
Process how pypeit should use any pre-existing calibration files.
If loading is requested but the calibration file (``cal_file``) does
not exist, ``self.success`` is set to False, and None is returned.
Parameters
----------
frame : :obj:`dict`
A dictionary with two elements: ``type`` is the string
defining the frame type and ``class`` is the pypeit class
used to load the pre-existing calibration file.
cal_file : :obj:`str`, `Path`_
Path to the calibration file.
force : :obj:`str`
Defines how to treat a pre-existing calibration file. Must be one
of the following options:
- ``'remake'``: Force the calibration be remade.
- ``'reload'``: Reload the frame if it exists.
- ``None``: Load the existing frame if it exists and
``self.reuse_calibs=True``.
Returns
-------
:obj:`object`
Either the loaded calibration object or None.
"""
if force not in [None, 'remake', 'reload']:
raise PypeItError(f'`force` keyword must be None, remake, or reload, not {force}')
if force == 'remake':
return None
_cal_file = Path(cal_file).absolute()
if force == 'reload' and not _cal_file.exists():
log.warning(f"{_cal_file} does not exist; cannot reload "
f"{frame['class'].__name__} calibration.")
self.success = False
return None
if force == 'reload' or (self.reuse_calibs and _cal_file.exists()):
return frame['class'].from_file(_cal_file, chk_version=self.chk_version)
[docs]
def run_the_steps(self, stop_at_step:str=None, reload_only:bool=False,
status_only:bool=False):
"""
Run full the full recipe of calibration steps.
"""
# State (use the safe_* wrappers so state I/O can never abort a run).
# A status-only pass is a *read* (e.g. the Dashboard deriving status on
# launch): it must not write the state file, or it would leave the
# last step marked 'running' / current_step set (Stage 5 R2, S5-Q12
# option A). Only a real run persists state.
if self.state is not None:
self.state.current_det = self.det
self.state.current_calibID = self.calib_ID
if not status_only:
self.state.safe_write()
for step in self.steps:
self.success = True
if reload_only:
force = 'reload'
elif stop_at_step is not None and step == stop_at_step:
force = 'remake'
log.info(f"Calibrations will stop at {stop_at_step}")
else:
force = None
# Run the step
if self.state is not None:
self.state.safe_update_calib(step, self.calib_ID, self.det, 'status', 'running')
# Skip the 'running' write for a status-only read (S5-Q12 A);
# a real run writes it as the live-monitoring feed.
if not status_only:
self.state.safe_write()
try:
getattr(self, f'get_{step}')(force=force)
except Exception:
# A step that raises (an unrecoverable error mid-build, e.g. a
# PypeItError thrown deep in get_flats) must not leave the state
# stuck at 'running': otherwise the Dashboard / pypeit_status
# would keep showing the step as in-progress (orange) forever,
# even though the run has died (bug report 000). Mark it 'fail'
# and persist so the failure is visible, then re-raise so the
# run still aborts with a non-zero exit code.
if self.state is not None:
self.state.safe_update_calib(step, self.calib_ID, self.det,
'status', 'fail')
if not status_only:
self.state.safe_write()
raise
# Update state
if self.state is not None:
if not self.success:
# A step can be "not successful" two ways: a genuine build
# failure (-> 'fail'), or — in reload/status-only mode — a
# calibration that simply has not been built yet (its file
# is missing; see _reload_or_remake). The latter is
# 'undone' (required-but-not-done in the Dashboard), not a
# failure (Stage 5 R1 #1).
# NOTE: must match the BaseCalibState.status Literal
# ('fail'/'undone', not 'failed') or the file fails to
# reload.
missing_status = 'undone' if reload_only else 'fail'
self.state.safe_update_calib(step, self.calib_ID,
self.det, 'status',
missing_status)
else:
# Run status method; metric collection is non-essential
# to the reduction, so never let it crash the run
try:
getattr(self, f'{step}_state')()
except Exception as e:
log.warning(f"Failed to collect state metrics for "
f"step '{step}': {e}")
# Write?
if not status_only:
# State writing is non-essential: log and continue on
# failure rather than crashing the reduction
self.state.safe_write()
# Drop out?
if not self.success and not status_only:
self.failed_step = f'get_{step}'
return
# Stop?
if stop_at_step is not None and step == stop_at_step:
log.info(f"Calibrations stopping at {stop_at_step}")
return
log.info("Calibration complete and/or fully loaded!")
log.info("#######################################################################")
[docs]
def _chk_set(self, items):
"""
Check whether a needed attribute has previously been set
Args:
items (list): Attributes to check
"""
for item in items:
if getattr(self, item) is None:
raise PypeItError("Use self.set to specify '{:s}' prior to generating XX".format(item))
# This is specific to `self.ms*` attributes
[docs]
def _chk_objs(self, items):
"""
Check that the input items exist internally as attributes
Args:
items (list):
List of required items for the calibration step
Returns:
bool: True if all exist or if all were successfully loaded, False
"""
for obj in items:
if getattr(self, obj) is None:
# Strip ms
iobj = obj[2:] if obj[0:2] == 'ms' else obj
log.warning("You need to generate {:s} prior to this calibration..".format(obj))
log.warning("Use get_{:s}".format(iobj))
return False
return True
def __repr__(self):
# Generate sets string
txt = '<{:s}: frame={}, det={}, calib_ID={}'.format(self.__class__.__name__,
self.frame,
self.det,
self.calib_ID)
txt += '>'
return txt
[docs]
@staticmethod
def get_association(fitstbl, spectrograph, caldir, setup, calib_ID, det, must_exist=True,
subset=None, include_science=False, proc_only=False):
"""
Construct a dictionary with the association between raw files and
processed calibration frames.
Args:
fitstbl (:class:`~pypeit.metadata.PypeItMetaData`):
The class holding the metadata for all the frames to process.
spectrograph (:obj:`pypeit.spectrographs.spectrograph.Spectrograph`):
Spectrograph object
caldir (:obj:`str`, `Path`_):
Path for the processed calibration frames.
setup (:obj:`str`):
The setup/configuration of the association.
calib_ID (:obj:`str`, :obj:`int`):
The *single* calibration group of the association.
det (:obj:`int`, :obj:`tuple`):
The detector/mosaic of the association.
must_exist (:obj:`bool`, optional):
If True, only *existing* calibration frames in the association
are included. If False, the nominal set of processed
calibration frame file names are returned, regardless of whether
or not they exist.
subset (`numpy.ndarray`_, optional):
A boolean array selecting a subset of rows from ``fitstbl`` for
output.
include_science (:obj:`bool`, optional):
Include science and standard frames in the association. This
parameter is mutually exclusive with ``proc_only``; if both are
true, ``proc_only`` takes precedence.
proc_only (:obj:`bool`, optional):
If True, only return a dictionary with the names of the
processed calibration frames. The dictionary sets the
calibration directory to ``DIR``, and the other keys are the
capitalized versions of the calibration type keywords; e.g.,
``asn['ARC']`` is the processed arc frame. This parameter is
mutually exclusive with ``include_science``; if both are true,
``proc_only`` takes precedence.
Returns:
:obj:`dict`: The set of raw and processed calibration frames
associated with the selected calibration group. This only includes
the processed frames if ``proc_only`` is True, and it includes the
science/standard frames if ``include_science`` is True.
"""
if fitstbl.calib_groups is None:
raise PypeItError('Calibration groups have not been defined!')
if include_science and proc_only:
log.warning('Requested to include the science/standard frames and to only return the '
'processed calibration frames. Ignoring former request.')
# Set the calibrations path
_caldir = str(Path(caldir).absolute())
# This defines the classes used by each frametype that results in an
# output calibration frame:
frame_calibrations = {'align': [alignframe.Alignments],
'arc': [buildimage.ArcImage, wavecalib.WaveCalib],
'bias': [buildimage.BiasImage],
'dark': [buildimage.DarkImage],
'pixelflat': [flatfield.FlatImages],
'illumflat': [flatfield.FlatImages],
'lampoffflats': [flatfield.FlatImages],
'slitless_pixflat': [flatfield.FlatImages],
'trace': [edgetrace.EdgeTraceSet, slittrace.SlitTraceSet],
'tilt': [buildimage.TiltImage, wavetilts.WaveTilts]
}
# Get the name of the detector/mosaic
detname = spectrograph.get_det_name(det)
# Find the unique configuations in the metaddata
asn = {}
setups = fitstbl.unique_configurations(copy=True, rm_none=True)
if setup not in setups:
log.warning(f'Requested setup {setup} is invalid. Choose from {",".join(setups)}.')
return asn
# Subset to output
if subset is None:
subset = np.ones(len(fitstbl), dtype=bool)
in_setup = fitstbl.find_configuration(setup) & subset
if not any(in_setup):
# There are no frames in this configuration
return asn
# Find all the frames in this calibration group
in_grp = fitstbl.find_calib_group(calib_ID) & in_setup
if not any(in_grp):
# There are no frames in this calibration group
return asn
# Iterate through each frame type and add the raw and processed
# calibration frames
for frametype, calib_classes in frame_calibrations.items():
indx = fitstbl.find_frames(frametype) & in_grp
if not any(indx):
continue
if not (all(fitstbl['calib'][indx] == fitstbl['calib'][indx][0]) or
all([fitstbl['calib'][indx][0] in cc.split(',') for cc in fitstbl['calib'][indx]])):
log_str = f'All {frametype} frames in group {calib_ID} '
log_str += 'are not all associated with the same subset of calibration '
log_str += 'groups; calib for the first file is '
log_str += f'{fitstbl["calib"][indx][0]}.'
log.warning(log_str)
calib_key = CalibFrame.construct_calib_key(setup, fitstbl['calib'][indx][0], detname)
asn[frametype] = {}
asn[frametype]['raw'] = fitstbl.frame_paths(indx)
asn[frametype]['proc'] \
= [str(calib_class.construct_file_name(calib_key, calib_dir=_caldir))
for calib_class in frame_calibrations[frametype]]
if must_exist:
# Only include the processed calibration frames found on disk
asn[frametype]['proc'] \
= [file for file in asn[frametype]['proc'] if Path(file).exists()]
if proc_only:
# Trim down the dictionary to only include the calibration directory
# and the processed calibration frames. This is a bit of a hack so
# that this function can be used to construct some of the header
# keys for the spec2d files.
files = {}
for key, val in asn.items():
if not isinstance(val, dict) or 'proc' not in val:
continue
for file in val['proc']:
_file = Path(file).absolute()
# NOTE: This assumes the calib_type (i.e., the class
# attribute of the processed calibration frame) is the first
# element of the output file name. If we change the
# calibration frame naming convention, this will need to be
# updated.
calib_type = _file.name.split('_')[0].upper()
files['DIR'] = str(_file.parent)
files[calib_type] = _file.name
return files
if include_science:
# Include the raw science and standard frames associated with this
# calibration group. This does *not* include any processed
# spec2d/spec1d file names.
for frametype in ['science', 'standard']:
indx = fitstbl.find_frames(frametype) & in_grp
if not any(indx):
continue
asn[frametype] = fitstbl.frame_paths(indx)
return asn
[docs]
@staticmethod
def association_summary(ofile, fitstbl, spectrograph, caldir, subset=None, det=None,
overwrite=False):
"""
Write a file listing the associations between the processed calibration
frames and their source raw files for every setup and every calibration
group.
Args:
ofile (:obj:`str`, `Path`_):
Full path to the output file.
fitstbl (:class:`~pypeit.metadata.PypeItMetaData`):
The class holding the metadata for all the frames to process.
spectrograph (:obj:`pypeit.spectrographs.spectrograph.Spectrograph`):
Spectrograph object
caldir (:obj:`str`, `Path`_):
Path for the processed calibration frames.
subset (`numpy.ndarray`_, optional):
A boolean array selecting a subset of rows from ``fitstbl`` for
output.
det (:obj:`int`, :obj:`tuple`, optional):
The specific detector (or mosaic) to use when constructing the
output processed calibration group file names. If None, a
placeholder is used.
overwrite (:obj:`bool`, optional):
Overwrite any existing file of the same name.
"""
if fitstbl.calib_groups is None:
raise PypeItError('Calibration groups have not been defined!')
_ofile = Path(ofile).absolute()
if _ofile.exists() and not overwrite:
raise PypeItError(f'{_ofile} exists! To overwrite, set overwrite=True.')
_det = 1 if det is None else det
detname = spectrograph.get_det_name(_det)
# Subset to output
if subset is None:
subset = np.ones(len(fitstbl), dtype=bool)
# Find the unique configuations in the metaddata
setups = fitstbl.unique_configurations(copy=True, rm_none=True)
asn = {}
# Iterate through each setup
for setup in setups.keys():
asn[setup] = {}
asn[setup]['--'] = copy.deepcopy(setups[setup])
in_setup = fitstbl.find_configuration(setup) & subset
if not any(in_setup):
continue
# Iterate through each calibration group
for calib_ID in fitstbl.calib_groups:
# Find all the frames in this calibration group
in_grp = fitstbl.find_calib_group(calib_ID) & in_setup
if not any(in_grp):
continue
asn[setup][calib_ID] \
= Calibrations.get_association(fitstbl, spectrograph, caldir, setup,
calib_ID, _det, must_exist=False,
subset=subset, include_science=True)
# Write it
with open(_ofile, 'w') as ff:
ff.write('# Auto-generated calibration association file using PypeIt version: '
f' {__version__}\n')
ff.write(f'# UTC {datetime.datetime.now(datetime.UTC).isoformat(timespec="milliseconds")}\n')
if det is None:
ff.write(f'# NOTE: {detname} is a placeholder for the reduced detectors/mosaics\n')
ff.write(yaml.dump(utils.yamlify(asn)))
log.info(f'Calibration association file written to: {_ofile}')
[docs]
@staticmethod
def default_steps():
"""
This defines the steps for calibrations and their order
Note that the order matters!
Returns:
list: Calibration steps, in order of execution
"""
return []
[docs]
class MultiSlitCalibrations(Calibrations):
"""
Calibration class for performing multi-slit calibrations (and also long-slit
and echelle). See :class:`Calibrations` for arguments.
.. note::
Calibrations are not sufficiently different yet to warrant a different
class for echelle reductions. This may change if a different order is
eventually required for the set of processing steps (see
:func:`default_steps`).
"""
[docs]
@staticmethod
def default_steps():
"""
This defines the calibration steps and their order.
Returns:
:obj:`list`: Calibration steps, in order of execution.
"""
# Order matters! And the name must match a viable "get_{step}" method
# in Calibrations.
# TODO: Does the bpm need to be done after the dark?
return ['bias', 'dark', 'bpm', 'slits', 'arc', 'tiltimg',
'wv_calib', 'tilts', 'scattlight', 'flats']
[docs]
class IFUCalibrations(Calibrations):
"""
Child of Calibrations class for performing IFU calibrations. See
:class:`Calibrations` for arguments.
"""
[docs]
@staticmethod
def default_steps():
"""
This defines the steps for calibrations and their order
Returns:
list: Calibration steps, in order of execution
"""
# Order matters!
return ['bias', 'dark', 'bpm', 'arc', 'tiltimg', 'slits', 'wv_calib', 'tilts', 'align',
'scattlight', 'flats']
[docs]
def check_for_calibs(par, fitstbl, raise_error=True, cut_cfg=None):
"""
Perform a somewhat quick and dirty check to see if the user
has provided all of the calibration frametype's to reduce
the science frames
Args:
par (:class:`~pypeit.par.pypeitpar.PypeItPar`):
fitstbl (:class:`~pypeit.metadata.PypeItMetaData`, None):
The class holding the metadata for all the frames in this
PypeIt run.
raise_error (:obj:`bool`, optional):
If True, crash out
cut_cfg (`numpy.ndarray`_, optional):
Also cut on this restricted configuration (mainly for chk_calibs)
Returns:
bool: True if we passed all checks
"""
if cut_cfg is None:
cut_cfg = np.ones(len(fitstbl), dtype=bool)
pass_calib = True
# Find the science frames
is_science = fitstbl.find_frames('science')
# Frame indices
frame_indx = np.arange(len(fitstbl))
for calib_ID in fitstbl.calib_groups:
in_grp = fitstbl.find_calib_group(calib_ID)
if not np.any(is_science & in_grp & cut_cfg):
continue
grp_science = frame_indx[is_science & in_grp & cut_cfg]
u_combid = np.unique(fitstbl['comb_id'][grp_science])
# TODO It does not appear comb_id is used for anything
# Maybe remove this for loop?
for j, comb_id in enumerate(u_combid):
frames = np.where(fitstbl['comb_id'] == comb_id)[0]
# Arc, tilt, science
for ftype in ['arc', 'tilt', 'science', 'trace']:
rows = fitstbl.find_frames(ftype, calib_ID=calib_ID, index=True)
if len(rows) == 0:
# Fail
msg = f'No frames of type={ftype} provided. Add them to your PypeIt file ' \
'if this is a standard run!'
pass_calib = False
if raise_error:
raise PypeItError(msg)
else:
log.warning(msg)
# Explore science frame
for key, ftype in zip(['use_biasimage', 'use_darkimage', 'use_pixelflat',
'use_illumflat'], ['bias', 'dark', 'pixelflat', 'illumflat']):
if par['scienceframe']['process'][key]:
rows = fitstbl.find_frames(ftype, calib_ID=calib_ID, index=True)
if len(rows) == 0:
# Allow for pixelflat inserted
if ftype == 'pixelflat' \
and par['calibrations']['flatfield']['pixelflat_file'] is not None:
continue
# Allow for no pixelflat but slitless_pixflat needs to exist
elif ftype == 'pixelflat' \
and len(fitstbl.find_frame_files('slitless_pixflat', calib_ID=calib_ID)) > 0:
continue
# Otherwise fail
add_msg = ' or slitless_pixflat' if ftype == 'pixelflat' else ''
msg = f'No frames of type={ftype}{add_msg} provided for the *{key}* processing ' \
'step. Add them to your PypeIt file!'
pass_calib = False
if raise_error:
raise PypeItError(msg)
else:
log.warning(msg)
if pass_calib:
log.info("Congrats!! You passed the calibrations inspection!!")
return pass_calib
[docs]
def required_calibs(par, fitstbl, spectrograph, run_state):
"""
Determine which calibration steps are required for the given PypeIt file
and update the state object accordingly.
This is based on the logic in :func:`check_for_calibs` but instead of
raising errors for missing frames, it sets the ``required`` flag on
each calibration step entry in the state.
The mapping from frametypes to calibration steps is:
- ``arc`` frames → ``arc`` and ``wv_calib`` steps
- ``tilt`` frames → ``tiltimg`` and ``tilts`` steps
- ``trace`` frames → ``slits`` step
- ``bias`` frames → ``bias`` step (if ``use_biasimage`` is True)
- ``dark`` frames → ``dark`` step (if ``use_darkimage`` is True)
- ``pixelflat``/``illumflat`` frames → ``flats`` step (if
``use_pixelflat`` or ``use_illumflat`` is True)
- ``scattlight`` frames → ``scattlight`` step (if
``subtract_scattlight`` is True)
- ``align`` frames → ``align`` step (IFU only)
Args:
par (:class:`~pypeit.par.pypeitpar.PypeItPar`):
The full parameter set for the reduction.
fitstbl (:class:`~pypeit.metadata.PypeItMetaData`):
The class holding the metadata for all the frames in this
PypeIt run.
spectrograph (:class:`~pypeit.spectrographs.spectrograph.Spectrograph`):
The spectrograph object for the instrument being reduced.
run_state (:class:`~pypeit.state.RunPypeItState`):
The state object to update. Each calibration step entry will
have its ``required`` field set to True or False.
Returns:
:class:`~pypeit.state.RunPypeItState`: The updated state object.
"""
# Determine which calibration class to use for the step list
if spectrograph.pypeline in ['MultiSlit', 'Echelle']:
steps = MultiSlitCalibrations.default_steps()
else:
steps = IFUCalibrations.default_steps()
# Find detectors
detectors = spectrograph.select_detectors(
subset=par['rdx']['detnum'] if par['rdx']['slitspatnum'] is None
else par['rdx']['slitspatnum'])
# Mandatory frametypes and their corresponding steps
# arc -> arc, wv_calib
# tilt -> tiltimg, tilts
# trace -> slits
mandatory_map = {
'arc': ['arc', 'wv_calib'],
'tilt': ['tiltimg', 'tilts'],
'trace': ['slits'],
}
# Conditional frametypes from par['scienceframe']['process']
# and their corresponding steps
conditional_map = {
'use_biasimage': ('bias', ['bias']),
'use_darkimage': ('dark', ['dark']),
'use_pixelflat': ('pixelflat', ['flats']),
'use_illumflat': ('illumflat', ['flats']),
}
# Process
proc_par = par['scienceframe']['process']
for calib_ID in fitstbl.calib_groups:
in_grp = fitstbl.find_calib_group(calib_ID)
if not any(in_grp):
continue
for det in detectors:
# Build set of required steps for this calib_ID/det
required_steps = set()
# Mandatory frametypes
for ftype, step_list in mandatory_map.items():
rows = fitstbl.find_frames(ftype, calib_ID=calib_ID, index=True)
if len(rows) > 0:
required_steps.update(step_list)
# Conditional frametypes
for key, (ftype, step_list) in conditional_map.items():
if proc_par[key]:
rows = fitstbl.find_frames(ftype, calib_ID=calib_ID, index=True)
if len(rows) > 0:
required_steps.update(step_list)
elif ftype == 'pixelflat':
# Allow for pixelflat_file parameter
if par['calibrations']['flatfield']['pixelflat_file'] is not None:
required_steps.update(step_list)
# Allow for slitless_pixflat frames
elif len(fitstbl.find_frame_files(
'slitless_pixflat', calib_ID=calib_ID)) > 0:
required_steps.update(step_list)
# Scattered light
if proc_par['subtract_scattlight']:
rows = fitstbl.find_frames('scattlight', calib_ID=calib_ID, index=True)
if len(rows) > 0:
required_steps.add('scattlight')
# Alignment (IFU)
if spectrograph.pypeline == 'SlicerIFU':
rows = fitstbl.find_frames('align', calib_ID=calib_ID, index=True)
if len(rows) > 0:
required_steps.add('align')
# Update state for each step
for step in steps:
if step == 'bpm':
# bpm is always generated at runtime, skip
continue
required = step in required_steps
run_state.update_calib(step, calib_ID, det, 'required', required)
return run_state