testCat_star.py 15 KB
Newer Older
Fang Yuedong's avatar
Fang Yuedong committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
import os
import galsim
import random
import numpy as np
import h5py as h5
import healpy as hp
import astropy.constants as cons
import traceback
from astropy.coordinates import spherical_to_cartesian
from astropy.table import Table
from scipy import interpolate
from datetime import datetime

from ObservationSim.MockObject import CatalogBase, Star, Galaxy, Quasar, Stamp
from ObservationSim.MockObject._util import tag_sed, getObservedSED, getABMAG, integrate_sed_bandpass, comoving_dist
from ObservationSim.Astrometry.Astrometry_util import on_orbit_obs_position

import astropy.io.fits as fitsio
from ObservationSim.MockObject._util import seds, sed_assign, extAv

# (TEST)
from astropy.cosmology import FlatLambdaCDM
from astropy import constants
from astropy import units as U

try:
    import importlib.resources as pkg_resources
except ImportError:
    # Try backported to PY<37 'importlib_resources'
    import importlib_resources as pkg_resources

NSIDE = 128

class Catalog(CatalogBase):
    """An user customizable class for reading in catalog(s) of objects and SEDs.
    
    NOTE: must inherit the "CatalogBase" abstract class

    ...
    
    Attributes
    ----------
    cat_dir : str
        a directory that contains the catalog file(s)
    star_path : str
        path to the star catalog file
    star_SED_path : str
        path to the star SED data
    objs : list
        a list of ObservationSim.MockObject (Star, Galaxy, or Quasar)
        NOTE: must have "obj" list when implement your own Catalog
    
    Methods
    ----------
    load_sed(obj, **kwargs):
        load the corresponding SED data for one object
    load_norm_filt(obj):
        load the filter throughput for the input catalog's photometric system.
    """
    def __init__(self, config, chip, pointing, chip_output, filt, **kwargs):
        """Constructor method.
        
        Parameters
        ----------
        config : dict
            configuration dictionary which is parsed from the input YAML file
        chip: ObservationSim.Instrument.Chip
            an ObservationSim.Instrument.Chip instance, can be used to identify the band etc.
        pointing: ObservationSim.Config.Pointing
            an ObservationSim.Config.Pointing instance, can be used to configure the astrometry module
        chip_output: ObservationSim.Config.ChipOutput
            an ObservationSim.Config.ChipOutput instance, can be used to setup the output format
        filt: ObservationSim.Instrument.Filter
            an ObservationSim.Instrument.Filter instance, can be used to identify the filter type
        **kwargs : dict
            other needed input parameters (in key-value pairs), please modify corresponding
            initialization call in "ObservationSim.py" as you need.
        
        Returns
        ----------
        None
        """
        
        super().__init__()
        self.cat_dir = os.path.join(config["data_dir"], config["catalog_options"]["input_path"]["cat_dir"])
        self.seed_Av = config["catalog_options"]["seed_Av"]

        #self.cosmo = FlatLambdaCDM(H0=67.66, Om0=0.3111)

        self.chip_output = chip_output
        self.filt = filt
        self.logger = chip_output.logger

        with pkg_resources.path('Catalog.data', 'SLOAN_SDSS.g.fits') as filter_path:
            self.normF_star = Table.read(str(filter_path))
        
        self.config = config
        self.chip = chip
        self.pointing = pointing

        self.max_size = 0.

        if "star_cat" in config["catalog_options"]["input_path"] and config["catalog_options"]["input_path"]["star_cat"] and config["catalog_options"]["star_yes"]:
            self.star_path = os.path.join(self.cat_dir, config["catalog_options"]["input_path"]["star_cat"])

            self.star_SED_path = os.path.join(self.cat_dir, config["catalog_options"]["SED_templates_path"]["star_SED"])
            self._load_SED_lib_star()
        

        if "rotateEll" in config["catalog_options"]:
            self.rotation = float(int(config["catalog_options"]["rotateEll"]/45.))
        else:
            self.rotation = 0.

        # Update output .cat header with catalog specific output columns
        self._add_output_columns_header()

        self._get_healpix_list()
        self._load()
    
    def _add_output_columns_header(self):
        self.add_hdr = " model_tag teff logg feh"
        self.add_hdr += " bulgemass diskmass detA e1 e2 kappa g1 g2 size galType veldisp "

        self.add_fmt = " %10s %8.4f %8.4f %8.4f"
        self.add_fmt += " %8.4f %8.4f %8.4f %8.4f %8.4f %8.4f %8.4f %8.4f %8.4f %4d %8.4f "
127
        self.chip_output.update_output_header(additional_column_names=self.add_hdr)
