<commit_before>"""Retrieve run information describing files to process in a pipeline.

This handles two methods of getting processing information: from a Galaxy
next gen LIMS system or an on-file YAML configuration.
"""
import collections
from contextlib import closing
import copy
import glob
import itertools
import os
import string

import toolz as tz
import yaml

from bcbio import install, utils
from bcbio.bam import ref
from bcbio.log import logger
from bcbio.distributed import objectstore
from bcbio.illumina import flowcell
from bcbio.pipeline import alignment, config_utils, genome
from bcbio.pipeline import datadict as dd
from bcbio.provenance import diagnostics, programs, versioncheck
from bcbio.provenance import data as provenancedata
from bcbio.variation import effects, genotype, population, joint, vcfutils
from bcbio.variation.cortex import get_sample_name
from bcbio.bam.fastq import open_fastq

ALLOWED_CONTIG_NAME_CHARS = set(list(string.digits) + list(string.ascii_letters) + ["-", "_", "*", ":", "."])
ALGORITHM_NOPATH_KEYS = ["variantcaller", "realign", "recalibrate",
                         "phasing", "svcaller", "hetcaller", "jointcaller", "tools_off", "mixup_check"]

def organize(dirs, config, run_info_yaml, sample_names=None, add_provenance=True,
             integrations=None):
    """Organize run information from a passed YAML file or the Galaxy API.

    Creates the high level structure used for subsequent processing.

    sample_names is a list of samples to include from the overall file, for cases
    where we are running multiple pipelines from the same configuration file.
    """
    if integrations is None: integrations = {}
    logger.info("Using input YAML configuration: %s" % run_info_yaml)
    assert run_info_yaml and os.path.exists(run_info_yaml), \
        "Did not find input sample YAML file: %s" % run_info_yaml
    run_details = _run_info_from_yaml(dirs, run_info_yaml, config, sample_names)
    remote_retriever = None
    for iname, retriever in integrations.iteritems():
        if iname in config:
            run_details = retriever.add_remotes(run_details, config[iname])
            remote_retriever = retriever
    out = []
    for item in run_details:
        item["dirs"] = dirs
        if "name" not in item:
            item["name"] = ["", item["description"]]
        elif isinstance(item["name"], basestring):
            description = "%s-%s" % (item["name"], clean_name(item["description"]))
            item["name"] = [item["name"], description]
            item["description"] = description
        # add algorithm details to configuration, avoid double specification
        item["resources"] = _add_remote_resources(item["resources"])
        item["config"] = config_utils.update_w_custom(config, item)
        item.pop("algorithm", None)
        item = add_reference_resources(item, remote_retriever)
        # Create temporary directories and make absolute, expanding environmental variables
        tmp_dir = tz.get_in(["config", "resources", "tmp", "dir"], item)
        if tmp_dir:
            # if no environmental variables, make and normalize the directory
            # otherwise we normalize later in distributed.transaction:
            if os.path.expandvars(tmp_dir) == tmp_dir:
                tmp_dir = utils.safe_makedir(os.path.expandvars(tmp_dir))
                tmp_dir = genome.abs_file_paths(tmp_dir)
            item["config"]["resources"]["tmp"]["dir"] = tmp_dir
        out.append(item)
    out = _add_provenance(out, dirs, config, add_provenance)
    return out

def normalize_world(data):
    """Normalize a data object, useful after serializetion via CWL.
    """
    data = _normalize_files(data)
    return data

def _add_provenance(items, dirs, config, add_provenance=True):
    if add_provenance:
        p = programs.write_versions(dirs, config=config)
        d = provenancedata.write_versions(dirs, items)
        versioncheck.testall(items)
        p_db = diagnostics.initialize(dirs)
    out = []
    for item in items:
        if add_provenance:
            entity_id = diagnostics.store_entity(item)
            item["config"]["resources"]["program_versions"] = p
            item["provenance"] = {"programs": p, "entity": entity_id,
                                  "db": p_db, "data": d}
        out.append([item])
    return out

def setup_directories(work_dir, fc_dir, config, config_file):
    fastq_dir, galaxy_dir, config_dir = _get_full_paths(flowcell.get_fastq_dir(fc_dir)
                                                        if fc_dir else None,
                                                        config, config_file)
    # check default install for tool data if not found locally
    if not os.path.exists(os.path.join(galaxy_dir, "tool-data")):
        _, config_file = config_utils.load_system_config(work_dir=work_dir)
        if os.path.exists(os.path.join(os.path.dirname(config_file), "tool-data")):
            galaxy_dir = os.path.dirname(config_file)
    return {"fastq": fastq_dir, "galaxy": galaxy_dir,
            "work": work_dir, "flowcell": fc_dir, "config": config_dir}

