Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
import joblib
from astropy.io import fits
from csst.core.processor import CsstProcessor
from csst.msc.data_manager import CsstMscDataManager
from csst_proto import flip_image
class CsstProcFlipImage(CsstProcessor):
""" This processor absorbs photons """
def __init__(self, dm: CsstMscDataManager, n_jobs: int = 1):
super(CsstProcFlipImage, self).__init__()
self.dm = dm
self.n_jobs = n_jobs
def run(self, debug: bool = False):
""" run this processor
Parameters
----------
debug:
if True, use debug mode
Returns
-------
"""
img_flipped_list = joblib.Parallel(n_jobs=self.n_jobs)(
joblib.delayed(CsstProcFlipImage.run_one_chip)(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_chip(dm, ccd_id):
""" run for one chip
this function
"""
# 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