Fang Yuedong's avatar
Fang Yuedong committed
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378

    def _get_healpix_list(self):
        self.sky_coverage = self.chip.getSkyCoverageEnlarged(self.chip.img.wcs, margin=0.2)
        ra_min, ra_max, dec_min, dec_max = self.sky_coverage.xmin, self.sky_coverage.xmax, self.sky_coverage.ymin, self.sky_coverage.ymax
        ra = np.deg2rad(np.array([ra_min, ra_max, ra_max, ra_min]))
        dec = np.deg2rad(np.array([dec_max, dec_max, dec_min, dec_min]))
        # vertices = spherical_to_cartesian(1., dec, ra)
        self.pix_list = hp.query_polygon(
            NSIDE,
            hp.ang2vec(np.radians(90.) - dec, ra),
            inclusive=True
        )
        # self.pix_list = hp.query_polygon(NSIDE, np.array(vertices).T, inclusive=True)
        if self.logger is not None:
            msg = str(("HEALPix List: ", self.pix_list))
            self.logger.info(msg)
        else:
            print("HEALPix List: ", self.pix_list)

    def load_norm_filt(self, obj):
        """Load the corresponding thourghput for the input magnitude "param["mag_use_normal"]".

        NOTE: if the input magnitude is already in CSST magnitude, simply return None

        Parameters
        ----------
        obj : ObservationSim.MockObject
            the object to get thourghput data for

        Returns
        ----------
        norm_filt : Astropy.Table
            the throughput Table with two columns (namely, "WAVELENGTH", "SENSITIVITY"):
                norm_filt["WAVELENGTH"] : wavelengthes in Angstroms
                norm_filt["SENSITIVITY"] : efficiencies
        """
        if obj.type == "star":
            return self.normF_star
        else:
            return None

    def _load_SED_lib_star(self):
        self.tempSED_star = h5.File(self.star_SED_path,'r')

    def _load_stars(self, stars, pix_id=None):
        nstars = len(stars['sourceID'])
        # Apply astrometric modeling
        ra_arr = stars["RA"][:]
        dec_arr = stars["Dec"][:]
        pmra_arr = stars['pmra'][:]
        pmdec_arr = stars['pmdec'][:]
        rv_arr = stars['RV'][:]
        parallax_arr = stars['parallax'][:]
        if self.config["obs_setting"]["enable_astrometric_model"]:
            ra_list = ra_arr.tolist()
            dec_list = dec_arr.tolist()
            pmra_list = pmra_arr.tolist()
            pmdec_list = pmdec_arr.tolist()
            rv_list = rv_arr.tolist()
            parallax_list = parallax_arr.tolist()
            dt = datetime.utcfromtimestamp(self.pointing.timestamp)
            date_str = dt.date().isoformat()
            time_str = dt.time().isoformat()
            ra_arr, dec_arr = on_orbit_obs_position(
                input_ra_list=ra_list,
                input_dec_list=dec_list,
                input_pmra_list=pmra_list,
                input_pmdec_list=pmdec_list,
                input_rv_list=rv_list,
                input_parallax_list=parallax_list,
                input_nstars=nstars,
                input_x=self.pointing.sat_x,
                input_y=self.pointing.sat_y,
                input_z=self.pointing.sat_z,
                input_vx=self.pointing.sat_vx,
                input_vy=self.pointing.sat_vy,
                input_vz=self.pointing.sat_vz,
                input_epoch="J2000",
                input_date_str=date_str,
                input_time_str=time_str
            )
        for istars in range(nstars):
            # # (TEST)
            # if istars > 100:
            #     break

            param = self.initialize_param()
            param['ra'] = ra_arr[istars]
            param['dec'] = dec_arr[istars]
            param['ra_orig'] = stars["RA"][istars]
            param['dec_orig'] = stars["Dec"][istars]
            param['pmra'] = pmra_arr[istars]
            param['pmdec'] = pmdec_arr[istars]
            param['rv'] = rv_arr[istars]
            param['parallax'] = parallax_arr[istars]
            if not self.chip.isContainObj(ra_obj=param['ra'], dec_obj=param['dec'], margin=200):
                continue
            param['mag_use_normal'] = stars['app_sdss_g'][istars]
            # if param['mag_use_normal'] >= 26.5:
            #     continue
            self.ids += 1
            param['id'] = stars['sourceID'][istars]
            param['sed_type'] = stars['sourceID'][istars]
            param['model_tag'] = stars['model_tag'][istars]
            param['teff'] = stars['teff'][istars]
            param['logg'] = stars['grav'][istars]
            param['feh'] = stars['feh'][istars]
            param['z'] = 0.0
            param['star'] = 1   # Star
            obj = Star(param, logger=self.logger)

            # Append additional output columns to the .cat file
            obj.additional_output_str = self.add_fmt%(param["model_tag"], param['teff'], param['logg'], param['feh'],
                                                    0., 0., 0., 0., 0., 0., 0., 0., 0., -1, 0.)

            self.objs.append(obj)

    def _load(self, **kwargs):
        """Read in all objects in from the catalog file(s).

        This is a must implemented method which is used to read in all objects, and
        then convert them to ObservationSim.MockObject (Star, Galaxy, or Quasar).
        
        Currently, 
        the model of ObservationSim.MockObject.Star class requires:
            param["star"] : int
                specify the object type: 0: galaxy, 1: star, 2: quasar
            param["id"] : int
                ID number of the object
            param["ra"] : float
                Right ascension (in degrees)
            param["dec"] : float
                Declination (in degrees)
            param["mag_use_normal"]: float
                the absolute magnitude in a particular filter
                NOTE: if that filter is not the corresponding CSST filter, the 
                load_norm_filt(obj) function must be implemented to load the filter
                throughput of that particular photometric system
        
        the model of ObservationSim.MockObject.Galaxy class requires:
            param["star"] : int
                specify the object type: 0: galaxy, 1: star, 2: quasar
            param["id"] : int
                ID number of the object
            param["ra"] : float
                Right ascension (in degrees)
            param["dec"] : float
                Declination (in degrees)
            param["mag_use_normal"]: float
                the absolute magnitude in a particular filter
                NOTE: if that filter is not the corresponding CSST filter, the 
                load_norm_filt(obj) function must be implemented to load the filter
                throughput of that particular photometric system
            param["bfrac"] : float
                the bulge fraction
            param["hlr_bulge"] : float
                the half-light-radius of the bulge
            param["hlr_disk"] : float
                the half-light-radius of the disk
            param["e1_bulge"], param["e2_bulge"] : float
                the ellipticity of the bulge components
            param["e1_disk"], param["e2_disk"] : float
                the ellipticity of the disk components
            (Optional parameters):
            param['disk_sersic_idx']: float
                Sersic index for galaxy disk component
            param['bulge_sersic_idx']: float
                Sersic index for galaxy bulge component
            param['g1'], param['g2']: float
                Reduced weak lensing shear components (valid for shear type: catalog)
        the model of ObservationSim.MockObject.Galaxy class requires:
            Currently a Quasar is modeled as a point source, just like a Star.

        NOTE: To construct an object, according to its type, just call:
            Star(param), Galaxy(param), or Quasar(param)

        NOTE: All constructed objects should be appened to "self.objs".

        NOTE: Any other parameters can also be set within "param" dict:
            Used to calculate required quantities and/or SEDs etc.

        Parameters
        ----------
        **kwargs : dict
            other needed input parameters (in key-value pairs), please modify corresponding
            initialization call in "ObservationSim.py" as you need.

        Returns
        ----------
        None
        """
        self.objs = []
        self.ids = 0
        
        #if "star_cat" in self.config["input_path"] and self.config["input_path"]["star_cat"] and not self.config["run_option"]["galaxy_only"]:
        if "star_cat" in self.config["catalog_options"]["input_path"] and self.config["catalog_options"]["input_path"]["star_cat"] and self.config["catalog_options"]["star_yes"]:
            star_cat = h5.File(self.star_path, 'r')['stars']
            for pix in self.pix_list:
                try:
                    stars = star_cat[str(pix)]
                    self._load_stars(stars, pix_id=pix)
                    del stars
                except Exception as e:
                    self.logger.error(str(e))
                    print(e)

        if self.logger is not None:
            self.logger.info("number of objects in catalog: %d"%(len(self.objs)))
        else:
            print("number of objects in catalog: ", len(self.objs))

    def load_sed(self, obj, **kwargs):
        """Load the corresponding SED data for a particular obj.

        Parameters
        ----------
        obj : ObservationSim.MockObject
            the object to get SED data for
        **kwargs : dict
            other needed input parameters (in key-value pairs), please modify corresponding
            initialization call in "ObservationSim.py" as you need.

        Returns
        ----------
        sed : Astropy.Table
            the SED Table with two columns (namely, "WAVELENGTH", "FLUX"):
                sed["WAVELENGTH"] : wavelength in Angstroms
                sed["FLUX"] : fluxes in photons/s/m^2/A
            NOTE: the range of wavelengthes must at least cover [2450 - 11000] Angstorms
        """

        if obj.type == 'star':
            _, wave, flux = tag_sed(
                h5file=self.tempSED_star,
                model_tag=obj.param['model_tag'],
                teff=obj.param['teff'],
                logg=obj.param['logg'],
                feh=obj.param['feh']
            )
        else:
            raise ValueError("Object type not known")
        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'))
        
        del wave
        del flux
        return sed