<commit_before>"""Prepare bcbio workflows from input YAML files.

This organizes the metadata and other information about workflows,
providing the necessary information to translate into CWL. The goal is to
eventually replace pipeline/main.py workflows with generalized
versions of this code.
"""
import copy
import collections
import pprint

import toolz as tz

def variant(variables):
    """Variant calling workflow implementation in CWL.

    The logic here is general, aside from `steps` and could be re-used for
    non-variable pipelines. This is an attempt to get the approach right before
    supporting everything with CWL.

    file_vs and std_vs are the list of world variables, split into those that
    reference files (and need declaration at each step) and those that don't
    and can be safely passed to each step. This function keeps track of these
    as they're added and updated by steps, passing this information on to
    creation of the CWL files.
    """
    file_vs, std_vs = _split_variables([_flatten_nested_input(v) for v in variables])
    def s(name, parallel, inputs, outputs, programs=None, noinputs=None):
        Step = collections.namedtuple("Step", "name parallel inputs outputs programs noinputs")
        if programs is None: programs = []
        if noinputs is None: noinputs = []
        return Step(name, parallel, inputs, outputs, programs, noinputs)
    def w(name, parallel, workflow, internal, noinputs=None):
        Workflow = collections.namedtuple("Workflow", "name parallel workflow internal noinputs")
        if noinputs is None: noinputs = []
        return Workflow(name, parallel, workflow, internal, noinputs)
    align = [s("prep_align_inputs", "single-split",
               [["files"]],
               [_cwl_file_world(["files"], ".gbi"),
                _cwl_nonfile_world(["config", "algorithm", "quality_format"]),
                _cwl_nonfile_world(["align_split"], allow_missing=True)]),
             s("process_alignment", "single-parallel",
               [["files"], ["reference", "fasta", "base"],
                ["reference", "aligner", "indexes"]],
               [_cwl_file_world(["work_bam"]),
                _cwl_file_world(["align_bam"]),
                _cwl_file_world(["hla", "fastq"], allow_missing=True),
                _cwl_file_world(["work_bam-plus", "disc"], ".bai"),
                _cwl_file_world(["work_bam-plus", "sr"], ".bai")],
               ["aligner", "samtools", "sambamba"]),
             s("merge_split_alignments", "single-merge",
               [["work_bam"], ["align_bam"], ["work_bam-plus", "disc"], ["work_bam-plus", "sr"],
                ["hla", "fastq"]],
               [_cwl_file_world(["align_bam"], ".bai"),
                _cwl_file_world(["work_bam-plus", "disc"], ".bai"),
                _cwl_file_world(["work_bam-plus", "sr"], ".bai"),
                _cwl_file_world(["hla", "fastq"], allow_missing=True)],
               ["biobambam"],
               noinputs=[["align_split"], ["config", "algorithm", "quality_format"]])]
    vc = [s("get_parallel_regions", "batch-split",
            [["batch_rec"]],
            [_cwl_nonfile_world(["region"])]),
          s("variantcall_batch_region", "batch-parallel",
            [["batch_rec"]],
            [_cwl_file_world(["vrn_file_region"], ".tbi"),
             _cwl_nonfile_world(["region"])]),
          s("concat_batch_variantcalls", "batch-merge",
            [["batch_rec"], ["vrn_file_region"]],
            [_cwl_file_world(["vrn_file"], ".tbi")]),
          s("postprocess_variants", "batch-single",
            [["batch_rec"], ["vrn_file"]],
            [_cwl_file_world(["vrn_file"], ".tbi")],
            noinputs=[["region"]]),
          s("compare_to_rm", "batch-single",
            [["batch_rec"], ["vrn_file"]],
            [_cwl_file_world(["validate", "summary"]),
             _cwl_file_world(["validate", "tp"], ".tbi"),
             _cwl_file_world(["validate", "fp"], ".tbi"),
             _cwl_file_world(["validate", "fn"], ".tbi")],
            noinputs=[["region"]])]
    steps = [w("alignment", "multi-parallel", align,
               [["align_split"], ["files"], ["work_bam"], ["config", "algorithm", "quality_format"]]),
             s("prep_samples", "multi-parallel",
               [["config", "algorithm", "variant_regions"],
                ["reference", "fasta", "base"]],
               [_cwl_file_world(["config", "algorithm", "variant_regions"], allow_missing=True),
                _cwl_file_world(["config", "algorithm", "variant_regions_merged"], allow_missing=True)]),
             s("postprocess_alignment", "multi-parallel",
               [["align_bam"], ["config", "algorithm", "variant_regions_merged"],
                ["reference", "fasta", "base"]],
               [_cwl_nonfile_world(["config", "algorithm", "coverage_interval"]),
                _cwl_file_world(["regions", "callable"]),
                _cwl_file_world(["regions", "sample_callable"]),
                _cwl_file_world(["regions", "nblock"]),
                _cwl_file_world(["regions", "highdepth"], allow_missing=True),
                _cwl_file_world(["regions", "offtarget_stats"])]),
             s("call_hla", "multi-parallel",
               [["hla", "fastq"]],
               [_cwl_nonfile_world(["hla", "hlacaller"], allow_missing=True),
                _cwl_file_world(["hla", "call_file"], allow_missing=True)]),
             s("combine_sample_regions", "multi-combined",
               [["regions", "callable"], ["regions", "nblock"],
                ["reference", "fasta", "base"]],
               [_cwl_file_world(["config", "algorithm", "callable_regions"]),
                _cwl_file_world(["config", "algorithm", "non_callable_regions"]),
                _cwl_nonfile_world(["config", "algorithm", "callable_count"], "int")]),
             s("batch_for_variantcall", "multi-batch",
               [["align_bam"], ["config", "algorithm", "callable_regions"], ["regions", "callable"],
                ["config", "algorithm", "variant_regions"],
                ["config", "algorithm", "validate"], ["config", "algorithm", "validate_regions"],
                ["reference", "fasta", "base"], ["reference", "rtg"],
                ["genome_resources", "variation", "cosmic"], ["genome_resources", "variation", "dbsnp"]],
               "batch_rec",
               noinputs=[["hla", "hlacaller"], ["config", "algorithm", "callable_count"]]),
             w("variantcall", "multi-parallel", vc,
               [["region"], ["vrn_file_region"]],
               noinputs=[["hla", "hlacaller"], ["config", "algorithm", "callable_count"]]),
             s("pipeline_summary", "multi-parallel",
               [["align_bam"], ["reference", "fasta", "base"]],
               [_cwl_file_world(["summary", "qc"])],
               ["samtools", "bamtools"]),
             s("coverage_report", "multi-parallel",
               [["align_bam"],
                ["reference", "fasta", "base"],
                ["config", "algorithm", "coverage"],
                ["config", "algorithm", "variant_regions"], ["regions", "offtarget_stats"]],
               [_cwl_file_world(["coverage", "all"], allow_missing=True),
                _cwl_file_world(["coverage", "problems"], allow_missing=True)]),
             s("qc_report_summary", "multi-combined",
               [["align_bam"],
                ["reference", "fasta", "base"],
                ["summary", "qc"], ["coverage", "all"], ["coverage", "problems"]],
               [_cwl_file_world(["coverage", "report"], allow_missing=True)])
             ]
    parallel_ids = []
    for step in steps:
        if hasattr(step, "workflow"):
            wf_inputs = []
            wf_outputs = []
            wf_steps = []
            for i, wf_step in enumerate(step.workflow):
                inputs, parallel_ids, nested_inputs = _get_step_inputs(wf_step, file_vs, std_vs, parallel_ids, step)
                outputs, file_vs, std_vs = _get_step_outputs(wf_step, wf_step.outputs, file_vs, std_vs)
                parallel_ids = _find_split_vs(outputs, wf_step.parallel)
                wf_steps.append(("step", wf_step.name, wf_step.parallel, [_clean_record(x) for x in inputs],
                                 outputs, wf_step.programs))
                wf_inputs = _merge_wf_inputs(inputs, wf_inputs, wf_outputs, step.internal, wf_step.parallel,
                                             nested_inputs)
                wf_outputs = _merge_wf_outputs(outputs, wf_outputs, wf_step.parallel)
            yield "wf_start", wf_inputs
            for wf_step in wf_steps:
                yield wf_step
            wf_outputs = [v for v in wf_outputs
                          if v["id"] not in set(["#%s" % _get_string_vid(x) for x in step.internal])]
            yield "upload", wf_outputs
            wf_outputs, file_vs, std_vs = _get_step_outputs(step, wf_outputs, file_vs, std_vs)
            yield "wf_finish", step.name, step.parallel, wf_inputs, wf_outputs
            file_vs = _extract_from_subworkflow(file_vs, step)
            std_vs = _extract_from_subworkflow(std_vs, step)
        else:
            inputs, parallel_ids, nested_inputs = _get_step_inputs(step, file_vs, std_vs, parallel_ids)
            outputs, file_vs, std_vs = _get_step_outputs(step, step.outputs, file_vs, std_vs)
            parallel_ids = _find_split_vs(outputs, step.parallel)
            yield "step", step.name, step.parallel, inputs, outputs, step.programs
    # Final outputs
    outputs = [["align_bam"], ["summary", "qc"], ["config", "algorithm", "callable_regions"]]
    outputs = [["align_bam"]]
    yield "upload", [_get_upload_output(x, file_vs) for x in outputs]

