-
Notifications
You must be signed in to change notification settings - Fork 0
/
kmos_functions.py
372 lines (293 loc) · 14.9 KB
/
kmos_functions.py
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
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
import glob
import logging
import subprocess
from subprocess import Popen, list2cmdline
import shutil
import readline
import os
readline.set_completer_delims(' \t\n;')
readline.parse_and_bind("tab: complete")
def multiple_log(message):
print(message)
logging.info(message)
def do_catg_from_file(file_entry):
"""Get the DO_CATG for a given file"""
if file_entry['tpl_id'] == "KMOS_spec_cal_dark" :
return "DARK"
elif file_entry['tpl_id'] == "KMOS_spec_cal_calunitflat" :
if file_entry['dpr_type'] == "FLAT,OFF" :
return "FLAT_OFF"
if file_entry['dpr_type'] == "FLAT,LAMP" :
return "FLAT_ON"
elif file_entry['tpl_id'] == "KMOS_spec_cal_wave" :
if file_entry['dpr_type'] == "WAVE,OFF" :
return "ARC_OFF"
if file_entry['dpr_type'] == "WAVE,LAMP" :
return "ARC_ON"
elif file_entry['tpl_id'] == "KMOS_spec_cal_stdstar":
return "STD"
elif file_entry['tpl_id'] == "KMOS_spec_obs_stare":
return "SCIENCE"
else:
raise NameError("Unknown DO_CATG")
return
def tpl_id_to_recipe_name( tpl_id ) :
"Get the recipe name from the tpl.id"
if tpl_id == "KMOS_spec_cal_dark" :
return "kmos_dark"
if tpl_id == "KMOS_spec_cal_calunitflat" :
return "kmos_flat"
if tpl_id == "KMOS_spec_cal_wave" :
return "kmos_wave_cal"
return "Unknown_TPL_ID"
def exec_commands_seq(cmds):
"Exec commands sequentially"
for cmd in cmds:
multiple_log("\tRun in {0} : {1} ...".format(cmd['dir'], list2cmdline(cmd['cmd'])))
cwd = os.getcwd()
os.chdir(cmd['dir'])
#subprocess.check_call(cmd['cmd'], stdout=open(os.devnull, 'wb'), stderr=open(os.devnull, 'wb'))
subprocess.check_call(cmd['cmd'])
os.chdir(cwd)
def exec_commands_par(cmds):
"Exec commands in parallel in multiple process"
def done(p):
return p.poll() is not None
def success(p):
return p.returncode == 0
def fail():
sys.exit(1)
max_task = os.sysconf('SC_NPROCESSORS_ONLN')
processes = []
while True:
while cmds and len(processes) < max_task:
cmd = cmds.pop()
multiple_log("\tRun in {0} : {1} ...".format(cmd['dir'], list2cmdline(cmd['cmd'])))
processes.append(Popen(cmd['cmd'],cwd=cmd['dir'],stdout=open(os.devnull, 'wb'),stderr=open(os.devnull, 'wb')))
for p in processes:
if done(p):
if success(p):
processes.remove(p)
else:
fail()
if not processes and not cmds:
break
else:
time.sleep(0.05)
def exec_commands(cmds, par):
"Exec commands"
# Empty list
if not cmds:
return
if par:
exec_commands_par(cmds)
else:
exec_commands_seq(cmds)
def SAM_create_sof(filelist, recipe_name, calibration_location=None, reduced_dark_folder=None, reduced_flat_folder=None, kmos_static_calib_directory='/Data/KCLASH/Data/Static_Cals/cal', location_for_sof=None, telluric_directory=None):
"Creates a SOF file. This is saved in location_for_sof, or else the same location as the rest of the calibration files is location_for_sof is None"
###############
# Verifications
if len(filelist) < 1 :
raise NameError("Empty List")
return
# Same band ?
if len(set([x['band'] for x in filelist])) != 1 :
raise NameError("Different bands in this sof")
return
# Same tpl_id ?
if len(set([x['tpl_id'] for x in filelist])) != 1 :
raise NameError("Different TPL_IDs in this sof")
return
#Commented this our- SPV 15/08/17
# # Same tpl_start ?
# if len(set(map(lambda x: x['tpl_start'], filelist))) != 1 :
# raise NameError("Different TPL_STARTs in this sof")
# return
###############
# SOF file name
sof_basename = "{}_{}.sof".format(recipe_name, filelist[0]['tpl_start'])
#Unless told otherwise, make the .sof file in the same location as the rest of rthe calibration files
if location_for_sof is None:
sof_name = "{}/{}".format(calibration_location, sof_basename)
else:
sof_name = "{}/{}".format(location_for_sof, sof_basename)
# Open the file
sof = open(sof_name, "wb")
# Write RAW Files
for file_entry in filelist:
sof.write("{}\t{}\n".format(file_entry['name'], do_catg_from_file(file_entry)))
band_lc = filelist[0]['band'].lower()
band_3uc = filelist[0]['band']*3
assert recipe_name in ['kmos_dark', 'kmos_flat', 'kmos_wave_cal', 'kmos_illumination', 'kmos_std_star', 'kmos_sci_red', 'kmos_combine'], 'Recipe name not understood!'
# Add Calibrations
if recipe_name == 'kmos_dark':
#Don't need to add anything here
pass
elif recipe_name == 'kmos_flat' :
if reduced_dark_folder is None:
fname='{}/BADPIXEL_DARK.fits'.format(calibration_location)
check_file_exists(fname)
sof.write('{} BADPIXEL_DARK\n'.format(fname))
else:
fname='{}/badpixel_dark.fits'.format(reduced_dark_folder)
check_file_exists(fname)
sof.write('{} BADPIXEL_DARK\n'.format(fname))
elif recipe_name == 'kmos_wave_cal':
sof.write("{}/kmos_wave_ref_table.fits REF_LINES\n".format(kmos_static_calib_directory))
sof.write("{}/kmos_wave_band.fits WAVE_BAND\n".format(kmos_static_calib_directory))
sof.write("{}/kmos_ar_ne_list_{}.fits ARC_LIST\n".format(kmos_static_calib_directory, band_lc))
if reduced_flat_folder is None:
sof.write("{}/FLAT_EDGE_{}.fits FLAT_EDGE\n".format(calibration_location, band_3uc))
sof.write("{}/XCAL_{}.fits XCAL\n".format(calibration_location, band_3uc))
sof.write("{}/YCAL_{}.fits YCAL\n".format(calibration_location, band_3uc))
else:
sof.write("{}/FLAT_EDGE_{}.fits FLAT_EDGE\n".format(reduced_flat_folder))
sof.write("{}/XCAL_{}.fits XCAL\n".format(reduced_flat_folder))
sof.write("{}/YCAL_{}.fits YCAL\n".format(reduced_flat_folder))
elif recipe_name == 'kmos_illumination':
#Write the sof for the illumination correction. Requires XCAL, YCAL, LCAL, FLAT_EDGE and the static WAVE_BAND
sof.write("{}/kmos_wave_band.fits WAVE_BAND\n".format(kmos_static_calib_directory))
sof.write("{}/LCAL_{}.fits LCAL\n".format(calibration_location, band_3uc))
if reduced_flat_folder is None:
sof.write("{}/FLAT_EDGE_{}.fits FLAT_EDGE\n".format(calibration_location, band_3uc))
sof.write("{}/XCAL_{}.fits XCAL\n".format(calibration_location, band_3uc))
sof.write("{}/YCAL_{}.fits YCAL\n".format(calibration_location, band_3uc))
else:
sof.write("{}/FLAT_EDGE_{}.fits FLAT_EDGE\n".format(reduced_flat_folder))
sof.write("{}/XCAL_{}.fits XCAL\n".format(reduced_flat_folder))
sof.write("{}/YCAL_{}.fits YCAL\n".format(reduced_flat_folder))
elif recipe_name == 'kmos_std_star':
#Write the sof for the illumination correction. Requires XCAL, YCAL, LCAL, FLAT_EDGE and the static WAVE_BAND
sof.write("{}/kmos_wave_band.fits WAVE_BAND\n".format(kmos_static_calib_directory))
sof.write("{}/kmos_spec_type.fits SPEC_TYPE_LOOKUP\n".format(kmos_static_calib_directory))
sof.write("{}/kmos_atmos_iz.fits ATMOS_MODEL\n".format(kmos_static_calib_directory))
sof.write("{}/LCAL_{}.fits LCAL\n".format(calibration_location, band_3uc))
sof.write("{}/ILLUM_CORR_{}.fits ILLUM_CORR\n".format(calibration_location, band_3uc))
sof.write("{}/MASTER_FLAT_{}.fits MASTER_FLAT\n".format(calibration_location, band_3uc))
sof.write("{}/XCAL_{}.fits XCAL\n".format(calibration_location, band_3uc))
sof.write("{}/YCAL_{}.fits YCAL\n".format(calibration_location, band_3uc))
elif recipe_name == 'kmos_sci_red':
#Write the sof for the illumination correction. Requires XCAL, YCAL, LCAL, FLAT_EDGE and the static WAVE_BAND
#assert telluric_directory is not None, 'Must have a telluric file location for the science reduction!'
sof.write("{}/LCAL_{}.fits LCAL\n".format(calibration_location, band_3uc))
sof.write("{}/ILLUM_CORR_{}.fits ILLUM_CORR\n".format(calibration_location, band_3uc))
sof.write("{}/MASTER_FLAT_{}.fits MASTER_FLAT\n".format(calibration_location, band_3uc))
sof.write("{}/XCAL_{}.fits XCAL\n".format(calibration_location, band_3uc))
sof.write("{}/YCAL_{}.fits YCAL\n".format(calibration_location, band_3uc))
sof.write("{}/kmos_wave_band.fits WAVE_BAND\n".format(kmos_static_calib_directory))
sof.write("{}/kmos_oh_spec_iz.fits OH_SPEC\n".format(kmos_static_calib_directory))
if telluric_directory:
sof.write("{}/TELLURIC_IZIZIZ.fits TELLURIC\n".format(telluric_directory))
elif recipe_name == 'kmos_combine':
#We don't need any other files for kmos_combine
pass
else:
raise NameError('Kmos recipe "{}" not understood!'.format(recipe_name))
# Close sof file
sof.close()
return sof_basename
def reduce_darks(destination_directory, dark_files, options):
"""
Reduce a set of darks.
Arguments:
destination_directory: directory the .sof file will be created in and where esorex will be run
dark_files: List of raw dark files to reduce
options: Command line options
"""
# Generate exection command
cmds=[]
recipe_name='kmos_dark'
sof_name = SAM_create_sof(dark_files, recipe_name, location_for_sof=destination_directory)
log_file = "esorex_"+recipe_name+".log"
base_cmd = ['esorex', '--suppress-prefix=TRUE', '--log-file='+log_file, '--log-dir=.', recipe_name, sof_name]
cmds.append({'cmd': base_cmd, 'dir': destination_directory})
exec_commands(cmds, options.parallel)
def reduce_calibs(destination_directory, filelist, recipe_name, options, reduced_dark_folder=None, reduced_flat_folder=None):
"""
Reduce a set of calibrations.
Arguments:
destination_directory: directory the .sof file will be created in
filelist: List of raw files to reduce
recipe name: The name of the esorex recipe to call.
options: Command line options
reduced_dark_folder. Default is None. If we've already reduced darks, which folder are they in?
reduced_flat_folder. Default is None. If we've already reduced the flats, which folder are they in?
"""
# Generate exection command
cmds=[]
sof_name = SAM_create_sof(filelist, recipe_name, calibration_location=destination_directory, reduced_dark_folder=reduced_dark_folder, reduced_flat_folder=reduced_flat_folder, location_for_sof=destination_directory)
log_file = "esorex_"+recipe_name+".log"
if recipe_name=='kmos_illumination':
base_cmd = ['esorex', '--suppress-prefix=TRUE', '--log-file='+log_file, '--log-dir=.', recipe_name, '--pix_scale=0.2', sof_name]
else:
base_cmd = ['esorex', '--suppress-prefix=TRUE', '--log-file='+log_file, '--log-dir=.', recipe_name, sof_name]
cmds.append({'cmd': base_cmd, 'dir': destination_directory})
exec_commands(cmds, options.parallel)
def reduce_std_star(destination_directory, calibration_directory, filelist, recipe_name, options, reduced_dark_folder=None, reduced_flat_folder=None, kmos_static_calib_directory='/Data/KCLASH/Data/Static_Cals/cal/'):
"""
Reduce a standard star observation.
Arguments:
destination_directory: directory the .sof file will be created in
calibration_directory: Locatin of the calibration files
filelist: List of raw files to reduce
recipe name: The name of the esorex recipe to call.
options: Command line options
reduced_dark_folder. Default is None. If we've already reduced darks, which folder are they in?
reduced_flat_folder. Default is None. If we've already reduced the flats, which folder are they in?
kmos_static_calib_directory: Location of *static* calibration files (e.g arc lines, atmospheric models, etc)
"""
#Check the destination directory exists, and make it if not
if not os.path.exists(os.path.abspath(destination_directory)):
os.makedirs(os.path.abspath(destination_directory))
# Generate exection command
cmds=[]
sof_name = SAM_create_sof(filelist, recipe_name, calibration_location=calibration_directory, location_for_sof=destination_directory)
log_file = "esorex_"+recipe_name+".log"
base_cmd = ['esorex', '--suppress-prefix=TRUE', '--log-file='+log_file, '--log-dir=.', recipe_name, sof_name]
cmds.append({'cmd': base_cmd, 'dir': destination_directory})
exec_commands(cmds, options.parallel)
def reduce_science(destination_directory, calibration_directory, telluric_directory, filelist, recipe_name, options, esorex_flags=None, log_file_name=None, reduced_dark_folder=None, reduced_flat_folder=None, kmos_static_calib_directory='/Data/KCLASH/Data/Static_Cals/cal/'):
"""
WRITE DOCS
"""
#Check the destination directory exists, and make it if not
if not os.path.exists(os.path.abspath(destination_directory)):
os.makedirs(os.path.abspath(destination_directory))
# Generate exection command
cmds=[]
sof_name = SAM_create_sof(filelist, recipe_name, calibration_location=calibration_directory, location_for_sof=destination_directory, telluric_directory=telluric_directory)
if log_file_name is None:
log_file = "esorex_"+recipe_name+".log"
else:
log_file=log_file_name
base_cmd = ['esorex', '--suppress-prefix=TRUE', '--log-file='+log_file, '--log-dir=.', recipe_name, sof_name]
if esorex_flags is not None:
for flag in esorex_flags:
base_cmd.insert(-1, flag)
cmds.append({'cmd': base_cmd, 'dir': destination_directory})
exec_commands(cmds, options.parallel)
def call_kmos_combine(destination_directory, filelist, options, esorex_flags=None, log_file_name=None):
"""
WRITE DOCS
"""
# Generate exection command
cmds=[]
recipe_name='kmos_combine'
sof_name = SAM_create_sof(filelist, recipe_name, location_for_sof=destination_directory)
if log_file_name is None:
log_file = "esorex_"+recipe_name+".log"
else:
log_file=log_file_name
log_file = "esorex_"+recipe_name+".log"
base_cmd = ['esorex', '--suppress-prefix=TRUE', '--log-file='+log_file, '--log-dir=.', recipe_name, sof_name]
if esorex_flags is not None:
for flag in esorex_flags:
base_cmd.insert(-1, flag)
cmds.append({'cmd': base_cmd, 'dir': destination_directory})
exec_commands(cmds, options.parallel)
def check_file_exists(fname):
#Check that these files exist:
if not os.path.isfile(fname):
raise NameError("{} doesn't seem to exist!".format(fname))
else:
return 0