pep 8 check.

import numpy as np

import scipy.ndimage as nd

from astropy.io import fits

import matplotlib.pyplot as plt

from .config import config, S

from .utils import region_replace, random_seed_select

MAG_SYSTEM = config['mag_system']

solar_spectrum = S.FileSpectrum(config['solar_spectrum'])

solar_spectrum.convert('photlam')

def sky_frame_maker(band, skybg, platescale, shape):

    """

    generate a sky background frame.

        self.nonlinear_coefficient = -0.1

        self.detector_name = 'EMCCD'

        self.ccd_label= 'CCD201-20'

        self.detector_name = 'CCD201-20-EM'

        self.ccd_label = 'EMCCD'

        self.pitch_size = 13

    def config_init(self):

        """initialize the camera.

            If the config is set, call this function to update the config.

        cti_trail = cte_201(self.em_cte, start=0, length=10)

        self.cti_trail = cti_trail / cti_trail.sum()

    def em_fix_fuc_fit(self, emgain):

        """Calculate the emgain fix coeficient to fix the gamma distribution.

        The coeficient is from fixing of ideal emgain distribution.

        """

        emgain = np.array([emgain]).flatten()

        p = [0.01014486, -0.00712984, -0.17163414,  0.09523666, -0.53926089]

        def kernel(em):

            log_em = np.log10(em)

            loglog_g = np.log10(log_em)

                output.append(kernel(em))

        return np.array(output)

    def bias_frame(self):

        """Generate bias frame

        The bias frame contains vertical, horizontal, peper-salt noise, bias drift effect.

            heat = self.volt * self.heat_speed

        self.ccd_temp = heat + self.cooler_temp + (self.ccd_temp - self.cooler_temp) * np.exp(-dt * self.temper_speed)

        if self.ccd_temp < self.cooler_temp: # 

        if self.ccd_temp < self.cooler_temp:

            self.ccd_temp = self.cooler_temp

        self.system_time += dt

        volt_coe_a = -0.01828

        volt_coe_b = 43.61

        volt_func = lambda es: volt_coe_a * es + volt_coe_b

        def volt_func(es):

            return volt_coe_a * es + volt_coe_b

        self.volt = volt_func(em_set)

            image_shutter = np.random.poisson(image_shutter)

            image[:, self.pscan1+self.ldark:-self.oscan1-self.rdark] += image_shutter

        if self.switch['cic']:

            cic_frame = np.zeros((self.dark_shape[0], self.bias_shape[1])) + self.cic

            image[self.pscan2:-self.oscan2, :] += np.random.poisson(cic_frame)

        em_fix = self.em_fix_fuc_fit(emgain) * emgain

        image = np.random.gamma(image, em_fix) + image * (emgain - em_fix)

        if self.switch['em_blooming']:

            image = self.emregester_blooming(image)

import os, yaml

import os

import yaml

import warnings

from datetime import datetime

import numpy as np

    ----------

    config_aim : str

        config_aim file path

    """

    with open(config_aim, 'r') as f:

        refdata_list = yaml.load(f, Loader=yaml.FullLoader)

config['datamodel'] = f'{cpism_refdata}/io/csst-cpic-l0.yaml'

config['log_dir'] = f'{cpism_refdata}/log'

config['log_level'] = f'info'

config['output'] = f'./'

config['log_level'] = 'info'

config['output'] = './'

config['sp2teff_model'] = f'{cpism_refdata}/target_model/sptype2teff_lut.json'

config['dm_pickle'] = f'{cpism_refdata}/optics/dm_model.pkl'

config['pysyn_refdata'] = f'{cpism_refdata}/starmodel/grp/redcat/trds'

config['nsample'] = 5

update_able_keys = [

    'apm_file', 'actuator_file', 'aberration', 'log_dir', 'log_level', 'catalog_folder', 'nsample', 'csst_format', 'output', 'check_fits_header'

    'apm_file', 'actuator_file', 'aberration',

    'log_dir', 'log_level', 'catalog_folder',

    'nsample', 'csst_format', 'output', 'check_fits_header'

]

def replace_cpism_refdata(

        config: dict,

        output: str = '$') -> None:

with warnings.catch_warnings():  # pragma: no cover

    warnings.filterwarnings("ignore")

    import pysynphot as S

    _ = S  # S will be used in other modules, but I need to use it once here to pass the lint.

def setup_config(new_config):

    config['default_band'] = list(config['bands'].keys())[0]

    config['default_filter'] = config['bands'][config['default_band']]

setup_config({})

