corgidrp.check#

Module to hold input-checking functions to minimize repetition

Copied over from the II&T pipeline

Attributes#

`logger`
`string_types`
`int_types`
`first_frame_keywords_default`
`last_frame_keywords_default`
`averaged_keywords_default`
`deleted_keywords_default`
`invalid_keywords_default`
`calculated_value_keywords_default`
`any_true_keywords_default`

Exceptions#

CheckException

Common base class for all non-exit exceptions.

Functions#

`real_positive_scalar`(var, vname, vexc)	Checks whether an object is a real positive scalar.
`real_array`(var, vname, vexc)	Checks whether an object is a real numpy array, or castable to one.
`oneD_array`(var, vname, vexc)	Checks whether an object is a 1D numpy array, or castable to one.
`twoD_array`(var, vname, vexc)	Checks whether an object is a 2D numpy array, or castable to one.
`twoD_square_array`(var, vname, vexc)	Checks whether an object is a 2D square array_like.
`threeD_array`(var, vname, vexc)	Checks whether an object is a 3D numpy array, or castable to one.
`real_scalar`(var, vname, vexc)	Checks whether an object is a real scalar.
`real_nonnegative_scalar`(var, vname, vexc)	Checks whether an object is a real nonnegative scalar.
`positive_scalar_integer`(var, vname, vexc)	Checks whether an object is a positive scalar integer.
`nonnegative_scalar_integer`(var, vname, vexc)	Checks whether an object is a nonnegative scalar integer.
`scalar_integer`(var, vname, vexc)	Checks whether an object is a scalar integer (no sign dependence).
`string`(var, vname, vexc)	Checks whether an object is a string.
`boolean`(var, vname, vexc)	Checks whether an object is a bool.
`dictionary`(var, vname, vexc)	Checks whether an object is a dictionary.
`check_filename_convention`(filename, expected_pattern)	Check if filename follows the expected naming convention.
`check_dimensions`(data, expected_shape[, frame_info, ...])	Check if data has expected dimensions.
`verify_hdu_count`(hdul, expected_count[, frame_info, ...])	Verify that the number of HDUs in the FITS file is as expected.
`verify_header_keywords`(header, required_keywords[, ...])	Verify that required header keywords are present and have expected values.
`validate_binary_table_fields`(hdu1, required_fields[, ...])	Validate binary table fields with consistent error reporting.
`get_latest_cal_file`(e2eoutput_path, pattern[, logger])	Get the most recent calibration file matching the pattern.
`generate_fits_excel_documentation`(fits_filepath, ...)	Generate an Excel file documenting the structure and headers of a FITS file.
`merge_headers`(input_dataset[, first_frame_keywords, ...])	Merge headers from multiple input frames into a single header set.
`fix_hdrs_for_tvac`(list_of_fits, output_dir[, ...])	Overwrite FITS headers with mock defaults while preserving certain values from originals.
`compare_to_mocks_hdrs`(fits_file[, header_template])	Check FITS headers against those expected from mocks.py (which is sourced ultimately from l1.csv).
`hdr_type_conform`(orig_pri_hdr, orig_img_hdr[, ...])	Correct FITS header type against that expected from mocks.py (which is sourced ultimately from l1.csv).

Module Contents#

corgidrp.check.logger[source]#

exception corgidrp.check.CheckException[source]#

Bases: Exception

Common base class for all non-exit exceptions.

corgidrp.check.string_types[source]#

corgidrp.check.int_types[source]#

corgidrp.check.real_positive_scalar(var, vname, vexc)[source]#

Checks whether an object is a real positive scalar.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.real_array(var, vname, vexc)[source]#

Checks whether an object is a real numpy array, or castable to one.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.oneD_array(var, vname, vexc)[source]#

Checks whether an object is a 1D numpy array, or castable to one.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.twoD_array(var, vname, vexc)[source]#

Checks whether an object is a 2D numpy array, or castable to one.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.twoD_square_array(var, vname, vexc)[source]#

Checks whether an object is a 2D square array_like.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.threeD_array(var, vname, vexc)[source]#

Checks whether an object is a 3D numpy array, or castable to one.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.real_scalar(var, vname, vexc)[source]#

Checks whether an object is a real scalar.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.real_nonnegative_scalar(var, vname, vexc)[source]#

