<commit_before>"""Definitions of workflows for translation into common workflow language.

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.

Ideally we could translate the specification of these workflows into a YAML-based
DSL (some sort of CWL-lite), instead of writing them in Python code.

The variable 'workflows' provides a dictionary to retrieve the steps and outputs
for each of the defined workflows.
"""
import collections
from bcbio.pipeline import datadict as dd

def s(name, parallel, inputs, outputs, image, programs=None, disk=None, cores=None, unlist=None,
      no_files=False):
    """Represent a step in a workflow.

    name -- The run function name, which must match a definition in distributed/multitasks
    inputs -- List of input keys required for the function. Each key is of the type:
      ["toplevel", "sublevel"] -- an argument you could pass to toolz.get_in.
    outputs -- List of outputs with information about file type. Use cwlout functions
    programs -- Required programs for this step, used to define resource usage.
    disk -- Information about disk usage requirements, specified as multipliers of
            input files. Ensures enough disk present when that is a limiting factor
            when selecting cloud node resources.
    cores -- Maximum cores necessary for this step, for non-multicore processes.
    unlist -- Variables being unlisted by this process. Useful for parallelization splitting and
      batching from multiple variables, like variant calling.
    no_files -- This step does not require file access.
    parallel -- Parallelization approach. There are three different levels of parallelization,
      each with subcomponents:

      1. multi -- Multiple samples, parallelizing at the sample level. Used in top-level workflow.
        - multi-parallel -- Run individual samples in parallel.
        - multi-combined -- Run all samples together.
        - multi-batch -- Run all samples together, converting into batches of grouped samples.
      2. single -- A single sample, used in sub-workflows.
        - single-split -- Split a sample into sub-components (by read sections).
        - single-parallel -- Run sub-components of a sample in parallel.
        - single-merge -- Merge multiple sub-components into a single sample.
        - single-single -- Single sample, single item, nothing fancy.
      3. batch -- Several related samples (tumor/normal, or populations). Used in sub-workflows.
        - batch-split -- Split a batch of samples into sub-components (by genomic region).
        - batch-parallel -- Run sub-components of a batch in parallel.
        - batch-merge -- Merge sub-components back into a single batch.
        - batch-single -- Run on a single batch.
    """
    Step = collections.namedtuple("Step", "name parallel inputs outputs image programs disk cores unlist no_files")
    if programs is None: programs = []
    if unlist is None: unlist = []
    return Step(name, parallel, inputs, outputs, image, programs, disk, cores, unlist, no_files)

def w(name, parallel, workflow, internal):
    """A workflow, allowing specification of sub-workflows for nested parallelization.

    name and parallel are documented under the Step (s) function.
    workflow -- a list of Step tuples defining the sub-workflow
    internal -- variables used in the sub-workflow but not exposed to subsequent steps
    """
    Workflow = collections.namedtuple("Workflow", "name parallel workflow internal")
    return Workflow(name, parallel, workflow, internal)

def et(name, parallel, inputs, outputs, expression):
    """Represent an ExpressionTool that reorders inputs using javascript.
    """
    ExpressionTool = collections.namedtuple("ExpressionTool", "name inputs outputs expression parallel")
    return ExpressionTool(name, inputs, outputs, expression, parallel)

def cwlout(key, valtype=None, extensions=None, fields=None, exclude=None):
    """Definition of an output variable, defining the type and associated secondary files.
    """
    out = {"id": key}
    if valtype:
        out["type"] = valtype
    if fields:
        out["fields"] = fields
    if extensions:
        out["secondaryFiles"] = extensions
    if exclude:
        out["exclude"] = exclude
    return out