def which_focalplane(band):

    """

    Return the name of the focalplane which the band belongs to.

    band: str

        The name of the band.

    Returns

    --------

    str

    return 'vis'

    # raise ValueError(f"未知的波段{band}")

def iso_time(time):

    """Transfer relative time to iso time format

    """

    if isinstance(time, str):

        _  = datetime.fromisoformat(time)

        _ = datetime.fromisoformat(time)  # check if it is a iso time

        return time

    utc0 = config['utc0']

    time = datetime.fromtimestamp(time0 + time)

    return time.isoformat()

def relative_time(time):

    """Transfer iso time format to relative time in seconds

    utc0 = config['utc0']

    time0 = datetime.timestamp(datetime.fromisoformat(utc0))

    return datetime.timestamp(datetime.fromisoformat(time)) - time0

 No newline at end of file

import yaml, os, re

import yaml

import os

from datetime import datetime

import numpy as np

import pandas as pd

from astropy.io import fits

from astropy.coordinates import SkyCoord

import astropy.units as u

from .config import __version__, which_focalplane

from .utils import Logger

log_level = config['log_level']

header_check = config['check_fits_header']

def set_up_logger(log_dir):

    if not os.path.exists(log_dir):

        os.makedirs(log_dir)

    return Logger(log_dir+'/cpism_pack.log', log_level).logger

log = set_up_logger(config['log_dir'])

def check_and_update_fits_header(header):

    """

    Check the header keywords and update the description according to the data model.

            else:

                key_type = 'ukn'

            # print(f"keyword: {keyword} type: {key_type}, datamodel: {dtype}, value: {value}")

            if key_type != dtype[0:3]:

                if key_type == 'int' and dtype[0:3] == 'flo':

                    header[keyword] = float(header[keyword])

                    # print('transfer int to float with header keyword:', keyword)

                else:

                    print_warning(

                        f"Keyword {keyword} has wrong type in [{hdu}]. {dtype} expected, {key_type} found.")

    return obstype

def datetime_obj_to_iso(time_obj):

    """

    transfer datetime object to iso format used in csst fits file

    example '2014-02-10T12:32:12.4'

    Parameters

    ----------

    time_obj: datetime.datetime

        The datetime object.

    Returns

    -------

    str

        The iso format of the datetime object.

    """

    isotime = time_obj.isoformat(sep='T', timespec='milliseconds')

    subsec = str(round(float(isotime[19:22]), 1))

    return isotime[:18] + subsec

def datetime_obj_to_mjd(time_obj):

    """

    transfer datetime object to mean julian date (MJD).

    header['NEXTEND'] = 1  # + parameters['nframe']

    # header['GROUPS'] = False

    header['DATE'] = datetime.now().isoformat(timespec='seconds')

    header['DATE'] = datetime_obj_to_iso(datetime.now())

    heaer_filename = filename[:-4]

    if len(heaer_filename) > 68:

        heaer_filename = heaer_filename[:68]

        cstar = obs_info['target']['cstar']

    radec = SkyCoord(cstar['ra'], cstar['dec'])

    target_name = radec.to_string('hmsdms')

    target_name = re.sub(R'[hdms\s]', '', target_name)

    ra_str = radec.ra.to_string(unit=u.hour, sep='', precision=1, pad=True)

    dec_str = radec.dec.to_string(unit=u.deg, sep='', alwayssign=True, pad=True)

    target_name = ra_str + dec_str

    header['OBJECT'] = cstar.get('name', target_name)

    header['TARGET'] = target_name

    header['OBSID'] = str(obsid)

    # telescope information

    header['REFFRAME'] = 'CSSTGSC-1.0'

    header['DATE-OBS'] = exp_start.isoformat(timespec='seconds')

    header['DATE-OBS'] = datetime_obj_to_iso(exp_start)

    header['SATESWV'] = '1'

    header['EXPSTART'] = datetime_obj_to_mjd(exp_start)

    header['BUNIT'] = 'ADU'

    header['FILTER'] = obs_info['band']

    header['DETSN'] = '0'

    header['DETSN'] = '00000000000'

    header['DETNAME'] = camera_config['detector_name']

    header['CHIPLAB'] = camera_config['ccd_label']

    header['DEWTEMP'] = float(camera_config['cooler_temp'])

    header['DEWTEMP'] = float(camera_config['cooler_temp']) + 273.15

    frame_info = obs_info['frame_info'][index]

    header['CHIPTEMP'] = float(frame_info['chiptemp'])

    header['CHIPTEMP'] = float(frame_info['chiptemp']) + 273.15

    header['DETSIZE'] = f"{imgszx} * {imgszy}"

    header['IMGINDEX'] = index + 1

    shift = obs_info['shift']

    header['shift'] = f"x:{shift[0]},y:{shift[1]}"

    fullname = os.path.join(output_folder, filename)

    print(fullname)

    if not os.path.exists(output_folder):

import argparse, sys, tqdm, time, os, yaml

import argparse

import tqdm

import time

import os

import yaml

from glob import glob

from datetime import datetime

