Source code for shapepipe.modules.split_exp_package.split_exp

"""SPLIT EXPOSURE.

Class to split single-exposure multi-CCD mosaic images into single-exposure
single-CCD files, one HDU per CCD.

This module splits the different CCDs (HDUs in FITS files) of a
single exposure into separate files.

:Author: Axel Guinot

"""

import numpy as np
import sip_tpv as stp
from astropy.io import fits
from astropy.wcs import WCS

from shapepipe.pipeline import file_io


[docs]class SplitExposures(object): """Split Exposures. Parameters ---------- input_file_list : list Input file paths, typically image, weight, and flag output_dir : str Output directory file_number_string : str Input file identified output_suffix : str Output file suffix n_hdu : int Number of HDUs (CCDs) of input files """ def __init__( self, input_file_list, output_dir, file_number_string, output_suffix, n_hdu ): self._input_file_list = input_file_list self._output_dir = output_dir self._file_number_string = file_number_string self._output_suffix = output_suffix self._n_hdu = n_hdu
[docs] def process(self): """Process. Process the splitting of single-exposure images. """ for exp_path, output_suffix in zip( self._input_file_list, self._output_suffix ): transf_int = 'flag' in output_suffix transf_coord = 'image' in output_suffix save_header = 'image' in output_suffix self.create_hdus( exp_path, output_suffix, transf_coord, transf_int, save_header )
[docs] def create_hdus( self, exp_path, output_suffix, transf_coord, transf_int, save_header ): """Create HDUs. Split a single exposures CCDs into separate files. Parameters ---------- exp_path : str Path to the single exposure output_suffix : str Suffix for the output file transf_coord : bool Transform the WCS (``pv`` to ``sip``) if ``True`` transf_int : bool Set data types to int if ``True`` save_header : bool Save WCS information if ``True`` """ header_file = np.zeros(self._n_hdu, dtype='O') for idx in range(1, self._n_hdu + 1): h = fits.getheader(exp_path, idx) if transf_coord: stp.pv_to_sip(h) d = fits.getdata(exp_path, idx) if transf_int: d = d.astype(np.int16) file_name = ( f'{self._output_dir}/{output_suffix}' + f'{self._file_number_string}-{str(idx-1)}.fits' ) new_file = file_io.FITSCatalogue( file_name, open_mode=file_io.BaseCatalogue.OpenMode.ReadWrite ) new_file.save_as_fits(data=d, image=True, image_header=h) if save_header: try: w = WCS(h) except Exception: print(f'WCS error for file {exp_path}') raise header_file[idx - 1] = {'WCS': w, 'header': h.tostring()} if save_header: file_name = ( f'{self._output_dir}/headers{self._file_number_string}.npy' ) np.save(file_name, header_file)