Source code for checkQC.config


from pkg_resources import Requirement, resource_filename
import logging
from checkQC.exceptions import ConfigEntryMissing

import yaml


log = logging.getLogger(__name__)


[docs]class ConfigFactory(object): """ The ConfigFactory provides methods for creating a Config instance. """
[docs] @staticmethod def from_config_path(config_path): """ Creates a Config instance from the provided config_path :param config_path: path to the configuration, or None. If no config_path is provided the default config file will be used \ :returns: Config instance based on the specified file path """ config_file_content = ConfigFactory._get_config_file(config_path) return Config(config_file_content)
@staticmethod def _get_config_file(config_path): """ Load the content of the config file. If no config_path is specified, get the default of config file. :param config_path: path to the config file or None :returns: the content of the config file """ try: if not config_path: config_path = resource_filename(Requirement.parse('checkQC'), 'checkQC/default_config/config.yaml') log.info("No config file specified, using default config from {}.".format(config_path)) with open(config_path) as stream: return yaml.safe_load(stream) except FileNotFoundError as e: log.error("Could not find config file: {}".format(e)) raise e
[docs] @staticmethod def get_logging_config_dict(config_path): """ Loads the specified logger config. This is useful when CheckQC is used more like a library, so that the default logging configuration can be overridden. :param config_path: Path to the logging config. :returns: The content of the logging config file. """ try: if not config_path: config_path = resource_filename(Requirement.parse('checkQC'), 'checkQC/default_config/logger.yaml') log.info("No logging config file specified, using default config from {}.".format(config_path)) with open(config_path) as stream: return yaml.safe_load(stream) except FileNotFoundError as e: log.error("Could not find config file: {}".format(e)) raise e
[docs]class Config(object): """ A Config object wraps the configuration for all handlers, so that the correct config can be passed to a handler depending on e.g. which read length and run type has been used for a sequencing run. """ def __init__(self, config): """ Create a Config instance. :param config: content of the config file """ self._config = config def _get_matching_handler(self, instrument_and_reagent_type, read_length, use_closest_read_length=False): """ Get the handler matching the provided parameters. :param instrument_and_reagent_type: the instrument and run type, e.g. 'hiseq2500_rapidhighoutput_v4' :param read_length: either as a range, e.g. '50-70' or a single value, e.g. '50' :returns: A dict corresponding to the handler config :raises: ConfigEntryMissing if instrument, reagent type and read length detected is missing from config """ try: config_read_lengths = list(map(str, self._config[instrument_and_reagent_type].keys())) for config_read_length in config_read_lengths: if "-" in config_read_length: split_read_length = config_read_length.split("-") low_break = int(split_read_length[0]) high_break = int(split_read_length[1]) if low_break <= int(read_length) <= high_break: return self._config[instrument_and_reagent_type][config_read_length]["handlers"] else: if int(read_length) == int(config_read_length): return self._config[instrument_and_reagent_type][int(config_read_length)]["handlers"] if use_closest_read_length: closest_read_length = self._find_closest_read_length(config_read_lengths, read_length) log.info(f"Read length {read_length} not find in config. Using closest read length: {closest_read_length}") return self._config[instrument_and_reagent_type][closest_read_length]["handlers"] raise ConfigEntryMissing("Could not find a config entry matching read length '{}' on " "instrument '{}'. Please check the provided " "config.".format(read_length, instrument_and_reagent_type)) except KeyError: raise ConfigEntryMissing("Could not find a config entry for instrument '{}' " "with read length '{}'. Please check the provided config " "file ".format(instrument_and_reagent_type, read_length)) def _find_closest_read_length(self, config_read_lengths, read_length): """ Find the closest read length in the config :param config_read_lengths: dict with config read lengths for a specific intrument and reagent type :param read_length: either as a range, e.g. '50-70' or a single value, e.g. '50' :returns: the closest read length, as a string (if interval) or int (if single value) """ distance = {} for config_read_length in sorted(config_read_lengths, reverse=True): if "-" in config_read_length: split_read_length = config_read_length.split("-") distance[config_read_length] = min(abs(int(read_length) - int(split_read_length[0])), abs(int(read_length) - int(split_read_length[1]))) else: distance[config_read_length] = abs(int(read_length) - int(config_read_length)) closest_read_length = min(distance, key=distance.get) if "-" not in closest_read_length: return int(closest_read_length) else: return closest_read_length def _add_default_config(self, current_handler_config): """ Add the default handlers specified in the config. :param current_handler_config: a list of handlers. This will be mutated. :returns: The provided list with the default configs added to it. """ current_handler_names = set(map(lambda x: x["name"], current_handler_config)) default_handlers = self._config["default_handlers"] for default_handler in default_handlers: if not default_handler["name"] in current_handler_names: current_handler_config.append(default_handler) return current_handler_config def _downgrade_errors(self, current_handler_config, downgrade_errors_for): """ Downgrade errors to warnings for specific handlers :param current_handler_config: a list of handlers. :param downgrade_errors_for: a tuple of handlers for which errors should be downgraded :returns: The provided list where some handlers might have been modified """ downgraded_handler_config = [] for handler in current_handler_config: if handler["name"] in downgrade_errors_for and handler["error"] != "unknown": log.info("Downgrading errors for {}".format(handler["name"])) handler["warning"] = handler["error"] handler["error"] = "unknown" downgraded_handler_config.append(handler) else: downgraded_handler_config.append(handler) return downgraded_handler_config
[docs] def get_handler_configs(self, instrument_and_reagent_type, read_length, downgrade_errors_for=(), use_closest_read_length=False): """ Get the handler configurations for the specified parameters. :param instrument_and_reagent_type: type of instrument and reagents to match from config :param read_length: give the read length either as str or int :returns: the corresponding handler configuration(s) """ try: handler_config = self._get_matching_handler(instrument_and_reagent_type, read_length, use_closest_read_length) handler_config_with_defaults = self._add_default_config(handler_config) downgraded_handler_config_with_defaults = self._downgrade_errors( handler_config_with_defaults, downgrade_errors_for) return downgraded_handler_config_with_defaults except ConfigEntryMissing as e: log.error("Could not find a config entry for instrument '{}' " "with read length '{}'. Please check the provided config " "file ".format(instrument_and_reagent_type, read_length)) raise e
def __getitem__(self, key): return self._config[key]
[docs] def get(self, key, default=None): try: return self.__getitem__(key) except KeyError: return default