import traceback

    emset: int

        EM gain setting value. 1023(0x3FF) for ~1.0× EM gain.

    obsid: int

        observation ID. Start from 4 for CPIC, 01 for science observation. See the input of io.obsid_parser for more details.

        observation ID. Start from 4 for CPIC, 01 for science observation.

        See the input of io.obsid_parser for more details.

    rotation: float

        rotation of the telescope. in unit of degree. 0 means North is up.

    shift: list

        print(f'\r  Done [{time.time() - start_time:.1f}s]                             ')

    return image_cube

def quick_run_v2(

        target_str: str,

        band: str,

    """

    print(f'Quick Run: {target_str}')

    log.debug(

        f"""input parameters:

        target_str: {target_str}

    )

def deduplicate_names_add_count(names: list):

    """remove duplicate names and add count"""

    for i in range(len(names)-1, -1, -1):

    if not file_list:

        log.warning(f"No observation file found in {obs_file}")

    for ind_target, file in enumerate(file_list):

        try:

            with open(file, 'r') as fid:

            except Exception as e:

                log.error(f"{info_text} failed with {type(e).__name__}{e}.\n\n {traceback.format_exc()}")

def main(argv=None):

    """

    Command line interface of csst_cpic_sim

    parser_quickrun.add_argument('nframe', type=int, help='number of frames')

    parser_quickrun.add_argument('-b', '--band', type=str, default='f661', help='band, one of f565/f661/f743/f883')

    parser_quickrun.add_argument('-r', '--rotation', type=float, default=0, help='rotation angle [degree]')

    parser_quickrun.add_argument('-s', '--skybk', type=float, default=21, help='magnitude of sky background [mag/arcsec^2]')

    parser_quickrun.add_argument('-f', '--cr_frame', action='store_true', help='if True, cosmic ray frame will be added')

    parser_quickrun.add_argument('-e', '--emset', action='store_true', help='if True, emgain set value will be used as input')

    parser_quickrun.add_argument('-c', '--camera_effect', action='store_true', help='if True, camera effect will be added')

    parser_quickrun.add_argument(

        '-s', '--skybk', type=float, default=21,

        help='magnitude of sky background [mag/arcsec^2]')

    parser_quickrun.add_argument(

        '-f', '--cr_frame', action='store_true',

        help='if True, cosmic ray frame will be added')

    parser_quickrun.add_argument(

        '-e', '--emset', action='store_true',

        help='if True, emgain set value will be used as input')

    parser_quickrun.add_argument(

        '-c', '--camera_effect', action='store_true',

        help='if True, camera effect will be added')

    parser_quickrun.add_argument('-o', '--output', type=str, default='./', help='output folder')

    def quick_run_call(args):

import os, pickle

import os

import pickle

import numpy as np

from astropy.io import fits

aberration_distance = 80 * focal_length

aberration = SurfaceAberrationAtDistance(aberration, aberration_distance)

def single_band_masked_psf(

        wavelength: float,

        error: float = 0,

    wf = Wavefront(aperture, wavelength)

    shift = np.array(shift) * ARCSEC2RAD / 2

    tiptilt_mirror.actuators = shift

    wf = tiptilt_mirror(wf)

    wf = aberration(wf)

    wf = tiptilt_mirror(wf)

    first_focal = prop_full_frame(deformable_mirror(wf))

    strength = first_focal.intensity.shaped.sum()

    psf = second_focal.intensity.shaped

    return psf / strength

def single_band_shift_psf(

        wavelength: float,

        error: float = 0,

        shift: list = [0, 0]) -> np.ndarray:

    """CPIC PSF considering the focal plane mask.

    Parameters

    -----------

    wavelength : float

        observation wavelength in meter

    error : float

        deformable mirror control error in nm

    shift : list

        angular shift of the target in arcsec.

    Returns

    ----------

    psf : np.ndarray

        psf in the focal plane. Normalized as the input flux is 1.

        (Note that total flux of the psf is not 1, because it is masked)

    """

    error = np.random.normal(0, error*1e-9, actuator.shape)

    deformable_mirror.actuators = actuator + error

    wf = Wavefront(aperture, wavelength)

    shift = np.array(shift) * ARCSEC2RAD / 2

    tiptilt_mirror.actuators = shift

    wf = aberration(wf)

    wf = tiptilt_mirror(wf)

    first_focal = prop_full_frame(deformable_mirror(wf))

    image = np.array(first_focal.intensity.shaped)

    return image / image.sum()

def single_band_psf(

        wavelength: float,

        error: float = 0) -> np.ndarray:

    ----------

    psf : np.ndarray

        psf in the focal plane. Normalized. The total flux is 1.

    """

    error = np.random.normal(0, error*1e-9, actuator.shape)

    deformable_mirror.actuators = actuator + error

    image = np.array(first_focal.intensity.shaped)

    return image / image.sum()

# def single_band_psf(wavelength, error=0, aber_phase=None):

#     error = np.random.normal(0, error*1e-9, actuator.shape)

#     deformable_mirror.actuators = actuator + error