migrated from Chao

csst/__init__.py

+__version__ = "0.0.1"
\ No newline at end of file

csst/common/CsstException.py

+#import BaseException
+
+class CsstException(BaseException):
+    def __init__(self):
+        print('Exceoption raised.')
+    
\ No newline at end of file

csst/common/__init__.py

+__version__ = "0.0.1"
\ No newline at end of file

csst/common/data.py

+from collections import OrderedDict
+
+import astropy.io.fits as fits
+from astropy.io.fits import HDUList, PrimaryHDU
+
+from csst.msc.mscdata import CsstMscImgData
+from .CsstException import CsstException
+
+
+class CsstDataFactory:
+    """
+    This class is designed to create CsstData and its inherited classes according to different kinds of input data format.
+    """
+
+    def __init__(self):
+        pass
+
+    def createData(self, fitsfilename):
+        """
+
+        Parameters
+        ----------
+        fitsfilename:
+            the file name of fits files
+
+        Returns
+        -------
+
+        """
+
+        try:
+            hl = fits.open(fitsfilename)
+            instrument = hl[0].header.get('INSTRUME')  # strip or not?
+            detector = hl[0].header.get('DETECTOR')    # strip or not?
+            print(instrument, detector)
+            if instrument == 'MSC' and int(detector[3:5]) >= 6 and int(detector[3:5]) <= 25:
+                # multi-band imaging
+                data = CsstMscImgData(hl[0], hl[1])
+
+        except Exception as e:
+            print(e)
+        return data
+
+
+class CsstData:
+    _primary_hdu = []
+    _l0data = []  # HDUList
+    _l1hdr_global = []
+    _l1data = OrderedDict()  # dict object
+    _l2data = OrderedDict()  #
+    _auxdata = OrderedDict()
+
+    def __init__(self, primaryHDU, imgHDU):
+        print('create CsstData')
+        self._primary_hdu = primaryHDU
+        self._l0data = imgHDU
+
+    def get_l0data(self, *, copy):
+        '''
+        obtain level 0 data from CsstData class
+        copy: True: if the user want to copy the memory of the data to the new class;
+              False: only reference of the data memory is written to the new class
+        '''
+        if copy:
+            return self._l0data.data.copy()
+        else:
+            return self._l0data.data
+
+    def get_l0keyword(self, ext, key):
+        '''
+        obtain keywords of the fits header of level 0 image data from the CsstData class
+        ext: the index of extension. if it equals to 'pri', looking up keywords from primary session, otherwise from extension sessions
+        key: the name of the key
+        '''
+        if ext == 'pri':
+            try:
+                value = self._primary_hdu.header.get(key)
+            except Exception as e:
+                print(e)
+        elif ext == 'img':
+            try:
+                value = self._l0data.header.get(key)
+            except Exception as e:
+                print(e)
+        else:
+            raise CsstException
+        return value
+
+    def set_l1keyword(self, key, value):
+        print('check out whether ' + key + " is a valid key and " + value + " is valid value")
+
+    def set_l1data(self, img):
+        print('save image data to l2data')
+
+    def get_auxdata(self, name):
+        print('Parent class returns zero image.')
+        return np.zeros_like(get_l0data())
+
+    def save_l1data(self, imgtype, filename):
+        '''
+        asve level 1 image and auxilary data to data file 
+        imgtype
+        '''
+        print("save L1 image to a fits file with name " + filename)
+        try:
+            self._l1hdr_global.set('TYPE', imgtype, 'Type of Level 1 data')
+            pri_hdu = PrimaryHDU(header=self._l1hdr_global)
+            hdulist = HDUList([pri_hdu, self._l1data[imgtype]])
+            hdulist.writeto(filename)
+        except Exception as e:
+            print(e)

csst/common/fileio.py

-from astropy.io import fits
-
-
-class Header(fits.Header):
-    """ a custom Header class
-
-    Conf
-    ----
-    https://docs.astropy.org/en/stable/io/fits/api/headers.html#astropy.io.fits.Header
-
-    """
-
-    def __init__(self, cards=[], copy=False):
-        super(Header, self).__init__(cards=cards, copy=copy)
-
-
-test_hdr_str = """
-SIMPLE  =                    T / conforms to FITS standard
-BITPIX  =                    8 / array data type
-NAXIS   =                    0 / number of array dimensions
-EXTEND  =                    T
-"""
-
-if __name__ == "__main__":
-
-    hdr = Header.fromstring(test_hdr_str, sep='\n')
-    print(hdr['SIMPLE'], hdr['BITPIX'], len(hdr))

csst/common/processors.py

+
+from abc import ABCMeta, abstractmethod
+from enum import Enum
+
+class CsstProcStatus(Enum):
+    empty = -1
+    normal = 0
+    ioerror = 1
+    runtimeerror = 2
+    
+#     self['empty'].info = 'Not run yet.'
+#     self['normal'].info = 'This is a normal run.'
+#     self['ioerror'].info = 'This run is exceptionally stopped due to IO error.'
+#     self['runtimeerror'].info = 'This run is exceptionally stopped due to runtime error.'
+    
+class CsstProcessor(metaclass=ABCMeta):
+    @abstractmethod
+    def prepare(self, **kwargs):
+        pass
+    @abstractmethod
+    def run(self, data):
+        return self._status
+    @abstractmethod
+    def cleanup(self):
+        pass
+    
+ 
\ No newline at end of file

csst/msc/__init__.py

