Skip to content
GitLab
Commit 95f81f91 authored by GZhao's avatar GZhao
Browse files

v2.0beta update

parent b7f8c4fa
Pipeline #4357 failed with stage
in 0 seconds
Expand all Show whitespace changes
Inline Side-by-side
.gitignore
View file @ 95f81f91
......@@ -8,6 +8,10 @@ dist/
starmodel/
cpism_refdata/
*.egg-info
example/example_output
refdata/starmodel
refdata/target_model
# Other files and folders
.settings/
......@@ -26,6 +30,7 @@ docs/notebooks/image_files/_*
# unitest output
tests/.coverage
tests/htmlcov/
tests/*.xml
# Executables
*.swf
......
.gitignore copy 0 → 100644
View file @ 95f81f91
# Build and Release Folders
bin-debug/
bin-release/
[Oo]bj/
[Bb]in/
build/
dist/
starmodel/
cpism_refdata/
*.egg-info
# Other files and folders
.settings/
_*/
~*
.VSCodeCounter
.vscode
*.log
output/
*.log.*
/*.fits
testrun/
example/example_output/
docs/notebooks/image_files/_*
# unitest output
tests/.coverage
tests/htmlcov/
# Executables
*.swf
*.air
*.ipa
*.apk
# Project files, i.e. `.project`, `.actionScriptProperties` and `.flexProperties`
# should NOT be excluded as they contain compiler settings and other important
# information for Eclipse / Flash Builder.
CpicImgSim/__init__.py deleted 100644 → 0
View file @ b7f8c4fa
# from .main import quick_run, observation_simulation
from .optics import make_focus_image, focal_mask
from .target import star_photlam, planet_contrast, extract_target_x_y, spectrum_generator
from .camera import EMCCD, CosmicRayFrameMaker, sky_frame_maker
from .config import __version__
__all__ = [
"EMCCD",
"CosmicRayFrameMaker",
"sky_frame_maker",
"star_photlam",
"planet_contrast",
"extract_target_x_y",
"spectrum_generator",
"make_focus_image",
"focal_mask",
# "quick_run",
# "observation_simulation"
]
\ No newline at end of file
CpicImgSim/camera.py deleted 100644 → 0
View file @ b7f8c4fa
This diff is collapsed.
CpicImgSim/config.py deleted 100644 → 0
View file @ b7f8c4fa
import os, yaml
import warnings
cpism_refdata = os.environ.get('cpism_refdata', './cpism_refdata')
if not os.path.exists(cpism_refdata): # pragma: no cover
raise Exception(
"Can not find CPISM reference data.")
if not os.path.exists(os.environ.get('PYSYN_CDBS', './trd')): # pragma: no cover
raise Exception(
"Can not find PYSYN Stellar reference data.")
# we need to ignore the warning from pysynphot, because we only use the star models.
with warnings.catch_warnings(): # pragma: no cover
warnings.filterwarnings("ignore")
import pysynphot as S
solar_spectrum = S.FileSpectrum(
f"{os.environ['PYSYN_CDBS']}/grid/solsys/solar_spec.fits")
solar_spectrum.convert('photlam')
config_file = cpism_refdata + '/optics/optics_config.yaml'
if not os.path.exists(config_file): # pragma: no cover
raise FileNotFoundError(f"光学配置文件不存在({config_file})")
with open(config_file, 'r') as f:
optics_config = yaml.load(f, Loader=yaml.FullLoader)
MAG_SYSTEM = 'abmag'
__version__ = '2.0.0'
def which_focalplane(band):
"""
Return the name of the focalplane which the band belongs to.
Parameters
-----------
band: str
The name of the band.
from ['f565', 'f661', 'f743', 'f883', 'f940', 'f1265', 'f1425', 'f1542', 'wfs']
Returns
--------
str
The name of the focalplane.
'vis' or 'nir' or 'wfs'
Raises
-------
ValueError
If the band is not in ['f565', 'f661', 'f743', 'f883', 'f940', 'f1265', 'f1425', 'f1542', 'wfs']
"""
band = band.lower()
if band in ['f565', 'f661', 'f743', 'f883']:
return 'vis'
if band in ['f940', 'f1265', 'f1425', 'f1542']:
return 'nir'
if band in ['wfs']:
return 'wfs'
raise ValueError(f"未知的波段{band}")
\ No newline at end of file
CpicImgSim/io.py deleted 100644 → 0
View file @ b7f8c4fa
import numpy as np
from astropy.io import fits
import yaml
from CpicImgSim.config import cpism_refdata, __version__, which_focalplane
from CpicImgSim.utils import Logger
import os
from datetime import datetime, timedelta
from astropy.coordinates import SkyCoord
import re
import json
import pandas as pd
config_file = f'{cpism_refdata}/cpism_config.yaml'
with open(config_file, 'r') as fid:
config = yaml.load(fid, Loader=yaml.FullLoader)
output_dir = config['output_dir']
if config['relative_path']:
ref_dir_base = os.path.dirname(cpism_refdata)
output_dir = f'{ref_dir_base}/{output_dir}'
log_dir = output_dir + '/LOG'
tmp_dir = config['tmp_dir']
log_level = config['log_level']
header_check = config['check_fits_header']
for dir in ['', 'TMP', 'CAL', 'SCI', 'LOG']:
sub_dir = f"{output_dir}/{dir}"
if not os.path.exists(sub_dir):
os.makedirs(sub_dir)
tmp_folder_path = '.'
if tmp_dir == 'TMP':
tmp_folder_path = output_dir + '/TMP'
log = Logger(log_dir+'/cpism_pack.log', log_level).logger
def check_and_update_fits_header(header):
"""
Check the header keywords and update the description according to the data model.
Parameters
-----------
header: astropy.io.fits.header.Header
The header to be checked.
Returns
--------
None
"""
hdu = 'image'
if 'FILETYPE' in header.keys():
hdu = 'primary'
model_file = f"{cpism_refdata}/io/image_header.json"
if hdu == 'primary':
model_file = f"{cpism_refdata}/io/primary_header.json"
with open(model_file, 'r', encoding='utf-8') as fid:
data_model = json.load(fid)
dm_comment = {}
def print_warning(info):
if header_check:
log.warning(info)
# check existance and format of keyword in fits header
for keyword, comment, format, _, _ in data_model:
if pd.isnull(comment):
comment = ''
if len(comment) > 46:
comment = comment[:46]
print_warning(
f"Keyword {keyword} has a comment longer than 46 characters. It will be truncated to 46 characters.")
dm_comment[keyword] = comment
if keyword not in header.keys():
print_warning(f"Keyword {keyword} not found in [{hdu}] header.")
elif not pd.isnull(format):
value = header[keyword]
# check the type of the value, I for int, R for float, C for str
if isinstance(value, str):
type = 'C'
elif isinstance(value, float):
type = 'R'
elif isinstance(value, bool):
type = 'L'
elif isinstance(value, int):
type = 'I'
else:
type = 'U'
if type != format[0]:
print_warning(
f"Keyword {keyword} has wrong type in [{hdu}]. {format[0]} expected, {type} found.")
# check if there are extral keyword in header, and update the comment
for keyword in header.keys():
# print(keyword)
if keyword not in dm_comment.keys():
print_warning(
f"Keyword {keyword} not found in the [{hdu}] data model.")
else:
header[keyword] = (header[keyword], dm_comment[keyword])
return header
def obsid_parser(
obsid: int):
"""
Parse the obsid to get the obstype.
Parameters
----------
obsid: str
The obsid of the observation.
Obsid must be 11 digits and start with 5 for CPIC.
Returns
-------
str
The obstype of the observation.
Raises
------
ValueError
If the obsid is not 11 digits or does not start with 5.
"""
obsid = str(obsid)
if len(obsid) != 11:
raise ValueError('Obsid must be 11 digits.')
if obsid[0] != '5':
raise ValueError('Obsid must start with 5 for CPIC')
obstype_dict = {
'00': 'BIAS',
'01': 'DARK',
'02': 'FLAT',
'03': 'BKGD',
'04': 'LASR',
'10': 'SCIE',
'11': 'DENF',
'12': 'CALS',
'15': 'TEMP'
}
obstype = obstype_dict.get(obsid[1:3], 'DEFT')
return obstype
def datetime_obj_to_mjd(time_obj):
"""
transfer datetime object to mean julian date (MJD).
Parameters
----------
time_obj: datetime.datetime
The datetime object.
Returns
-------
float
The mean julian date (MJD).
"""
return (time_obj - datetime(1858, 11, 17)).total_seconds() / 86400
def primary_hdu(
obs_info: dict,
gnc_info: dict,
filename_output=False):
"""
Generate the primary hdu of the fits file.
Parameters
----------
obs_info: dict
The parameters of the observation. See `save_fits` function.
gnc_info: dict
The gnc information of the observation.
filename_output: bool (optional)
If True, the folder and the filename will be returned.
Returns
-------
fits.PrimaryHDU
The primary hdu of the fits file.
str, str
The folder and filename of the fits file. Only returned if filename_output is True.
Notes
-----
The gnc_info dict should contain the information of orbit and observation.
these informations are used to genrated the hdu header. Refer to the data model for more information.
"""
camera_config, _ = load_camera_and_optics_config(obs_info['band'])
obsid = obs_info['obsid']
exp_start = gnc_info.get(
'EXPSTART', datetime.now().isoformat(timespec='seconds'))
exp_start = datetime.fromisoformat(exp_start)
duartion = (obs_info['expt'] +
camera_config['readout_time']) * obs_info['nframe']
default_end = exp_start + timedelta(seconds=duartion)
exp_end = gnc_info.get('EXPEND', default_end.isoformat(timespec='seconds'))
exp_end = datetime.fromisoformat(exp_end)
filename = "CSST_CPIC"
filename += "_" + which_focalplane(obs_info['band']).upper()
filename += "_" + obsid_parser(obsid)
filename += "_" + exp_start.strftime("%Y%m%d%H%M%S")
filename += "_" + exp_end.strftime("%Y%m%d%H%M%S")
filename += f"_{obsid}_X_L0_V01.fits"
type_dir = 'CAL'
if str(obsid)[1] == '1':
type_dir = 'SCI'
mjd_dir = f"{datetime_obj_to_mjd(exp_start):.0f}"
folder = f"{type_dir}/{mjd_dir}"
header = fits.Header()
# General keywords
header['SIMPLE'] = True
header['BITPIX'] = 8
header['NAXIS'] = 0
header['EXTEND'] = True
header['NEXTEND'] = 1 # + parameters['nframe']
header['GROUPS'] = False
header['DATE'] = datetime.now().isoformat(timespec='seconds')
heaer_filename = filename[:-4]
if len(heaer_filename) > 68:
heaer_filename = heaer_filename[:68]
header['FILENAME'] = heaer_filename
header['FILETYPE'] = obsid_parser(obsid)
header['TELESCOP'] = 'CSST'
header['INSTRUME'] = 'CPIC'
header['RADECSYS'] = 'ICRS'
header['EQUINOX'] = 2000.0
header['FITSCREA'] = f'CPISM V{__version__}'
cstar = {'ra': '0d', 'dec': '0d'}
if obs_info['target'] != {}:
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)
header['OBJECT'] = cstar.get('name', target_name)
header['TARGET'] = target_name
header['OBSID'] = str(obsid)
header['OBJ_RA'] = radec.ra.degree
header['OBJ_DEC'] = radec.dec.degree
# telescope information
header['REFFRAME'] = 'CSSTGSC-1.0'
header['DATE-OBS'] = exp_start.isoformat(timespec='seconds')
header['SATESWV'] = 'SIMULATION'
header['EXPSTART'] = datetime_obj_to_mjd(exp_start)
header['CABSTART'] = gnc_info.get('CABSTART', header['EXPSTART'])
header['SUNANGL0'] = gnc_info.get('SUNANGL0', -1.0)
header['MOONANG0'] = gnc_info.get('MOONANG0', -1.0)
header['TEL_ALT0'] = gnc_info.get('TEL_ALT0', -1.0)
header['POS_ANG0'] = gnc_info.get(
'POS_ANG0', float(obs_info['rotation']))
header['POSI0_X'] = gnc_info.get('POSI0_X', -1.0)
header['POSI0_Y'] = gnc_info.get('POSI0_Y', -1.0)
header['POSI0_Z'] = gnc_info.get('POSI0_Z', -1.0)
header['VELO0_X'] = gnc_info.get('VELO0_X', -1.0)
header['VELO0_Y'] = gnc_info.get('VELO0_Y', -1.0)
header['VELO0_Z'] = gnc_info.get('VELO0_Z', -1.0)
header['EULER0_1'] = gnc_info.get('EULER0_1', -1.0)
header['EULER0_2'] = gnc_info.get('EULER0_2', -1.0)
header['EULER0_3'] = gnc_info.get('EULER0_3', -1.0)
header['RA_PNT0'] = gnc_info.get('RA_PNT0', header['OBJ_RA'])
header['DEC_PNT0'] = gnc_info.get('DEC_PNT0', header['OBJ_DEC'])
header['EXPEND'] = datetime_obj_to_mjd(exp_end)
header['CABEND'] = gnc_info.get('CABEDN', header['EXPEND'])
header['SUNANGL1'] = gnc_info.get('SUNANGL1', header['SUNANGL0'])
header['MOONANG1'] = gnc_info.get('MOONANG1', header['MOONANG0'])
header['TEL_ALT1'] = gnc_info.get('TEL_ALT1', header['TEL_ALT0'])
header['POS_ANG1'] = gnc_info.get('POS_ANG1', header['POS_ANG0'])
header['POSI1_X'] = gnc_info.get('POSI1_X', header['POSI0_x'])
header['POSI1_Y'] = gnc_info.get('POSI1_Y', header['POSI0_y'])
header['POSI1_Z'] = gnc_info.get('POSI1_Z', header['POSI0_z'])
header['VELO1_X'] = gnc_info.get('VELO1_X', header['VELO0_x'])
header['VELO1_Y'] = gnc_info.get('VELO1_Y', header['VELO0_y'])
header['VELO1_Z'] = gnc_info.get('VELO1_Z', header['VELO0_z'])
header['EULER1_1'] = gnc_info.get('EULER1_1', header['EULER0_1'])
header['EULER1_2'] = gnc_info.get('EULER1_2', header['EULER0_2'])
header['EULER1_3'] = gnc_info.get('EULER1_3', header['EULER0_3'])
header['RA_PNT1'] = gnc_info.get('RA_PNT1', header['RA_PNT0'])
header['DEC_PNT1'] = gnc_info.get('DEC_PNT1', header['DEC_PNT0'])
header['EXPTIME'] = (exp_end - exp_start).total_seconds()
header['EPOCH'] = float(exp_start.year)
header['CHECKSUM'] = '0000000000000000'
header['DATASUM'] = '0000000000'
check_and_update_fits_header(header)
# other information
hdu = fits.PrimaryHDU(header=header)
hdu.add_checksum()
if filename_output:
return hdu, folder, filename
else:
return hdu
def load_camera_and_optics_config(band):
"""
Load camera and optics configuration from reference data.
Parameters
----------
band : str
Band name.
Returns camera_config, optics_config : dict, dict
"""
camera = which_focalplane(band)
if camera == 'vis':
config_file = 'emccd_config.yaml'
elif camera == 'nir':
raise ValueError('NIR camera is not supported yet')
config_file = 'nir_config.yaml'
with open(f"{cpism_refdata}/camera/{config_file}", 'r') as fid:
camera_config = yaml.load(fid, Loader=yaml.FullLoader)
with open(f"{cpism_refdata}/optics/optics_config.yaml", 'r') as fid:
optics_config = yaml.load(fid, Loader=yaml.FullLoader)[camera]
return camera_config, optics_config
def frame_header(obs_info, index, bunch_start, primary_header={}):
"""
Generate the header for a single frame.
Parameters
----------
obs_info : dict
Dictionary of parameters. See `save_fits` function.
index : int
Frame index.
bunch_start : str
Start time of the bunch.
primary_header : dict (optional)
Primary header. default: {}
Returns
-------
astropy.io.fits.Header
"""
header = fits.Header()
camera_config, optics_config = load_camera_and_optics_config(
obs_info['band'])
plszx = camera_config['plszx']
plszy = camera_config['plszy']
pscan1 = camera_config['pscan1']
pscan2 = camera_config['pscan2']
oscan1 = camera_config['oscan1']
oscan2 = camera_config['oscan2']
udark = camera_config['udark']
bdark = camera_config['bdark']
ldark = camera_config['ldark']
rdark = camera_config['rdark']
imgszx = plszx + pscan1 + oscan1 + ldark + rdark
imgszy = plszy + pscan2 + oscan2 + udark + bdark
header['XTENSION'] = 'IMAGE'
header['BITPIX'] = 16
header['NAXIS'] = 2
header['NAXIS1'] = 1080
header['NAXIS2'] = 1050
header['EXTNAME'] = 'IMAGE'
header['EXTVER'] = 1
header['BSCALE'] = 1.0
header['BZERO'] = 32768.0
header['BUNIT'] = 'ADU'
header['FILTER'] = obs_info['band']
header['DETSN'] = '00000000000'
header['DETNAME'] = camera_config['detector_name']
header['CHIPLAB'] = camera_config['ccd_label']
header['CHIPTEMP'] = float(camera_config['chip_temp'])
header['DEWTEMP'] = float(camera_config['dewar_temp'])
header['DETSIZE'] = f"{imgszx} * {imgszy}"
header['IMGINDEX'] = index
frame_time = obs_info['expt'] + camera_config['readout_time']
bunch_start = datetime.fromisoformat(bunch_start)
expstart = bunch_start + timedelta(seconds=frame_time * (index - 1))
bunch_start_mjd = datetime_obj_to_mjd(bunch_start)
ra0 = primary_header.get('RA_PNT0', -1.0)
dec0 = primary_header.get('DEC_PNT0', -1.0)
pa0 = primary_header.get('POS_ANG0', -1.0)
cab0 = primary_header.get('CABSTART', bunch_start_mjd)
delta_t = frame_time * index
bunch_end = bunch_start + timedelta(seconds=delta_t)
bunch_end_mjd = datetime_obj_to_mjd(bunch_end)
ra1 = primary_header.get('RA_PNT1', ra0)
dec1 = primary_header.get('DEC_PNT1', dec0)
pa1 = primary_header.get('POS_ANG1', pa0)
cab1 = primary_header.get('CABEND', bunch_end_mjd)
img_cab = datetime_obj_to_mjd(expstart)
ratio = (img_cab - cab0)/(cab1 - cab0)
ra = ra0 + (ra1 - ra0) * ratio
dec = dec0 + (dec1 - dec0) * ratio
pa = pa0 + (pa1 - pa0) * ratio
header['IMG_EXPT'] = expstart.isoformat(timespec='seconds')
header['IMG_CABT'] = header['IMG_EXPT']
header['IMG_DUR'] = float(obs_info['expt'])
header['IMG_PA'] = ra
header['IMG_RA'] = dec
header['IMG_DEC'] = pa
header['DATASECT'] = f"{plszx} * {plszy}"
header['PIXSCAL'] = optics_config['platescale']
header['PIXSIZE'] = camera_config['pitch_size']
header['NCHAN'] = 1
header['PSCAN1'] = pscan1
header['PSCAN2'] = pscan2
header['OSCAN1'] = oscan1
header['OSCAN2'] = oscan2
header['UDARK'] = udark
header['BDARK'] = bdark
header['LDARK'] = ldark
header['RDARK'] = rdark
# WCS
cstar = {'ra': '0d', 'dec': '0d'}
if obs_info['target'] != {}:
cstar = obs_info['target']['cstar']
radec = SkyCoord(cstar['ra'], cstar['dec'])
shift = obs_info['shift']
platescale = optics_config['platescale']
rotation = np.radians(obs_info['rotation'])
header['WCSAXES'] = 2
header['CRPIX1'] = (plszx + 1)/2 + pscan1 + ldark + shift[0] / platescale
header['CRPIX2'] = (plszy + 1)/2 + pscan2 + udark + shift[0] / platescale
header['CRVAL1'] = radec.ra.degree
header['CRVAL2'] = radec.dec.degree
header['CTYPE1'] = 'RA---TAN'
header['CTYPE2'] = 'DEC--TAN'
header['CD1_1'] = np.cos(rotation)
header['CD1_2'] = -np.sin(rotation)
header['CD2_1'] = np.sin(rotation)
header['CD2_2'] = np.cos(rotation)
header['others'] = 'other'
# Readout information
header['EMGAIN'] = float(obs_info['emgain'])
header['GAIN'] = float(camera_config['ph_per_adu'])
header['DET_BIAS'] = float(camera_config['bias_level'])
header['RON'] = float(camera_config['readout_noise'])
header['READTIME'] = float(camera_config['readout_time'])
header['ROSPEED'] = float(camera_config['readout_speed'])
# CPIC information
header['LS_STAT'] = 'OFF'
header['IWA'] = optics_config['mask_width'] / 2
header['WFSINFO1'] = -1.0
header['WFSINFO2'] = -1.0
header['CHECKSUM'] = '0000000000000000'
header['DATASUM'] = '0000000000'
header = check_and_update_fits_header(header)
return header
def save_fits_simple(images, obs_info):
"""
Save the image to a fits file with a simple header to TMP directory.
Parameters
----------
images : numpy.ndarray
Image array to be written.
obs_info : dict
Dictionary of observation informations. See `save_fits` function.
Returns
----------
Filename of the saved fits file.
"""
target = obs_info['target']
target_info = 'NO_TARGET'
if 'cstar' in target.keys():
target_info = ''
target_info = f"S{target['cstar']['magnitude']:.1f}"
target_info += f"_P{len(target.get('planets', '[]'))}"
name = target_info
if 'name' in target.keys():
name = target['name']
name = name.replace('/', '_')
name = name.replace(',', '_')
now = datetime.now()
time = now.strftime("%Y%m%d%H%M%S")
filename = f"{name}_{time}.fits"
header = fits.Header()
header['skybg'] = obs_info['skybg']
header['name'] = name
header['exptime'] = obs_info['expt']
header['nframe'] = obs_info['nframe']
header['band'] = obs_info['band']
header['emgain'] = obs_info['emgain']
header['obsid'] = obs_info['obsid']
header['rotation'] = obs_info['rotation']
shift = obs_info['shift']
header['shift'] = f"x:{shift[0]},y:{shift[1]}"
fullname = f"{tmp_folder_path}/{filename}"
fits.writeto(fullname, images, overwrite=True, header=header)
return fullname
def save_fits(images, obs_info, gnc_info, csst_format=True):
"""
Save the image to a fits file.
Parameters
----------
images : numpy.ndarray
Image array to be saved.
obs_info : dict
Dictionary of observation informations.
Must contain the following keys
- band: str
- Band of the image.
- expt: float
- Exposure time of the each image.
- nframe: int
- Number of frames in the image.
- emgain: int
- EM gain of the camera.
- obsid: str
- Observation ID. Obsid must be 11 digits and start with 5 for CPIC. See pharse_obsid() for details.
- rotation: float
- Rotation angle of the image.
- shift: list
- Shift of the image.
gnc_info : dict
Dictionary of GNC information.
Contains the keywords in the primary header. See primary_hdu() for details.
csst_format : bool, optional
Whether to save the fits file in CSST format, by default True.
"""
if not csst_format:
save_fits_simple(images, obs_info)
return
hdu_header, folder, filename = primary_hdu(obs_info, gnc_info, True)
hdu_list = fits.HDUList([hdu_header])
if len(images.shape) == 2:
images = np.array([images])
for index in range(images.shape[0]):
header = frame_header(
obs_info,
index + 1,
hdu_header.header['DATE-OBS'],
primary_header=hdu_header.header
)
frame_hdu = fits.ImageHDU(images[index, :, :], header=header)
frame_hdu.add_checksum()
hdu_list.append(frame_hdu)
folder = f"{output_dir}/{folder}"
if not os.path.exists(folder):
os.makedirs(folder)
hdu_list.writeto(f"{folder}/{filename}", overwrite=True)
CpicImgSim/main.py deleted 100644 → 0
View file @ b7f8c4fa
import numpy as np
import re
from .target import spectrum_generator
from .optics import make_focus_image, focal_mask, optics_config
from .psf_simulation import simulate_psf
from .camera import EMCCD, CosmicRayFrameMaker, sky_frame_maker
from .io import save_fits, log
from .config import which_focalplane
def psf_function(band, cstar_spectrum, shape, error=0.1):
cstar = True
if shape < 300:
cstar = False
return simulate_psf(error, band, cstar_spectrum, nsample=1, cstar=cstar)
def observation_simulation(
target: dict,
skybg: float,
expt: float,
nframe: int,
band: str,
emgain: float,
obsid: int = 51900000000,
rotation: float = 0,
shift: list = [0, 0],
gnc_info: dict = {},
csst_format: bool = True,
psf_function: callable = psf_function):
"""
Simulate the observation. All-In-One function of the package.
Parameters
-----------
target: dict
The target information. See target.py for details.
skybg: float
magnitude of the skybackground at the input b and. (abmag system)
expt: float
exposure time in second.
nframe: int
number of frames to be simulated.
band: str
the band of the observation. (e.g. 'f661')
emgain: float
the EM gain of the camera.
obsid: int
the observation id. Default is 51900000000.
rotation: float
the rotation angle of the target in degree. Default is 0.
shift: list
the shift of the target in arcsec. Default is [0, 0].
gnc_info: dict
the gnc information. Default is {}. See io.py for details.
csst_format: bool
whether to save the fits file in CSST format. Default is True.
psf_function: callable
the function to generate the psf. See optics.py for details.
Returns
-----------
np.ndarray of the simulated images with shape (nframe, 1088, 1050).
"""
target_list = []
if 'cstar' in target.keys():
target_list = spectrum_generator(target)
focal_name = which_focalplane(band)
this_focal_config = optics_config[focal_name]
telescope_config = optics_config['telescope']
area = telescope_config['aperature_area']
if focal_name == 'vis':
camera = EMCCD()
else:
raise ValueError('Only VIS focal plane is supported.')
platescale = this_focal_config['platescale']
iwa = this_focal_config['mask_width'] / 2
crmaker = CosmicRayFrameMaker()
images = []
params = {
'target': target,
'skybg': skybg,
'expt': expt,
'nframe': nframe,
'band': band,
'emgain': emgain,
'obsid': obsid,
'rotation': rotation,
'shift': shift,
}
paramstr = ', '.join([f'{k}={v}' for k, v in params.items()])
log.debug(f"parameters: {paramstr}")
for i in range(nframe):
log.info(f'Simulation Running: Frame {i+1}/{nframe}')
focal_frame = make_focus_image(
band,
target_list,
psf_function,
platesize=camera.flat_shape,
rotation=rotation,
init_shifts=shift,
)
if skybg is None or skybg > 100:
sky_bkg_frame = 0
else:
sky_bkg_frame = sky_frame_maker(
band,
skybg,
platescale,
camera.flat_shape
)
focal_frame = (focal_frame + sky_bkg_frame) * area
focal_frame = focal_mask(focal_frame, iwa, platescale)
cr_frame = crmaker.make_cr_frame(camera.dark_shape, expt)
img = camera.readout(
focal_frame,
emgain,
expt,
image_cosmic_ray=cr_frame
)
images.append(img)
images = np.array(images)
save_fits(images, params, gnc_info, csst_format=csst_format)
return images
def quick_run(
target_str: str,
skymag: float,
band: str,
expt: float,
nframe: int,
emgain: float,
rotation: float = 0,
shift: list = [0, 0]) -> np.ndarray:
"""
A quick run function to simulate the observation.
Parameters
-----------
target_str: str
The target information in string format.
In the format of "\*5.1/25.3(1.3,1.5)/22.1(2.3,-4.5)" which means a central star
with magnitude 5.1, and two substellar with magnitude 25.3 and 22.1, respectively.
The first number in the parenthesis is the x position in arcsec, and the second is the y position.
skybg: float
magnitude of the skybackground at the input band. (abmag system)
band: str
the band of the observation. (e.g. 'f661')
expt: float
exposure time in second.
nframe: int
number of frames to be simulated.
emgain: float
the EM gain of the camera.
rotation: float (optional)
the rotation angle of the target in degree. Default is 0.
shift: list (optional)
the shift of the target in arcsec. Default is [0, 0].
Returns
-----------
np.ndarray of the simulated images, with shape (nframe, 1088, 1050)
Notes
-----------
1. stars are assumed to be G0III with distance 10pc.
2. magnitude of the star and substellar are assumed to be in the same band.
3. Csst format is not supported.
4. The psf is assumed to be the default one.
5. fits file will be saved in the current directory.
"""
log.info(f'Quick Run: {target_str}')
target_dict = {
'name': 'cal',
}
if (target_str != '') and (target_str[0] == '*'):
objects = target_str[1:].split('/')
cstar_mag = float(objects[0])
cstar = {
'magnitude': cstar_mag,
'ra': '0d',
'dec': '0d',
'sptype': 'G0III',
'distance': 10,
'mag_input_band': band
}
stars = []
for sub_stellar in objects[1:]:
float_regex = R"[+-]?\d+(?:\.\d+)?"
match = re.match(
rf"({float_regex})\(({float_regex}),({float_regex})\)", sub_stellar)
if not match:
raise ValueError('Wrong format for sub stellar.')
mag = float(match.group(1))
x = float(match.group(2))
y = float(match.group(3))
pangle = np.arctan2(x, y) * 180 / np.pi
separation = np.sqrt(x**2 + y**2)
stars.append({
'magnitude': mag,
'pangle': pangle,
'separation': separation,
'sptype': 'G0III',
'mag_input_band': band
})
target_dict = {
'name': target_str[1:],
'cstar': cstar,
'stars': stars,
}
return observation_simulation(
target=target_dict,
skybg=skymag,
expt=expt,
nframe=nframe,
band=band,
emgain=emgain,
csst_format=False,
shift=shift,
rotation=rotation,
)
# observation_simulation(
# target={},
# skybg=15,
# expt=10,
# nframe=2,
# band='f661',
# emgain=30,
# obsid=50112345678,
# )
# quick_run('*5.1/25.3(0.8,0.8)', None, 'f661', 10, 1, 10)
# quick_run('*5/20(0.8,0.8)', None, 'f883', 10, 1, 10)
# # quick *5.1/25.3(1.3,1.5) expt nframe emgain band rotation shift
# # quick target_name expt nframe emgain band rotation shift
# # plan plan_file_or_folder
if __name__ == '__main__': # pragma: no cover
target_example = {
'cstar': {
'magnitude': 1,
'ra': '120d',
'dec': '40d',
'distance': 10,
'sptype': 'F0III',
},
'stars': [
{
'magnitude': 20,
'pangle': 60,
'separation': 1,
'sptype': 'F0III'
}
]
}
# quick_run('', 10, 'f661', 1, 1, 30)
# quick_run('*2.4/10(3,5)/15(-4,2)', 21, 'f661', 1, 1, 30)
# # normal target
observation_simulation(
target=target_example,
skybg=21,
expt=1,
nframe=2,
band='f661',
emgain=30,
obsid=51012345678,
)
# # bias
# observation_simulation(
# target=target_example,
# skybg=999,
# expt=1,
# nframe=2,
# band='f661',
# emgain=1,
# obsid=51012345678,
# shift=[3, 3],
# rotation=60
# )
# # bias-gain
# observation_simulation(
# target={},
# skybg=999,
# expt=0.01,
# nframe=2,
# band='f661',
# emgain=1000,
# obsid=50012345678,
# )
# # dark
# observation_simulation(
# target={},
# skybg=999,
# expt=100,
# nframe=2,
# band='f661',
# emgain=30,
# obsid=50112345678,
# )
# # flat
# observation_simulation(
# target={},
# skybg=15,
# expt=10,
# nframe=2,
# band='f661',
# emgain=30,
# obsid=50112345678,
# )
CpicImgSim/optics.py deleted 100644 → 0
View file @ b7f8c4fa
import os
import yaml
import time
import scipy as sp
import numpy as np
from CpicImgSim.config import cpism_refdata, which_focalplane, S # S is synphot
from CpicImgSim.config import optics_config
from CpicImgSim.utils import region_replace
from CpicImgSim.io import log
from astropy.convolution import convolve_fft
from scipy.signal import fftconvolve
FILTERS = {
'f565': S.FileBandpass(f'{cpism_refdata}/throughtput/f565_total.fits'),
'f661': S.FileBandpass(f'{cpism_refdata}/throughtput/f661_total.fits'),
'f743': S.FileBandpass(f'{cpism_refdata}/throughtput/f743_total.fits'),
'f883': S.FileBandpass(f'{cpism_refdata}/throughtput/f883_total.fits'),
'f940': S.FileBandpass(f'{cpism_refdata}/throughtput/f940_total.fits'),
'f1265': S.FileBandpass(f'{cpism_refdata}/throughtput/f1265_total.fits'),
'f1425': S.FileBandpass(f'{cpism_refdata}/throughtput/f1425_total.fits'),
'f1542': S.FileBandpass(f'{cpism_refdata}/throughtput/f1542_total.fits'),
}
def filter_throughput(filter_name):
"""
Totally throughput of the each CPIC band.
Including the throughput of the filter, telescope, cpic, and camera QE.
If the filter_name is not supported, return the throughput of the default filter(f661).
Parameters
-----------
filter_name: str
The name of the filter.
One of ['f565', 'f661'(default), 'f743', 'f883', 'f940', 'f1265', 'f1425', 'f1542']
Returns
--------
synphot.Bandpass
The throughput of the filter.
"""
filter_name = filter_name.lower()
filter_name = 'f661' if filter_name == 'default' else filter_name
if filter_name not in FILTERS.keys():
log.warning(f"滤光片名称错误({filter_name})，返回默认滤光片(f661)透过率")
filter_name = 'f661'
return FILTERS[filter_name]
def example_psf_func(band, spectrum, frame_size, error=0.1):
"""
Example psf generating function.
Parameters
-------------
band: str
The name of the band.
spectrum: synphot.Spectrum or synphot.SourceSpectrum
The spectrum of the target.
frame_size: int
The size of the frame.
error: float
Phase RMS error.
Returns
---------------
2D array
psf image with shape of `frame_size`
"""
pass
def example_monochromatic_psf(wavelength, error=0.1):
pass
def rotate_and_shift(shift, rotation, init_shifts):
rotation_rad = rotation / 180 * np.pi
return np.array([
shift[0] * np.cos(rotation_rad) + shift[1] * np.sin(rotation_rad),
-shift[0] * np.sin(rotation_rad) + shift[1] * np.cos(rotation_rad)
]) + np.array(init_shifts)
from scipy.ndimage import rotate
def ideal_focus_image(
bandpass: S.spectrum.SpectralElement,
targets: list,
platescale,
platesize: list = [1024, 1024],
init_shifts: list = [0, 0],
rotation: float = 0,):
focal_image = np.zeros(platesize)
focal_shape = np.array(platesize)[::-1] # x, y
if not targets:
return focal_image
for target in targets:
sub_x, sub_y, sub_spectrum, sub_image = target
sub_shift = rotate_and_shift([sub_x, sub_y], rotation, init_shifts) / platescale
sed = (sub_spectrum * bandpass).integrate()
if sub_image is None:
x = (focal_shape[0] - 1)/2 + sub_shift[0]
y = (focal_shape[1] - 1)/2 + sub_shift[1]
int_x = int(x)
int_y = int(y)
if int_x < 0 or int_x >= focal_shape[0] - 1 or int_y < 0 or int_y >= focal_shape[1] - 1:
continue
dx1 = x - int_x
dx0 = 1 - dx1
dy1 = y - int_y
dy0 = 1 - dy1
sub = np.array([
[dx0*dy0, dx1*dy0],
[dx0*dy1, dx1*dy1]]) * sed
focal_image[int_y: int_y+2, int_x: int_x+2] += sub
else:
# sub_image = sub_image
sub_image = np.abs(rotate(sub_image, rotation, reshape=False))
sub_image = sub_image / sub_image.sum()
sub_img_shape = np.array(sub_image.shape)[::-1]
sub_shift += (focal_shape-1)/2 - (sub_img_shape-1)/2
focal_image = region_replace(
focal_image,
sub_image * sed,
sub_shift,
subpix=True
)
return focal_image
from scipy.signal import fftconvolve
def sp_convole_fft(image, kernal):
kernal = kernal / kernal.sum()
# y0 = kernal.shape[0] // 2
# x0 = kernal.shape[1] // 2
outimg = fftconvolve(image, kernal, mode='same')
# return outimg[y0:y0+image.shape[0], x0:x0+image.shape[1]]
return outimg
def convolve_psf(
band: str,
targets: list,
psf_function: callable,
init_shifts: list = [0, 0],
rotation: float = 0,
nsample: int = 5,
error: float = 1,
platesize: list = [1024, 1024]) -> np.ndarray :
config = optics_config[which_focalplane(band)]
platescale = config['platescale']
filter = filter_throughput(band)
wave = filter.wave
throughput = filter.throughput
min_wave = wave[0]
max_wave = wave[-1]
all_fp_image = []
for i_wave in range(nsample):
d_wave = (max_wave - min_wave) / nsample
wave0 = min_wave + i_wave * d_wave
wave1 = min_wave + (i_wave + 1) * d_wave
center_wavelength = (wave0 + wave1) / 2 * 1e-10
i_throughput = throughput.copy()
i_throughput[(wave > wave1) | (wave < wave0)] = 0
i_band = S.ArrayBandpass(wave, i_throughput, waveunits='angstrom')
i_fp_image = ideal_focus_image(i_band, targets, platescale, platesize, init_shifts, rotation)
psf = psf_function(center_wavelength, error=error)
t0 = time.time()
# c_fp_image = convolve_fft(i_fp_image, psf, allow_huge=True)
c_fp_image = sp_convole_fft(i_fp_image, psf)
print(f"Convolution time: {time.time()-t0}")
all_fp_image.append(c_fp_image)
return np.array(all_fp_image).mean(axis=0)
def make_focus_image(
band: str,
targets: list,
psf_function: callable,
init_shifts: list = [0, 0],
rotation: float = 0,
platesize: list = [1024, 1024]) -> np.ndarray:
"""
Make the focus image of the targets.
Parameters
-----------
band: str
The name of the band.
targets: list
The list of the targets.
Each element of the list is a tuple of (x, y, spectrum).
- x, y: float
- The position of the target in the focal plane.
- spectrum: synphot.Spectrum or synphot.SourceSpectrum
- The spectrum of the target.
psf_function: callable
The function to generate the PSF, with same parameters and return as `example_psf_func`.
init_shifts: list
The initial shifts of the center targets. Unit: arcsec.
The default is [0, 0].
rotation: float
The rotation of the focal plane. Unit: degree.
The default is 0 degree.
platesize: list
The size of the focal plane. Unit: pixel.
The default is [1024, 1024].
Returns
--------
np.ndarray
The focus image of the targets.
2D array with the shape of platesize.
"""
config = optics_config[which_focalplane(band)]
platescale = config['platescale']
focal_image = np.zeros(platesize)
if not targets:
return focal_image
cstar_x, cstar_y, cstar_spectrum = targets[0]
cstar_shift = rotate_and_shift([cstar_x, cstar_y]) / platescale
error_value = 0 # nm
cstar_psf = psf_function(band, cstar_spectrum, config['cstar_frame_size'],
error=error_value)
platesize = np.array(platesize)[::-1]
psf_shape = np.array(cstar_psf.shape)[::-1]
cstar_shift += (platesize-1)/2 - (psf_shape-1)/2
focal_image = region_replace(
focal_image,
cstar_psf,
cstar_shift,
padded_in=False,
padded_out=False,
subpix=True)
for i_target in range(1, len(targets)):
sub_x, sub_y, sub_spectrum = targets[i_target]
pdout = False if i_target == len(targets)-1 else True
pdin = False if i_target == 1 else True
log.debug(f"input target {sub_x=:}, {sub_y=:}")
sub_shift = rotate_and_shift([sub_x, sub_y], rotation, init_shifts) / platescale
log.debug(f"after rotate and shift {sub_shift=:}")
sub_psf = psf_function(
band,
sub_spectrum,
config['substellar_frame_size'],
error=error_value
)
psf_shape = np.array(sub_psf.shape)[::-1]
sub_shift += (platesize-1)/2 - (psf_shape-1)/2
log.debug(f"input shift of region_replace: {sub_shift=:}")
focal_image = region_replace(
focal_image,
sub_psf,
sub_shift,
padded_in=pdin,
padded_out=pdout,
subpix=True
)
return focal_image
def focal_mask(image, iwa, platescale, throughtput=1e-6):
"""
Mask the image outside the inner working angle.
Parameters
-----------
image: np.ndarray
The image to be masked.
iwa: float
The inner working angle. Unit: arcsec.
platescale: float
The platescale of the image. Unit: arcsec/pixel.
throughtput: float
The throughtput of the mask. The default is 1e-6.
Returns
--------
np.ndarray
The masked image.
"""
xx, yy = np.mgrid[0:image.shape[0], 0:image.shape[1]]
center = np.array([(image.shape[0]-1)/2, (image.shape[1]-1)/2])
mask = (abs(xx - center[0]) < iwa /
platescale) | (abs(yy - center[1]) < iwa / platescale)
image_out = image.copy()
image_out[mask] *= throughtput
return image_out
CpicImgSim/psf_simulation.py deleted 100644 → 0
View file @ b7f8c4fa
import numpy as np
from astropy.io import fits
from hcipy import Field, Wavefront, DeformableMirror, FraunhoferPropagator
from hcipy import SurfaceApodizer, SurfaceAberrationAtDistance
from hcipy import make_pupil_grid, make_circular_aperture, make_focal_grid
from hcipy import make_xinetics_influence_functions
from hcipy import read_fits
from .config import cpism_refdata, S
from .optics import filter_throughput
# initial psf simulation
apm, apm_header = fits.getdata(
f'{cpism_refdata}/optics/apm.fits', header=True)
actuator = read_fits(f'{cpism_refdata}/optics/actuator.fits')
surface_aberration = read_fits(
f'{cpism_refdata}/optics/initial_phase_aberration.fits')
wavelength = 625e-9 # m
pupil_diameter = 2 # m
focal_length = pupil_diameter * 83
pupil_grid = make_pupil_grid(apm.shape[0], apm.shape[0] * apm_header['PHRATE'])
aperture = make_circular_aperture(pupil_diameter)(pupil_grid)
aperture = aperture * Field(apm.flatten(), pupil_grid)
emccd_pixel_size = 13e-6 # m
arcsec2rad = np.radians(1 / 3600)
emccd_pixel_scale = emccd_pixel_size / \
(arcsec2rad * focal_length) # arcsec / pixel
spatial_resolution = focal_length * wavelength / \
pupil_diameter # meter per airy disk
q = spatial_resolution / emccd_pixel_size # pixels per airy disk
num_airy = 512 / q # airy disk per frame (2 * 512 = 1024 pix)
focal_full_frame = make_focal_grid(
q, num_airy, spatial_resolution=spatial_resolution)
prop_full_frame = FraunhoferPropagator(
pupil_grid, focal_full_frame, focal_length)
num_airy = 128 / q # make a small frame for the planets
focal_sub_frame = make_focal_grid(
q, num_airy, spatial_resolution=spatial_resolution)
prop_sub_frame = FraunhoferPropagator(
pupil_grid, focal_sub_frame, focal_length)
num_actuators_across = 32
# dm spacing is little smaller than pupil
actuator_spacing = 0.95 / 32 * pupil_diameter
influence_functions = make_xinetics_influence_functions(
pupil_grid, num_actuators_across, actuator_spacing)
deformable_mirror = DeformableMirror(influence_functions)
aberration = SurfaceApodizer(
surface_sag=surface_aberration.flatten(), refractive_index=-1)
# arbitrary distance for the aberration to propagate
aberration_distance = 80 * focal_length
aberration = SurfaceAberrationAtDistance(aberration, aberration_distance)
def single_band_psf(wavelength, waveerror=0, aber_phase=None):
error = np.random.normal(0, waveerror*1e-9, actuator.shape)
deformable_mirror.actuators = actuator + error
wf = Wavefront(aperture, wavelength)
wf = aberration(wf)
if aber_phase is not None:
other_error = SurfaceApodizer(
surface_sag=aber_phase.flatten(), refractive_index=-1)
wf = other_error(wf)
img = prop_full_frame(deformable_mirror(wf)).intensity.shaped
return img
def simulate_psf(error_value, band, spectrum, nsample=5, cstar=True, aber_phase=None):
filter = filter_throughput(band)
wave = filter.wave
throughput = filter.throughput
min_wave = wave[0]
max_wave = wave[-1]
sed = []
sed_center_wavelength = []
for i_wave in range(nsample):
d_wave = (max_wave - min_wave) / nsample
wave0 = min_wave + i_wave * d_wave
wave1 = min_wave + (i_wave + 1) * d_wave
sed_center_wavelength.append((wave0 + wave1) / 2 * 1e-10)
i_throughput = throughput.copy()
i_throughput[(wave > wave1) | (wave < wave0)] = 0
i_band = S.ArrayBandpass(wave, i_throughput, waveunits='angstrom')
i_sed = (spectrum * i_band).integrate()
sed.append(i_sed)
error = np.random.normal(0, error_value*1e-9, actuator.shape)
imgs = []
deformable_mirror.actuators = actuator + error
prop = prop_full_frame
if not cstar:
prop = prop_sub_frame
for i_sample in range(nsample):
wf = Wavefront(aperture, sed_center_wavelength[i_sample])
wf = aberration(wf)
if aber_phase is not None:
other_error = SurfaceApodizer(
surface_sag=aber_phase.flatten(), refractive_index=-1)
wf = other_error(wf)
img = prop(deformable_mirror(wf)).intensity.shaped
imgs.append(img / img.sum() * sed[i_sample])
return np.array(imgs).sum(axis=0)
CpicImgSim/target.py deleted 100644 → 0
View file @ b7f8c4fa
This diff is collapsed.
CpicImgSim/utils.py deleted 100644 → 0
View file @ b7f8c4fa
import numpy as np
import scipy.ndimage as nd
import logging
import random
# DO NOT IMPORT CPICIMGSIM MODULES HERE
class Logger(object):
def __init__(self, filename, level='INFO'):
self.logger = logging.getLogger('cpism_log')
self.logger.setLevel(logging.DEBUG)
shinfo = logging.StreamHandler()
onlyinfo = logging.Filter()
onlyinfo.filter = lambda record: (record.levelno < logging.WARNING)
fmtstr = '%(message)s'
shinfo.setFormatter(logging.Formatter(fmtstr)) # 设置屏幕上显示的格式
shinfo.setLevel(logging.INFO)
shinfo.addFilter(onlyinfo)
sh = logging.StreamHandler()
fmtstr = '!%(levelname)s!: %(message)s [%(filename)s - %(funcName)s (line: %(lineno)d)]: '
sh.setFormatter(logging.Formatter(fmtstr)) # 设置屏幕上显示的格式
sh.setLevel(logging.WARNING)
th = logging.FileHandler(filename) # 往文件里写入#指定间隔时间自动生成文件的处理器
fmtstr = '%(asctime)s %(filename)s [%(funcName)s] - %(levelname)s: %(message)s'
th.setFormatter(logging.Formatter(fmtstr)) # 设置文件里写入的格式
th.setLevel(logging.__dict__.get(level.upper()))
self.logger.addHandler(shinfo)
self.logger.addHandler(sh)
self.logger.addHandler(th)
def random_seed_select(seed=-1):
"""
Select a random seed for numpy.random and return it.
"""
if seed == -1:
seed = random.randint(0, 2**32-1)
np.random.seed(seed)
return seed
def region_replace(
background: np.ndarray,
front: np.ndarray,
shift: list,
bmask: float = 1.0,
fmask: float = 1.0,
padded_in: bool = False,
padded_out: bool = False,
subpix: bool = False
):
"""
replace a region of the background with the front image.
Parameters
----------
background: np.ndarray
The background image.
front: np.ndarray
The front image.
shift: list
The [x, y] shift of the front image. Unit: pixel.
Relative to the lower-left corner of the background image.
[0, 0] means the lower-left corner of the front image is at the lower-left corner of the background image.
bmask: float
The mask of the background image. Default: 1.0
0.0 means the background image is masked.
1.0 means the background image is fully added.
fmask: float
The mask of the front image. Default: 1.0
0.0 means the front image is masked (not added).
1.0 means the front image is fully added.
padded_in: bool
Whether the input background image is padded. Default: False
In the function, the background image is padded by the size of the front image.
If True, means the background image is padded.
padded_out: bool
Whether the output image is padded. Default: False
In the function, the background image is padded by the size of the front image.
If True, means the output image is padded.
padded_in and padded_out are designed for the case that replace_region fuction is called multiple times.
subpix: bool
Whether the shift is subpixel. Default: False
If True, the shift is subpixel, using scipy.ndimage.shift to shift the front image.
If False, the shift is integer, using numpy slicing to shift the front image.
Returns
-------
np.ndarray
The output image.
shape = background.shape if padded_out = False
shape = background.shape + 2 * front.shape if padded_out = True
"""
int_shift = np.array(shift).astype(int)
b_sz = np.array(background.shape)
f_sz = np.array(front.shape)
if padded_in:
padded = background
b_sz = b_sz - f_sz * 2
else:
padded = np.pad(background, ((f_sz[0], f_sz[0]), (f_sz[1], f_sz[1])))
if np.any((int_shift < -b_sz) | (int_shift > b_sz)):
if padded_out:
return padded
return background
if subpix:
subs = np.array(shift) - int_shift
front = nd.shift(front, (subs[0], subs[1]))
int_shift += f_sz
roi_y = int_shift[1]
roi_x = int_shift[0]
padded[roi_y: roi_y+f_sz[0], roi_x:roi_x+f_sz[1]] *= bmask
padded[roi_y: roi_y+f_sz[0], roi_x:roi_x+f_sz[1]] += fmask * front
if padded_out:
return padded
return padded[f_sz[0]:b_sz[0]+f_sz[0], f_sz[1]:b_sz[1]+f_sz[1]]
csst_cpic_sim/__init__.py
View file @ 95f81f91
from .main import quick_run, observation_simulation
from .optics import make_focus_image, focal_mask
# from .main import quick_run, observation_simulation
from .optics import focal_mask
from .target import star_photlam, planet_contrast, extract_target_x_y, spectrum_generator
from .camera import EMCCD, CosmicRayFrameMaker, sky_frame_maker
from .camera import CosmicRayFrameMaker, sky_frame_maker
from .config import __version__
__all__ = [
"EMCCD",
"CosmicRayFrameMaker",
"sky_frame_maker",
"star_photlam",
"planet_contrast",
"extract_target_x_y",
"spectrum_generator",
"make_focus_image",
"focal_mask",
"quick_run",
"observation_simulation"
]
\ No newline at end of file
csst_cpic_sim/camera.py
View file @ 95f81f91
This diff is collapsed.
csst_cpic_sim/config.py
View file @ 95f81f91
import os, yaml
import warnings
from datetime import datetime
import numpy as np
cpism_refdata = '/nfsdata/share/simulation-unittest/cpic_sim/cpism_refdata/'
print(cpism_refdata)
if not os.path.exists(cpism_refdata): # pragma: no cover
raise Exception(
"Can not find CPISM reference data.")
os.environ['PYSYN_CDBS'] = '/nfsdata/share/simulation-unittest/cpic_sim/starmodel/grp/redcat/trds/'
if not os.path.exists(os.environ.get('PYSYN_CDBS', '/home/ubuntu/Downloads/cpic-img-sim-master/cpic-img-sim-master/trd/grp/redcat/trds/')): # pragma: no cover
raise Exception(
"Can not find PYSYN Stellar reference data.")
config_aim = os.path.dirname(os.path.dirname(__file__))
config_aim = os.path.join(config_aim, 'data/refdata_path.yaml')
config_set = False
def set_config(refdata_path=None):
if refdata_path is None:
print("input cpism refencence data folder")
refdata_path = input()
refdata_path = os.path.abspath(refdata_path)
with open(config_aim, 'w') as f:
yaml.dump(refdata_path, f)
return refdata_path
try:
with open(config_aim, 'r') as f:
cpism_refdata = yaml.load(f, Loader=yaml.FullLoader)
if not os.path.isdir(cpism_refdata):
raise FileNotFoundError('cpism refdata path not found')
config_set = True
except FileNotFoundError:
warnings.warn(f'refdata not setup yet, set it before use')
cpism_refdata = set_config()
config = {}
config['cpism_refdata'] = cpism_refdata
config['utc0'] = '2024-05-01T00:00:00'
config['hybrid_model'] = f'{cpism_refdata}/target_model/hybrid_model.fits'
config['bcc_model'] = f'{cpism_refdata}/target_model/bccmodels'
config['mag_system'] = 'abmag'
config['apm_file'] = f'{cpism_refdata}/optics/apm.fits'
config['actuator_file'] = f'{cpism_refdata}/optics/actuator.fits'
config['aberration'] = f'{cpism_refdata}/optics/initial_phase_aberration.fits'
config['mask_width'] = 0.4
config['check_fits_header'] = False
config['bands'] = {
'f661': f'{cpism_refdata}/throughtput/f661_total.fits',
'f743': f'{cpism_refdata}/throughtput/f743_total.fits',
'f883': f'{cpism_refdata}/throughtput/f883_total.fits',
'f565': f'{cpism_refdata}/throughtput/f565_total.fits',
}
config['diameter'] = 2 # in meters
config['platescale'] = 0.016153
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['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['catalog_folder'] = f'{cpism_refdata}/demo_catalog'
config['csst_format'] = True
config['nsample'] = 5
update_able_keys = [
'apm_file', 'actuator_file', 'aberration', 'log_dir', 'log_level', 'catalog_folder', 'nsample', 'csst_format', 'output', 'check_fits_header'
]
def replace_cpism_refdata(config, output='$'):
aim = cpism_refdata
target = '${cpism_refdata}'
if output != '$':
aim, target = target, aim
for key, value in config.items():
if isinstance(value, str):
config[key] = value.replace(aim, target)
if isinstance(value, dict):
replace_cpism_refdata(value, output)
with open(cpism_refdata + '/cpism_config.yaml', 'r') as f:
new_config = yaml.load(f, Loader=yaml.FullLoader)
replace_cpism_refdata(new_config, None)
config.update(new_config)
os.environ['PYSYN_CDBS'] = config['pysyn_refdata']
__version__ = '2.0.0'
# we need to ignore the warning from pysynphot, because we only use the star models.
with warnings.catch_warnings(): # pragma: no cover
warnings.filterwarnings("ignore")
import pysynphot as S
solar_spectrum = S.FileSpectrum(
f"{os.environ['PYSYN_CDBS']}/grid/solsys/solar_spec.fits")
solar_spectrum.convert('photlam')
config_file = cpism_refdata + '/optics/optics_config.yaml'
if not os.path.exists(config_file): # pragma: no cover
raise FileNotFoundError(f"光学配置文件不存在({config_file})")
with open(config_file, 'r') as f:
optics_config = yaml.load(f, Loader=yaml.FullLoader)
def setup_config(new_config):
config.update(new_config)
config['utc0_float'] = datetime.timestamp(datetime.fromisoformat(config['utc0']))
config['solar_spectrum'] = f"{os.environ['PYSYN_CDBS']}/grid/solsys/solar_spec.fits"
config['aperature_area'] = (config['diameter'] * 50)**2 * np.pi # cm^2
config['default_band'] = list(config['bands'].keys())[0]
config['default_filter'] = config['bands'][config['default_band']]
MAG_SYSTEM = 'abmag'
__version__ = '1.0.0'
setup_config({})
def which_focalplane(band):
"""
......@@ -61,3 +130,24 @@ def which_focalplane(band):
return 'wfs'
raise ValueError(f"未知的波段{band}")
def iso_time(time):
if isinstance(time, str):
_ = datetime.fromisoformat(time)
return time
utc0 = config['utc0']
time0 = datetime.timestamp(datetime.fromisoformat(utc0))
time = datetime.fromtimestamp(time0 + time)
return time.isoformat()
def relative_time(time):
if isinstance(time, float):
return time
if isinstance(time, int):
return float(int)
utc0 = config['utc0']
time0 = datetime.timestamp(datetime.fromisoformat(utc0))
return datetime.timestamp(datetime.fromisoformat(time)) - time0
\ No newline at end of file
csst_cpic_sim/io.py
View file @ 95f81f91
import yaml, os, re
from datetime import datetime
import numpy as np
import pandas as pd
from astropy.io import fits
import yaml
from .config import cpism_refdata, __version__, which_focalplane
from .utils import Logger
import os
from datetime import datetime, timedelta
from astropy.coordinates import SkyCoord
import re
import json
import pandas as pd
config_file = f'{cpism_refdata}/cpism_config.yaml'
with open(config_file, 'r') as fid:
config = yaml.load(fid, Loader=yaml.FullLoader)
output_dir = config['output_dir']
if config['relative_path']:
ref_dir_base = os.path.dirname(cpism_refdata)
output_dir = f'{ref_dir_base}/{output_dir}'
from .config import __version__, which_focalplane
from .utils import Logger
from .config import config, iso_time
log_dir = output_dir + '/LOG'
tmp_dir = config['tmp_dir']
default_output_dir = config['output']
log_level = config['log_level']
header_check = config['check_fits_header']
for dir in ['', 'TMP', 'CAL', 'SCI', 'LOG']:
sub_dir = f"{output_dir}/{dir}"
if not os.path.exists(sub_dir):
os.makedirs(sub_dir)
tmp_folder_path = '.'
if tmp_dir == 'TMP':
tmp_folder_path = output_dir + '/TMP'
log_dir = config['log_dir']
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log = Logger(log_dir+'/cpism_pack.log', log_level).logger
......@@ -53,17 +34,19 @@ def check_and_update_fits_header(header):
None
"""
hdu = 'image'
if 'FILETYPE' in header.keys():
hdu = 'primary'
model_file = f"{cpism_refdata}/io/image_header.json"
if hdu == 'primary':
model_file = f"{cpism_refdata}/io/primary_header.json"
# model_file = f"{cpism_refdata}/io/image_header.json"
# if hdu == 'primary':
# model_file = f"{cpism_refdata}/io/primary_header.json"
model_file = config['datamodel']
with open(model_file, 'r', encoding='utf-8') as fid:
data_model = json.load(fid)
data_model = yaml.load(fid, Loader=yaml.FullLoader)
if 'FILETYPE' in header.keys():
header_model =data_model['HDU0']
hdu = 'hdu0'
else:
header_model = data_model['HDU1']
hdu = 'hdu1'
dm_comment = {}
def print_warning(info):
......@@ -71,37 +54,42 @@ def check_and_update_fits_header(header):
log.warning(info)
# check existance and format of keyword in fits header
for keyword, comment, format, _, _ in data_model:
for _, content in header_model.items():
comment = content['comment']
keyword = content['key']
dtype = content['dtype']
if pd.isnull(comment):
comment = ''
if len(comment) > 46:
comment = comment[:46]
if len(comment) > 47:
# comment = comment[:46]
print_warning(
f"Keyword {keyword} has a comment longer than 46 characters. It will be truncated to 46 characters.")
f"Keyword {keyword} has a comment longer than 47 characters.")
dm_comment[keyword] = comment
if keyword not in header.keys():
print_warning(f"Keyword {keyword} not found in [{hdu}] header.")
print_warning(f"Keyword {keyword} not found in header.")
elif not pd.isnull(format):
elif not pd.isnull(dtype):
value = header[keyword]
# check the type of the value, I for int, R for float, C for str
if isinstance(value, str):
type = 'C'
key_type = 'str'
elif isinstance(value, float):
type = 'R'
key_type = 'flo'
elif isinstance(value, bool):
type = 'L'
key_type = 'boo'
elif isinstance(value, int):
type = 'I'
key_type = 'int'
elif isinstance(value, type(header['COMMENT'])):
key_type = 'str'
else:
type = 'U'
key_type = 'ukn'
if type != format[0]:
if key_type != dtype[0:3]:
print_warning(
f"Keyword {keyword} has wrong type in [{hdu}]. {format[0]} expected, {type} found.")
f"Keyword {keyword} has wrong type in [{hdu}]. {dtype} expected, {key_type} found.")
# check if there are extral keyword in header, and update the comment
for keyword in header.keys():
......@@ -109,7 +97,7 @@ def check_and_update_fits_header(header):
if keyword not in dm_comment.keys():
print_warning(
f"Keyword {keyword} not found in the [{hdu}] data model.")
else:
elif keyword != 'COMMENT': # comment keyword is not allowed to be updated
header[keyword] = (header[keyword], dm_comment[keyword])
return header
......@@ -140,19 +128,19 @@ def obsid_parser(
if len(obsid) != 11:
raise ValueError('Obsid must be 11 digits.')
if obsid[0] != '5':
raise ValueError('Obsid must start with 5 for CPIC')
if obsid[0] != '4':
raise ValueError('Obsid must start with 4 for CPIC')
obstype_dict = {
'00': 'BIAS',
'01': 'DARK',
'02': 'FLAT',
'03': 'BKGD',
'04': 'LASR',
'10': 'SCIE',
'11': 'DENF',
'12': 'CALS',
'15': 'TEMP'
'20': 'BIAS',
'21': 'DARK',
'22': 'FLAT',
'23': 'BKG',
'24': 'LASER',
'01': 'SCI',
'02': 'DSF',
'10': 'CALS',
'00': 'TEMP'
}
obstype = obstype_dict.get(obsid[1:3], 'DEFT')
return obstype
......@@ -204,19 +192,14 @@ def primary_hdu(
these informations are used to genrated the hdu header. Refer to the data model for more information.
"""
camera_config, _ = load_camera_and_optics_config(obs_info['band'])
# camera_config, _ = load_camera_and_optics_config(obs_info['band'])
obsid = obs_info['obsid']
exp_start = gnc_info.get(
'EXPSTART', datetime.now().isoformat(timespec='seconds'))
exp_start = obs_info.get('EXPSTART')
exp_start = datetime.fromisoformat(exp_start)
duartion = (obs_info['expt'] +
camera_config['readout_time']) * obs_info['nframe']
default_end = exp_start + timedelta(seconds=duartion)
exp_end = gnc_info.get('EXPEND', default_end.isoformat(timespec='seconds'))
exp_end = obs_info['EXPEND']
exp_end = datetime.fromisoformat(exp_end)
filename = "CSST_CPIC"
......@@ -227,7 +210,7 @@ def primary_hdu(
filename += f"_{obsid}_X_L0_V01.fits"
type_dir = 'CAL'
if str(obsid)[1] == '1':
if int(f'{obsid}'[1:3]) <= 10:
type_dir = 'SCI'
mjd_dir = f"{datetime_obj_to_mjd(exp_start):.0f}"
......@@ -240,7 +223,7 @@ def primary_hdu(
header['NAXIS'] = 0
header['EXTEND'] = True
header['NEXTEND'] = 1 # + parameters['nframe']
header['GROUPS'] = False
# header['GROUPS'] = False
header['DATE'] = datetime.now().isoformat(timespec='seconds')
heaer_filename = filename[:-4]
......@@ -252,7 +235,8 @@ def primary_hdu(
header['INSTRUME'] = 'CPIC'
header['RADECSYS'] = 'ICRS'
header['EQUINOX'] = 2000.0
header['FITSCREA'] = f'CPISM V{__version__}'
header['FITSSWV'] = f'CPISM V{__version__}'
header['COMMENT'] = ''
cstar = {'ra': '0d', 'dec': '0d'}
if obs_info['target'] != {}:
......@@ -264,16 +248,19 @@ def primary_hdu(
header['OBJECT'] = cstar.get('name', target_name)
header['TARGET'] = target_name
header['OBSID'] = str(obsid)
header['OBJ_RA'] = radec.ra.degree
header['OBJ_DEC'] = radec.dec.degree
header['RA_OBJ'] = radec.ra.degree
header['DEC_OBJ'] = radec.dec.degree
# telescope information
header['REFFRAME'] = 'CSSTGSC-1.0'
header['DATE-OBS'] = exp_start.isoformat(timespec='seconds')
header['SATESWV'] = 'SIMULATION'
header['SATESWV'] = '1'
header['EXPSTART'] = datetime_obj_to_mjd(exp_start)
header['CABSTART'] = gnc_info.get('CABSTART', header['EXPSTART'])
cabstart = gnc_info.get('CABSTART', exp_start.isoformat(timespec='seconds'))
cabstart = iso_time(cabstart)
cabstart_mjd = datetime_obj_to_mjd(datetime.fromisoformat(cabstart))
header['CABSTART'] = cabstart_mjd
header['SUNANGL0'] = gnc_info.get('SUNANGL0', -1.0)
header['MOONANG0'] = gnc_info.get('MOONANG0', -1.0)
header['TEL_ALT0'] = gnc_info.get('TEL_ALT0', -1.0)
......@@ -288,10 +275,14 @@ def primary_hdu(
header['EULER0_1'] = gnc_info.get('EULER0_1', -1.0)
header['EULER0_2'] = gnc_info.get('EULER0_2', -1.0)
header['EULER0_3'] = gnc_info.get('EULER0_3', -1.0)
header['RA_PNT0'] = gnc_info.get('RA_PNT0', header['OBJ_RA'])
header['DEC_PNT0'] = gnc_info.get('DEC_PNT0', header['OBJ_DEC'])
header['RA_PNT0'] = gnc_info.get('RA_PNT0', header['RA_OBJ'])
header['DEC_PNT0'] = gnc_info.get('DEC_PNT0', header['DEC_OBJ'])
header['EXPEND'] = datetime_obj_to_mjd(exp_end)
cabend = gnc_info.get('CABEND', exp_end.isoformat(timespec='seconds'))
cabend = iso_time(cabend)
cabend_mjd = datetime_obj_to_mjd(datetime.fromisoformat(cabend))
header['CABEND'] = gnc_info.get('CABEDN', header['EXPEND'])
header['SUNANGL1'] = gnc_info.get('SUNANGL1', header['SUNANGL0'])
header['MOONANG1'] = gnc_info.get('MOONANG1', header['MOONANG0'])
......@@ -309,8 +300,8 @@ def primary_hdu(
header['RA_PNT1'] = gnc_info.get('RA_PNT1', header['RA_PNT0'])
header['DEC_PNT1'] = gnc_info.get('DEC_PNT1', header['DEC_PNT0'])
header['EXPTIME'] = (exp_end - exp_start).total_seconds()
header['EPOCH'] = float(exp_start.year)
header['EXPTIME'] = (exp_end - exp_start).total_seconds()
header['CHECKSUM'] = '0000000000000000'
header['DATASUM'] = '0000000000'
......@@ -327,35 +318,7 @@ def primary_hdu(
return hdu
def load_camera_and_optics_config(band):
"""
Load camera and optics configuration from reference data.
Parameters
----------
band : str
Band name.
Returns camera_config, optics_config : dict, dict
"""
camera = which_focalplane(band)
if camera == 'vis':
config_file = 'emccd_config.yaml'
elif camera == 'nir':
raise ValueError('NIR camera is not supported yet')
config_file = 'nir_config.yaml'
with open(f"{cpism_refdata}/camera/{config_file}", 'r') as fid:
camera_config = yaml.load(fid, Loader=yaml.FullLoader)
with open(f"{cpism_refdata}/optics/optics_config.yaml", 'r') as fid:
optics_config = yaml.load(fid, Loader=yaml.FullLoader)[camera]
return camera_config, optics_config
def frame_header(obs_info, index, bunch_start, primary_header={}):
def frame_header(obs_info, index, primary_header, camera_dict={}):
"""
Generate the header for a single frame.
......@@ -365,10 +328,8 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
Dictionary of parameters. See `save_fits` function.
index : int
Frame index.
bunch_start : str
Start time of the bunch.
primary_header : dict (optional)
Primary header. default: {}
primary_header : dict
Primary header
Returns
-------
......@@ -377,9 +338,9 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
"""
header = fits.Header()
camera_config, optics_config = load_camera_and_optics_config(
obs_info['band'])
# camera_config, optics_config = load_camera_and_optics_config(
# obs_info['band'])
camera_config = camera_dict
plszx = camera_config['plszx']
plszy = camera_config['plszy']
pscan1 = camera_config['pscan1']
......@@ -398,48 +359,49 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
header['NAXIS'] = 2
header['NAXIS1'] = 1080
header['NAXIS2'] = 1050
header['PCOUNT'] = 0
header['GCOUNT'] = 1
header['BSCALE'] = 1
header['BZERO'] = 32768
header['EXTNAME'] = 'IMAGE'
header['EXTVER'] = 1
header['BSCALE'] = 1.0
header['BZERO'] = 32768.0
header['BUNIT'] = 'ADU'
header['FILTER'] = obs_info['band']
header['DETSN'] = '00000000000'
header['DETSN'] = '0'
header['DETNAME'] = camera_config['detector_name']
header['CHIPLAB'] = camera_config['ccd_label']
header['CHIPTEMP'] = float(camera_config['chip_temp'])
header['DEWTEMP'] = float(camera_config['dewar_temp'])
header['DEWTEMP'] = float(camera_config['cooler_temp'])
frame_info = obs_info['frame_info'][index]
header['CHIPTEMP'] = float(frame_info['chiptemp'])
header['DETSIZE'] = f"{imgszx} * {imgszy}"
header['IMGINDEX'] = index
frame_time = obs_info['expt'] + camera_config['readout_time']
bunch_start = datetime.fromisoformat(bunch_start)
expstart = bunch_start + timedelta(seconds=frame_time * (index - 1))
bunch_start_mjd = datetime_obj_to_mjd(bunch_start)
header['IMGINDEX'] = index + 1
utc0_float = config['utc0_float']
ra0 = primary_header.get('RA_PNT0', -1.0)
dec0 = primary_header.get('DEC_PNT0', -1.0)
pa0 = primary_header.get('POS_ANG0', -1.0)
cab0 = primary_header.get('CABSTART', bunch_start_mjd)
delta_t = frame_time * index
bunch_end = bunch_start + timedelta(seconds=delta_t)
bunch_end_mjd = datetime_obj_to_mjd(bunch_end)
cab0 = primary_header.get('CABSTART', obs_info['frame_info'][0]['expt_start'])
ra1 = primary_header.get('RA_PNT1', ra0)
dec1 = primary_header.get('DEC_PNT1', dec0)
pa1 = primary_header.get('POS_ANG1', pa0)
cab1 = primary_header.get('CABEND', bunch_end_mjd)
cab1 = primary_header.get('CABEND', cab0)
frame_stamp = frame_info['expt_start'] + utc0_float
frame_mjd = datetime_obj_to_mjd(datetime.fromtimestamp(frame_stamp))
img_cab = datetime_obj_to_mjd(expstart)
cab0 = iso_time(cab0)
cab1 = iso_time(cab1)
cab0 = datetime_obj_to_mjd(datetime.fromisoformat(cab0))
cab1 = datetime_obj_to_mjd(datetime.fromisoformat(cab1))
ratio = (img_cab - cab0)/(cab1 - cab0)
ratio = (frame_mjd - cab0)/(cab1 - cab0)
ra = ra0 + (ra1 - ra0) * ratio
dec = dec0 + (dec1 - dec0) * ratio
pa = pa0 + (pa1 - pa0) * ratio
header['IMG_EXPT'] = expstart.isoformat(timespec='seconds')
header['IMG_EXPT'] = datetime.fromtimestamp(frame_stamp).isoformat(timespec='seconds')
header['IMG_CABT'] = header['IMG_EXPT']
header['IMG_DUR'] = float(obs_info['expt'])
......@@ -448,8 +410,8 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
header['IMG_DEC'] = pa
header['DATASECT'] = f"{plszx} * {plszy}"
header['PIXSCAL'] = optics_config['platescale']
header['PIXSIZE'] = camera_config['pitch_size']
header['PIXSCAL'] = frame_info['platescale']
header['PIXSIZE'] = float(camera_config['pitch_size'])
header['NCHAN'] = 1
header['PSCAN1'] = pscan1
header['PSCAN2'] = pscan2
......@@ -467,7 +429,7 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
radec = SkyCoord(cstar['ra'], cstar['dec'])
shift = obs_info['shift']
platescale = optics_config['platescale']
platescale = frame_info['platescale']
rotation = np.radians(obs_info['rotation'])
header['WCSAXES'] = 2
......@@ -481,7 +443,6 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
header['CD1_2'] = -np.sin(rotation)
header['CD2_1'] = np.sin(rotation)
header['CD2_2'] = np.cos(rotation)
header['others'] = 'other'
# Readout information
header['EMGAIN'] = float(obs_info['emgain'])
......@@ -493,7 +454,7 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
# CPIC information
header['LS_STAT'] = 'OFF'
header['IWA'] = optics_config['mask_width'] / 2
header['IWA'] = frame_info['iwa']
header['WFSINFO1'] = -1.0
header['WFSINFO2'] = -1.0
......@@ -505,7 +466,7 @@ def frame_header(obs_info, index, bunch_start, primary_header={}):
return header
def save_fits_simple(images, obs_info):
def save_fits_simple(images, obs_info, output_folder=None):
"""
Save the image to a fits file with a simple header to TMP directory.
......@@ -551,12 +512,20 @@ def save_fits_simple(images, obs_info):
shift = obs_info['shift']
header['shift'] = f"x:{shift[0]},y:{shift[1]}"
if output_folder is None:
fullname = f"{tmp_folder_path}/{filename}"
else:
fullname = f"{output_folder}/{filename}"
if os.path.exists(output_folder) is False:
os.makedirs(output_folder)
log.debug(f"Output folder {output_folder} is created.")
log.debug(f"save fits file to {fullname}")
fits.writeto(fullname, images, overwrite=True, header=header)
return fullname
def save_fits(images, obs_info, gnc_info, csst_format=True):
def save_fits(images, obs_info, gnc_info, camera_dict={}, csst_format=True, output_folder=None):
"""
Save the image to a fits file.
......@@ -592,7 +561,7 @@ def save_fits(images, obs_info, gnc_info, csst_format=True):
"""
if not csst_format:
save_fits_simple(images, obs_info)
save_fits_simple(images, obs_info, output_folder=output_folder)
return
hdu_header, folder, filename = primary_hdu(obs_info, gnc_info, True)
......@@ -604,16 +573,23 @@ def save_fits(images, obs_info, gnc_info, csst_format=True):
for index in range(images.shape[0]):
header = frame_header(
obs_info,
index + 1,
hdu_header.header['DATE-OBS'],
primary_header=hdu_header.header
index,
hdu_header.header,
camera_dict=camera_dict,
)
frame_hdu = fits.ImageHDU(images[index, :, :], header=header)
frame_hdu.add_checksum()
hdu_list.append(frame_hdu)
folder = f"{output_dir}/{folder}"
if output_folder is None:
folder = f"{default_output_dir}/{folder}"
else:
folder = f"{output_folder}/{folder}"
if not os.path.exists(folder):
os.makedirs(folder)
log.debug(f'make new folder {folder}')
hdu_list.writeto(f"{folder}/{filename}", overwrite=True)
full_path = f"{folder}/{filename}"
log.debug(f'save fits file: {full_path}')
hdu_list.writeto(full_path, overwrite=True)
csst_cpic_sim/main.py
View file @ 95f81f91
This diff is collapsed.
csst_cpic_sim/optics.py
View file @ 95f81f91
import os
import yaml
import numpy as np
from .config import cpism_refdata, which_focalplane, S # S is synphot
from .config import optics_config
from scipy.signal import fftconvolve
from scipy.ndimage import rotate
from .config import config, S # S is synphot
from .utils import region_replace
from .io import log
from .psf_simulation import single_band_masked_psf, single_band_psf
FILTERS = {
"f565": S.FileBandpass(f"{cpism_refdata}/throughtput/f565_total.fits"),
"f661": S.FileBandpass(f"{cpism_refdata}/throughtput/f661_total.fits"),
"f743": S.FileBandpass(f"{cpism_refdata}/throughtput/f743_total.fits"),
"f883": S.FileBandpass(f"{cpism_refdata}/throughtput/f883_total.fits"),
"f940": S.FileBandpass(f"{cpism_refdata}/throughtput/f940_total.fits"),
"f1265": S.FileBandpass(f"{cpism_refdata}/throughtput/f1265_total.fits"),
"f1425": S.FileBandpass(f"{cpism_refdata}/throughtput/f1425_total.fits"),
"f1542": S.FileBandpass(f"{cpism_refdata}/throughtput/f1542_total.fits")
}
FILTERS = {}
for key, value in config['bands'].items():
FILTERS[key] = S.FileBandpass(value)
default_band = config['default_band']
def filter_throughput(filter_name):
"""
Totally throughput of each CPIC band.
Totally throughput of the each CPIC band.
Including the throughput of the filter, telescope, cpic, and camera QE.
If the filter_name is not supported, return the throughput of the default filter(f661).
......@@ -39,142 +33,132 @@ def filter_throughput(filter_name):
"""
filter_name = filter_name.lower()
filter_name = "f661" if filter_name == "default" else filter_name
filter_name = default_band if filter_name == 'default' else filter_name
if filter_name not in FILTERS.keys():
log.warning(f"滤光片名称错误({filter_name})，返回默认滤光片(f661)透过率")
filter_name = "f661"
log.warning(f"滤光片名称错误({filter_name})，返回默认滤光片({default_band})透过率")
filter_name = default_band
return FILTERS[filter_name]
def example_psf_func(band, spectrum, frame_size, error=0.1):
"""
Example psf generating function.
def rotate_and_shift(shift, rotation, init_shifts):
rotation_rad = rotation / 180 * np.pi
return np.array([
shift[0] * np.cos(rotation_rad) + shift[1] * np.sin(rotation_rad),
-shift[0] * np.sin(rotation_rad) + shift[1] * np.cos(rotation_rad)
]) + np.array(init_shifts)
Parameters
-------------
band: str
The name of the band.
spectrum: synphot.Spectrum or synphot.SourceSpectrum
The spectrum of the target.
frame_size: int
The size of the frame.
error: float
Phase RMS error.
Returns
---------------
2D array
psf image with shape of `frame_size`
def ideal_focus_image(
bandpass: S.spectrum.SpectralElement,
targets: list,
platescale,
platesize: list = [1024, 1024],
init_shifts: list = [0, 0],
rotation: float = 0,):
"""
pass
focal_image = np.zeros(platesize)
focal_shape = np.array(platesize)[::-1] # x, y
if not targets:
return focal_image
for target in targets:
sub_x, sub_y, sub_spectrum, sub_image = target
sub_shift = rotate_and_shift([sub_x, sub_y], rotation, init_shifts) / platescale
sed = (sub_spectrum * bandpass).integrate()
if sub_image is None:
x = (focal_shape[0] - 1)/2 + sub_shift[0]
y = (focal_shape[1] - 1)/2 + sub_shift[1]
int_x = int(x)
int_y = int(y)
if int_x < 0 or int_x >= focal_shape[0] - 1 or int_y < 0 or int_y >= focal_shape[1] - 1:
continue
dx1 = x - int_x
dx0 = 1 - dx1
dy1 = y - int_y
dy0 = 1 - dy1
sub = np.array([
[dx0*dy0, dx1*dy0],
[dx0*dy1, dx1*dy1]]) * sed
focal_image[int_y: int_y+2, int_x: int_x+2] += sub
else:
# sub_image = sub_image
sub_image = np.abs(rotate(sub_image, rotation, reshape=False))
sub_image = sub_image / sub_image.sum()
sub_img_shape = np.array(sub_image.shape)[::-1]
sub_shift += (focal_shape-1)/2 - (sub_img_shape-1)/2
focal_image = region_replace(
focal_image,
sub_image * sed,
sub_shift,
subpix=True
)
return focal_image
def make_focus_image(
def focal_convolve(
band: str,
targets: list,
psf_function: callable,
init_shifts: list = [0, 0],
rotation: float = 0,
platesize: list = [1024, 1024],
) -> np.ndarray:
"""
Make the focus image of the targets.
nsample: int = 5,
error: float = 0,
platesize: list = [1024, 1024]) -> np.ndarray :
Parameters
-----------
band: str
The name of the band.
targets: list
The list of the targets.
Each element of the list is a tuple of (x, y, spectrum).
- x, y: float
- The position of the target in the focal plane.
- spectrum: synphot.Spectrum or synphot.SourceSpectrum
- The spectrum of the target.
psf_function: callable
The function to generate the PSF, with same parameters and return as `example_psf_func`.
init_shifts: list
The initial shifts of the center targets. Unit: arcsec.
The default is [0, 0].
rotation: float
The rotation of the focal plane. Unit: degree.
The default is 0 degree.
platesize: list
The size of the focal plane. Unit: pixel.
The default is [1024, 1024].
# config = optics_config[which_focalplane(band)]
platescale = config['platescale']
Returns
--------
np.ndarray
The focus image of the targets.
2D array with the shape of platesize.
"""
# telescope_config = optics_config['telescope']
area = config['aperature_area']
config = optics_config[which_focalplane(band)]
platescale = config["platescale"]
filter = filter_throughput(band)
wave = filter.wave
throughput = filter.throughput
min_wave = wave[0]
max_wave = wave[-1]
focal_image = np.zeros(platesize)
if not targets:
return focal_image
platescale = config['platescale']
iwa = config['mask_width'] / 2
def rotate_and_shift(shift):
rotation_rad = rotation / 180 * np.pi
return np.array(
[
shift[0] * np.cos(rotation_rad) + shift[1] * np.sin(rotation_rad),
-shift[0] * np.sin(rotation_rad) + shift[1] * np.cos(rotation_rad),
]
) + np.array(init_shifts)
if abs(init_shifts[0]) > 4 or abs(init_shifts[1]) > 4:
print('Input shifts are too large, and are set to zero')
init_shifts = [0, 0]
cstar_x, cstar_y, cstar_spectrum = targets[0]
cstar_shift = rotate_and_shift([cstar_x, cstar_y]) / platescale
all_fp_image = []
if not targets:
return np.zeros((platesize[1], platesize[0]))
error_value = 0 # nm
for i_wave in range(nsample):
d_wave = (max_wave - min_wave) / nsample
wave0 = min_wave + i_wave * d_wave
wave1 = min_wave + (i_wave + 1) * d_wave
center_wavelength = (wave0 + wave1) / 2 * 1e-10
cstar_psf = psf_function(
band, cstar_spectrum, config["cstar_frame_size"], error=error_value
)
i_throughput = throughput.copy()
i_throughput[(wave > wave1) | (wave < wave0)] = 0
i_band = S.ArrayBandpass(wave, i_throughput, waveunits='angstrom')
platesize = np.array(platesize)[::-1]
psf_shape = np.array(cstar_psf.shape)[::-1]
cstar_shift += (platesize - 1) / 2 - (psf_shape - 1) / 2
i_fp_image = ideal_focus_image(i_band, targets[1:], platescale, platesize, init_shifts, rotation)
psf = single_band_psf(center_wavelength, error=error)
focal_image = region_replace(
focal_image,
cstar_psf,
cstar_shift,
padded_in=False,
padded_out=False,
subpix=True,
)
_, _, cstar_sp, _ = targets[0]
cstar_flux = (cstar_sp * i_band).integrate()
cstar_psf = single_band_masked_psf(center_wavelength, error=error, shift=init_shifts)
for i_target in range(1, len(targets)):
sub_x, sub_y, sub_spectrum = targets[i_target]
pdout = False if i_target == len(targets) - 1 else True
pdin = False if i_target == 1 else True
log.debug(f"input target {sub_x=:}, {sub_y=:}")
sub_shift = rotate_and_shift([sub_x, sub_y]) / platescale
log.debug(f"after rotate and shift {sub_shift=:}")
sub_psf = psf_function(
band, sub_spectrum, config["substellar_frame_size"], error=error_value
)
psf_shape = np.array(sub_psf.shape)[::-1]
sub_shift += (platesize - 1) / 2 - (psf_shape - 1) / 2
log.debug(f"input shift of region_replace: {sub_shift=:}")
focal_image = region_replace(
focal_image,
sub_psf,
sub_shift,
padded_in=pdin,
padded_out=pdout,
subpix=True,
)
c_fp_image = fftconvolve(i_fp_image, psf, mode='same')
c_fp_image = focal_mask(c_fp_image, iwa, platescale)
c_fp_image = c_fp_image + cstar_flux * cstar_psf
return focal_image
all_fp_image.append(c_fp_image * area) # trans to photon/second
return np.array(all_fp_image).sum(axis=0)
def focal_mask(image, iwa, platescale, throughtput=1e-6):
......@@ -197,11 +181,10 @@ def focal_mask(image, iwa, platescale, throughtput=1e-6):
np.ndarray
The masked image.
"""
xx, yy = np.mgrid[0 : image.shape[0], 0 : image.shape[1]]
center = np.array([(image.shape[0] - 1) / 2, (image.shape[1] - 1) / 2])
mask = (abs(xx - center[0]) < iwa / platescale) | (
abs(yy - center[1]) < iwa / platescale
)
xx, yy = np.mgrid[0:image.shape[0], 0:image.shape[1]]
center = np.array([(image.shape[0]-1)/2, (image.shape[1]-1)/2])
mask = (abs(xx - center[0]) < iwa /
platescale) | (abs(yy - center[1]) < iwa / platescale)
image_out = image.copy()
image_out[mask] *= throughtput
return image_out
csst_cpic_sim/psf_simulation.py
View file @ 95f81f91
This diff is collapsed.
csst_cpic_sim/target.py
View file @ 95f81f91
This diff is collapsed.
csst_cpic_sim/utils.py
View file @ 95f81f91
......@@ -2,6 +2,7 @@ import numpy as np
import scipy.ndimage as nd
import logging
import random
import matplotlib.pyplot as plt
# DO NOT IMPORT CPICIMGSIM MODULES HERE
......@@ -125,3 +126,16 @@ def region_replace(
return padded
return padded[f_sz[0]:b_sz[0]+f_sz[0], f_sz[1]:b_sz[1]+f_sz[1]]
def psf_imshow(psf, vmin=1e-8, vmax=0.1, log=True, region=1):
focal_img = psf.copy()
focal_img = (focal_img - focal_img.min()) / (focal_img.max() - focal_img.min())
if log:
focal_img = np.log10(focal_img * 9 + 1)
plt.imshow(focal_img, origin='lower', cmap='gray', vmin=vmin, vmax=vmax)
shape = psf.shape
plt.xlim(shape[1] * (1 - region) / 2, shape[1] * (1 + region) / 2)
plt.ylim(shape[0] * (1 - region) / 2, shape[0] * (1 + region) / 2)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment