# Licensed under a 3-clause BSD style license - see LICENSE.rst
from collections import OrderedDict
import os
from astropy import log
import pandas
from sofia_redux.instruments import flitecam as fdrp
from sofia_redux.toolkit.utilities.darus import get_file_from_darus
from sofia_redux.toolkit.utilities.func import goodfile
__all__ = ['getcalpath']
# DaRUS dataset holding FLITECAM reference files.
# https://darus.uni-stuttgart.de/dataset.xhtml?persistentId=doi:10.18419/DARUS-6126
DARUS_DOI = '10.18419/DARUS-6126'
[docs]
def getcalpath(header):
"""
Return the path of the ancillary files used for the pipeline.
Looks up default calibration files and other reference data in
the calibration data path, based on the characteristics of the
observation, as recorded in the input header.
The output format is designed to be compatible with FORCAST
configurations.
Parameters
----------
header : astropy.io.fits.Header
Returns
-------
collections.OrderedDict
name : str
Name of the config mode.
gmode : int
The grism mode to be used during the reduction.
cnmode : int
The chop/nod mode to be used during the reduction.
dateobs : int
The observation date, in YYYYMMDD format.
obstype : str
Observation type.
srctype : str
Source type.
spectel : str
Name of the spectral element used for the observation.
slit : str
Name of the slit used for the observation.
pathcal : str
Base file path for calibration data.
kwfile : str
File path to the keyword definition file.
linfile : str
File path to the linearity correction file.
maskfile : str
File path to the grism order mask file.
wavefile : str
File path to the grism wavelength calibration file.
resolution : float
Spectral resolution for the grism mode
respfile : str
File path to the grism instrument response file.
slitfile : float
File path to the grism slit function file.
linefile : str
File path to a line list for wavelength calibration.
wmin : int
Approximate minimum wavelength for the instrument. Used
to identify matching ATRAN files.
wmax : int
Approximate maximum wavelength for the instrument. Used
to identify matching ATRAN files.
error : bool
True if there was an error retrieving the files.
"""
path = os.path.join(os.path.dirname(fdrp.__file__), 'data')
result = OrderedDict(
(('name', ''), ('gmode', -1), ('cnmode', ''), ('dateobs', 99999999),
('obstype', ''), ('srctype', ''), ('spectel', ''), ('slit', ''),
('pathcal', ''), ('kwfile', ''), ('linfile', ''),
('maskfile', ''), ('wavefile', ''),
('respfile', ''), ('slitfile', ''), ('linefile', ''),
('waveshift', 0.), ('resolution', 0.),
('wmin', 1), ('wmax', 6),
('error', False)))
result['spectel'] = header.get('SPECTEL1', '').upper().strip()
result['slit'] = header.get('SPECTEL2', '').upper().strip()
date = header.get('DATE-OBS', '').replace('T', ' ').replace('-', ' ')
date = date.split()
dateobs = 99999999
if len(date) >= 3:
try:
dateobs = int(''.join(date[0:3]))
except ValueError:
pass
result['dateobs'] = dateobs
instcfg = header.get('INSTCFG', 'UNKNOWN').strip().upper()
if instcfg in ['SPECTROSCOPY', 'GRISM']:
name = 'GRI'
gmode = 1
else:
name = 'IMA'
gmode = -1
result['name'] = name
result['gmode'] = gmode
# also read and store the sky mode, obstype, and srctype
# from the header
result['cnmode'] = header.get('INSTMODE', 'UNKNOWN').strip().upper()
result['obstype'] = header.get('OBSTYPE', 'OBJECT').strip().upper()
result['srctype'] = header.get('SRCTYPE', 'UNKNOWN').strip().upper()
# read the top-level default file
result['pathcal'] = os.path.join(path, '')
calfile_default = os.path.join(path, 'caldefault.txt')
if not os.path.isfile(calfile_default):
msg = f"Problem reading default file " \
f"{calfile_default}"
log.warning(msg)
result['error'] = True
return result
calcols = ['name', 'kwfile', 'linfile']
df = pandas.read_csv(calfile_default, sep=r'\s+',
comment='#', names=calcols, index_col=0)
table = df[(df.index >= dateobs) & (df['name'] == result['name'])]
if len(table) == 0:
msg = f"Problem reading defaults for " \
f"{result['name']}on date {dateobs} from {calfile_default}"
log.warning(msg)
result['error'] = True
return result
# take the first applicable date
row = table[table.index == table.index.min()].sort_index().iloc[0]
for f in calcols:
if f.endswith('file') and f in row and row[f] != '.':
# get it from the local data directory if placed there manually
localdatafile = os.path.join(path, row[f])
if goodfile(localdatafile):
result[f] = localdatafile
log.info(f"Using {row[f]} from {path}")
else:
# for public distributions, it may need
# to be downloaded from DaRUS
result[f] = _download_cache_file(row[f])
# Read additional grism defaults into result
if result['gmode'] > 0:
_get_grism_cal(path, result)
return result
def _get_grism_cal(pathcal, result):
"""
Return the path of the ancillary files used for the grism pipeline.
Looks up additional default calibration files and other reference data
needed for grism observations. The input parameters are assumed to be
passed from the getcalpath function: they should already be formatted
as necessary.
Parameters
----------
header : astropy.io.fits.Header
pathcal : str
Absolute file path to calibration data directory that contains
the default calibration files.
result : collections.OrderedDict
The configuration structure to update with grism values.
"""
spectel = result['spectel']
slit = result['slit']
dateobs = result['dateobs']
pathcal = os.path.join(pathcal, 'grism')
calfile_default = os.path.join(pathcal, 'caldefault.txt')
if not os.path.isfile(calfile_default):
msg = f"Problem reading default file {calfile_default}."
log.warning(msg)
result['error'] = True
return
calcols = ['spectel', 'slit', 'maskfile', 'wavefile',
'respfile', 'linefile', 'waveshift', 'resolution']
df = pandas.read_csv(calfile_default, sep=r'\s+',
comment='#', names=calcols, index_col=0)
table = df[(df.index >= dateobs)
& (df['spectel'] == spectel)
& (df['slit'] == slit)]
if len(table) == 0:
msg = 'getcalpath - Problem reading defaults ' \
'for %s, %s ' % (spectel, slit)
msg += 'on date %s from %s' % (dateobs, calfile_default)
log.warning(msg + '\nRoutine failed. Returning default dict')
result['error'] = True
return
# take the first applicable date
row = table[table.index == table.index.min()].sort_index().iloc[0]
for f in calcols:
if f.endswith('file') and f in row and row[f] != '.':
# for source distributions, the file should be in
# the standard calpath
localdatafile = os.path.join(pathcal, row[f])
if goodfile(localdatafile):
result[f] = localdatafile
log.info(f"Using {row[f]} from {pathcal}")
else:
# for public distributions, it may need
# to be downloaded from DaRUS
result[f] = _download_cache_file(row[f])
elif f in ['waveshift', 'resolution'] and f in row and row[f] != '.':
try:
result[f] = float(row[f])
except ValueError:
msg = f'Problem reading {f} for {spectel}, {slit} ' \
f'on date {dateobs} from {calfile_default}'
log.warning(msg)
result['error'] = True
def _download_cache_file(filename):
basename = os.path.basename(filename)
try:
cache_file = get_file_from_darus(DARUS_DOI, basename)
log.info(f"Using {basename} cached as {cache_file}")
except OSError:
# OSError covers the expected exceptions: HTTPError if the DaRUS
# request fails, FileNotFoundError if the file is not in the dataset,
# and download_file failures. Return basename only if the file
# can't be downloaded; pipeline will issue clearer errors later.
cache_file = basename
log.warning(f'File {cache_file} could not be downloaded from DaRUS '
f'dataset {DARUS_DOI}')
return cache_file
