Fixes to passing normal id to purple process. Also harmonizing pbrun fq2bam settings, providing GATK best practices -K flag.

Author: kevinmhadi

conf/modules.config

@@ -197,6 +197,7 @@ process {
     }
 
     withName: '.*PARABRICKS_FQ2BAM.*' {
+        ext.args = '--bwa-options "-Y -K 100000000"'
 		publishDir       = [
             mode: params.publish_dir_mode,
             path: { 
@@ -292,8 +293,9 @@ process {
     }
 
     withName: 'BWAMEM.*_MEM|SENTIEON_BWAMEM' {
-        // Using -B 3 for tumor samples
-        ext.args         = { meta.status.toString() == "1" ? "-K 100000000 -Y -B 3 -R ${meta.read_group}" : "-K 100000000 -Y -R ${meta.read_group}" }
+        // // Using -B 3 for tumor samples
+        // ext.args         = { meta.status.toString() == "1" ? "-K 100000000 -Y -B 3 -R ${meta.read_group}" : "-K 100000000 -Y -R ${meta.read_group}" }
+        ext.args = { meta.status.toString() == "1" ? "-K 100000000 -Y -R ${meta.read_group}" : "-K 100000000 -Y -R ${meta.read_group}" }
     }
 
     withName: 'MERGE_BAM|INDEX_MERGE_BAM' {

conf/resource.config

@@ -83,6 +83,13 @@ process {
         memory          = { check_max( 64.GB * task.attempt, 'memory' ) }
         time            = { check_max( 48.h  * task.attempt, 'time'    ) }
     }
+
+    withName: '.*PURPLE.*' {
+        cpus            = { check_max( 2 * task.attempt, 'cpus' ) }
+        memory          = { check_max( 48.GB * task.attempt, 'memory' ) }
+        time            = { check_max( 4.h  * task.attempt, 'time'    ) }
+    }
+
     withName: 'GRIDSS_GRIDSS' {
         cpus            = { check_max( 8 * task.attempt, 'cpus' ) }
         memory          = { check_max( 32.GB * task.attempt, 'memory' ) }

modules/local/gridss/somaticFilter/main.nf

@@ -25,6 +25,7 @@ process GRIPSS_SOMATIC_FILTER {
     task.ext.when == null || task.ext.when
 
     script:
+    def reference_arg = meta.containsKey('normal_id') ? "-reference ${meta.normal_id}" : ''
     def args          = task.ext.args ?: ''
     def prefix        = task.ext.prefix ?: "${meta.id}"
     def VERSION       = '2.3.4' // WARN: Version information not provided by tool on CLI. Please update this string when bumping container versions.
@@ -41,7 +42,9 @@ process GRIPSS_SOMATIC_FILTER {
          -ref_genome ${fasta} \\
          -ref_genome_version ${pon_gridss_ref_genome_version} \\
          -output_dir ./ \\
-         -sample ${meta.sample}
+         -sample ${meta.sample} \\
+         ${reference_arg} \\
+         ${args} \\
 
     cat <<-END_VERSIONS > versions.yml
     "${task.process}":

nextflow.config

@@ -241,7 +241,7 @@ params {
     vep_spliceregion = null // spliceregion plugin disabled within VEP
 
 	// Signatures
-	filter_ffpe_impact = false // Default is to not run FFPE impact signatures
+	filter_ffpe_impact = false // Default is to not filter out FFPE impact signatures
 
 
 	// MultiQC options

subworkflows/local/bam_svcalling_gridss/main.nf

@@ -47,6 +47,7 @@ workflow BAM_SVCALLING_GRIDSS {
 workflow GRIDSS_SOMATIC_FILTER_STEP {
     take:
     vcf
+    inputs_unlaned
 
     main:
     // pondir_gridss = WorkflowNfcasereports.create_file_channel(params.pon_gridss)
@@ -61,6 +62,26 @@ workflow GRIDSS_SOMATIC_FILTER_STEP {
     somatic_all             = Channel.empty()
     somatic_high_confidence = Channel.empty()
 
+    vcf = vcf.map { it -> 
+        [ it[0].patient, [ it[0], it[1], it[2] ] ] // meta.patient, [vcf, vcf_index]
+    }
+    .join(
+        inputs_unlaned.filter {it -> it.meta.status.toString() == "0"}.map { it ->  [ it.meta.patient, [ it.meta + [ normal_id: it.meta.sample ] ] ]}.unique(),
+        remainder: true
+    )
+    .dump(tag: "right after join vcf_joined_with_inputs for GRIDSS_SOMATIC_FILTER_STEP", pretty: true)
+    .map { it -> // patient, [ meta (vcf), vcf (vcf), vcf_tbi (vcf) ], [ meta with normal id (inputs) ]
+        def (_key, existing, meta_input_lst) = (it + [null, null])[0..2]
+        def (meta_existing, vcf_existing, tbi_existing) = existing ?: [null, null, null]
+        def meta_input = meta_input_lst ? meta_input_lst[0] : null // should only be one entry in the list since we unique by patient, but just in case we take the first
+        if (meta_input != null) {
+            meta_existing = meta_existing + [ normal_id: meta_input.normal_id ]
+        }
+        [ meta_existing, vcf_existing, tbi_existing ]
+    }
+    .dump(tag: "vcf_joined_with_inputs for GRIDSS_SOMATIC_FILTER_STEP", pretty: true)
+    .filter { it -> it[0] != null } // filter out any entries that don't have meta (shouldn't be any since we join with remainder: true, but just in case)
+    
     GRIPSS_SOMATIC_FILTER(
         vcf, 
         pon_gridss_bedpe_svs,

subworkflows/local/steps.nf

@@ -1985,7 +1985,6 @@ workflow PURPLE_STEP {
 
     // need a channel with patient and meta for merging with rest
     purple_inputs_for_merge = inputs_unlaned
-        .filter { it -> it.meta.status.toString() == "1" }
         .filter { it -> 
             (it.ploidy instanceof List && it.ploidy.isEmpty())
             || (it.purity instanceof List && it.purity.isEmpty())
@@ -2037,15 +2036,15 @@ workflow PURPLE_STEP {
             }
             .dump(tag: "meta_purple merged", pretty: true)
 
-    purple_inputs_cobalt_dir = purple_inputs_for_merge
+    purple_inputs_cobalt_dir = meta_purple_branched.tumor
         .join(cobalt_dir_for_merge)
         .map { it -> [ it[0], it[2] ] } // patient, cobalt_dir
 
-    purple_inputs_amber_dir = purple_inputs_for_merge
+    purple_inputs_amber_dir = meta_purple_branched.tumor
         .join(amber_dir_for_merge)
         .map { it -> [ it[0], it[2] ] } // patient, amber_dir
 
-    purple_inputs_sv = purple_inputs_for_merge.map { it -> [ it[0], [], [] ] }.unique{ it -> it[0] }
+    purple_inputs_sv = meta_purple_branched.tumor.map { it -> [ it[0], [], [] ] }.unique{ it -> it[0] }
     if (params.purple_use_svs) {
         sample_purple_use_svs = inputs_unlaned
             .filter { it -> it.meta.status.toString() == "1" }
@@ -2069,7 +2068,7 @@ workflow PURPLE_STEP {
                 println "purple_use_svs for sample ${it.meta.sample} set to ${value}"
                 [ it.meta.patient, [ value ] ]
             }
-        purple_inputs_sv = purple_inputs_for_merge.map { it -> [ it[0], [ it[1] ] ] } // patient, [meta]
+        purple_inputs_sv = meta_purple_branched.tumor.filter { it -> it[1].status.toString() == "1" }.map { it -> [ it[0], [ it[1] ] ] } // patient, [meta]
             .join(vcf_from_sv_calling_for_merge.map { it -> [ it[0] ] + [ it[1..-1] ] })
             .join(sample_purple_use_svs)
             .map { it -> // patient, meta, vcftbi_list, use_svs_list
@@ -2087,18 +2086,18 @@ workflow PURPLE_STEP {
                 // .map { it -> [ it[0], it[2], it[3] ] } // patient, vcf, tbi
     }
 
-    purple_inputs_snv = purple_inputs_for_merge.map { it -> [ it[0], [], [] ] }.unique{ it -> it[0] }
+    purple_inputs_snv = meta_purple_branched.tumor.map { it -> [ it[0], [], [] ] }.unique{ it -> it[0] }
     if (params.purple_use_smlvs) {
         println "Using Purple small variants"
-        purple_inputs_snv = purple_inputs_for_merge
+        purple_inputs_snv = meta_purple_branched.tumor.filter { it -> it[1].status.toString() == "1" }
             .join(filtered_somatic_vcf_for_merge)
             .map { it -> [ it[0], it[2], it[3] ] } // patient, vcf, tbi
     }
 
-    purple_inputs_snv_germline = purple_inputs_for_merge.map { it -> [ it[0], [], [] ] }.unique{ it -> it[0] }
+    purple_inputs_snv_germline = meta_purple_branched.tumor.map { it -> [ it[0], [], [] ] }.unique{ it -> it[0] }
 
     if (! params.tumor_only && params.purple_use_smlvs) {
-        purple_inputs_snv_germline = purple_inputs_for_merge
+        purple_inputs_snv_germline = meta_purple_branched.tumor.filter { it -> it[1].status.toString() == "1" }
             .join(germline_vcf_for_merge)
             .map { it -> [ it[0], it[2], it[3] ] } // patient, vcf, tbi
     }

workflows/main_workflow.nf

@@ -909,7 +909,8 @@ workflow NFGOS {
         //     }
 
         GRIDSS_SOMATIC_FILTER_STEP(
-            vcf_raw_from_gridss_gridss
+            vcf_raw_from_gridss_gridss,
+            inputs_unlaned
         )
 
         versions = versions.mix(GRIDSS_SOMATIC_FILTER_STEP.out.versions)