def _alignment(checkpoints):
    align = [s("prep_align_inputs", "single-split" if checkpoints["align_split"] else "single-single",
               [["alignment_rec"]],
               [cwlout("process_alignment_rec", "record",
                       fields=[cwlout(["files"], ["null", {"type": "array", "items": "File"}], [".gbi"]),
                               cwlout(["config", "algorithm", "quality_format"], ["string", "null"]),
                               cwlout(["align_split"], ["string", "null"])])],
               "bcbio-vc", ["grabix", "htslib", "biobambam", "atropos;env=python3",
                            "optitype", "razers3=3.5.0"],  # Includes HLA callers for general docker inclusion
               disk={"files": 1.5}),
             s("process_alignment", "single-parallel" if checkpoints["align_split"] else "single-single",
               [["alignment_rec"], ["process_alignment_rec"]],
               [cwlout(["work_bam"], ["File", "null"], [".bai"]),
                cwlout(["align_bam"], ["File", "null"], [".bai"]),
                cwlout(["hla", "fastq"], ["null", {"type": "array", "items": "File"}]),
                cwlout(["work_bam_plus", "disc"], ["File", "null"], [".bai"]),
                cwlout(["work_bam_plus", "sr"], ["File", "null"], [".bai"])],
               "bcbio-vc", ["bwa", "bwakit", "grabix", "minimap2", "novoalign", "snap-aligner=1.0dev.97",
                            "sentieon", "samtools", "pysam>=0.13.0", "sambamba", "fgbio", "umis",
                            "biobambam", "seqtk", "samblaster", "variantbam"],
               disk={"files": 2})]
    if checkpoints["align_split"]:
        align += [s("merge_split_alignments", "single-merge",
                  [["alignment_rec"], ["work_bam"], ["align_bam"],
                   ["work_bam_plus", "disc"], ["work_bam_plus", "sr"], ["hla", "fastq"]],
                  [cwlout(["align_bam"], ["File", "null"], [".bai"]),
                   cwlout(["work_bam_plus", "disc"], ["File", "null"], [".bai"]),
                   cwlout(["work_bam_plus", "sr"], ["File", "null"], [".bai"]),
                   cwlout(["hla", "fastq"], ["null", {"type": "array", "items": "File"}])],
                  "bcbio-vc", ["biobambam", "samtools", "variantbam"],
                  disk={"files": 3.5})]
    return align

def _variant_hla(checkpoints):
    """Add hla analysis to workflow, if configured.
    """
    if not checkpoints.get("hla"):
        return [], []
    hla = [s("hla_to_rec", "multi-batch",
             [["hla", "fastq"],
              ["config", "algorithm", "hlacaller"]],
               [cwlout("hla_rec", "record")],
               "bcbio-vc", cores=1, no_files=True),
           s("call_hla", "multi-parallel",
             [["hla_rec"]],
             [cwlout(["hla", "hlacaller"], ["string", "null"]),
              cwlout(["hla", "call_file"], ["File", "null"])],
             "bcbio-vc", ["optitype", "razers3=3.5.0"])]
    return hla, []