def _get_full_paths(fastq_dir, config, config_file):
    """Retrieve full paths for directories in the case of relative locations.
    """
    if fastq_dir:
        fastq_dir = utils.add_full_path(fastq_dir)
    config_dir = utils.add_full_path(os.path.dirname(config_file))
    galaxy_config_file = utils.add_full_path(config.get("galaxy_config", "universe_wsgi.ini"),
                                             config_dir)
    return fastq_dir, os.path.dirname(galaxy_config_file), config_dir

# ## Remote resources

def _add_remote_resources(resources):
    """Retrieve remote resources like GATK/MuTect jars present in S3.
    """
    out = copy.deepcopy(resources)
    for prog, info in resources.iteritems():
        for key, val in info.iteritems():
            if key == "jar" and objectstore.is_remote(val):
                store_dir = utils.safe_makedir(os.path.join(os.getcwd(), "inputs", "jars", prog))
                fname = objectstore.download(val, store_dir, store_dir)
                version_file = os.path.join(store_dir, "version.txt")
                if not utils.file_exists(version_file):
                    version = install.get_gatk_jar_version(prog, fname)
                    with open(version_file, "w") as out_handle:
                        out_handle.write(version)
                else:
                    with open(version_file) as in_handle:
                        version = in_handle.read().strip()
                del out[prog][key]
                out[prog]["dir"] = store_dir
                out[prog]["version"] = version
    return out

# ## Genome reference information

def add_reference_resources(data, remote_retriever=None):
    """Add genome reference information to the item to process.
    """
    aligner = data["config"]["algorithm"].get("aligner", None)
    if remote_retriever:
        data["reference"] = remote_retriever.get_refs(data["genome_build"], aligner, data["config"])
    else:
        data["reference"] = genome.get_refs(data["genome_build"], aligner, data["dirs"]["galaxy"], data)
        _check_ref_files(data["reference"], data)
    # back compatible `sam_ref` target
    data["sam_ref"] = utils.get_in(data, ("reference", "fasta", "base"))
    ref_loc = utils.get_in(data, ("config", "resources", "species", "dir"),
                           utils.get_in(data, ("reference", "fasta", "base")))
    if remote_retriever:
        data = remote_retriever.get_resources(data["genome_build"], ref_loc, data)
    else:
        data["genome_resources"] = genome.get_resources(data["genome_build"], ref_loc, data)
    if effects.get_type(data) == "snpeff" and "snpeff" not in data["reference"]:
        data["reference"]["snpeff"] = effects.get_snpeff_files(data)
    data = _fill_validation_targets(data)
    data = _fill_prioritization_targets(data)
    # Re-enable when we have ability to re-define gemini configuration directory
    if False:
        if population.do_db_build([data], need_bam=False):
            data["reference"]["gemini"] = population.get_gemini_files(data)
    return data

def _check_ref_files(ref_info, data):
    problems = []
    for contig in ref.file_contigs(ref_info["fasta"]["base"], data["config"]):
        cur_problems = set([])
        for char in list(contig.name):
            if char not in ALLOWED_CONTIG_NAME_CHARS:
                cur_problems.add(char)
        if len(cur_problems) > 0:
            problems.append("Found non-allowed characters in chromosome name %s: %s" %
                            (contig.name, " ".join(list(cur_problems))))
    if len(problems) > 0:
        msg = ("\nProblems with input reference file %s\n" % ref_info["fasta"]["base"])
        raise ValueError(msg + "\n".join(problems) + "\n")

def _fill_validation_targets(data):
    """Fill validation targets pointing to globally installed truth sets.
    """
    ref_file = dd.get_ref_file(data)
    sv_targets = zip(itertools.repeat("svvalidate"),
                     tz.get_in(["config", "algorithm", "svvalidate"], data, {}).keys())
    for vtarget in [list(xs) for xs in [["validate"], ["validate_regions"]] + sv_targets]:
        val = tz.get_in(["config", "algorithm"] + vtarget, data)
        if val and not os.path.exists(val) and not objectstore.is_remote(val):
            installed_val = os.path.normpath(os.path.join(os.path.dirname(ref_file), os.pardir, "validation", val))
            if os.path.exists(installed_val):
                data = tz.update_in(data, ["config", "algorithm"] + vtarget, lambda x: installed_val)
            else:
                raise ValueError("Configuration problem. Validation file not found for %s: %s" %
                                 (vtarget, val))
    return data<commit_msg>Raise helpful error if genome_build not specified. Fixes #1277<commit_after>"""Retrieve run information describing files to process in a pipeline.

