Skip to content
GitLab
An error occurred while loading the file. Please try again.
Switch branch/tag
Find file BlameHistoryPermalink
Pointing.py 5.59 KiB
Open in Web IDE
1 
import os
2 
import shutil
3 
import yaml
4 
import galsim
5 
import numpy as np
6 
from astropy.time import Time
7
8 
from observation_sim.config._util import get_obs_id
9 
import observation_sim.instruments._util as _util
10
11
12 
class Pointing(object):
13 
    def __init__(self, id=0, ra=0., dec=0., img_pa=0., timestamp=1621915200, sat_x=0., sat_y=0., sat_z=0., sun_x=0., sun_y=0., sun_z=0., sat_vx=0., sat_vy=0., sat_vz=0., exp_time=150., pointing_type='SCI', pointing_type_code='101', pointing_id='00000001', obs_config_file=None, t_shutter_open=1.3, t_shutter_close=1.3):
14 
        self.id = id
15 
        self.ra = ra
16 
        self.dec = dec
17 
        self.img_pa = img_pa * galsim.degrees
18 
        self.timestamp = timestamp
19 
        self.sat_x, self.sat_y, self.sat_z = sat_x, sat_y, sat_z
20 
        self.sun_x, self.sun_y, self.sun_z = sun_x, sun_y, sun_z
21 
        self.sat_vx, self.sat_vy, self.sat_vz = sat_vx, sat_vy, sat_vz
22 
        self.exp_time = exp_time
23 
        self.pointing_type = pointing_type
24 
        self.pointing_type_code = pointing_type_code
25 
        self.obs_id = pointing_id
26 
        self.survey_field_type = 'WIDE'
27 
        self.jdt = 0.
28 
        self.obs_config_file = obs_config_file
29 
        self.t_shutter_open = t_shutter_open
30 
        self.t_shutter_close = t_shutter_close
31 
        self.output_dir = "."
32 
        if self.obs_config_file is not None:
33 
            with open(self.obs_config_file, "r") as stream:
34 
                try:
35 
                    self.obs_param = yaml.safe_load(stream)
36 
                except yaml.YAMLError as exc:
37 
                    print(exc)
38 
            if self.obs_param["obs_type"]:
39 
                self.pointing_type = self.obs_param["obs_type"]
40 
            if self.obs_param["obs_type_code"]:
41 
                self.pointing_type_code = self.obs_param["obs_type_code"]
42 
            if self.obs_param["obs_id"]:
43 
                self.obs_id = str(self.obs_param["obs_id"])
44
45 
    def get_full_depth_exptime(self, filter_type):
46 
        if self.survey_field_type == 'WIDE':
47 
            if filter_type in _util.SPEC_FILTERS:
48 
                return 150. * 4
49 
            else:
50 
                if filter_type.lower() in ['nuv', 'y']:
51 
                    return 150. * 4
52 
                elif filter_type.lower() in ['u', 'g', 'r', 'i', 'z']:
53 
                    return 150. * 2
54 
                else:
55 
                    return max(150., self.exp_time)  # [TODO] for FGS
56 
        elif self.survey_field_type == 'DEEP':
57 
            if filter_type in _util.SPEC_FILTERS:
58 
                return 250. * 4 * 4
59 
            else:
60 
                if filter_type.lower() in ['nuv', 'y']:
61 
                    return 250. * 4 * 4
62 
                elif filter_type.lower() in ['u', 'g', 'r', 'i', 'z']:
63 
                    return 250. * 2 * 4
64 
                else:
65 
                    return max(150., self.exp_time)  # [TODO] for FGS
66
67 
    def read_pointing_columns(self, columns, id=0, t=1621915200, pointing_type='SCI'):
68 
        self.id = id
69 
        col_len = len(columns)
70 
        self.ra = float(columns[0])
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 127 128 129 130 131 132 133 134 135 136 137
self.dec = float(columns[1]) self.img_pa = float(columns[4]) * galsim.degrees # self.pointing_type = pointing_type if col_len > 5: jdt = np.double(columns[5]) t_temp = Time(jdt, format='jd') self.jdt = jdt self.timestamp = t_temp.unix self.sat_x = float(columns[6]) self.sat_y = float(columns[7]) self.sat_z = float(columns[8]) self.sun_x = float(columns[9]) self.sun_y = float(columns[10]) self.sun_z = float(columns[11]) self.sat_vx = float(columns[15]) self.sat_vy = float(columns[16]) self.sat_vz = float(columns[17]) self.exp_time = float(columns[18]) is_deep = float(columns[19]) # [TODO] Can also define other survey types if is_deep == 2: self.survey_field_type = "DEEP" if not self.obs_config_file: self.obs_config_file = str(columns[20]) with open(self.obs_config_file, "r") as stream: try: self.obs_param = yaml.safe_load(stream) except yaml.YAMLError as exc: print(exc) self.pointing_type_code = columns[21][0:3] self.obs_id = columns[21][3:] self.pointing_type = self.obs_param["obs_type"] else: self.timestamp = t def make_output_pointing_dir(self, overall_config, copy_obs_config=False): run_dir = os.path.join( overall_config["work_dir"], overall_config["run_name"]) if not os.path.exists(run_dir): try: os.makedirs(run_dir, exist_ok=True) except OSError: pass self.output_prefix = get_obs_id( img_type=self.pointing_type, project_cycle=overall_config["project_cycle"], run_counter=overall_config["run_counter"], pointing_id=self.obs_id, pointing_type_code=self.pointing_type_code) self.output_dir = os.path.join(run_dir, self.output_prefix) if not os.path.exists(self.output_dir): try: os.makedirs(self.output_dir, exist_ok=True) except OSError: pass if copy_obs_config and self.obs_config_file: obs_config_output_path = os.path.join( self.output_dir, os.path.basename(self.obs_config_file)) if not os.path.exists(obs_config_output_path): try: shutil.copy(self.obs_config_file, self.output_dir) except OSError: pass