+__version__ = "0.0.1"
\ No newline at end of file

csst/msc/instrument.py

+from collections import OrderedDict
+from abc import ABCMeta, abstractmethod
+from csst.common.data import CsstData
+from enum import Enum
+import numpy as np
+from csst.common.processors import CsstProcStatus, CsstProcessor
+
+class CsstMscInstrumentProc(CsstProcessor):
+    _status = CsstProcStatus.empty
+    _switches = {'crosstalk':False,'nonlinear':False,'deepcr':False,'cti':False, 'brighterfatter':False}
+    
+    def __init__(self):
+        pass
+    
+    def _do_crosstalk(self):
+        if self._switches['crosstalk']:
+            print('Crosstalk correction')
+        
+    def _do_nonlinear(self):
+        if self._switches['nonlinear']:
+            print('Nonlinear effect correction')
+        
+    def _do_deepcr(self):
+        if self._switches['deepcr']:
+            print('Deep CR operation')
+        else:
+            print('Laplace CR correction')
+        
+    def _do_cti(self):
+        if self._switches['cti']:
+            print('CTI effect correction')
+        
+    def _do_brighterfatter(self):
+        if self._switches['brighterfatter']:
+            print('Brighter-Fatter effect correction')
+    
+    
+    def prepare(self, **kwargs):
+        for name in kwargs:
+            self._switches[name] = kwargs[name]
+            
+    def run(self, data):
+        if type(data).__name__=='CsstMscImgData' or type(data).__name__=='CsstMscSlsData':
+            self.__l1img = data.get_l0data(copy = True)
+            self.__weightimg = np.random.uniform(0,1,(9216,9232))
+            self.__flagimg = np.random.uniform(0,1,(9216,9232))
+            
+            flat = data.get_flat()
+            bias = data.get_bias()
+            dark = data.get_dark()
+            print('Flat and bias correction')
+            self._do_crosstalk()
+            self._do_nonlinear()
+            self._do_cti()
+            self._do_deepcr()
+            self._do_brighterfatter()
+            
+            print('fake to finish the run and save the results back to CsstData')
+            
+            data.set_l1data('sci',self.__l1img)
+            data.set_l1data('weight',self.__weightimg)
+            data.set_l1data('flag',self.__flagimg)
+
+            print('Update keywords')
+            data.set_l1keyword('SOMEKEY','some value','Test if I can append the header')
+            
+            self._status = CsstProcStatus.normal
+        else:
+            self._status = CsstProcStatus.ioerror
+        return self._status
+        
+    def cleanup(self):
+        pass
+

csst/msc/mscdata.py

+from collections import OrderedDict
+import astropy.io.fits as fits
+from astropy.io.fits import HDUList, PrimaryHDU, ImageHDU
+from astropy.io.fits.header import Header
+from csst.common.data import CsstData
+
+class CsstMscData(CsstData):
+    _l1img_types = {'sci':True,'weight':True,'flag':True}
+    def __init__(self, primaryHDU, imgHDU):
+        print('create CsstMscData')
+        super().__init__(primaryHDU, imgHDU)
+        self._l1hdr_global = primaryHDU.header.copy()
+#         self._l1hdr_global['SIMPLE']  =  'T' #/ conforms to FITS standard
+#         self._l1hdr_global['NAXIS']  =  0
+        self._l1data['sci'] =ImageHDU()
+        self._l1data['weight'] = ImageHDU()
+        self._l1data['flag'] = ImageHDU()
+        
+    def set_flat(self, flatimg):
+        self._auxdata['flat'] = flatimg
+    
+    def set_bias(self, biasimg):
+        self._auxdata['bias'] = biasimg
+        
+    def set_dark(self, darkimg):
+        self._auxdata['dark'] = darkimg
+    
+    def set_badpixel(self, badpixelimg):
+        self._auxdata['badpixel'] = badpixelimg
+        
+    def get_flat(self):
+        return self._auxdata['flat']
+    
+    def get_bias(self):
+        return self._auxdata['bias']
+    
+    def get_dark(self):
+        return self._auxdata['dark']
+    
+    def get_badpixel(self):
+        return self._auxdata['badpixel']
+    
+    def init_l0data(self):
+        pass
+#         hdr_global = Header(self._primary_hdu.header, copy = True)
+#         hdr_l1 = Header(self._l0data.header, copy = True)
+#         hdu_pri = PrimaryHDU(hdr = hdr_global)
+#         hdu_img = self._l0data.copy()
+        
+    def set_l1keyword(self, key, value, comment=''):
+        print('check out whether '+key+" is a valid key and "+value+" is valid value")
+        self._l1hdr_global.set(key,value,comment)
+        
+    def set_l1data(self, imgtype, img):
+        try:
+            if self._l1img_types[imgtype]:
+                self._l1data[imgtype].data = img.copy()
+        except Exception as e:
+            print(e)
+        print('save image data to l1data')
+        
+    def save_l1data(self, imgtype, filename):
+        print('check '+imgtype+' is validate')
+        try:
+            if self._l1img_types[imgtype]:
+                super().save_l1data(imgtype, filename)
+        except Exception as e:
+            print(e)
+
+class CsstMscImgData(CsstMscData):
+    def __init__(self, primaryHDU, imgHDU):
+        print('create CsstMscImgData')
+        super().__init__(primaryHDU, imgHDU)
+        

genuml.sh

+pyreverse -ASmy -o png csst/common