Checks whether an object is a real nonnegative scalar.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.positive_scalar_integer(var, vname, vexc)[source]#

Checks whether an object is a positive scalar integer.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.nonnegative_scalar_integer(var, vname, vexc)[source]#

Checks whether an object is a nonnegative scalar integer.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.scalar_integer(var, vname, vexc)[source]#

Checks whether an object is a scalar integer (no sign dependence).

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.string(var, vname, vexc)[source]#

Checks whether an object is a string.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.boolean(var, vname, vexc)[source]#

Checks whether an object is a bool.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.dictionary(var, vname, vexc)[source]#

Checks whether an object is a dictionary.

Parameters:

var – variable to check
vname – string to output in case of error for debugging
vexc – Exception to raise in case of error for debugging

Returns:

returns var

corgidrp.check.check_filename_convention(filename, expected_pattern, frame_info='', logger=None, data_level='l2b')[source]#

Check if filename follows the expected naming convention.

Parameters:

filename (str) – Filename to check
expected_pattern (str) – Expected pattern (e.g., ‘cgi_*_l2b.fits’)
frame_info (str) – Additional info for logging (e.g., “Frame 0”)
logger – Logger instance to use. If None, uses module logger.
data_level (str) – data level to be considered.

Returns:

True if filename matches convention

Return type:

bool

corgidrp.check.check_dimensions(data, expected_shape, frame_info='', logger=None)[source]#

Check if data has expected dimensions.

Parameters:

data (numpy.ndarray) – Data array to check
expected_shape (tuple) – Expected shape tuple
frame_info (str) – Additional info for logging (e.g., “Frame 0”)
logger – Logger instance to use. If None, uses module logger.

Returns:

True if dimensions match

Return type:

bool

corgidrp.check.verify_hdu_count(hdul, expected_count, frame_info='', logger=None)[source]#

Verify that the number of HDUs in the FITS file is as expected.

Parameters:

hdul (astropy.io.fits.HDUList) – FITS HDUList object
expected_count (int) – Expected number of HDUs
frame_info (str) – Additional info for logging (e.g., “Frame 0”)
logger – Logger instance to use. If None, uses module logger.

Returns:

True if HDU count matches expected count

Return type:

bool

corgidrp.check.verify_header_keywords(header, required_keywords, frame_info='', logger=None)[source]#

Verify that required header keywords are present and have expected values.

Parameters:

header (astropy.io.fits.Header) – FITS header object
required_keywords (dict or list) – Dictionary of {keyword: expected_value} or list of keywords
frame_info (str) – Additional info for logging (e.g., “Frame 0”)
logger – Logger instance to use. If None, uses module logger.

Returns:

True if all keywords are valid

Return type:

bool

corgidrp.check.validate_binary_table_fields(hdu1, required_fields, logger=None)[source]#

Validate binary table fields with consistent error reporting.

Parameters:

hdu1 (astropy.io.fits.BinTableHDU) – FITS binary table HDU
required_fields (list) – List of required field names
logger – Logger instance to use. If None, uses module logger.

Returns:

True if all fields are valid

Return type:

bool

corgidrp.check.get_latest_cal_file(e2eoutput_path, pattern, logger=None)[source]#

Get the most recent calibration file matching the pattern.

Parameters:

e2eoutput_path (str) – Directory to search for calibration files
pattern (str) – Pattern to match (e.g., ‘*_dpm_cal.fits’)
logger – Logger instance to use. If None, uses module logger.

Returns:

Path to the most recent calibration file

Return type:

str

corgidrp.check.generate_fits_excel_documentation(fits_filepath, output_excel_path)[source]#

Generate an Excel file documenting the structure and headers of a FITS file.

Parameters:

fits_filepath (str) – Path to the FITS file to document
output_excel_path (str) – Path where the Excel file should be saved

Returns:

Path to the generated Excel file

Return type:

str

Raises:

ImportError – If pandas is not available
FileNotFoundError – If the FITS file doesn’t exist

corgidrp.check.first_frame_keywords_default = ['MJDSRT', 'SCTSRT'][source]#

corgidrp.check.last_frame_keywords_default = ['VISITID', 'MJDEND', 'SCTEND', 'NAXIS', 'NAXIS1', 'NAXIS2', 'NAXIS3', 'NAXIS4'][source]#

