Source code for ariastrotools.instrument


import numpy as np

from .utils import extract_allexts


[docs] class Handle_NEID: """ Class to handle NEID spectrograph FITS data. This class provides methods for reading NEID FITS files, applying barycentric corrections to the wavelength solution, and performing basic blaze correction on the extracted orders. Attributes ---------- name : str Instrument identifier, set to "NEID". Methods ------- __init__(): Initializes the NEID handler. getfull_data(fname): Reads all extensions from a FITS file and returns data and headers. barycorr(wl_array, header): Applies barycentric correction to the wavelength array using header keywords `SSBZxxx`. process_data(fname): Reads the FITS file, applies barycentric and blaze corrections, and returns corrected data and headers. """ name = "NEID" def __init__(self): """Initialize the NEID handler instance.""" pass
[docs] def fits_extensions(self): fluxext = [1, 2, 3] varext = [4, 5, 6] wlext = [7, 8, 9] extra = [12] tables = { 13: { "value_cols": ["VALUE"], "uncertainty_cols": ["UNCERTAINTY"], "combine_cols": [] } } # Other than spectra, if any other qty # need to be combined, those extension # can be added here. return fluxext, varext, wlext, extra, tables
[docs] def getfull_data(self, fname): """ Extract all extensions from a NEID FITS file. Parameters ---------- fname : str Path to the FITS file. Returns ------- datadict : dict Dictionary mapping extension keywords to numpy arrays containing the data. headerdict : dict Dictionary mapping extension keywords to FITS headers. """ datadict, headerdict = extract_allexts(fname) return datadict, headerdict
[docs] def telluric_correction(self, datadict): """ Apply telluric correction to the science spectrum. The science flux is divided by the combined telluric transmission, computed as the product of the two telluric components stored in the ``TELLURIC`` array. The corresponding variance is scaled by the square of the transmission to preserve uncertainty propagation. After the correction, the telluric transmission is replaced with a unity array of the same shape to indicate that the telluric correction has already been applied. Parameters ---------- datadict : dict Dictionary containing the extracted science data. The following keys are required: - ``'SCIFLUX'`` : ndarray Science flux array. - ``'SCIVAR'`` : ndarray Variance corresponding to ``SCIFLUX``. - ``'TELLURIC'`` : ndarray Telluric transmission array of shape ``(n_orders, n_pixels, 2)``, where the final dimension contains two multiplicative telluric components. Returns ------- dict The input dictionary with the following updates: - ``'SCIFLUX'`` replaced by the telluric-corrected flux. - ``'SCIVAR'`` replaced by the propagated variance. - ``'TELLURIC'`` replaced by a unity transmission array with the same shape as the original telluric array. """ tellurics = datadict['TELLURIC'] full_tellurics = tellurics[:, :, 0] * tellurics[:, :, 1] flux = datadict['SCIFLUX'] var = datadict['SCIVAR'] corr_flux = flux / full_tellurics corr_var = var / full_tellurics ** 2 datadict['SCIFLUX'] = corr_flux datadict['SCIVAR'] = corr_var datadict['TELLURIC'] = np.ones_like(tellurics) return datadict
[docs] def barycorr(self, wl_array, header): """ Apply barycentric correction to the wavelength array. For each spectral order, the barycentric correction factor is retrieved from the header keywords ``SSBZxxx`` (where xxx is the order number, zero-padded to 3 digits). The factor is then applied multiplicatively to the wavelength array. Parameters ---------- wl_array : ndarray 2D array of wavelength values (orders x pixels). header : astropy.io.fits.Header FITS header containing the barycentric correction keywords. Returns ------- corr_wl_array : ndarray Wavelength array corrected for barycentric motion. header : astropy.io.fits.Header Updated header with ``SSBZxxx`` values reset to zero. """ n_orders = wl_array.shape[0] zfacts = [] for index in range(n_orders): order = 173 - index strnum = str(order) while len(strnum) < 3: strnum = '0' + strnum zfact = header['SSBZ'+strnum] header['SSBZ'+strnum] = 0 zfacts.append(float(zfact)) zfacts = np.array(zfacts) corr_wl_array = (wl_array.T * (1+zfacts)).T return corr_wl_array, header
[docs] def process_data(self, fname, contnorm=False): """ Process a NEID FITS file. This method reads the science data from a NEID FITS file, applies the barycentric correction to the wavelength arrays, and optionally continuum-normalizes the science spectra. Parameters ---------- fname : str Path to the NEID FITS file. contnorm : bool, optional If ``True``, continuum-normalize the science flux and variance using :func:`continuum_normalize`. Default is ``False``. Returns ------- datadict : dict Dictionary containing the processed data arrays. The science flux and variance are stored as ``float64`` arrays, and the wavelength arrays are updated with the barycentric correction. If ``contnorm`` is ``True``, the science flux and variance are continuum-normalized. headerdict : dict Dictionary containing the updated FITS headers. The primary header is modified to include the barycentric correction keywords. """ datadict, headerdict = self.getfull_data(fname) sci_ext = [1, 2, 3] var_ext = [4, 5, 6] wl_ext = [7, 8, 9] header_kws = list(datadict.keys()) for n, ext in enumerate(sci_ext): flux_kw = header_kws[sci_ext[n]] var_kw = header_kws[var_ext[n]] wl_kw = header_kws[wl_ext[n]] flux = datadict[flux_kw].astype(np.float64) var = datadict[var_kw].astype(np.float64) wl = datadict[wl_kw].astype(np.float64) header_ext = headerdict[header_kws[0]] corr_wl, corr_header = self.barycorr(wl, header_ext) headerdict[header_kws[0]] = corr_header datadict[flux_kw] = flux datadict[var_kw] = var datadict[wl_kw] = corr_wl if contnorm: from .spectral_utils import continuum_normalize datadict = continuum_normalize(datadict, sci_ext, var_ext, wl_ext) return datadict, headerdict
[docs] def req_qtys(self): """ Dictonary format: {Name of extension: List of keys from that extension} """ qty = {'CCFS': ['CCFRVMOD', 'BISMOD', 'FWHMMOD']} for order in range(52, 174, 1): key = "CCFRV" if order < 100: key_order = key + "0" + str(order) else: key_order = key + str(order) qty['CCFS'].append(key_order) return qty
instrument_dict = { "NEID": Handle_NEID } # End