testCat_star.py

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 "
        self.chip_output.update_output_header(additional_column_names=self.add_hdr)

    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