Skip to content
GitLab
Commit 8df06b27 authored by Fang Yuedong's avatar Fang Yuedong
Browse files

Merge branch 'sim_scheduler' into develop

parents 81e2570f 93270bbf
Hide whitespace changes
Inline Side-by-side
tests/PSFMatsIDW_coverage.py deleted 100644 → 0
View file @ 81e2570f
import unittest
import sys,os,math
from itertools import islice
import mpi4py.MPI as MPI
import numpy as np
import matplotlib.pyplot as plt
import matplotlib as mpl
mpl.use('Agg')
import scipy.io
from scipy import ndimage
#sys.path.append("/public/home/weichengliang/lnData/CSST_new_framwork/csstPSF_v4_20210326")
# sys.path.append("../")
from ObservationSim.PSF.PSFInterp import PSFConfig as myConfig
from ObservationSim.PSF.PSFInterp import PSFUtil as myUtil
NPSF = 400
##############################
###计算PSF椭率###
def psfSecondMoments(psfMat, cenX, cenY, pixSize=1):
apr = 0.5 #arcsec, 0.5角秒内测量
fl = 28. #meters
pxs = 2.5 #microns
apr = np.deg2rad(apr/3600.)*fl*1e6
apr = apr/pxs
apr = np.int(np.ceil(apr))
I = psfMat
ncol = I.shape[1]
nrow = I.shape[0]
w = 0.0
w11 = 0.0
w12 = 0.0
w22 = 0.0
for icol in range(ncol):
for jrow in range(nrow):
x = icol*pixSize - cenX
y = jrow*pixSize - cenY
rr = np.sqrt(x*x + y*y)
wgt= 0.0
if rr <= apr:
wgt = 1.0
w += I[jrow, icol]*wgt
w11 += x*x*I[jrow, icol]*wgt
w12 += x*y*I[jrow, icol]*wgt
w22 += y*y*I[jrow, icol]*wgt
w11 /= w
w12 /= w
w22 /= w
sz = w11 + w22
e1 = (w11 - w22)/sz
e2 = 2.0*w12/sz
return sz, e1, e2
##############################
##############################
'''
def assignTasks(npsf, NTasks, ThisTask):
npsfPerTasks = int(npsf/NTasks)
iStart= 0 + npsfPerTasks*ThisTask
iEnd = npsfPerTasks + npsfPerTasks*ThisTask
if ThisTask == NTasks:
iEnd = npsf
return iStart, iEnd
'''
#def test_psfIDW(iccd, iwave, psfPath, ThisTask, NTasks):
def test_psfIDW(iccd, iwave, psfPath):
nccd = 30
npsfA = 900
npsfB = 400
#iStart, iEnd = assignTasks(400, NTasks, ThisTask)
psfPathA = psfPath+'_30x30field'
psfPathB = psfPath+'_20x20field'
imxA = np.zeros(npsfA)
imyA = np.zeros(npsfA)
psfA = np.zeros([npsfA, 512, 512])
imxB = np.zeros(npsfB)
imyB = np.zeros(npsfB)
psfB = np.zeros([npsfB, 512, 512])
#for ipsf in range(iStart+1, iEnd+1):
for ipsf in range(1, npsfA+1):
print('ipsfA:', ipsf, end='\r', flush=True)
psfInfo = myConfig.LoadPSF(iccd, iwave, ipsf, psfPathA, InputMaxPixelPos=True, PSFCentroidWgt=True, VPSF=False)
imxA[ipsf-1] = psfInfo['image_x'] + psfInfo['centroid_x']
imyA[ipsf-1] = psfInfo['image_y'] + psfInfo['centroid_y']
psfA[ipsf-1, :, :] = psfInfo['psfMat']
for ipsf in range(1, npsfB+1):
print('ipsfB:', ipsf, end='\r', flush=True)
psfInfo = myConfig.LoadPSF(iccd, iwave, ipsf, psfPathB, InputMaxPixelPos=True, PSFCentroidWgt=True, VPSF=False)
imxB[ipsf-1] = psfInfo['image_x'] + psfInfo['centroid_x']
imyB[ipsf-1] = psfInfo['image_y'] + psfInfo['centroid_y']
psfB[ipsf-1, :, :] = psfInfo['psfMat']
#myConfig.psfMaker_IDW(px, py, PSFMat, cen_col, cen_row, IDWindex=2, OnlyNeighbors=True, hoc=None, hoclist=None, PSFCentroidWgt=False)
for ipsf in range(npsfB):
print('ipsf:', ipsf, end='\r', flush=True)
px = imxB[ipsf]
py = imyB[ipsf]
cen_col = imxA
cen_row = imyA
PSFMat = psfA
psfIDW = myConfig.psfMaker_IDW(px, py, PSFMat, cen_col, cen_row, IDWindex=2, OnlyNeighbors=True, hoc=None, hoclist=None, PSFCentroidWgt=False)
np.save('figs/psfIDW/psfIDW_{:}_{:}_{:}'.format(iccd, iwave, ipsf+1), psfIDW)
#def test_psfEll(iccd, iwave, ThisTask, NTasks):
def test_psfEll(iccd, iwave):
nccd = 30
npsf = 400
#iStart, iEnd = assignTasks(npsf, NTasks, ThisTask)
#imx = np.zeros(npsf)
#imy = np.zeros(npsf)
psf_e1 = np.zeros(npsf)
psf_e2 = np.zeros(npsf)
psf_sz = np.zeros(npsf)
#for ipsf in range(iStart+1, iEnd+1):
for ipsf in range(1, 401):
if ipsf > 1:
continue
print('ipsf-{:}'.format(ipsf), end='\r')
#psfInfo = myConfig.LoadPSF(iccd, iwave, ipsf, psfPath, InputMaxPixelPos=True, PSFCentroidWgt=True, VPSF=False)
#psfMat = psfInfo['psfMat']
#imx[ipsf-1] = psfInfo['image_x']+psfInfo['centroid_x']
#imy[ipsf-1] = psfInfo['image_y']+psfInfo['centroid_y']
psfMat = np.load('figs/psfIDW/psfIDW_{:}_{:}_{:}.npy'.format(iccd, iwave, ipsf)) #psfInfo['psfMat']
cenX = 256
cenY = 256
sz, e1, e2 = psfSecondMoments(psfMat, cenX, cenY, pixSize=1)
psf_e1[ipsf-1] = e1
psf_e2[ipsf-1] = e2
psf_sz[ipsf-1] = sz
#######
#comm.barrier()
#imx = comm.allreduce(imx, op=MPI.SUM)
#imy = comm.allreduce(imy, op=MPI.SUM)
#psf_e1 = comm.allreduce(psf_e1, op=MPI.SUM)
#psf_e2 = comm.allreduce(psf_e2, op=MPI.SUM)
#psf_sz = comm.allreduce(psf_sz, op=MPI.SUM)
#comm.barrier()
#if ThisTask == 0:
# arr = [psf_e1, psf_e2, psf_sz]
# np.save('/public/home/weichengliang/lnData/CSST_new_framwork/csstPSF_v4_20210326/test4report/data/psfEll20IDW_{:}_{:}'.format(iccd, iwave), arr)
'''
def test_psfResidualCalc(iccd, iwave, ipsf, psfPath):
psfInfo = myConfig.LoadPSF(iccd, iwave, ipsf, psfPath+'_20x20field', InputMaxPixelPos=True, PSFCentroidWgt=True, VPSF=False)
psfMatORG = psfInfo['psfMat']
psfMatIDW = np.load('/public/home/weichengliang/lnData/CSST_new_framwork/csstPSF_v4_20210326/test4report/figs/psfIDW/psfIDW_{:}_{:}_{:}.npy'.format(iccd, iwave, ipsf))
_,tREE80 = myUtil.psfEncircle(psfMatORG, fraction=0.8, psfSampleSizeInMicrons=2.5, focalLengthInMeters=28, cenPix=None)
tREE80_pix = np.int32(np.ceil(tREE80/(0.074/2/2))[0])
#print(tREE80, np.ceil(tREE80/(0.074/2/2)), tREE80_pix)
timg0 = psfMatORG[256-tREE80_pix:256+tREE80_pix, 256-tREE80_pix:256+tREE80_pix]
timg1 = psfMatIDW[256-tREE80_pix:256+tREE80_pix, 256-tREE80_pix:256+tREE80_pix]
#print("residual::", np.max((timg1-timg0)/timg0), np.min((timg1-timg0)/timg0), np.mean((timg1-timg0)/timg0))
return np.mean((timg1-timg0)/timg0)
'''
def test_psfResidualPlot(iccd, iwave, ipsf, psfPath):
psfInfo = myConfig.LoadPSF(iccd, iwave, ipsf, psfPath+'_20x20field', InputMaxPixelPos=True, PSFCentroidWgt=True, VPSF=False)
psfMatORG = psfInfo['psfMat']
psfMatIDW = np.load('figs/psfIDW/psfIDW_{:}_{:}_{:}.npy'.format(iccd, iwave, ipsf))
npix = psfMatORG.shape[0]
pixCutEdge= int(npix/2-15)
img0 = psfMatORG[pixCutEdge:npix-pixCutEdge, pixCutEdge:npix-pixCutEdge]
img1 = psfMatIDW[pixCutEdge:npix-pixCutEdge, pixCutEdge:npix-pixCutEdge]
imgX = (img1 - img0)/img0
img0 = np.log10(img0)
img1 = np.log10(img1)
imgX = np.log10(np.abs(imgX))
fig = plt.figure(figsize=(18,4))
ax = plt.subplot(1,3,1)
plt.imshow(img0, origin='lower', vmin=-7, vmax=-1.3)
plt.plot([npix/2-pixCutEdge, npix/2-pixCutEdge],[0, (npix/2-pixCutEdge)*2-1],'w--')
plt.plot([0, (npix/2-pixCutEdge)*2-1],[npix/2-pixCutEdge, npix/2-pixCutEdge],'w--')
plt.annotate('ORG', [0,(npix/2-pixCutEdge)*2-5], c='w', size=15)
cticks=[-7, -6, -5, -4, -3, -2, -1]
cbar = plt.colorbar(ticks=cticks)
cbar.ax.set_yticklabels(['$10^{-7}$', '$10^{-6}$', '$10^{-5}$','$10^{-4}$','$10^{-3}$','$10^{-2}$', '$10^{-1}$'])
print(img0.min(), img0.max())
ax = plt.subplot(1,3,2)
plt.imshow(img1, origin='lower', vmin=-7, vmax=-1.3)
plt.plot([npix/2-pixCutEdge, npix/2-pixCutEdge],[0, (npix/2-pixCutEdge)*2-1],'w--')
plt.plot([0, (npix/2-pixCutEdge)*2-1],[npix/2-pixCutEdge, npix/2-pixCutEdge],'w--')
plt.annotate('IDW', [0,(npix/2-pixCutEdge)*2-5], c='w', size=15)
cticks=[-7, -6, -5, -4, -3, -2, -1]
cbar = plt.colorbar(ticks=cticks)
cbar.ax.set_yticklabels(['$10^{-7}$', '$10^{-6}$', '$10^{-5}$','$10^{-4}$','$10^{-3}$','$10^{-2}$', '$10^{-1}$'])
print(img1.min(), img1.max())
ax = plt.subplot(1,3,3)
plt.imshow(imgX, origin='lower', vmin =-3, vmax =np.log10(3e-1))
plt.plot([npix/2-pixCutEdge, npix/2-pixCutEdge],[0, (npix/2-pixCutEdge)*2-1],'w--')
plt.plot([0, (npix/2-pixCutEdge)*2-1],[npix/2-pixCutEdge, npix/2-pixCutEdge],'w--')
#plt.annotate('(IDW-ORG)/ORG', [0,(npix/2-pixCutEdge)*2-5], c='w', size=15)
cticks=[-5, -4, -3, -2, -1]
cbar = plt.colorbar(ticks=cticks)
cbar.ax.set_yticklabels(['$10^{-5}$','$10^{-4}$','$10^{-3}$','$10^{-2}$', '$10^{-1}$'])
print(np.max((psfMatORG-psfMatIDW)))
plt.savefig('figs/psfResidual_iccd{:}.pdf'.format(iccd))
def test_psfEllPlot(OVERPLOT=False):
#if ThisTask == 0:
if True:
prefix = 'psfEll20'
iccd = 1
iwave= 1
data = np.load('data/'+prefix+'_1_1.npy')
imx= data[0]
imy= data[1]
psf_e1 = data[2]
psf_e2 = data[3]
print(np.shape(imx))
npsf = np.shape(imx)[0]
plt.cla()
plt.close("all")
fig = plt.figure(figsize=(12,12))
plt.plot(imx, imy, 'r.')
plt.savefig('figs/psfPos.pdf')
#######
fig = plt.figure(figsize=(12, 12))
plt.subplots_adjust(wspace=0.1, hspace=0.1)
ax = plt.subplot(1, 1, 1)
for ipsf in range(npsf):
plt.plot(imx[ipsf], imy[ipsf], 'b.')
ang = np.arctan2(psf_e2[ipsf], psf_e1[ipsf])/2
ell = np.sqrt(psf_e1[ipsf]**2 + psf_e2[ipsf]**2)
ell *= 15
lcos = ell*np.cos(ang)
lsin = ell*np.sin(ang)
plt.plot([imx[ipsf]-lcos, imx[ipsf]+lcos],[imy[ipsf]-lsin, imy[ipsf]+lsin],'b', lw=2)
###########
ang = 0.
ell = 0.05
ell*= 15
lcos = ell*np.cos(ang)
lsin = ell*np.sin(ang)
#plt.plot([imx[898]-lcos, imx[898]+lcos],[imy[898]+5.-lsin, imy[898]+5.+lsin],'k', lw=2)
#plt.annotate('{:}'.format(ell/15), (imx[898]-2., imy[898]+6.), xycoords='data', fontsize=10)
plt.xlabel('CCD X (mm)')
plt.ylabel('CCD Y (mm)')
if OVERPLOT == True:
prefix = 'psfEll20IDW'
data = np.load('data/'+prefix+'_1_1.npy')
#imx= data[0]
#imy= data[1]
psf_e1 = data[0]
psf_e2 = data[1]
npsf = np.shape(imx)[0]
for ipsf in range(npsf):
#plt.plot(imx[ipsf], imy[ipsf], 'r.')
ang = np.arctan2(psf_e2[ipsf], psf_e1[ipsf])/2
ell = np.sqrt(psf_e1[ipsf]**2 + psf_e2[ipsf]**2)
ell *= 15
lcos = ell*np.cos(ang)
lsin = ell*np.sin(ang)
plt.plot([imx[ipsf]-lcos, imx[ipsf]+lcos],[imy[ipsf]-lsin, imy[ipsf]+lsin],'r', lw=1)
plt.gca().set_aspect(1)
if OVERPLOT == True:
prefix = 'psfEllOPIDW'
plt.savefig('figs/'+prefix+'_iccd{:}.pdf'.format(iccd))
'''
def test_psfdEllPlot():
if ThisTask == 0:
prefix = 'psfEll20'
iccd = 1
iwave= 1
data = np.load('data/'+prefix+'_1_1.npy')
imx= data[0]
imy= data[1]
psf_e1 = data[2]
psf_e2 = data[3]
psf_sz = data[4]
print(np.shape(imx))
npsf = np.shape(imx)[0]
ellX = np.sqrt(psf_e1**2 + psf_e2**2)
angX = np.arctan2(psf_e2, psf_e1)/2
angX = np.rad2deg(angX)
szX = psf_sz
##############################
prefix = 'psfEll20IDW'
data = np.load('data/'+prefix+'_1_1.npy')
#imx= data[0]
#imy= data[1]
psf_e1 = data[0]
psf_e2 = data[1]
psf_sz = data[2]
ellY = np.sqrt(psf_e1**2 + psf_e2**2)
angY = np.arctan2(psf_e2, psf_e1)/2
angY = np.rad2deg(angY)
szY = psf_sz
##############################
fig=plt.figure(figsize=(15, 4))
ax = plt.subplot(1,3,1)
plt.hist(ellX, bins=20, color='b', alpha=0.5)
plt.hist(ellY, bins=20, color='r', alpha=0.5)
plt.xlabel('$\epsilon$')
plt.ylabel('PDF')
ax = plt.subplot(1,3,2)
plt.hist((ellY-ellX)/ellX, bins=20, color='r', alpha=0.5)
plt.xlabel('$(\epsilon_{\\rm IDW}-\epsilon_{\\rm ORG})/\epsilon_{\\rm ORG}$')
plt.ylabel('PDF')
ax = plt.subplot(1,3,3)
plt.hist((angY-angX)/angX, bins=20, color='r', alpha=0.5, range=[-0.1, 0.1])
plt.xlabel('$(\\alpha_{\\rm IDW}-\\alpha_{\\rm ORG})/\\alpha_{\\rm ORG}$')
plt.ylabel('PDF')
plt.savefig('figs/psfEllOPIDWPDF.pdf')
fig=plt.figure(figsize=(4, 4))
plt.hist((szY-szX)/szX, bins=20, color='r', alpha=0.5)
plt.xlabel('$(R_{\\rm IDW}-R_{\\rm ORG})/R_{\\rm ORG}$')
plt.ylabel('PDF')
plt.savefig('figs/psfEllOPIDWPDF_dsz.pdf')
'''
def test_psfdEllabsPlot(iccd):
#if ThisTask == 0:
if True:
prefix = 'psfEll20'
#iccd = 1
#iwave= 1
data = np.load('data/'+prefix+'_{:}_1.npy'.format(iccd))
imx= data[0]
imy= data[1]
psf_e1 = data[2]
psf_e2 = data[3]
psf_sz = data[4]
print(np.shape(imx))
npsf = np.shape(imx)[0]
ellX = np.sqrt(psf_e1**2 + psf_e2**2)
angX = np.arctan2(psf_e2, psf_e1)/2
angX = np.rad2deg(angX)
szX = psf_sz
##############################
prefix = 'psfEll20IDW'
data = np.load('data/'+prefix+'_{:}_1.npy'.format(iccd))
#imx= data[0]
#imy= data[1]
psf_e1 = data[0]
psf_e2 = data[1]
psf_sz = data[2]
ellY = np.sqrt(psf_e1**2 + psf_e2**2)
angY = np.arctan2(psf_e2, psf_e1)/2
angY = np.rad2deg(angY)
szY = psf_sz
##############################
fig=plt.figure(figsize=(6, 5))
grid = plt.GridSpec(3,1,left=0.15, right=0.95, wspace=None, hspace=0.02)
#plt.subplots_adjust(left=None,bottom=None,right=None,top=None,wspace=None,hspace=0.02)
ax = plt.subplot(grid[0:2,0])
plt.plot([0.01,0.1],[0.01,0.1], 'k--', lw=1. )
plt.scatter(ellX, ellY, color='b', alpha=1., s=3., edgecolors='None')
plt.xlim([0.015, 0.085])
plt.ylim([0.015, 0.085])
plt.ylabel('$\epsilon_{\\rm IDW}$')
plt.gca().axes.get_xaxis().set_visible(False)
ax = plt.subplot(grid[2,0])
plt.plot([0.015,0.085],[0.,0.], 'k--', lw=1. )
plt.scatter(ellX, (ellY-ellX), color='b', s=3., edgecolors='None')
plt.xlim([0.015, 0.085])
plt.ylim([-0.0018, 0.0018])
plt.xlabel('$\epsilon_{\\rm ORG}$')
plt.ylabel('$\Delta$')
plt.savefig('figs/psfEllOPIDWPDF_{:}.pdf'.format(iccd))
fig=plt.figure(figsize=(4, 4))
plt.hist((szY-szX)/szX, bins=20, color='r', alpha=0.5)
plt.xlabel('$(R_{\\rm IDW}-R_{\\rm ORG})/R_{\\rm ORG}$')
plt.ylabel('PDF')
plt.savefig('figs/psfEllOPIDWPDF_dsz_{:}.pdf'.format(iccd))
class PSFMatsIDW_coverage(unittest.TestCase):
def test_psfIDW_(self):
#comm = MPI.COMM_WORLD
#ThisTask = comm.Get_rank()
#NTasks = comm.Get_size()
iccd = 1
iwave= 1
ipsf = 400
psfPath = '/data/simudata/CSSOSDataProductsSims/data/csstPSFdata/CSSOS_psf_20210326/CSST_psf_ciomp'
#test_psfIDW(iccd, iwave, psfPath, ThisTask, NTasks)
test_psfIDW(iccd, iwave, psfPath)
'''
for iccd in range(7, 10):
res = np.zeros(400)
for ipsf in range(1,401):
print(ipsf, end="\r")
res[ipsf-1] = test_psfResidualCalc(iccd, iwave, ipsf, psfPath)
#fig = plt.figure(figsize=(6,6))
#plt.hist(np.abs(res), bins=50)
#plt.xlim([0,1])
#plt.savefig('figs/psfResidualREE80PDF.pdf')
print("{:}:".format(iccd), res[res<=0.01].size/400*100)
'''
test_psfResidualPlot(iccd, iwave, ipsf, psfPath)
#test_psfEll(iccd, iwave, ThisTask, NTasks)
test_psfEll(iccd, iwave)
test_psfEllPlot(OVERPLOT=True)
#test_psfdEllPlot()
test_psfdEllabsPlot(iccd)
##############################
##############################
##############################
if __name__=='__main__':
unittest.main()
tests/test_BF_CTE.py 0 → 100644
View file @ 8df06b27
import unittest
import sys,os,math
from itertools import islice
import numpy as np
import copy
import ctypes
import galsim
import yaml
from astropy.io import fits
from ObservationSim.Instrument import Chip, Filter, FilterParam, FocalPlane
from ObservationSim.Instrument.Chip import ChipUtils as chip_utils
#from ObservationSim.sim_steps import add_brighter_fatter_CTE
from ObservationSim.Instrument.Chip.libCTI.CTI_modeling import CTI_sim
try:
import importlib.resources as pkg_resources
except ImportError:
# Try backported to PY<37 'importlib_resources'
import importlib_resources as pkg_resources
### test FUNCTION --- START ###
def add_brighter_fatter(img):
#Inital dynamic lib
try:
with pkg_resources.files('ObservationSim.Instrument.Chip.libBF').joinpath("libmoduleBF.so") as lib_path:
lib_bf = ctypes.CDLL(lib_path)
except AttributeError:
with pkg_resources.path('ObservationSim.Instrument.Chip.libBF', "libmoduleBF.so") as lib_path:
lib_bf = ctypes.CDLL(lib_path)
lib_bf.addEffects.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.POINTER(ctypes.c_float), ctypes.POINTER(ctypes.c_float), ctypes.c_int]
# Set bit flag
bit_flag = 1
bit_flag = bit_flag | (1 << 2)
nx, ny = img.array.shape
nn = nx * ny
arr_ima= (ctypes.c_float*nn)()
arr_imc= (ctypes.c_float*nn)()
arr_ima[:]= img.array.reshape(nn)
arr_imc[:]= np.zeros(nn)
lib_bf.addEffects(nx, ny, arr_ima, arr_imc, bit_flag)
img.array[:, :] = np.reshape(arr_imc, [nx, ny])
del arr_ima, arr_imc
return img
### test FUNCTION --- END ###
def defineCCD(iccd, config_file):
with open(config_file, "r") as stream:
try:
config = yaml.safe_load(stream)
#for key, value in config.items():
# print (key + " : " + str(value))
except yaml.YAMLError as exc:
print(exc)
chip = Chip(chipID=iccd, config=config)
chip.img = galsim.ImageF(400, 200) #galsim.ImageF(chip.npix_x, chip.npix_y)
focal_plane = FocalPlane(chip_list=[iccd])
chip.img.wcs= focal_plane.getTanWCS(192.8595, 27.1283, -113.4333*galsim.degrees, chip.pix_scale)
return chip
def defineFilt(chip):
filter_param = FilterParam()
filter_id, filter_type = chip.getChipFilter()
filt = Filter(
filter_id=filter_id,
filter_type=filter_type,
filter_param=filter_param,
ccd_bandpass=chip.effCurve)
bandpass_list = filt.bandpass_sub_list
return filt
class detModule_coverage(unittest.TestCase):
def __init__(self, methodName='runTest'):
super(detModule_coverage, self).__init__(methodName)
##self.dataPath = "/public/home/chengliang/CSSOSDataProductsSims/csst-simulation/tests/UNIT_TEST_DATA" ##os.path.join(os.getenv('UNIT_TEST_DATA_ROOT'), 'csst_fz_gc1')
self.dataPath = os.path.join(os.getenv('UNIT_TEST_DATA_ROOT'), 'csst_fz_msc')
self.iccd = 1
def test_add_brighter_fatter(self):
config_file = os.path.join(self.dataPath, 'config_test.yaml')
chip = defineCCD(self.iccd, config_file)
filt = defineFilt(chip)
print(chip.chipID)
print(chip.cen_pix_x, chip.cen_pix_y)
#objA-lowSFB
obj = galsim.Gaussian(sigma=0.2, flux=1000)
arr = obj.drawImage(nx=64, ny=64, scale=0.074).array
chip.img.array[(100-32):(100+32),(200-32):(200+32)] = arr[:,:]
img_old = copy.deepcopy(chip.img)
img_new = add_brighter_fatter(img=chip.img)
arr1= img_old.array
arr2= img_new.array
deltaA_max = np.max(np.abs(arr2-arr1))
print('deltaA-max:', np.max(np.abs(arr2-arr1)))
print('deltaA-min:', np.min(np.abs(arr2-arr1)))
#objB-highSFB
obj = galsim.Gaussian(sigma=0.2, flux=10000)
arr = obj.drawImage(nx=64, ny=64, scale=0.074).array
chip.img.array[(100-32):(100+32),(200-32):(200+32)] = arr[:,:]
img_old = copy.deepcopy(chip.img)
img_new = add_brighter_fatter(img=chip.img)
arr3= img_old.array
arr4= img_new.array
deltaB_max = np.max(np.abs(arr4-arr3))
print('deltaB-max:', np.max(np.abs(arr4-arr3)))
print('deltaB-min:', np.min(np.abs(arr4-arr3)))
self.assertTrue( deltaB_max > deltaA_max )
def test_apply_CTE(self):
config_file = os.path.join(self.dataPath, 'config_test.yaml')
chip = defineCCD(self.iccd, config_file)
filt = defineFilt(chip)
print(chip.chipID)
print(chip.cen_pix_x, chip.cen_pix_y)
print(" Apply CTE Effect")
nx,ny,noverscan,nsp,nmax = 4608,4616,84,3,10
ntotal = 4700
beta,w,c = 0.478,84700,0
t = np.array([0.74,7.7,37],dtype=np.float32)
rho_trap = np.array([0.6,1.6,1.4],dtype=np.float32)
trap_seeds = np.array([0,100,1000],dtype=np.int32)
release_seed = 500
image = fits.getdata(os.path.join(self.dataPath, "testCTE_image_before.fits")).astype(np.int32)
#get_trap_map(trap_seeds,nx,ny,nmax,rho_trap,beta,c,".")
#bin2fits("trap.bin",".",nsp,nx,ny,nmax)
image_cti = CTI_sim(image,nx,ny,noverscan,nsp,nmax,beta,w,c,t,rho_trap,trap_seeds,release_seed)
fits.writeto(os.path.join(self.dataPath, "testCTE_image_after.fits"),data=image_cti,overwrite=True)
if __name__ == '__main__':
unittest.main()
tests/test_PSFmodule.py 0 → 100644
View file @ 8df06b27
import unittest
import sys,os,math
from itertools import islice
import numpy as np
import galsim
import yaml
from ObservationSim.Instrument import Chip, Filter, FilterParam, FocalPlane
from ObservationSim.PSF.PSFInterp import PSFInterp
def defineCCD(iccd, config_file):
with open(config_file, "r") as stream:
try:
config = yaml.safe_load(stream)
#for key, value in config.items():
# print (key + " : " + str(value))
except yaml.YAMLError as exc:
print(exc)
chip = Chip(chipID=iccd, config=config)
chip.img = galsim.ImageF(chip.npix_x, chip.npix_y)
focal_plane = FocalPlane(chip_list=[iccd])
chip.img.wcs= focal_plane.getTanWCS(192.8595, 27.1283, -113.4333*galsim.degrees, chip.pix_scale)
return chip
def defineFilt(chip):
filter_param = FilterParam()
filter_id, filter_type = chip.getChipFilter()
filt = Filter(
filter_id=filter_id,
filter_type=filter_type,
filter_param=filter_param,
ccd_bandpass=chip.effCurve)
bandpass_list = filt.bandpass_sub_list
return bandpass_list
class PSFInterpModule_coverage(unittest.TestCase):
def __init__(self, methodName='runTest'):
super(PSFInterpModule_coverage, self).__init__(methodName)
self.dataPath = os.path.join(os.getenv('UNIT_TEST_DATA_ROOT'), 'csst_fz_msc')
self.iccd = 8
def test_loadPSFSet(self):
config_file = os.path.join(self.dataPath, 'config_test.yaml')
chip = defineCCD(self.iccd, config_file)
bandpass = defineFilt(chip)
print(chip.chipID)
print(chip.cen_pix_x, chip.cen_pix_y)
pathTemp = self.dataPath #"/public/share/yangxuliu/CSSOSDataProductsSims/psfCube/set1_dynamic/"
psfModel= PSFInterp(chip, npsf=900, PSF_data_file=pathTemp, PSF_data_prefix="", HocBuild=True, LOG_DEBUG=True)
x, y = 4096, 4096 #imgPos[iobj, :] # try get the PSF at some location (1234, 1234) on the chip
x = x+chip.bound.xmin
y = y+chip.bound.ymin
pos_img = galsim.PositionD(x, y)
psf,_ = psfModel.get_PSF(chip=chip, pos_img=pos_img, bandpass=0, galsimGSObject=True)
psfA = psfModel.get_PSF(chip=chip, pos_img=pos_img, bandpass=bandpass[0], galsimGSObject=False)
psfB = psfModel.get_PSF(chip=chip, pos_img=pos_img, findNeighMode='hoclistFind', bandpass=bandpass[0], galsimGSObject=False)
self.assertTrue( psf != None )
self.assertTrue( np.max(np.abs(psfA-psfB))<1e-6 )
if __name__ == '__main__':
unittest.main()
tests/test_SpecDisperse.py
View file @ 8df06b27
...@@ -6,7 +6,7 @@ ...@@ -6,7 +6,7 @@
import unittest import unittest
from ObservationSim.MockObject.SpecDisperser import rotate90, SpecDisperser from ObservationSim.MockObject.SpecDisperser import rotate90, SpecDisperser
from ObservationSim.Config import ChipOutput, Config from ObservationSim.Config import ChipOutput
from ObservationSim.Instrument import Telescope, Chip, FilterParam, Filter, FocalPlane from ObservationSim.Instrument import Telescope, Chip, FilterParam, Filter, FocalPlane
from ObservationSim.MockObject import MockObject, Star from ObservationSim.MockObject import MockObject, Star
from ObservationSim.PSF import PSFGauss from ObservationSim.PSF import PSFGauss
...@@ -424,8 +424,6 @@ class TestSpecDisperse(unittest.TestCase): ...@@ -424,8 +424,6 @@ class TestSpecDisperse(unittest.TestCase):
print(key + " : " + str(value)) print(key + " : " + str(value))
except yaml.YAMLError as exc: except yaml.YAMLError as exc:
print(exc) print(exc)
# config = Config.read_config(configFn)
# path_dict = Config.config_dir(config,work_dir=work_dir, data_dir=data_dir)
filter_param = FilterParam() filter_param = FilterParam()
......
tests/test_darknoise_func.py 0 → 100644
View file @ 8df06b27
import unittest
import sys,os,math
from itertools import islice
import numpy as np
import galsim
import yaml
from ObservationSim.Instrument import Chip, Filter, FilterParam, FocalPlane
#from ObservationSim.Instrument.Chip import ChipUtils as chip_utils
### test FUNCTION --- START ###
def get_base_img(img, chip, read_noise, readout_time, dark_noise, exptime=150., InputDark=None):
if InputDark == None:
# base_level = read_noise**2 + dark_noise*(exptime+0.5*readout_time)
## base_level = dark_noise*(exptime+0.5*readout_time)
base_level = dark_noise*(exptime)
base_img1 = base_level * np.ones_like(img.array)
else:
base_img1 = np.zeros_like(img.array)
ny = int(chip.npix_y/2)
nx = chip.npix_x
arr = np.arange(ny).reshape(ny, 1)
arr = np.broadcast_to(arr, (ny, nx))
base_img2 = np.zeros_like(img.array)
base_img2[:ny, :] = arr
base_img2[ny:, :] = arr[::-1,:]
base_img2[:,:] = base_img2[:,:]*(readout_time/ny)*dark_noise
return base_img1+base_img2
### test FUNCTION --- END ###
def defineCCD(iccd, config_file):
with open(config_file, "r") as stream:
try:
config = yaml.safe_load(stream)
#for key, value in config.items():
# print (key + " : " + str(value))
except yaml.YAMLError as exc:
print(exc)
chip = Chip(chipID=iccd, config=config)
chip.img = galsim.ImageF(chip.npix_x, chip.npix_y)
focal_plane = FocalPlane(chip_list=[iccd])
chip.img.wcs= focal_plane.getTanWCS(192.8595, 27.1283, -113.4333*galsim.degrees, chip.pix_scale)
return chip
def defineFilt(chip):
filter_param = FilterParam()
filter_id, filter_type = chip.getChipFilter()
filt = Filter(
filter_id=filter_id,
filter_type=filter_type,
filter_param=filter_param,
ccd_bandpass=chip.effCurve)
bandpass_list = filt.bandpass_sub_list
return bandpass_list
class detModule_coverage(unittest.TestCase):
def __init__(self, methodName='runTest'):
super(detModule_coverage, self).__init__(methodName)
self.dataPath = os.path.join(os.getenv('UNIT_TEST_DATA_ROOT'), 'csst_fz_msc')
self.iccd = 1
def test_add_dark(self):
config_file = os.path.join(self.dataPath, 'config_test.yaml')
chip = defineCCD(self.iccd, config_file)
bandpass = defineFilt(chip)
print(chip.chipID)
print(chip.cen_pix_x, chip.cen_pix_y)
exptime=150.
base_img = get_base_img(img=chip.img, chip=chip, read_noise=chip.read_noise, readout_time=chip.readout_time, dark_noise=chip.dark_noise, exptime=exptime, InputDark=None)
ny = int(chip.npix_y/2)
self.assertTrue( np.abs(np.max(base_img) - (exptime*chip.dark_noise+(ny-1)*(chip.readout_time/ny)*chip.dark_noise )) < 1e-6 )
self.assertTrue( np.min(base_img) == 3 )
base_img = get_base_img(img=chip.img, chip=chip, read_noise=chip.read_noise, readout_time=chip.readout_time, dark_noise=chip.dark_noise, exptime=150., InputDark="testTag")
self.assertTrue( np.abs(np.max(base_img) - ((ny-1)*(chip.readout_time/ny)*chip.dark_noise )) < 1e-6 )
if __name__ == '__main__':
unittest.main()
tests/test_imaging.py 0 → 100644
View file @ 8df06b27
import unittest
from scipy import interpolate
import astropy.constants as cons
from astropy.table import Table
import h5py as h5
import sys,os,math
from itertools import islice
import numpy as np
import galsim
import yaml
import copy
from astropy.cosmology import FlatLambdaCDM
from astropy import constants
from astropy import units as U
from ObservationSim.MockObject._util import getObservedSED
from ObservationSim.Instrument import Chip, Filter, FilterParam, FocalPlane
from ObservationSim.PSF.PSFInterp import PSFInterp
from ObservationSim.MockObject._util import integrate_sed_bandpass, getNormFactorForSpecWithABMAG, getABMAG
def convert_sed(mag, sed, target_filt, norm_filt=None):
bandpass = target_filt.bandpass_full
if norm_filt is not None:
norm_thr_rang_ids = norm_filt['SENSITIVITY'] > 0.001
else:
norm_filt = Table(
np.array(np.array([bandpass.wave_list*10.0, bandpass.func(bandpass.wave_list)])).T, names=(['WAVELENGTH', 'SENSITIVITY'])
)
norm_thr_rang_ids = norm_filt['SENSITIVITY'] > 0.001
sedNormFactor = getNormFactorForSpecWithABMAG(ABMag=mag,
spectrum=sed,
norm_thr=norm_filt,
sWave=np.floor(norm_filt[norm_thr_rang_ids][0][0]),
eWave=np.ceil(norm_filt[norm_thr_rang_ids][-1][0]))
sed_photon = copy.copy(sed)
sed_photon = np.array([sed_photon['WAVELENGTH'], sed_photon['FLUX']*sedNormFactor]).T
sed_photon = galsim.LookupTable(x=np.array(sed_photon[:, 0]), f=np.array(sed_photon[:, 1]), interpolant='nearest')
# Get magnitude
sed_photon = galsim.SED(sed_photon, wave_type='A', flux_type='1', fast=False)
interFlux = integrate_sed_bandpass(sed=sed_photon, bandpass=bandpass)
mag_csst = getABMAG(
interFlux=interFlux,
bandpass=bandpass
)
if target_filt.survey_type == "photometric":
return sed_photon, mag_csst, interFlux
elif target_filt.survey_type == "spectroscopic":
del sed_photon
return sed, mag_csst, interFlux
def _load_gals(file_path):
gals_cat = h5.File(file_path, 'r')['galaxies']
for ikeys in gals_cat.keys():
gals = gals_cat[ikeys]
param = {}
igals = 9
param['z'] = gals['redshift'][igals]
param['mag_use_normal'] = gals['mag_csst_g'][igals]
print(param['mag_use_normal'] )
param['e1'] = gals['ellipticity_true'][igals][0]
param['e2'] = gals['ellipticity_true'][igals][1]
param['e1_disk'] = param['e1']
param['e2_disk'] = param['e2']
param['e1_bulge'] = param['e1']
param['e2_bulge'] = param['e2']
# Masses
param['bulgemass'] = gals['bulgemass'][igals]
param['diskmass'] = gals['diskmass'][igals]
param['size'] = gals['size'][igals]
# Sersic index
param['disk_sersic_idx'] = 1.
param['bulge_sersic_idx'] = 4.
# Sizes
param['bfrac'] = param['bulgemass']/(param['bulgemass'] + param['diskmass'])
if param['bfrac'] >= 0.6:
param['hlr_bulge'] = param['size']
param['hlr_disk'] = param['size'] * (1. - param['bfrac'])
else:
param['hlr_disk'] = param['size']
param['hlr_bulge'] = param['size'] * param['bfrac']
# SED coefficients
param['coeff'] = gals['coeff'][igals]
# TEST no redening and no extinction
param['av'] = 0.0
param['redden'] = 0
pcs = h5.File("/public/share/yangxuliu/CSSOSDataProductsSims/data_50sqDeg/sedlibs/"+"pcs.h5", "r")
lamb = h5.File("/public/share/yangxuliu/CSSOSDataProductsSims/data_50sqDeg/sedlibs/"+"lamb.h5", "r")
lamb_gal = lamb['lamb'][()]
pcs = pcs['pcs'][()]
cosmo = FlatLambdaCDM(H0=67.66, Om0=0.3111)
factor = 10**(-.4 * cosmo.distmod(param['z']).value)
flux = np.matmul(pcs, param['coeff']) * factor
# if np.any(flux < 0):
# raise ValueError("Glaxy %s: negative SED fluxes"%obj.id)
flux[flux < 0] = 0.
sedcat = np.vstack((lamb_gal, flux)).T
sed_data = getObservedSED(
sedCat=sedcat,
redshift=param['z'],
av=param["av"],
redden=param["redden"]
)
wave, flux = sed_data[0], sed_data[1]
speci = interpolate.interp1d(wave, flux)
lamb = np.arange(2000, 11001+0.5, 0.5)
y = speci(lamb)
# erg/s/cm2/A --> photon/s/m2/A
all_sed = y * lamb / (cons.h.value * cons.c.value) * 1e-13
sed = Table(np.array([lamb, all_sed]).T, names=('WAVELENGTH', 'FLUX'))
param["sed"] = sed
del wave
del flux
return param
def defineCCD(iccd, config_file):
with open(config_file, "r") as stream:
try:
config = yaml.safe_load(stream)
#for key, value in config.items():
# print (key + " : " + str(value))
except yaml.YAMLError as exc:
print(exc)
chip = Chip(chipID=iccd, config=config)
chip.img = galsim.ImageF(chip.npix_x, chip.npix_y)
focal_plane = FocalPlane(chip_list=[iccd])
chip.img.wcs= focal_plane.getTanWCS(192.8595, 27.1283, -113.4333*galsim.degrees, chip.pix_scale)
return chip
def defineFilt(chip):
filter_param = FilterParam()
filter_id, filter_type = chip.getChipFilter()
filt = Filter(
filter_id=filter_id,
filter_type=filter_type,
filter_param=filter_param,
ccd_bandpass=chip.effCurve)
return filt
class imagingModule_coverage(unittest.TestCase):
def __init__(self, methodName='runTest'):
super(imagingModule_coverage, self).__init__(methodName)
self.dataPath = "/public/home/chengliang/CSSOSDataProductsSims/csst-simulation/tests/UNIT_TEST_DATA" ##os.path.join(os.getenv('UNIT_TEST_DATA_ROOT'), 'csst_fz_gc1')
self.iccd = 8
def test_imaging(self):
config_file = os.path.join(self.dataPath, 'config_test.yaml')
chip = defineCCD(self.iccd, config_file)
bandpass = defineFilt(chip)
filt = defineFilt(chip)
print(chip.chipID)
print(chip.cen_pix_x, chip.cen_pix_y)
obj = _load_gals("UNIT_TEST_DATA/galaxies_C6_bundle000287.h5")
sed_data = obj['sed']
norm_filt = None
obj_sed, obj_mag, obj_flux = convert_sed(
mag=obj["mag_use_normal"],
sed=sed_data,
target_filt=filt,
norm_filt=norm_filt,
)
pupil_area= np.pi * (0.5 * 2.)**2
exptime = 150.
nphotons_tot = obj_flux*pupil_area * exptime #getElectronFluxFilt(filt, tel, exptime)
full = integrate_sed_bandpass(sed=obj_sed, bandpass=filt.bandpass_full)
print(full, nphotons_tot, obj_mag)
for i in range(4):
sub = integrate_sed_bandpass(sed=obj_sed, bandpass=filt.bandpass_sub_list[i])
ratio = sub / full
nphotons = ratio * nphotons_tot
disk = galsim.Sersic(n=obj['disk_sersic_idx'], half_light_radius=obj['hlr_disk'], flux=1.0)
disk_shape = galsim.Shear(g1=obj['e1_disk'], g2=obj['e2_disk'])
disk = disk.shear(disk_shape)
gal_temp = disk
gal_temp = gal_temp.withFlux(nphotons)
psf = galsim.Gaussian(sigma=0.1, flux=1)
gal_temp = galsim.Convolve(psf, gal_temp)
if i == 0:
gal = gal_temp
else:
gal = gal + gal_temp
print(gal)
self.assertTrue( gal != None )
if __name__ == '__main__':
unittest.main()
tests/test_prescan_overscan_func.py 0 → 100644
View file @ 8df06b27
import unittest
import sys,os,math
from itertools import islice
import numpy as np
import galsim
import yaml
from ObservationSim.Instrument import Chip, Filter, FilterParam, FocalPlane
#from ObservationSim.Instrument.Chip import ChipUtils as chip_utils
### test FUNCTION --- START ###
def AddPreScan(GSImage, pre1=27, pre2=4, over1=71, over2=80, nsecy = 2, nsecx=8):
img= GSImage.array
ny, nx = img.shape
dx = int(nx/nsecx)
dy = int(ny/nsecy)
imgt=np.zeros([int(nsecy*nsecx), int(ny/nsecy+pre2+over2), int(nx/nsecx+pre1+over1)])
for iy in range(nsecy):
for ix in range(nsecx):
if iy % 2 == 0:
tx = ix
else:
tx = (nsecx-1)-ix
ty = iy
chunkidx = int(tx+ty*nsecx) #chunk1-[1,2,3,4], chunk2-[5,6,7,8], chunk3-[9,10,11,12], chunk4-[13,14,15,16]
imgtemp = np.zeros([int(ny/nsecy+pre2+over2), int(nx/nsecx+pre1+over1)])
if int(chunkidx/4) == 0:
imgtemp[pre2:pre2+dy, pre1:pre1+dx] = img[iy*dy:(iy+1)*dy, ix*dx:(ix+1)*dx]
imgt[chunkidx, :, :] = imgtemp
if int(chunkidx/4) == 1:
imgtemp[pre2:pre2+dy, over1:over1+dx] = img[iy*dy:(iy+1)*dy, ix*dx:(ix+1)*dx]
imgt[chunkidx, :, :] = imgtemp #[:, ::-1]
if int(chunkidx/4) == 2:
imgtemp[over2:over2+dy, over1:over1+dx] = img[iy*dy:(iy+1)*dy, ix*dx:(ix+1)*dx]
imgt[chunkidx, :, :] = imgtemp #[::-1, ::-1]
if int(chunkidx/4) == 3:
imgtemp[over2:over2+dy, pre1:pre1+dx] = img[iy*dy:(iy+1)*dy, ix*dx:(ix+1)*dx]
imgt[chunkidx, :, :] = imgtemp #[::-1, :]
imgtx1 = np.hstack(imgt[:nsecx:, :, :]) #hstack chunk(1,2)-[1,2,3,4,5,6,7,8]
imgtx2 = np.hstack(imgt[:(nsecx-1):-1, :, :]) #hstack chunk(4,3)-[16,15,14,13,12,11,,10,9]
newimg = galsim.Image(int(nx+(pre1+over1)*nsecx), int(ny+(pre2+over2)*nsecy), init_value=0)
newimg.array[:, :] = np.concatenate([imgtx1, imgtx2]) #vstack chunk(1,2) & chunk(4,3)
newimg.wcs = GSImage.wcs
return newimg
### test FUNCTION --- END ###
def defineCCD(iccd, config_file):
with open(config_file, "r") as stream:
try:
config = yaml.safe_load(stream)
#for key, value in config.items():
# print (key + " : " + str(value))
except yaml.YAMLError as exc:
print(exc)
chip = Chip(chipID=iccd, config=config)
chip.img = galsim.ImageF(chip.npix_x, chip.npix_y)
focal_plane = FocalPlane(chip_list=[iccd])
chip.img.wcs= focal_plane.getTanWCS(192.8595, 27.1283, -113.4333*galsim.degrees, chip.pix_scale)
return chip
def defineFilt(chip):
filter_param = FilterParam()
filter_id, filter_type = chip.getChipFilter()
filt = Filter(
filter_id=filter_id,
filter_type=filter_type,
filter_param=filter_param,
ccd_bandpass=chip.effCurve)
bandpass_list = filt.bandpass_sub_list
return bandpass_list
class detModule_coverage(unittest.TestCase):
def __init__(self, methodName='runTest'):
super(detModule_coverage, self).__init__(methodName)
self.dataPath = os.path.join(os.getenv('UNIT_TEST_DATA_ROOT'), 'csst_fz_msc')
self.iccd = 1
def test_add_prescan_overscan(self):
config_file = os.path.join(self.dataPath, 'config_test.yaml')
chip = defineCCD(self.iccd, config_file)
bandpass = defineFilt(chip)
print(chip.chipID)
print(chip.cen_pix_x, chip.cen_pix_y)
chip.img = AddPreScan(GSImage=chip.img,
pre1=chip.prescan_x,
pre2=chip.prescan_y,
over1=chip.overscan_x,
over2=chip.overscan_y)
self.assertTrue( (chip.prescan_x+chip.overscan_x)*8+chip.npix_x == np.shape(chip.img.array)[1] )
self.assertTrue( (chip.prescan_y+chip.overscan_y)*2+chip.npix_y == np.shape(chip.img.array)[0] )
if __name__ == '__main__':
unittest.main()
tools/getPSF.py 0 → 100644
View file @ 8df06b27
import os
import numpy as np
import ObservationSim.PSF.PSFInterp as PSFInterp
from ObservationSim.Instrument import Chip, Filter, FilterParam
import yaml
import galsim
import astropy.io.fits as fitsio
# Setup PATH
SIMPATH = "/share/simudata/CSSOSDataProductsSims/data/CSSTSimImage_C8/testRun_FGS"
config_filename= SIMPATH+"/config_C6_fits.yaml"
cat_filename = SIMPATH+"/MSC_00000000/MSC_10106100000000_chip_40_filt_FGS.cat"
# Read cat file
catFn = open(cat_filename,"r")
line = catFn.readline()
print(cat_filename,'\n',line)
imgPos = []
chipID = -1
for line in catFn:
line = line.strip()
columns = line.split()
if chipID == -1:
chipID = int(columns[1])
else:
assert chipID == int(columns[1])
ximg = float(columns[3])
yimg = float(columns[4])
imgPos.append([ximg, yimg])
imgPos = np.array(imgPos)
nobj = imgPos.shape[0]
print('chipID, nobj::', chipID, nobj)
# Read config file
with open(config_filename, "r") as stream:
try:
config = yaml.safe_load(stream)
for key, value in config.items():
print (key + " : " + str(value))
except yaml.YAMLError as exc:
print(exc)
# Setup Chip
chip = Chip(chipID=chipID, config=config)
print('chip.bound::', chip.bound.xmin, chip.bound.xmax, chip.bound.ymin, chip.bound.ymax)
for iobj in range(nobj):
print("\nget psf for iobj-", iobj, '\t', 'bandpass:', end=" ", flush=True)
# Setup Position on focalplane
x, y = imgPos[iobj, :] # try get the PSF at some location (1234, 1234) on the chip
x = x+chip.bound.xmin
y = y+chip.bound.ymin
pos_img = galsim.PositionD(x, y)
# Setup sub-bandpass
# (There are 4 sub-bandpasses for each PSF sample)
filter_param = FilterParam()
filter_id, filter_type = chip.getChipFilter()
filt = Filter(
filter_id=filter_id,
filter_type=filter_type,
filter_param=filter_param,
ccd_bandpass=chip.effCurve)
bandpass_list = filt.bandpass_sub_list
for i in range(len(bandpass_list)):
print(i, end=" ", flush=True)
bandpass = bandpass_list[i] # say you want to access the PSF for the sub-bandpass at the blue end for that chip
# Get corresponding PSF model
psf_model = PSFInterp(chip=chip, npsf=100, PSF_data_file=config["psf_setting"]["psf_dir"])
psf = psf_model.get_PSF(chip=chip, pos_img=pos_img, bandpass=bandpass, galsimGSObject=False)
if True:
fn = "psf_{:}.{:}.{:}.fits".format(chipID, iobj, i)
if fn != None:
if os.path.exists(fn):
os.remove(fn)
hdu = fitsio.PrimaryHDU()
hdu.data = psf
hdu.header.set('pixScale', 5)
hdu.writeto(fn)
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