-
Notifications
You must be signed in to change notification settings - Fork 11
/
Copy pathbuild_ql_calibs
executable file
·259 lines (213 loc) · 11.6 KB
/
build_ql_calibs
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
127
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
#!/usr/bin/env python3
#
# See top-level LICENSE.rst file for Copyright information
#
# -*- coding: utf-8 -*-
"""
This script builds the QL calibrations. It can copy files from a previous run of
the dev suite, but if those files aren't found it will the relevant dev suite
tests to generate them.
"""
from pathlib import Path
import os
import sys
import shutil
import glob
import traceback
from collections import namedtuple
from test_scripts.test_main import build_test_setup
from test_scripts.pypeit_tests import PypeItQuickLookTest
import argparse
import warnings
from IPython import embed
from pypeit import version as pypeit_version
from pypeit.spectrographs import util as spec_util
from pypeit import utils
def parser(options=None):
parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument('instrument', type=str, default="all",
help='Instrument for which to generate calibration frames for quicklook '
'scripts. Options are "keck_nires", "keck_mosfire", '
'"keck_lris_red_mark4", or "all". Defaults to "all".')
parser.add_argument('-s', '--setup', type=str, default="all",
help='Instrument setup for which to generate calibration frames for '
'quicklook scripts. Options depend on instrument. Defaults to "all".')
parser.add_argument('--redux_dir', type=str, help='Full path to the REDUX dir; '
'default is REDUX_OUT in current directory')
parser.add_argument('-o', '--output_dir', type=str,
help='Full path to the destination directory for the QL calibrations.'
'Defaults to the "QL_CALIB" environment variable or "QL_CALIB" in '
'the current directory')
parser.add_argument('-fc', '--force_copy', action='store_true', default=False,
help='Copy the files even if they already exist in the destination')
parser.add_argument('-fb', '--force_build', action='store_true', default=False,
help='Always rebuild the calibrations via the dev-suite, even if they '
'already exist in the destination')
return parser.parse_args() if options is None else parser.parse_args(options)
def main():
pargs = parser()
# Set the redux_dir where the calibrations generated by the dev suite live.
redux_dir = Path('REDUX_OUT').resolve() if pargs.redux_dir is None \
else Path(pargs.redux_dir).resolve()
# Set the output dir where the calibrations should be placed.
ql_calib = Path(os.getenv('QL_CALIB', default='QL_CALIB') if pargs.output_dir is None
else pargs.output_dir).resolve()
# Build the list of files to copy and where to copy them to.
# This includes updating any filenames to match what the quicklook
# scripts expect, and may include generating the files uisng the dev suite.
source_files = []
dest_files = []
# NOTE: NIRES only has one setup
if pargs.instrument == 'all' or pargs.instrument == 'keck_nires':
if pargs.setup != 'all':
warnings.warn('NIRES only has one setup.')
nires_dest_dir = ql_calib / 'keck_nires_A' / 'Calibrations'
nires_dest_dir.mkdir(parents=True, exist_ok=True)
# Source dev-suite directory for all NIRES calibrations
#nires_setup = 'ABBA_wstandard'
# NOTE: Use this example instead because all of the QL tests currently
# use it!
nires_setup = 'ABpat_wstandard'
keck_nires_path = redux_dir / 'keck_nires' / nires_setup / 'Calibrations'
found_files = sorted(keck_nires_path.glob('*')) if keck_nires_path.exists() else []
if len(found_files) == 0 or pargs.force_build:
warnings.warn('NIRES calibrations missing or re-building forced. Processing...')
run_devsuite('keck_nires', nires_setup, redux_dir)
found_files = sorted(keck_nires_path.glob('*'))
if len(found_files) == 0:
raise ValueError('Could not generate NIRES calibrations.')
source_files += found_files
dest_files += [nires_dest_dir / f.name for f in found_files]
source_files += [list(keck_nires_path.parent.glob('*.pypeit'))[0]]
dest_files += [nires_dest_dir.parent / 'keck_nires_A.pypeit']
if pargs.instrument == 'all' or pargs.instrument == 'keck_mosfire':
if pargs.setup == 'all' or pargs.setup == "Y_long":
spec = spec_util.load_spectrograph("keck_mosfire")
mosfire_dest_dir = ql_calib / 'keck_mosfire_A' / 'Calibrations'
mosfire_dest_dir.mkdir(parents=True, exist_ok=True)
keck_mosfire_path = redux_dir / 'keck_mosfire' / pargs.setup / 'Calibrations'
files = ['Arc_A_2_DET01.fits',
'Flat_A_all_DET01.fits',
'Slits_A_all_DET01.fits.gz',
'Tilts_A_2_DET01.fits',
'WaveCalib_A_2_DET01.fits']
found_files = [keck_mosfire_path / f for f in files
if Path(keck_mosfire_path / f).exists()]
if len(found_files) != len(files) or pargs.force_build:
print('Could not find all MOSFIRE calibrations, attempting to generate them.')
run_devsuite('keck_mosfire', 'Y_long', redux_dir)
found_files = [keck_mosfire_path / f for f in files
if Path(keck_mosfire_path / f).exists()]
if len(found_files) != len(files):
raise ValueError('Could not generate MOSFIRE calibrations.')
source_files += found_files
dest_files += [mosfire_dest_dir / f.name for f in found_files]
source_files += [list(keck_mosfire_path.parent.glob('*.pypeit'))[0]]
dest_files += [mosfire_dest_dir.parent / 'keck_mosfire_A.pypeit']
# TODO: Removing these for now, after deprecation of ql_multislit.py. May
# recover much of this code, if we add a parent archive directory for these
# instruments.
# if pargs.instrument == 'all' or pargs.instrument == 'keck_lris_red_mark4':
# if pargs.setup == 'all' or pargs.setup == "long_600_10000_d680":
#
# dev_suite_output = os.path.join(redux_dir, 'keck_lris_red_mark4', 'long_600_10000_d680')
# rdx_calibs = os.path.join(dev_suite_output, "Calibrations")
# rdx_science = os.path.join(dev_suite_output, "Science")
#
#
# # Find the calibrations
# bias_file = utils.find_single_file(os.path.join(rdx_calibs, 'Bias*'))
# slits_file = utils.find_single_file(os.path.join(rdx_calibs, 'Slits*'))
# tilts_file = utils.find_single_file(os.path.join(rdx_calibs, 'Tilts*'))
# wave_calib = utils.find_single_file(os.path.join(rdx_calibs, 'WaveCalib*'))
# std_spec1d = utils.find_single_file(os.path.join(rdx_science,
# 'spec1d_*00127-GD153*.fits'))
# sens_file = utils.find_single_file(os.path.join(dev_suite_output, 'sens_*.fits'))
#
# if bias_file is None or slits_file is None or tilts_file is None \
# or wave_calib is None or std_spec1d is None or sens_file is None:
# print(f"Couldn't find all QL calibrations in {dev_suite_output}, attempting to "
# "generate them.")
# run_devsuite("keck_lris_red_mark4", 'long_600_10000_d680', redux_dir)
#
# bias_file = utils.find_single_file(os.path.join(rdx_calibs, 'Bias*'))
# slits_file = utils.find_single_file(os.path.join(rdx_calibs, 'Slits*'))
# tilts_file = utils.find_single_file(os.path.join(rdx_calibs, 'Tilts*'))
# wave_calib = utils.find_single_file(os.path.join(rdx_calibs, 'WaveCalib*'))
# std_spec1d = utils.find_single_file(os.path.join(rdx_science,
# 'spec1d_*00127-GD153*.fits'))
# sens_file = utils.find_single_file(os.path.join(dev_suite_output, 'sens_*.fits'))
# if bias_file is None or slits_file is None or tilts_file is None \
# or wave_calib is None or std_spec1d is None or sens_file is None:
# print("Failed to generate keck_lris_red_mark4 calibrations.")
# sys.exit(1)
#
# source_files = [bias_file, slits_file, tilts_file, wave_calib, std_spec1d, sens_file]
#
# # Find the destination directory.
# # Use a spec2d to get the QL calib_dir as it will have the
# # header information from the RAW science file
# spec = spec_util.load_spectrograph("keck_lris_red_mark4")
# spec2d = utils.find_single_file(os.path.join(rdx_science, "spec2d*"))
# if spec2d is None:
# print(f"Couldn't find spec2d file in {rdx_science} to read")
# sys.exit(1)
# dest_dir = os.path.join(ql_calib, spec.get_ql_calib_dir(spec2d))
# os.makedirs(dest_dir, exist_ok=True)
#
# dest_files = [os.path.join(dest_dir, os.path.basename(file)) for file in source_files]
# Copy the files
for source_file, dest_file in zip(source_files, dest_files):
copy_me(source_file, dest_file, force=pargs.force_copy)
def find_source_files(source_filters):
"""Expand any wildcards to see if any exist"""
found_files = []
for filter in source_filters:
found_files += glob.glob(filter)
return found_files
def copy_me(src, dst, force=False):
"""
Copy a file under certain conditions.
If the destination already exists, it will only be copied if its timestamp
is more recent than the source, unless forced.
Args:
src (:obj:`str`, Path):
Existing file to copy.
dst (:obj:`str`, Path):
Destination for file copy.
force (:obj:`bool`, optional):
Force the copy
"""
# Compare date stamp
if force or not Path(dst).exists() or Path(dst).stat().st_mtime < Path(src).stat().st_mtime:
shutil.copy2(src, dst)
print(f'Generated/over-wrote {dst}')
def run_devsuite(instrument, setup, redux_dir):
"""
Run the dev-suite tests needed to generate quick-look calibrations for an
instrument and setup.
"""
# The parameters to simulate the command line arguments to pypeit_test
DevSuiteParams = namedtuple("DevSuiteParams", ["tests", "prep_only", "do_not_reuse_calibs",
"outputdir", "coverage"])
absolute_outputdir = os.path.abspath(redux_dir)
pargs = DevSuiteParams(tests=["reduce", "afterburn"], prep_only=False,
do_not_reuse_calibs=False, outputdir=absolute_outputdir,
coverage=False)
test_setup = build_test_setup(pargs, instrument, setup, True, True, False)
for test in test_setup.tests:
# A quick look test will run build_ql_calibs, potentially creating an
# infinite recursion. So skip those tests
if isinstance(test, PypeItQuickLookTest):
continue
print(f"Running test {instrument}/{setup} {test.description}")
passed = test.run()
if not passed:
print(f"Failed running dev suite tests for {instrument}/{setup}")
for message in test.error_msgs:
print(message)
print(f"Also be sure to chek the log at {test.logfile}")
break
if __name__ == '__main__':
# Giddy up
main()