# Virus Health Check: a validation tool for HETDEX/VIRUS data
# Copyright (C) 2015, 2016, 2017, 2018, 2019 "The HETDEX collaboration"
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Driver implementations"""
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import configparser
import os
import json
from collections import defaultdict
from astropy.io import fits
import pyhetdex.doc.docstring as doc
import pyhetdex.tools.files.file_tools as ft
import pyhetdex.tools.processes as proc
from pyhetdex.tools import six_ext
from libvhc import factories
import libvhc.config as vhcconf
import libvhc.loggers as vhclog
import libvhc.html as vhchtml
import libvhc.utils as vhcutils
import libvhc.loaders as vhcload
import libvhc.reference_file_parser as vhcrfp
from . import exceptions
# Driver helper functionalities
DRIVER_FILE = "v_driver.txt"
"Name of the default driver file"
DRIVER_FILE_COUNTER = "v_driver_{0}.txt"
"Name of the driver file written when executing vhc"
[docs]@doc.format_docstring(DRIVER_FILE)
def get_drivers(path, recipe, conf, log, out_driver_file=None):
"""Load the drivers and try to read the driver names, get the default
ones for the ``recipe`` or use the list of drivers from the configuration
file
Parameters
----------
path : string
path where to look for the driver file
recipe : string
name of the recipe in use
conf : instance of :class:`pyhetdex.tools.configuration.ConfigParser`
configuration object
log : :class:`logging.Logger`
logger to use
out_driver_file : string, optional
if given, the drivers are written to this file.
Returns
-------
drivers : list of strings
name of the drivers to execute
drivers_dict : dictionary
map between driver names and callable implementing them
"""
# load the drivers
drivers_dict = vhcload.load_drivers(log)
# get the configuration entries about the drivers
config_drivers = conf.get_list(recipe, 'drivers', use_default=True)
try:
override_driver_file = conf.getboolean(recipe, 'override_driver_file')
except configparser.NoOptionError:
override_driver_file = False
config_drivers_msg = 'Using the drivers from the configuration file'
# if there are drivers in the configuration file and the
# override_driver_file is set to True, use the configuration file entries
if config_drivers and override_driver_file:
log.info(config_drivers_msg)
drivers = config_drivers
else:
try:
with open(driver_file(path)) as f:
drivers = [l.strip("\n").strip() for l in f]
except six_ext.FileOpenError:
# if no driver file is found and there are drivers in the
# configuration, use them, instead of creating a default driver
# file
log.warning("No driver file found")
if config_drivers:
log.info(config_drivers_msg)
drivers = config_drivers
else:
# Make the driver file
drivers = default_drivers(path, recipe, log)
if out_driver_file is not None:
try:
with open(out_driver_file, 'w') as f:
f.writelines(d + "\n" for d in drivers)
log.info("drivers written to file '%s'", out_driver_file)
except six_ext.FileOpenError as e:
log.warning("%s", e)
log.info('The following drivers will be used: %s', ', '.join(drivers))
return drivers, drivers_dict
[docs]def driver_file(path):
"""Return the name of the driver file
Parameters
----------
path : string
path where the recipe file leaves
Returns
-------
string
name of the driver file
"""
return os.path.join(path, DRIVER_FILE)
[docs]def default_drivers(path, recipe, log):
"""Get the default drivers.
Parameters
----------
path : string
path provided to the vhc executable
recipe : string
recipe name
Returns
-------
drivers : list of string
default drivers for the given recipe
"""
try:
drivers = vhcutils.default_drivers[recipe]
log.info("Use default drivers for recipe '%s'", recipe)
except KeyError:
log.critical("No default drivers found for recipe '%s'.", recipe)
drivers = []
return drivers
# Driver definitions
# driver function template
[docs]def driver_signature(vcheck, path, argv, *args, **kwargs): # pragma: no cover
"""Prototype of a function implementing a VHC driver.
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently executing
path : string
path provided to the ``vhc`` executable.
argv : list of strings
remaining of the command line
args : list
positional arguments. Not used when calling from ``vhc``
kwargs : dictionary
keyword arguments. Not used when calling from ``vhc``
"""
pass
# arc frames checks
def _check_n_peaks_cmd(vcheck, fname, log, conf):
"""Build the call to the check the number of ark peaks
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the number of fibers reference value(s) from a configuration file
ifuslot = vhcutils.ifuslot(fname, section=vcheck.recipe)
ifuid = vhcutils.ifuid(ifuslot)
specid = vhcutils.specid(ifuslot)
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
linefile, tolerance = vhcrfp.picker("arcs").get_value(specid, channel,
amplifier)
distfile = vhcrfp.picker("distortion").get_value(specid, channel,
amplifier)
_, error = vhcrfp.picker("n_fibers").get_value(ifuid, channel, amplifier)
cmd_str = "checkarc {headkey} {json} "
cmd_str += " -L {}".format(linefile, distfile)
cmd_str += " --peak_tolerance {} --maxmissing {}".format(tolerance,
error + 1)
cmd_str += " {fname}"
return cmd_str
find_peaks = factories.command_func_factory(_check_n_peaks_cmd)
"""Check that at least one fiber has the correct number of arc peaks and that
they look like peaks. The files with the lines position and the maximum shift
in peaks allowed come from the :class:`arcs reference file parser
<libvhc.reference_file_parser._Arcs>`; the reference distortion file comes from
:class:`distortion reference file parser
<libvhc.reference_file_parser._Distortion>`; the maximum number of missing
fibers comes from :class:`n_fibers reference file parser
<libvhc.reference_file_parser._NFibers>`."""
# bias frames checks
def _check_bias_cmd(vcheck, fname, log, conf):
"""Build the call for the bias comparison check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the bias reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
vals = vhcrfp.picker("bias").get_value(specid, channel, amplifier)
mean, stddev, dev_mean, dev_stddev, _, _, _, _ = vals
# build checkbias call
cmd_str = "checkbias {headkey} {json} -B"
# expected mean and standard deviation
cmd_str += " --bias_mean {} --bias_stddev {}".format(mean, stddev)
# maximum deviation from them
cmd_str += " --max_bias_dev_mean {}".format(dev_mean)
cmd_str += " --max_bias_dev_rms {}".format(dev_stddev)
cmd_str += " {fname}"
return cmd_str
compare = factories.command_func_factory(_check_bias_cmd)
"""Check the bias in the files. The expected value of the bias mean and
variance, as well as their maximum allowed fractional shift come from
:class:`bias reference file parser <libvhc.reference_file_parser._Bias>`. If
the fits files have the ``BIASTRGT`` header keyword, use its value instead of
the bias mean from the reference file."""
def _check_bias_flatness_cmd(vcheck, fname, log, conf):
"""Build the call for the bias flatness check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the bias reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
vals = vhcrfp.picker("bias").get_value(specid, channel, amplifier)
_, _, _, _, maxdiff, maxdiff_rms, _, _ = vals
# build checkbias call
cmd_str = "checkbias {headkey} {json} -F"
# maximum relative deviation of average and stddev
cmd_str += " --maxdiff {} --maxdiff_rms {}".format(maxdiff, maxdiff_rms)
cmd_str += " {fname}"
return cmd_str
flatbias = factories.command_func_factory(_check_bias_flatness_cmd)
"""Check the flatness of the bias in the files. The maximum fractional
deviations from flatness comes from :class:`bias reference file parser
<libvhc.reference_file_parser._Bias>`."""
def _check_line_bias_cmd(vcheck, fname, log, conf):
"""Build the call for the line based bias check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the overscan reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
vals = vhcrfp.picker("bias").get_value(specid, channel, amplifier)
_, _, _, _, _, _, stddev, _ = vals
# build the checkheader call
cmd_str = "checkbias {headkey} {json} -L"
# expected mean and standard deviation
cmd_str += " --bias_line_stddev {}".format(stddev)
cmd_str += " {fname}"
return cmd_str
line_bias = factories.command_func_factory(_check_line_bias_cmd)
"""Check the bias for line structure in the files. Calculates the clipped
mean of each line of the bias, and then the stddev of the resulting 1-D array.
The maximum allowed value of the stddev comes from :class:`bias reference
file parser <libvhc.reference_file_parser._Bias>`."""
def _check_fft_bias_cmd(vcheck, fname, log, conf):
"""Build the call for the fourier transform bias check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the overscan reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
vals = vhcrfp.picker("bias").get_value(specid, channel, amplifier)
_, _, _, _, _, _, _, signal = vals
# build the checkheader call
cmd_str = "checkbias {headkey} {json} -A"
# expected mean and standard deviation
cmd_str += " --max_wavefft {}".format(signal)
cmd_str += " {fname}"
return cmd_str
fft_bias = factories.command_func_factory(_check_fft_bias_cmd)
"""Check the bias for wave structure in the files. Checks for a signal
in the fast fourier transform. The maximum allowed signal comes
from :class:`bias reference file
parser <libvhc.reference_file_parser._Bias>`."""
# common checks
def _get_exps(recipe, path, argv, conf):
'''Check that the number of exposures is correct.
Parameters
----------
recipe : string
name of the recipe
path : string
path provided to the ``vhc`` executable.
argv : list of strings
remaining of the command line
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
success : bool
whether the test is successful
exposures : list
list of exposure directories
expected_exposures : int
number of expected exposures
'''
if not argv:
expected_exposures = conf.getint(recipe, 'n_exposures')
exposure_dir = conf.get('common', 'exposure_dir')
exposures = list(ft.scan_dirs(path, matches='*' + exposure_dir,
recursive=False))
success = (expected_exposures == len(exposures))
else:
expected_exposures = 1
exposures = [os.path.join(path, argv[0]), ]
success = os.path.isdir(exposures[0])
if not success:
exposures = []
return success, exposures, expected_exposures
def _edit_header(fname, conf, header_key, result, comment):
'''Open the fits file in update mode and add the ``VHCCHECK`` and
``keyword`` header keywords
Parameters
----------
fname : string
name of the file to update
conf : instance of :class:`pyhetdex.tools.configuration.ConfigParser`
configuration object
header_key, result, comment : strings
name, value and comment of the header keyword
'''
vhc_check = "VHCCHECK"
with fits.open(fname, memmap=False, mode='update') as hdu:
header = hdu[0].header
header[header_key] = (result, comment)
header[vhc_check] = ('', "The VHC keys must be one of NOTRUN,"
" PASSED or FAILED.")
extra_value = vhcutils.extra_value_with_conf_revision(conf)
header[vhcutils.EXTRA_KEY] = (extra_value, vhcutils.EXTRA_COMMENT)
def _write_header_keywords(path, fits_files, conf, header_key, results):
'''Write the header keywords ``header_keyword`` with the overall result and the
``VHCCHECK`` one to mark the execution of VCH.
Parameters
----------
path : string
path provided to the ``vhc`` executable.
fits_files : list
files name where to write the header keywords
conf : instance of :class:`pyhetdex.tools.configuration.ConfigParser`
configuration object
header_key : string
name of the header keyword to be written
results : list of boolean
results of the tests
'''
key_value = 'PASSED' if all(results) else 'FAILED'
comment = '{} out of {} checks passed'.format(sum(results), len(results))
# write the keyword to all the fits files
worker = proc.get_worker(name=path)
for fname in fits_files:
worker(_edit_header, fname, conf, header_key, key_value, comment)
worker.get_results()
[docs]def n_files(vcheck, path, argv):
"""Check the number of exposures and, for each exposure, the number of
files.
If ``argv`` is an empty list, the number of exposures is taken from the
'n_exposures' option of the given recipe and compared with the number of
exposure directories found. Otherwise the existence of ``path
+ argv[0]`` is checked. The pattern for the exposure directory is taken
from the ``exposure_dir`` option in the ``[common]`` section.
Afterwards, all the fits files are searched for and compared with the
expected number: :data:`~libvhc.factories.FILES_IFU` * number of IFUS in
the fplane file.
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently executing
path : string
path provided to the ``vhc`` executable.
argv : list of strings
remaining of the command line
"""
resultdict = {'pathname': path,
'testname': 'n_files',
'checks': [],
'result': 'PASSED',
'error': ''}
log = vhclog.getLogger(name=path)
conf = vhcconf.get_config()
success, exposures, expected_exposures = _get_exps(vcheck.recipe, path,
argv, conf)
tdict = {'subtestname': 'n_exp',
'expected': expected_exposures,
'tolerance': 0,
'actual': len(exposures),
'result': 'PASSED' if success else 'FAILED'}
resultdict['checks'].append(tdict)
global_success = [success, ]
msg = ('Found "{}" exposures; "{}"'
'expected'.format(len(exposures), expected_exposures))
if success:
log.info(msg)
else:
log.error(msg)
vhchtml.add_fplane_test(vcheck, msg, success)
# Get the wildcard for the files expected given the recipe
rec_wildcard = vhcutils.recipe_match[vcheck.recipe]
n_ifus = len(vhcutils.get_fplane().ifus)
expected_files = factories.FILES_IFU * n_ifus
fits_files = []
for c, exp in enumerate(exposures, 1):
files = list(ft.scan_files(exp, matches=rec_wildcard))
fits_files.extend(files)
n_files = len(files)
success = (n_files == expected_files)
global_success.append(success)
msg = ('Found "{}" files in exposure "{}; "{}"'
'expected'.format(n_files, os.path.relpath(exp, path),
expected_files))
tdict = {'subtestname': 'n_files_exp%02d' % c,
'expected': expected_files,
'tolerance': 0,
'actual': n_files,
'result': 'PASSED' if success else 'FAILED'}
resultdict['checks'].append(tdict)
if success:
log.info(msg)
else:
log.error(msg)
vhchtml.add_fplane_test(vcheck, msg, success)
if conf.getboolean("headerkeys", "write_vhc_header_key"):
_write_header_keywords(path, fits_files, conf, 'VHCNFILE',
global_success)
if all(global_success):
resultdict['result'] = 'PASSED'
else:
resultdict['result'] = 'FAILED'
resultdict['error'] = 'failed %d out of %d n_files tests' \
% (global_success.count(False),
len(global_success))
log.info(vhclog.JSON_KEY + json.dumps(resultdict))
cmd_str = "checkheader {headkey} {json} -N -r {recipe} {fname}"
n_pixels = factories.command_line_factory(cmd_str)
"Check the number of pixels in the files"
cmd_str = "checkheader {headkey} {json} -E -r {recipe} {fname}"
exptime = factories.command_line_factory(cmd_str)
"Check the exposure time in the files"
def _check_overscan_cmd(vcheck, fname, log, conf):
"""Build the call for the overscan check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the overscan reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
try:
vals = vhcrfp.picker("overscan").get_value(specid, channel, amplifier)
mean, stddev, dev_mean, dev_stddev, _ = vals
except exceptions.VHCReferenceKeyError:
vals = vhcrfp.picker("bias").get_value(specid, channel, amplifier)
mean, stddev, dev_mean, dev_stddev, _, _, _, _ = vals
# build the checkheader call
cmd_str = "checkheader {headkey} {json} -O -r {recipe}"
# expected mean and standard deviation
cmd_str += " --ovc_mean {} --ovc_stddev {}".format(mean, stddev)
# maximum deviation from them
cmd_str += " --max_ovc_dev_mean {} --max_ovc_dev_rms {}".format(dev_mean,
dev_stddev)
cmd_str += " {fname}"
return cmd_str
check_overscan = factories.command_func_factory(_check_overscan_cmd)
"""Check the overscan in the files. The expected value of the overscan mean and
variance, as well as their maximum allowed fractional shift come from
:class:`overscan reference file parser
<libvhc.reference_file_parser._Overscan>`. If the combination of spectrograph
ID, channel and amplifier is not found on the file falls back to the
:class:`bias reference file parser <libvhc.reference_file_parser._Bias>`. If
the fits files have the ``BIASTRGT`` header keyword, use its value instead of
the overscan mean from the reference file."""
def _check_line_overscan_cmd(vcheck, fname, log, conf):
"""Build the call for the line based overscan check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the overscan reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
try:
vals = vhcrfp.picker("overscan").get_value(specid, channel, amplifier)
_, _, _, _, stddev = vals
except exceptions.VHCReferenceKeyError:
vals = vhcrfp.picker("bias").get_value(specid, channel, amplifier)
_, _, _, _, _, _, stddev, _ = vals
# build the checkheader call
cmd_str = "checkheader {headkey} {json} -L -r {recipe}"
# expected mean and standard deviation
cmd_str += " --ovsc_line_stddev {}".format(stddev)
cmd_str += " {fname}"
return cmd_str
check_line_overscan = factories.command_func_factory(_check_line_overscan_cmd)
"""Check the overscan in the files. Calculates the clipped mean of each line
of the overscan, and then the stddev of the resulting 1-D array. The maximum
allowed value of the stddev comes from :class:`overscan reference
file parser <libvhc.reference_file_parser._Overscan>`."""
def _check_dettemp_cmd(vcheck, fname, log, conf):
"""Build the call for the dettemp check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the overscan reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
vals = vhcrfp.picker("dettemp").get_value(specid, channel, amplifier)
max_dettemp, delta_setpoint = vals
# build the checkheader call
cmd_str = 'checkheader {headkey} {json} --dettemp -r {recipe}'
# maximum detector temperature and delta from setpoint
cmd_str += ' --max_dettemp {}'.format(max_dettemp)
cmd_str += ' --delta_setpoint {}'.format(delta_setpoint)
cmd_str += " {fname}"
return cmd_str
check_dettemp = factories.command_func_factory(_check_dettemp_cmd)
"""Check the detector temperature in the files. The maximum allowed temperature
and the difference between the set and the measured temperature come from
:class:`dettemp reference file parser
<libvhc.reference_file_parser._Dettemp>`"""
cmd_str = "checkheader {headkey} {json} --keywords -r {recipe} {fname}"
check_headerkeys = factories.command_line_factory(cmd_str)
"Check the required header keywords in the files"
cmd_str = "checkheader {headkey} {json} --bitstat -r {recipe} {fname}"
check_bitstat = factories.command_line_factory(cmd_str)
"Check for stuck bits in the files"
def _check_nullpixel_cmd(vcheck, fname, log, conf):
"""Build the call for the dettemp check
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the overscan reference value(s) from a configuration file
specid = vhcutils.specid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
max_null_pixels = vhcrfp.picker("nullpixel").get_value(specid, channel,
amplifier)
# build the checkheader call
cmd_str = 'checkheader {headkey} {json} --nullpix -r {recipe}'
# maximum detector temperature and delta from setpoint
cmd_str += ' --max_nullpix {}'.format(max_null_pixels)
cmd_str += " {fname}"
return cmd_str
check_nullpix = factories.command_func_factory(_check_nullpixel_cmd)
"""Check the number of null in the files. The maximum allowed number of null
pixels comes from :class:`nullpixel reference file parser
<libvhc.reference_file_parser._Nullpixel>`"""
cmd_str = "checkheader {headkey} {json} --repeat -r {recipe} {fname}"
check_repeat = factories.command_line_factory(cmd_str)
"Check that the ``REPEAT`` header keyword, if found, is not true."
def _saturation_cmd(vcheck, fname, log, conf):
"""Build the call to the check the saturation level
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the number of fibers reference value(s) from a configuration file
ifuid = vhcutils.ifuid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
saturation, max_n = vhcrfp.picker("saturation").get_value(ifuid, channel,
amplifier)
cmd_str = "checkflat {headkey} {json} "
cmd_str += " --saturation --saturation_level {}".format(saturation)
cmd_str += " --max_sat_pixels {}".format(max_n)
cmd_str += " {fname}"
return cmd_str
saturation = factories.command_func_factory(_saturation_cmd)
"""Check the saturation is below a given threshold. The saturation level and
the maximum allowed number of pixels above this level come from
:class:`saturation reference file parser
<libvhc.reference_file_parser._Saturation>`"""
# flat frames checks
def _check_n_fibers_cmd(vcheck, fname, log, conf):
"""Build the call to the check the number of fibers
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the number of fibers reference value(s) from a configuration file
ifuid = vhcutils.ifuid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
nfibers, error = vhcrfp.picker("n_fibers").get_value(ifuid, channel,
amplifier)
cmd_str = "checkflat {headkey} {json} "
cmd_str += " -F -f {} --maxmissing {}".format(nfibers, error)
cmd_str += " {fname}"
return cmd_str
n_fibers = factories.command_func_factory(_check_n_fibers_cmd)
"""Check the number of fibers in the files. The expected number of fibers and
the maximum allowed deviation come from :class:`n_fibers reference file parser
<libvhc.reference_file_parser._NFibers>`"""
def _check_min_flux_cmd(vcheck, fname, log, conf):
"""Build the call to the check that the fibers have enough flux
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the number of fibers reference value(s) from a configuration file
ifuid = vhcutils.ifuid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
nfibers, error = vhcrfp.picker("n_fibers").get_value(ifuid, channel,
amplifier)
min_flux = vhcrfp.picker("min_flux").get_value(ifuid, channel, amplifier)
cmd_str = "checkflat {headkey} {json} --flux "
cmd_str += " --minflux {} -f {} --maxmissing {}".format(min_flux, nfibers,
error)
cmd_str += " {fname}"
return cmd_str
min_flux = factories.command_func_factory(_check_min_flux_cmd)
"""Check the flux in the flat files is above a given threshold for all the
fibers. The threshold value comes from :class:`min_flux reference file parser
<libvhc.reference_file_parser._MinFlux>`, while the expected number of fibers
and the maximum allowed number of fibers below the threashold come from
:class:`n_fibers reference file parser
<libvhc.reference_file_parser._NFibers>`"""
def _check_row_cte_cmd(vcheck, fname, log, conf):
"""Build the call to the check for row shift cte issues
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the histogram limit reference value(s) from a configuration file
ifuid = vhcutils.ifuid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
hist_cte = vhcrfp.picker("hist_cte").get_value(ifuid, channel,
amplifier)
cmd_str = "checkcte {headkey} {json} "
cmd_str += "--n_hist {}".format(hist_cte)
cmd_str += " {fname}"
return cmd_str
row_cte = factories.command_func_factory(_check_row_cte_cmd)
"""Check the images for row shift problems. This is done by running ``yoda``
and checking the distribution of the position angles of the detected sources.
This check only makes sense running on science frames, since it relies on
cosmics in the images. It flags the images if the count of the 90 degree bins
of the position angle histogram exceed n_hist times the mean of the remaining
histogram. The n_hist value comes from :class:`n_hist reference file parser
<libvhc.reference_file_parser._nHist>`. If ``yoda`` is not found, the
driver is marked as ``NOTRUN``."""
cmd_str = "checkflat {headkey} {json} -C {fname}"
col_cte = factories.command_line_factory(cmd_str)
"Check for column shift cte issues in the files"
# hetdex_dithers checks
def _n_dithers_exp_dirs(vcheck, log, path, argv, conf):
'''Check the number of exposure directories
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently executing
log : instance of :class:`logging.LoggerAdapter`
logger
path : string
path provided to the ``vhc`` executable.
argv : list of strings
remaining of the command line
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
success : bool
whether the test is successful
exposures : list
list of exposure directories
expected_exposures : int
number of expected exposures
'''
# check that there are the expected number of exposures
success, exposures, expected_exposures = _get_exps(vcheck.recipe, path,
argv, conf)
msg = ('Found "{}" exposure directories; "{}"'
' expected'.format(len(exposures), expected_exposures))
if success:
log.info(msg)
else:
log.error(msg)
vhchtml.add_fplane_test(vcheck, msg, success)
return success, exposures, expected_exposures
def _n_dithers_dither_key(vcheck, log, exposures, path):
'''Check that for each exposure directory all the ``DITHER`` keywords have
the same value and log it.
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently executing
log : instance of :class:`logging.LoggerAdapter`
logger
exposures : list
list of exposure directories
path : string
path provided to the ``vhc`` executable.
Returns
-------
success : bool
whether the test is successful
dithers : list
dither numbers
fits_files : list
fits file names
'''
# Get the wildcard for the files expected given the recipe
rec_wildcard = vhcutils.recipe_match[vcheck.recipe]
# default success and message
success = True
msg = ''
dithers = set()
fits_files = []
for exp in exposures:
files = list(ft.scan_files(exp, matches=rec_wildcard))
fits_files.extend(files)
ditherdict = defaultdict(int)
for fname in files:
hh = fits.getheader(fname)
ditherdict[hh['DITHER']] += 1
if len(ditherdict) != 1:
# Found more than one dither value in one exposure directory
success = False
msg += ('Found "{}" dither numbers in exposure "{};'
' one expected. '.format(list(ditherdict.keys()),
os.path.relpath(exp, path)))
dithers.update(ditherdict) # save the dither numbers
if not msg:
msg = ('All the files in each exposure have a unique value of the'
' ``DITHER`` keyword')
if success:
log.info(msg)
else:
log.error(msg)
vhchtml.add_fplane_test(vcheck, msg, success)
return success, list(dithers), fits_files
def _n_dithers_n_dither_key(vcheck, log, dithers, expected_exposures):
'''Check that the number of ``DITHER`` keyword values is correct
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently executing
log : instance of :class:`logging.LoggerAdapter`
logger
dithers : list
dither numbers
expected_exposures : int
number of expected exposures
Returns
-------
success : bool
whether the test is successful
'''
success = True
msg = 'The values of the ``DITHER`` header keywords ({}) is correct'
msg = msg.format(dithers)
if expected_exposures == 1:
if len(dithers) != 1:
success = False
msg = 'Only one ``DITHER`` keyword values expected not {}'
msg = msg.format(dithers)
else:
expected_exposures = list(range(1, expected_exposures + 1))
if sorted(dithers) != expected_exposures:
success = False
msg = ('The expected values of the ``DITHER`` header keyword are'
' {} but {} have found'.format(expected_exposures,
dithers))
if success:
log.info(msg)
else:
log.error(msg)
vhchtml.add_fplane_test(vcheck, msg, success)
return success
[docs]def n_dithers(vcheck, path, argv):
"""Check the number of dithers for a shot, with the following criteria:
* there must be the correct number of exposure directories;
* all the files in a directory must have the same value of the ``DITHER``
header keyword;
* if dithering is performed using the dithering mechanism, in total there
must be as many values of ``DITHER`` as there are dither, otherwise the
values of ``DITHER`` must be the same across all the exposures; the
``dither_with_probe`` option of the ``[hetdex_dithers]`` configuration
section instructs the driver whether the dithering is done offsetting the
fiducial position in the drive probes (``true``) or using the dithering
mechanism (``false``).
Each of the three steps is logged independently
If ``argv`` is an empty list, the number of exposures is taken from the
'n_exposures' option of the given recipe and compared with the number of
exposure directories found. Otherwise the existence of ``path
+ argv[0]`` is checked. The pattern for the exposure directory is taken
from the ``exposure_dir`` option in the ``[common]`` section.
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently executing
path : string
path provided to the ``vhc`` executable.
argv : list of strings
remaining of the command line
"""
resultdict = {'pathname': path,
'testname': 'n_dithers',
'checks': [],
'result': 'PASSED',
'error': ''}
log = vhclog.getLogger(name=path)
conf = vhcconf.get_config()
global_success = []
# check that there are the expected number of exposures
_exp_check_out = _n_dithers_exp_dirs(vcheck, log, path, argv, conf)
success, exposures, expected_exposures = _exp_check_out
global_success.append(success)
tdict = {'subtestname': 'n_dithers_nexp',
'expected': expected_exposures,
'tolerance': 0,
'actual': len(exposures),
'result': 'PASSED' if success else 'FAILED'}
resultdict['checks'].append(tdict)
# check the DITHER header keyword
success, dithers, fits_files = _n_dithers_dither_key(vcheck, log,
exposures, path)
global_success.append(success)
tdict = {'subtestname': 'n_dithers_keyword',
'expected': len(exposures),
'tolerance': 0,
'actual': len(dithers),
'result': 'PASSED' if success else 'FAILED'}
resultdict['checks'].append(tdict)
# check that the number of ``DITHER`` header keywords is consistent with
# expectation
dither_with_probe = conf.getboolean('hetdex_dithers', 'dither_with_probe')
n_dither_keys = 1 if dither_with_probe else expected_exposures
success = _n_dithers_n_dither_key(vcheck, log, dithers, n_dither_keys)
global_success.append(success)
tdict = {'subtestname': 'n_dithers_probe',
'expected': n_dither_keys,
'tolerance': 0,
'actual': len(dithers),
'result': 'PASSED' if success else 'FAILED'}
resultdict['checks'].append(tdict)
if conf.getboolean("headerkeys", "write_vhc_header_key"):
_write_header_keywords(path, fits_files, conf, 'VHCNDITH',
global_success)
if all(global_success):
resultdict['result'] = 'PASSED'
else:
resultdict['result'] = 'FAILED'
resultdict['error'] = 'failed %d out of %d n_dither tests' \
% (global_success.count(False),
len(global_success))
log.info(vhclog.JSON_KEY + json.dumps(resultdict))
def _check_sky_level_cmd(vcheck, fname, log, conf):
"""Build the call to the check that the there is sky in the images
Parameters
----------
vcheck : instance of :class:`~libvhc.VCheck`
store the recipe name and the check currently
executing
fname : string
name of the file to check
log : instance of :class:`logging.LoggerAdapter`
logger
conf : instance of :class:`configparser.ConfigParser`
configuration object
Returns
-------
cmd : string
string of the command to execute
"""
# get the number of fibers reference value(s) from a configuration file
ifuid = vhcutils.ifuid(vhcutils.ifuslot(fname, section=vcheck.recipe))
channel = vhcutils.channel(fname, section=vcheck.recipe)
amplifier = vhcutils.amplifier(fname, section=vcheck.recipe)
sky_level = vhcrfp.picker("sky_level").get_value(ifuid, channel,
amplifier)
cmd_str = "checkhetdex {headkey} {json} --sky "
cmd_str += " --sky_level {} ".format(sky_level)
cmd_str += " {fname}"
return cmd_str
sky_level = factories.command_func_factory(_check_sky_level_cmd)
"""Check that some sky is found in the fibers. This is achieved by comparing
the average value of a central region of the chip against the overscan.
The minimum expected skylevel above overscan is read from
:class:`sky_level reference file parser
<libvhc.reference_file_parser._SkyLevel>`"""