def _variant_vc(checkpoints):
    """Add variant calling to workflow, if configured.
    """
    if not checkpoints.get("vc"):
        return [], []
    vc_wf = [s("get_parallel_regions", "batch-split",
               [["batch_rec"]],
               [cwlout(["region_block"], {"type": "array", "items": "string"})],
               "bcbio-vc",
               disk={"files": 2.0}, cores=1),
             s("variantcall_batch_region", "batch-parallel",
               [["batch_rec"], ["region_block"]],
               [cwlout(["vrn_file_region"], ["File", "null"], [".tbi"]),
                cwlout(["region_block"], {"type": "array", "items": "string"})],
               "bcbio-vc", ["bcftools", "bedtools", "freebayes=1.1.0.46",
                            "gatk", "gatk4", "sentieon",
                            "htslib", "picard", "platypus-variant", "pythonpy",
                            "samtools", "pysam>=0.13.0", "strelka", "vardict", "vardict-java",
                            "varscan", "vcfanno", "vcflib", "vt", "r=3.4.1", "perl"],
               disk={"files": 2.0}),
             s("concat_batch_variantcalls", "batch-merge",<commit_msg>CWL: include cbc solver in docker image<commit_after>"""Definitions of workflows for translation into common workflow language.

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.

Ideally we could translate the specification of these workflows into a YAML-based
DSL (some sort of CWL-lite), instead of writing them in Python code.

The variable 'workflows' provides a dictionary to retrieve the steps and outputs
for each of the defined workflows.
"""
import collections
from bcbio.pipeline import datadict as dd

def s(name, parallel, inputs, outputs, image, programs=None, disk=None, cores=None, unlist=None,
      no_files=False):
    """Represent a step in a workflow.

    name -- The run function name, which must match a definition in distributed/multitasks
    inputs -- List of input keys required for the function. Each key is of the type:
      ["toplevel", "sublevel"] -- an argument you could pass to toolz.get_in.
    outputs -- List of outputs with information about file type. Use cwlout functions
    programs -- Required programs for this step, used to define resource usage.
    disk -- Information about disk usage requirements, specified as multipliers of
            input files. Ensures enough disk present when that is a limiting factor
            when selecting cloud node resources.
    cores -- Maximum cores necessary for this step, for non-multicore processes.
    unlist -- Variables being unlisted by this process. Useful for parallelization splitting and
      batching from multiple variables, like variant calling.
    no_files -- This step does not require file access.
    parallel -- Parallelization approach. There are three different levels of parallelization,
      each with subcomponents:

      1. multi -- Multiple samples, parallelizing at the sample level. Used in top-level workflow.
        - multi-parallel -- Run individual samples in parallel.
        - multi-combined -- Run all samples together.
        - multi-batch -- Run all samples together, converting into batches of grouped samples.
      2. single -- A single sample, used in sub-workflows.
        - single-split -- Split a sample into sub-components (by read sections).
        - single-parallel -- Run sub-components of a sample in parallel.
        - single-merge -- Merge multiple sub-components into a single sample.
        - single-single -- Single sample, single item, nothing fancy.
      3. batch -- Several related samples (tumor/normal, or populations). Used in sub-workflows.
        - batch-split -- Split a batch of samples into sub-components (by genomic region).
        - batch-parallel -- Run sub-components of a batch in parallel.
        - batch-merge -- Merge sub-components back into a single batch.
        - batch-single -- Run on a single batch.
    """
    Step = collections.namedtuple("Step", "name parallel inputs outputs image programs disk cores unlist no_files")
    if programs is None: programs = []
    if unlist is None: unlist = []
    return Step(name, parallel, inputs, outputs, image, programs, disk, cores, unlist, no_files)

def w(name, parallel, workflow, internal):
    """A workflow, allowing specification of sub-workflows for nested parallelization.

    name and parallel are documented under the Step (s) function.
    workflow -- a list of Step tuples defining the sub-workflow
    internal -- variables used in the sub-workflow but not exposed to subsequent steps
    """
    Workflow = collections.namedtuple("Workflow", "name parallel workflow internal")
    return Workflow(name, parallel, workflow, internal)

def et(name, parallel, inputs, outputs, expression):
    """Represent an ExpressionTool that reorders inputs using javascript.
    """
    ExpressionTool = collections.namedtuple("ExpressionTool", "name inputs outputs expression parallel")
    return ExpressionTool(name, inputs, outputs, expression, parallel)

def cwlout(key, valtype=None, extensions=None, fields=None, exclude=None):
    """Definition of an output variable, defining the type and associated secondary files.
    """
    out = {"id": key}
    if valtype:
        out["type"] = valtype
    if fields:
        out["fields"] = fields
    if extensions:
        out["secondaryFiles"] = extensions
    if exclude:
        out["exclude"] = exclude
    return out

def _alignment(checkpoints):
    align = [s("prep_align_inputs", "single-split" if checkpoints["align_split"] else "single-single",
               [["alignment_rec"]],
               [cwlout("process_alignment_rec", "record",
                       fields=[cwlout(["files"], ["null", {"type": "array", "items": "File"}], [".gbi"]),
                               cwlout(["config", "algorithm", "quality_format"], ["string", "null"]),
                               cwlout(["align_split"], ["string", "null"])])],
               "bcbio-vc", ["grabix", "htslib", "biobambam", "atropos;env=python3",
                            "optitype", "razers3=3.5.0", "coincbc"],  # HLA deps for general docker inclusion
               disk={"files": 1.5}),
             s("process_alignment", "single-parallel" if checkpoints["align_split"] else "single-single",
               [["alignment_rec"], ["process_alignment_rec"]],
               [cwlout(["work_bam"], ["File", "null"], [".bai"]),
                cwlout(["align_bam"], ["File", "null"], [".bai"]),
                cwlout(["hla", "fastq"], ["null", {"type": "array", "items": "File"}]),
                cwlout(["work_bam_plus", "disc"], ["File", "null"], [".bai"]),
                cwlout(["work_bam_plus", "sr"], ["File", "null"], [".bai"])],
               "bcbio-vc", ["bwa", "bwakit", "grabix", "minimap2", "novoalign", "snap-aligner=1.0dev.97",
                            "sentieon", "samtools", "pysam>=0.13.0", "sambamba", "fgbio", "umis",
                            "biobambam", "seqtk", "samblaster", "variantbam"],
               disk={"files": 2})]
    if checkpoints["align_split"]:
        align += [s("merge_split_alignments", "single-merge",
                  [["alignment_rec"], ["work_bam"], ["align_bam"],
                   ["work_bam_plus", "disc"], ["work_bam_plus", "sr"], ["hla", "fastq"]],
                  [cwlout(["align_bam"], ["File", "null"], [".bai"]),
                   cwlout(["work_bam_plus", "disc"], ["File", "null"], [".bai"]),
                   cwlout(["work_bam_plus", "sr"], ["File", "null"], [".bai"]),
                   cwlout(["hla", "fastq"], ["null", {"type": "array", "items": "File"}])],
                  "bcbio-vc", ["biobambam", "samtools", "variantbam"],
                  disk={"files": 3.5})]
    return align

def _variant_hla(checkpoints):
    """Add hla analysis to workflow, if configured.
    """
    if not checkpoints.get("hla"):
        return [], []
    hla = [s("hla_to_rec", "multi-batch",
             [["hla", "fastq"],
              ["config", "algorithm", "hlacaller"]],
               [cwlout("hla_rec", "record")],
               "bcbio-vc", cores=1, no_files=True),
           s("call_hla", "multi-parallel",
             [["hla_rec"]],
             [cwlout(["hla", "hlacaller"], ["string", "null"]),
              cwlout(["hla", "call_file"], ["File", "null"])],
             "bcbio-vc", ["optitype", "razers3=3.5.0", "coincbc"])]
    return hla, []

def _variant_vc(checkpoints):
    """Add variant calling to workflow, if configured.
    """
    if not checkpoints.get("vc"):
        return [], []
    vc_wf = [s("get_parallel_regions", "batch-split",
               [["batch_rec"]],
               [cwlout(["region_block"], {"type": "array", "items": "string"})],
               "bcbio-vc",
               disk={"files": 2.0}, cores=1),
             s("variantcall_batch_region", "batch-parallel",
               [["batch_rec"], ["region_block"]],
               [cwlout(["vrn_file_region"], ["File", "null"], [".tbi"]),
                cwlout(["region_block"], {"type": "array", "items": "string"})],
               "bcbio-vc", ["bcftools", "bedtools", "freebayes=1.1.0.46",
                            "gatk", "gatk4", "sentieon",
                            "htslib", "picard", "platypus-variant", "pythonpy",
                            "samtools", "pysam>=0.13.0", "strelka", "vardict", "vardict-java",
                            "varscan", "vcfanno", "vcflib", "vt", "r=3.4.1", "perl"],
               disk={"files": 2.0}),
             s("concat_batch_variantcalls", "batch-merge",
{'fraction_non_alphanumeric': 0.1475763225355395, 'fraction_numerical': 0.005243532975996271, 'mean_word_length': 2.67087254063302, 'pattern_counts': {'":': 32, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3465800', 'n_tokens_mistral': 5020, 'n_tokens_neox': 4845, 'n_words': 1530}