corgidrp.check.averaged_keywords_default[source]#

corgidrp.check.deleted_keywords_default = ['LOCAMT', 'CYCLES', 'LASTEXP'][source]#

corgidrp.check.invalid_keywords_default = ['FTIMEUTC', 'PROXET', 'DATETIME'][source]#

corgidrp.check.calculated_value_keywords_default[source]#

corgidrp.check.any_true_keywords_default = ['OVEREXP'][source]#

corgidrp.check.merge_headers(input_dataset, first_frame_keywords=None, last_frame_keywords=None, averaged_keywords=None, deleted_keywords=None, invalid_keywords=None, calculated_value_keywords=None, any_true_keywords=None)[source]#

Merge headers from multiple input frames into a single header set.

Used when building combined frames or calibration products from a dataset of frames.

Frames are sorted by SCTSRT (ascending); the chronologically last frame provides the base headers and values for last_frame_keywords.

Header keywords are handled according to which set they belong to:

first_frame_keywords: use value from the chronologically first frame
last_frame_keywords: use value from the chronologically last frame
averaged_keywords: average across all frames (Z{i}VAR uses pooled-variance)
deleted_keywords: remove keywords from output headers entirely
invalid_keywords: assign -999 with type matching original (int/float/str)
calculated_value_keywords: compute value for output (eg FILETIME = current time)
any_true_keywords: if any frame has a true value for the keyword, use True. else False
all other keywords: must be identical across frames; raise error if not

Note/ TODO - Only the primary, image extension, error, and DQ headers are merged here. Any additional HDUs listed in Image.hdu_list are not handled.

Parameters:

input_dataset (corgidrp.data.Dataset) – Dataset of input frames to merge.
first_frame_keywords (list or set, optional) – Keywords to take from the first frame.
last_frame_keywords (list or set, optional) – Keywords to take from the last frame.
averaged_keywords (list or set, optional) – Keywords to average across frames.
deleted_keywords (list or set, optional) – Keywords to remove from output headers.
invalid_keywords (list or set, optional) – Keywords to set to -999 with type matching original (int/float/str).
calculated_value_keywords (list or set, optional) – Keywords whose value is computed for the merged output (e.g. FILETIME = current UTC time).
any_true_keywords (list or set, optional) – Keywords where output is True if any frame has a truthy value, else False (e.g. DESMEAR, CTI_CORR).

Returns:

(merged_pri_hdr, merged_ext_hdr, merged_err_hdr, merged_dq_hdr)

Return type:

tuple

corgidrp.check.fix_hdrs_for_tvac(list_of_fits, output_dir, header_template=None)[source]#

Overwrite FITS headers with mock defaults while preserving certain values from originals.

Used for TVAC (and similar) data. Writes updated files to output_dir; does not modify originals.

Parameters:

list_of_fits (list) – FITS file paths to update
output_dir (str) – Directory to write updated FITS files to
header_template (callable, optional) – Function returning headers. Defaults to mocks.create_default_L1_headers.

Returns:

Updated FITS file paths written to output_dir

Return type:

list

corgidrp.check.compare_to_mocks_hdrs(fits_file, header_template=None)[source]#

Check FITS headers against those expected from mocks.py (which is sourced ultimately from l1.csv). Raises error if any mismatches found, including keywords with the wrong type.

Parameters:

fits_file (list) – FITS file path to check
header_template (callable, optional) – Function returning headers. Defaults to mocks.create_default_L1_headers.

corgidrp.check.hdr_type_conform(orig_pri_hdr, orig_img_hdr, header_template=None)[source]#

Correct FITS header type against that expected from mocks.py (which is sourced ultimately from l1.csv).

Parameters:

orig_pri_hdr (fits.Header) – Original primary header to correct
orig_img_hdr (fits.Header) – Original image header to correct
header_template (callable, optional) – Function returning headers. Defaults to mocks.create_default_L1_headers.

Returns:

(adjusted_pri_hdr, adjusted_img_hdr) with types conformed to expected values: from mocks.py. Note that this function does not check for missing keywords, only type mismatches, and it preserves all original values (it does not assign mock values to missing keywords).

Return type:

tuple