how_this_code_will_be_used.py

import joblib
from astropy.io import fits
from csst.core.processor import CsstProcessor
from csst.msc.data_manager import CsstMscDataManager

from csst_proto.api import flip_image


class CsstProcFlipImage(CsstProcessor):
    """ This processor flips images """

    def __init__(self, dm: CsstMscDataManager, n_jobs: int = 1):
        """

        Parameters
        ----------
        dm:
            csst data manager, used to manage the input/output file paths
        n_jobs:
            number of jobs launched
        """
        super(CsstProcFlipImage, self).__init__()
        self.dm = dm
        self.n_jobs = n_jobs

    def run(self, debug: bool = False):
        """ flip images for all detectors

        Parameters
        ----------
        debug:
            if True, use debug mode

        Returns
        -------
        a list of flipped images
        """
        img_flipped_list = joblib.Parallel(n_jobs=self.n_jobs)(
            joblib.delayed(CsstProcFlipImage.run_one_detector)(this_ccd_id)
            for this_ccd_id in self.dm.target_ccd_ids)
        return img_flipped_list

    def prepare(self, **kwargs):
        """ prepare the environment """
        return

    def cleanup(self):
        """ clean up the environment """
        return

    @staticmethod
    def run_one_detector(dm: CsstMscDataManager, ccd_id: int = 6):
        """ run for one detector

        Parameters
        ----------
        dm:
            csst data manager, used to manage the input/output file paths
        ccd_id:
            detector id

        Returns
        -------
        the flipped image
        """
        # input file path
        fp_input = dm.l1_ccd(ccd_id, post="img.fits")
        img_input = fits.getdata(fp_input)
        # flip image
        img_flipped = flip_image(img_input)
        return img_flipped