def _merge_wf_inputs(new, out, wf_outputs, to_ignore, parallel, nested_inputs):
    """Merge inputs for a sub-workflow, adding any not present inputs in out.

    Skips inputs that are internally generated or generated and ignored, keeping<commit_msg>CWL: update to a variant based pipeline for tests<commit_after>"""Prepare bcbio workflows from input YAML files.

This organizes the metadata and other information about workflows,
providing the necessary information to translate into CWL. The goal is to
eventually replace pipeline/main.py workflows with generalized
versions of this code.
"""
import copy
import collections
import pprint

import toolz as tz

def variant(variables):
    """Variant calling workflow implementation in CWL.

    The logic here is general, aside from `steps` and could be re-used for
    non-variable pipelines. This is an attempt to get the approach right before
    supporting everything with CWL.

    file_vs and std_vs are the list of world variables, split into those that
    reference files (and need declaration at each step) and those that don't
    and can be safely passed to each step. This function keeps track of these
    as they're added and updated by steps, passing this information on to
    creation of the CWL files.
    """
    file_vs, std_vs = _split_variables([_flatten_nested_input(v) for v in variables])
    def s(name, parallel, inputs, outputs, programs=None, noinputs=None):
        Step = collections.namedtuple("Step", "name parallel inputs outputs programs noinputs")
        if programs is None: programs = []
        if noinputs is None: noinputs = []
        return Step(name, parallel, inputs, outputs, programs, noinputs)
    def w(name, parallel, workflow, internal, noinputs=None):
        Workflow = collections.namedtuple("Workflow", "name parallel workflow internal noinputs")
        if noinputs is None: noinputs = []
        return Workflow(name, parallel, workflow, internal, noinputs)
    align = [s("prep_align_inputs", "single-split",
               [["files"]],
               [_cwl_file_world(["files"], ".gbi"),
                _cwl_nonfile_world(["config", "algorithm", "quality_format"]),
                _cwl_nonfile_world(["align_split"], allow_missing=True)]),
             s("process_alignment", "single-parallel",
               [["files"], ["reference", "fasta", "base"],
                ["reference", "aligner", "indexes"]],
               [_cwl_file_world(["work_bam"]),
                _cwl_file_world(["align_bam"]),
                _cwl_file_world(["hla", "fastq"], allow_missing=True),
                _cwl_file_world(["work_bam-plus", "disc"], ".bai"),
                _cwl_file_world(["work_bam-plus", "sr"], ".bai")],
               ["aligner", "samtools", "sambamba"]),
             s("merge_split_alignments", "single-merge",
               [["work_bam"], ["align_bam"], ["work_bam-plus", "disc"], ["work_bam-plus", "sr"],
                ["hla", "fastq"]],
               [_cwl_file_world(["align_bam"], ".bai"),
                _cwl_file_world(["work_bam-plus", "disc"], ".bai"),
                _cwl_file_world(["work_bam-plus", "sr"], ".bai"),
                _cwl_file_world(["hla", "fastq"], allow_missing=True)],
               ["biobambam"],
               noinputs=[["align_split"], ["config", "algorithm", "quality_format"]])]
    vc = [s("get_parallel_regions", "batch-split",
            [["batch_rec"]],
            [_cwl_nonfile_world(["region"])]),
          s("variantcall_batch_region", "batch-parallel",
            [["batch_rec"]],
            [_cwl_file_world(["vrn_file_region"], ".tbi"),
             _cwl_nonfile_world(["region"])]),
          s("concat_batch_variantcalls", "batch-merge",
            [["batch_rec"], ["vrn_file_region"]],
            [_cwl_file_world(["vrn_file"], ".tbi")]),
          s("postprocess_variants", "batch-single",
            [["batch_rec"], ["vrn_file"]],
            [_cwl_file_world(["vrn_file"], ".tbi")],
            noinputs=[["region"]]),
          s("compare_to_rm", "batch-single",
            [["batch_rec"], ["vrn_file"]],
            [_cwl_file_world(["validate", "summary"]),
             _cwl_file_world(["validate", "tp"], ".tbi"),
             _cwl_file_world(["validate", "fp"], ".tbi"),
             _cwl_file_world(["validate", "fn"], ".tbi")],
            noinputs=[["region"]])]
    steps = [w("alignment", "multi-parallel", align,
               [["align_split"], ["files"], ["work_bam"], ["config", "algorithm", "quality_format"]]),
             s("prep_samples", "multi-parallel",
               [["config", "algorithm", "variant_regions"],
                ["reference", "fasta", "base"]],
               [_cwl_file_world(["config", "algorithm", "variant_regions"], allow_missing=True),
                _cwl_file_world(["config", "algorithm", "variant_regions_merged"], allow_missing=True)]),
             s("postprocess_alignment", "multi-parallel",
               [["align_bam"], ["config", "algorithm", "variant_regions_merged"],
                ["reference", "fasta", "base"]],
               [_cwl_nonfile_world(["config", "algorithm", "coverage_interval"]),
                _cwl_file_world(["regions", "callable"]),
                _cwl_file_world(["regions", "sample_callable"]),
                _cwl_file_world(["regions", "nblock"]),
                _cwl_file_world(["regions", "highdepth"], allow_missing=True),
                _cwl_file_world(["regions", "offtarget_stats"])]),
             #s("call_hla", "multi-parallel",
             #  [["hla", "fastq"]],
             #  [_cwl_nonfile_world(["hla", "hlacaller"], allow_missing=True),
             #   _cwl_file_world(["hla", "call_file"], allow_missing=True)]),
             s("combine_sample_regions", "multi-combined",
               [["regions", "callable"], ["regions", "nblock"],
                ["reference", "fasta", "base"]],
               [_cwl_file_world(["config", "algorithm", "callable_regions"]),
                _cwl_file_world(["config", "algorithm", "non_callable_regions"]),
                _cwl_nonfile_world(["config", "algorithm", "callable_count"], "int")]),
             s("batch_for_variantcall", "multi-batch",
               [["align_bam"], ["config", "algorithm", "callable_regions"], ["regions", "callable"],
                ["config", "algorithm", "variant_regions"],
                ["config", "algorithm", "validate"], ["config", "algorithm", "validate_regions"],
                ["reference", "fasta", "base"], ["reference", "rtg"],
                ["genome_resources", "variation", "cosmic"], ["genome_resources", "variation", "dbsnp"]],
               "batch_rec",
               noinputs=[["hla", "hlacaller"], ["config", "algorithm", "callable_count"]]),
             w("variantcall", "multi-parallel", vc,
               [["region"], ["vrn_file_region"]],
               noinputs=[["hla", "hlacaller"], ["config", "algorithm", "callable_count"]]),
             #s("pipeline_summary", "multi-parallel",
             #  [["align_bam"], ["reference", "fasta", "base"]],
             #  [_cwl_file_world(["summary", "qc"])],
             #  ["samtools", "bamtools"]),
             #s("coverage_report", "multi-parallel",
             #  [["align_bam"],
             #   ["reference", "fasta", "base"],
             #   ["config", "algorithm", "coverage"],
             #   ["config", "algorithm", "variant_regions"], ["regions", "offtarget_stats"]],
             #  [_cwl_file_world(["coverage", "all"], allow_missing=True),
             #   _cwl_file_world(["coverage", "problems"], allow_missing=True)]),
             #s("qc_report_summary", "multi-combined",
             #  [["align_bam"],
             #   ["reference", "fasta", "base"],
             #   ["summary", "qc"], ["coverage", "all"], ["coverage", "problems"]],
             #  [_cwl_file_world(["coverage", "report"], allow_missing=True)])
             ]
    parallel_ids = []
    for step in steps:
        if hasattr(step, "workflow"):
            wf_inputs = []
            wf_outputs = []
            wf_steps = []
            for i, wf_step in enumerate(step.workflow):
                inputs, parallel_ids, nested_inputs = _get_step_inputs(wf_step, file_vs, std_vs, parallel_ids, step)
                outputs, file_vs, std_vs = _get_step_outputs(wf_step, wf_step.outputs, file_vs, std_vs)
                parallel_ids = _find_split_vs(outputs, wf_step.parallel)
                wf_steps.append(("step", wf_step.name, wf_step.parallel, [_clean_record(x) for x in inputs],
                                 outputs, wf_step.programs))
                wf_inputs = _merge_wf_inputs(inputs, wf_inputs, wf_outputs, step.internal, wf_step.parallel,
                                             nested_inputs)
                wf_outputs = _merge_wf_outputs(outputs, wf_outputs, wf_step.parallel)
            yield "wf_start", wf_inputs
            for wf_step in wf_steps:
                yield wf_step
            wf_outputs = [v for v in wf_outputs
                          if v["id"] not in set(["#%s" % _get_string_vid(x) for x in step.internal])]
            yield "upload", wf_outputs
            wf_outputs, file_vs, std_vs = _get_step_outputs(step, wf_outputs, file_vs, std_vs)
            yield "wf_finish", step.name, step.parallel, wf_inputs, wf_outputs
            file_vs = _extract_from_subworkflow(file_vs, step)
            std_vs = _extract_from_subworkflow(std_vs, step)
        else:
            inputs, parallel_ids, nested_inputs = _get_step_inputs(step, file_vs, std_vs, parallel_ids)
            outputs, file_vs, std_vs = _get_step_outputs(step, step.outputs, file_vs, std_vs)
            parallel_ids = _find_split_vs(outputs, step.parallel)
            yield "step", step.name, step.parallel, inputs, outputs, step.programs
    # Final outputs
    outputs = [["align_bam"], ["summary", "qc"], ["config", "algorithm", "callable_regions"]]
    outputs = [["align_bam"], ["vrn_file"], ["validate", "summary"]]
    yield "upload", [_get_upload_output(x, file_vs) for x in outputs]

def _merge_wf_inputs(new, out, wf_outputs, to_ignore, parallel, nested_inputs):
    """Merge inputs for a sub-workflow, adding any not present inputs in out.

    Skips inputs that are internally generated or generated and ignored, keeping
{'fraction_non_alphanumeric': 0.1566705017257766, 'fraction_numerical': 0.0, 'mean_word_length': 2.6831245394252026, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5791988', 'n_tokens_mistral': 6156, 'n_tokens_neox': 5906, 'n_words': 1304}