This handles two methods of getting processing information: from a Galaxy
next gen LIMS system or an on-file YAML configuration.
"""
import collections
from contextlib import closing
import copy
import glob
import itertools
import os
import string

import toolz as tz
import yaml

from bcbio import install, utils
from bcbio.bam import ref
from bcbio.log import logger
from bcbio.distributed import objectstore
from bcbio.illumina import flowcell
from bcbio.pipeline import alignment, config_utils, genome
from bcbio.pipeline import datadict as dd
from bcbio.provenance import diagnostics, programs, versioncheck
from bcbio.provenance import data as provenancedata
from bcbio.variation import effects, genotype, population, joint, vcfutils
from bcbio.variation.cortex import get_sample_name
from bcbio.bam.fastq import open_fastq

ALLOWED_CONTIG_NAME_CHARS = set(list(string.digits) + list(string.ascii_letters) + ["-", "_", "*", ":", "."])
ALGORITHM_NOPATH_KEYS = ["variantcaller", "realign", "recalibrate",
                         "phasing", "svcaller", "hetcaller", "jointcaller", "tools_off", "mixup_check"]

def organize(dirs, config, run_info_yaml, sample_names=None, add_provenance=True,
             integrations=None):
    """Organize run information from a passed YAML file or the Galaxy API.

    Creates the high level structure used for subsequent processing.

    sample_names is a list of samples to include from the overall file, for cases
    where we are running multiple pipelines from the same configuration file.
    """
    if integrations is None: integrations = {}
    logger.info("Using input YAML configuration: %s" % run_info_yaml)
    assert run_info_yaml and os.path.exists(run_info_yaml), \
        "Did not find input sample YAML file: %s" % run_info_yaml
    run_details = _run_info_from_yaml(dirs, run_info_yaml, config, sample_names)
    remote_retriever = None
    for iname, retriever in integrations.iteritems():
        if iname in config:
            run_details = retriever.add_remotes(run_details, config[iname])
            remote_retriever = retriever
    out = []
    for item in run_details:
        item["dirs"] = dirs
        if "name" not in item:
            item["name"] = ["", item["description"]]
        elif isinstance(item["name"], basestring):
            description = "%s-%s" % (item["name"], clean_name(item["description"]))
            item["name"] = [item["name"], description]
            item["description"] = description
        # add algorithm details to configuration, avoid double specification
        item["resources"] = _add_remote_resources(item["resources"])
        item["config"] = config_utils.update_w_custom(config, item)
        item.pop("algorithm", None)
        item = add_reference_resources(item, remote_retriever)
        # Create temporary directories and make absolute, expanding environmental variables
        tmp_dir = tz.get_in(["config", "resources", "tmp", "dir"], item)
        if tmp_dir:
            # if no environmental variables, make and normalize the directory
            # otherwise we normalize later in distributed.transaction:
            if os.path.expandvars(tmp_dir) == tmp_dir:
                tmp_dir = utils.safe_makedir(os.path.expandvars(tmp_dir))
                tmp_dir = genome.abs_file_paths(tmp_dir)
            item["config"]["resources"]["tmp"]["dir"] = tmp_dir
        out.append(item)
    out = _add_provenance(out, dirs, config, add_provenance)
    return out

def normalize_world(data):
    """Normalize a data object, useful after serializetion via CWL.
    """
    data = _normalize_files(data)
    return data

def _add_provenance(items, dirs, config, add_provenance=True):
    if add_provenance:
        p = programs.write_versions(dirs, config=config)
        d = provenancedata.write_versions(dirs, items)
        versioncheck.testall(items)
        p_db = diagnostics.initialize(dirs)
    out = []
    for item in items:
        if add_provenance:
            entity_id = diagnostics.store_entity(item)
            item["config"]["resources"]["program_versions"] = p
            item["provenance"] = {"programs": p, "entity": entity_id,
                                  "db": p_db, "data": d}
        out.append([item])
    return out

def setup_directories(work_dir, fc_dir, config, config_file):
    fastq_dir, galaxy_dir, config_dir = _get_full_paths(flowcell.get_fastq_dir(fc_dir)
                                                        if fc_dir else None,
                                                        config, config_file)
    # check default install for tool data if not found locally
    if not os.path.exists(os.path.join(galaxy_dir, "tool-data")):
        _, config_file = config_utils.load_system_config(work_dir=work_dir)
        if os.path.exists(os.path.join(os.path.dirname(config_file), "tool-data")):
            galaxy_dir = os.path.dirname(config_file)
    return {"fastq": fastq_dir, "galaxy": galaxy_dir,
            "work": work_dir, "flowcell": fc_dir, "config": config_dir}

def _get_full_paths(fastq_dir, config, config_file):
    """Retrieve full paths for directories in the case of relative locations.
    """
    if fastq_dir:
        fastq_dir = utils.add_full_path(fastq_dir)
    config_dir = utils.add_full_path(os.path.dirname(config_file))
    galaxy_config_file = utils.add_full_path(config.get("galaxy_config", "universe_wsgi.ini"),
                                             config_dir)
    return fastq_dir, os.path.dirname(galaxy_config_file), config_dir

# ## Remote resources

def _add_remote_resources(resources):
    """Retrieve remote resources like GATK/MuTect jars present in S3.
    """
    out = copy.deepcopy(resources)
    for prog, info in resources.iteritems():
        for key, val in info.iteritems():
            if key == "jar" and objectstore.is_remote(val):
                store_dir = utils.safe_makedir(os.path.join(os.getcwd(), "inputs", "jars", prog))
                fname = objectstore.download(val, store_dir, store_dir)
                version_file = os.path.join(store_dir, "version.txt")
                if not utils.file_exists(version_file):
                    version = install.get_gatk_jar_version(prog, fname)
                    with open(version_file, "w") as out_handle:
                        out_handle.write(version)
                else:
                    with open(version_file) as in_handle:
                        version = in_handle.read().strip()
                del out[prog][key]
                out[prog]["dir"] = store_dir
                out[prog]["version"] = version
    return out

# ## Genome reference information

def add_reference_resources(data, remote_retriever=None):
    """Add genome reference information to the item to process.
    """
    aligner = data["config"]["algorithm"].get("aligner", None)
    if remote_retriever:
        data["reference"] = remote_retriever.get_refs(data["genome_build"], aligner, data["config"])
    else:
        data["reference"] = genome.get_refs(data["genome_build"], aligner, data["dirs"]["galaxy"], data)
        _check_ref_files(data["reference"], data)
    # back compatible `sam_ref` target
    data["sam_ref"] = utils.get_in(data, ("reference", "fasta", "base"))
    ref_loc = utils.get_in(data, ("config", "resources", "species", "dir"),
                           utils.get_in(data, ("reference", "fasta", "base")))
    if remote_retriever:
        data = remote_retriever.get_resources(data["genome_build"], ref_loc, data)
    else:
        data["genome_resources"] = genome.get_resources(data["genome_build"], ref_loc, data)
    if effects.get_type(data) == "snpeff" and "snpeff" not in data["reference"]:
        data["reference"]["snpeff"] = effects.get_snpeff_files(data)
    data = _fill_validation_targets(data)
    data = _fill_prioritization_targets(data)
    # Re-enable when we have ability to re-define gemini configuration directory
    if False:
        if population.do_db_build([data], need_bam=False):
            data["reference"]["gemini"] = population.get_gemini_files(data)
    return data

def _check_ref_files(ref_info, data):
    problems = []
    if not data["genome_build"]:
        problems.append("Did not find 'genome_build' for sample: %s" % dd.get_sample_name(data))
    else:
        for contig in ref.file_contigs(ref_info["fasta"]["base"], data["config"]):
            cur_problems = set([])
            for char in list(contig.name):
                if char not in ALLOWED_CONTIG_NAME_CHARS:
                    cur_problems.add(char)
            if len(cur_problems) > 0:
                problems.append("Found non-allowed characters in chromosome name %s: %s" %
                                (contig.name, " ".join(list(cur_problems))))
    if len(problems) > 0:
        msg = ("\nProblems with input reference file %s\n" % tz.get_in(["fasta", "base"], ref_info))
        raise ValueError(msg + "\n".join(problems) + "\n")

def _fill_validation_targets(data):
    """Fill validation targets pointing to globally installed truth sets.
    """
    ref_file = dd.get_ref_file(data)
    sv_targets = zip(itertools.repeat("svvalidate"),
                     tz.get_in(["config", "algorithm", "svvalidate"], data, {}).keys())
    for vtarget in [list(xs) for xs in [["validate"], ["validate_regions"]] + sv_targets]:
        val = tz.get_in(["config", "algorithm"] + vtarget, data)
        if val and not os.path.exists(val) and not objectstore.is_remote(val):
            installed_val = os.path.normpath(os.path.join(os.path.dirname(ref_file), os.pardir, "validation", val))
            if os.path.exists(installed_val):
                data = tz.update_in(data, ["config", "algorithm"] + vtarget, lambda x: installed_val)
            else:
                raise ValueError("Configuration problem. Validation file not found for %s: %s" %
                                 (vtarget, val))
    return data
{'fraction_non_alphanumeric': 0.11251631034828867, 'fraction_numerical': 0.000501856870420556, 'mean_word_length': 2.966361464968153, 'pattern_counts': {'":': 20, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4120008', 'n_tokens_mistral': 6004, 'n_tokens_neox': 5882, 'n_words': 1487}