SV Liftover Code

CHANGELOG.md

@@ -11,6 +11,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 ### Added
 
 - StableLift for sSNV liftover
+- StableLift for SV liftover
 
 ## [6.2.0] - 2024-11-22
 

config/input_handler.config

@@ -327,7 +327,10 @@ input_handler {
             'call-sCNA': [:],
             'recalibrate-BAM': ['BAM':'', 'contamination_table':''],
             'calculate-targeted-coverage': ['expanded-intervals': ''],
-            'convert-BAM2FASTQ': []
+            'convert-BAM2FASTQ': [],
+            'call-gSV': [:],
+            'call-sSV': [:],
+            'call-gSNP': [:]
         ];
 
         return default_data_map;

config/pipeline_selector.config

@@ -26,7 +26,9 @@ pipeline_selector {
             'call-sSV': ['recalibrate-BAM'],
             'call-sCNA': ['recalibrate-BAM'],
             'call-SRC': ['call-sSNV', 'call-sCNA'],
-            'StableLiftsSNV': ['call-sSNV']
+            'StableLiftsSNV': ['call-sSNV'],
+            'StableLiftgSV': ['call-gSV'],
+            'StableLiftsSV': ['call-sSV']
             ]
 
         if (input_type == 'BAM') {
@@ -137,22 +139,21 @@ pipeline_selector {
             'call_sSV': 'call-sSV',
             'call_sCNA': 'call-sCNA',
             'call_SRC': 'call-SRC',
-            'StableLift': 'StableLiftsSNV'
+            'StableLift': ['StableLiftsSNV', 'StableLiftgSV', 'StableLiftsSV']
         ];
 
         pipeline_name_map.each { pipeline_key, pipeline ->
             if (!params.pipeline_params.containsKey(pipeline_key)) {
                 params.pipeline_params[pipeline_key] = ['is_pipeline_enabled': false];
             }
 
-            params.pipeline_params[pipeline_key].is_pipeline_enabled = pipelines.contains(pipeline);
+            if (pipeline_key == 'StableLift') {
+                List requested_liftovers = pipeline.findAll{ element -> pipelines.contains(element) };
 
-            if (pipeline_key == 'StableLift' && pipelines.contains(pipeline)) {
-                if (!params.pipeline_params[pipeline_key].containsKey('lift_modes')) {
-                    params.pipeline_params[pipeline_key]['lift_modes'] = []
-                }
-
-                params.pipeline_params[pipeline_key]['lift_modes'] << pipeline
+                params.pipeline_params[pipeline_key]['is_pipeline_enabled'] = (requested_liftovers) ? true : false;
+                params.pipeline_params[pipeline_key]['lift_modes'] = (requested_liftovers) ?: [];
+            } else {
+                params.pipeline_params[pipeline_key].is_pipeline_enabled = pipelines.contains(pipeline);
             }
         }
     }

config/schema.yaml

@@ -126,6 +126,8 @@ requested_pipelines:
     - call-sCNA
     - call-SRC
     - StableLiftsSNV
+    # - StableLiftgSV
+    # - StableLiftsSV
   help: 'List of pipelines to run'
 override_realignment:
   type: 'Bool'

config/template.config

@@ -27,7 +27,7 @@ params {
 
     sample_mode = 'paired' // Choose from: 'single', 'paired', 'multi'
 
-    // Select pipeline(s) to run. Choices: 'align-DNA', 'recalibrate-BAM', 'generate-SQC-BAM', 'calculate-targeted-coverage', 'call-gSNP', 'call-mtSNV', 'call-sSNV', 'call-sSV', 'call-gSV', 'call-sCNA', 'StableLiftsSNV'
+    // Select pipeline(s) to run. Choices: 'align-DNA', 'recalibrate-BAM', 'generate-SQC-BAM', 'calculate-targeted-coverage', 'call-gSNP', 'call-mtSNV', 'call-sSNV', 'call-sSV', 'call-gSV', 'call-sCNA', 'StableLiftsSNV', 'StableLiftsSV', 'StableLiftgSV'
     requested_pipelines = ['align-DNA', 'recalibrate-BAM', 'generate-SQC-BAM', 'call-gSNP', 'call-mtSNV', 'call-sSNV', 'call-sSV', 'call-gSV', 'call-sCNA']
 
     // Override conversion to FASTQ and re-alignment with BAM input

module/StableLift/workflow.nf

@@ -21,13 +21,13 @@ workflow stable_lift {
             .map{ it ->
                 def samples = [];
                 params.sample_data.each { s, s_data ->
-                    if (s_data.state == 'normal') {
-                        return;
-                    }
                     params.StableLift.lift_modes.each { raw_mode ->
                         def mode = raw_mode.replace('StableLift', '');
+                        if (!s_data.containsKey("call-${mode}" as String)) {
+                            return;
+                        }
                         s_data["call-${mode}"].each { tool, data ->
-                            if (tool == 'BCFtools-Intersect') {
+                            if (['BCFtools-Intersect', 'Manta-gSV', 'Manta-sSV'].contains(tool)) {
                                 return;
                             }
                             samples.add([

module/call_gSV/create_YAML_call_gSV.nf

@@ -19,7 +19,7 @@ process create_YAML_call_gSV {
         )
 
     output:
-        path(input_yaml), emit: call_gsv_yaml
+        tuple val(sample_id), path(input_yaml), emit: call_gsv_yaml
 
     exec:
     input_yaml = "call_gSV_input.yaml"

module/call_gSV/identify_outputs.nf

@@ -0,0 +1,42 @@
+include { identify_file } from '../common'
+include { sanitize_string } from '../../external/pipeline-Nextflow-module/modules/common/generate_standardized_filename/main.nf'
+
+workflow identify_call_gsv_outputs {
+    take:
+    och_call_gsv
+
+    main:
+    och_call_gsv.map{ call_gsv_out ->
+        def raw_sample_id = call_gsv_out[0];
+        def sample_id = sanitize_string(raw_sample_id);
+        def gsv_output_dir = new File(call_gsv_out[1].toString());
+        def gsv_output_pattern = /(.*)-([\d\.]*)$/;
+
+        def output_info = [
+            'Manta': ['Manta-gSV', "Manta-*${sample_id}*candidateSV.vcf.gz"],
+            // 'DELLY': ['Delly2-gSV', "DELLY-*${sample_id}*gSV.vcf.gz"]
+        ]
+
+        def outputs_to_check = [];
+        def match = null;
+
+        gsv_output_dir.eachFile { file ->
+            match = (file.name =~ gsv_output_pattern);
+            if (match) {
+                outputs_to_check << [match[0][1], file.name];
+            }
+        }
+
+        outputs_to_check.each { output_tool, output_dir_name ->
+            params.sample_data[raw_sample_id]['call-gSV'][output_info[output_tool][0]] = identify_file("${gsv_output_dir}/${output_dir_name}/output/${output_info[output_tool][1]}");
+        }
+
+        return 'done';
+    }
+    .collect()
+    .map{ 'done' }
+    .set{ och_call_gsv_identified }
+
+    emit:
+    och_call_gsv_identified = och_call_gsv_identified
+}

module/call_gSV/run_call_gSV.nf

@@ -25,15 +25,17 @@ process run_call_gSV {
         pattern: "call-gSV-*/*"
 
     input:
-        path(input_yaml)
+        tuple val(sample_id), path(input_yaml)
 
     output:
+        tuple val(sample_id), path(output_directory), emit: identify_call_gsv_out, optional: true
         path "call-gSV-*/*", optional: true
         path ".command.*"
         val('done'), emit: complete
         env EXIT_CODE, emit: exit_code
 
     script:
+    output_directory = "call-gSV-*/${sample_id}"
     String params_to_dump = combine_input_with_params(params.call_gSV.metapipeline_arg_map, new File(input_yaml.toRealPath().toString()))
     String setup_commands = generate_graceful_error_controller(task.ext)
     String weblog_args = generate_weblog_args()

module/call_gSV/workflow.nf

@@ -5,6 +5,7 @@
 include { create_YAML_call_gSV } from "${moduleDir}/create_YAML_call_gSV"
 include { run_call_gSV } from "${moduleDir}/run_call_gSV" addParams( log_output_dir: params.metapipeline_log_output_dir )
 include { mark_pipeline_complete; mark_pipeline_exit_code } from "../pipeline_status"
+include { identify_call_gsv_outputs } from "./identify_outputs"
 
 /*
 * Main workflow for calling the call-gSV pipeline
@@ -16,60 +17,79 @@ workflow call_gSV {
     take:
         modification_signal
     main:
-        // Watch for pipeline ordering
-        Channel.watchPath( "${params.pipeline_status_directory}/*.ready" )
-            .until{ it -> it.name == "${params.this_pipeline}.ready" }
-            .ifEmpty('done')
-            .collect()
-            .map{ 'done' }
-            .set{ pipeline_predecessor_complete }
+        if (!params.call_gSV.is_pipeline_enabled) {
+            modification_signal.until{ it == 'done' }.ifEmpty('done')
+                .map{ it ->
+                    mark_pipeline_complete(params.this_pipeline);
+                    mark_pipeline_exit_code(params.this_pipeline, 0);
+                    return 'done';
+                }
+                .set{ completion_signal }
+        } else {
+            // Watch for pipeline ordering
+            Channel.watchPath( "${params.pipeline_status_directory}/*.ready" )
+                .until{ it -> it.name == "${params.this_pipeline}.ready" }
+                .ifEmpty('done')
+                .collect()
+                .map{ 'done' }
+                .set{ pipeline_predecessor_complete }
 
-        // Extract inputs from data structure
-        modification_signal.until{ it == 'done' }.ifEmpty('done')
-            .mix(pipeline_predecessor_complete)
-            .collect()
-            .map{ it ->
-                def samples = [];
-                params.sample_data.each { s, s_data ->
-                    samples.add(['patient': s_data['patient'], 'sample': s, 'state': s_data['state'], 'bam': s_data['recalibrate-BAM']['BAM']]);
-                };
-                return samples
-            }
-            .flatten()
-            .reduce(['normal': [] as Set, 'tumor': [] as Set]) { a, b ->
-                a[b.state] += b;
-                return a
-            }
-            .set{ ich }
+            // Extract inputs from data structure
+            modification_signal.until{ it == 'done' }.ifEmpty('done')
+                .mix(pipeline_predecessor_complete)
+                .collect()
+                .map{ it ->
+                    def samples = [];
+                    params.sample_data.each { s, s_data ->
+                        samples.add(['patient': s_data['patient'], 'sample': s, 'state': s_data['state'], 'bam': s_data['recalibrate-BAM']['BAM']]);
+                    };
+                    return samples
+                }
+                .flatten()
+                .reduce(['normal': [] as Set, 'tumor': [] as Set]) { a, b ->
+                    a[b.state] += b;
+                    return a
+                }
+                .set{ ich }
 
-        ich
-            .map{ it -> it.normal }
-            .flatten()
-            .unique{ [it.patient, it.sample, it.state] }
-            .map{ it -> [params.patient, it['sample'], it['bam']] }
-            .set{ input_ch_create_YAML }
+            ich
+                .map{ it -> it.normal }
+                .flatten()
+                .unique{ [it.patient, it.sample, it.state] }
+                .map{ it -> [params.patient, it['sample'], it['bam']] }
+                .set{ input_ch_create_YAML }
 
-        create_YAML_call_gSV(input_ch_create_YAML)
+            create_YAML_call_gSV(input_ch_create_YAML)
 
-        run_call_gSV(create_YAML_call_gSV.out.call_gsv_yaml)
+            run_call_gSV(create_YAML_call_gSV.out.call_gsv_yaml)
 
-        run_call_gSV.out.complete
-            .mix( pipeline_predecessor_complete )
-            .collect()
-            .map{ it ->
-                mark_pipeline_complete(params.this_pipeline);
-                return 'done';
-            }
-            .mix(
-                run_call_gSV.out.exit_code
-                    .map{ it -> (it as Integer) }
-                    .sum()
-                    .map { exit_code ->
-                        mark_pipeline_exit_code(params.this_pipeline, exit_code);
-                        return 'done';
-                    }
+            identify_call_gsv_outputs(
+                modification_signal.until{ it == 'done' }
+                    .mix( run_call_gSV.out.identify_call_gsv_out )
             )
-            .collect()
-            .map { it -> return 'done'; }
-            .set{ completion_signal }
+
+            run_call_gSV.out.complete
+                .mix( identify_call_gsv_outputs.out.och_call_gsv_identified )
+                .mix( pipeline_predecessor_complete )
+                .collect()
+                .map{ it ->
+                    mark_pipeline_complete(params.this_pipeline);
+                    return 'done';
+                }
+                .mix(
+                    run_call_gSV.out.exit_code
+                        .map{ it -> (it as Integer) }
+                        .sum()
+                        .map { exit_code ->
+                            mark_pipeline_exit_code(params.this_pipeline, exit_code);
+                            return 'done';
+                        }
+                )
+                .collect()
+                .map { it -> return 'done'; }
+                .set{ completion_signal }
+        }
+
+    emit:
+    completion_signal = completion_signal
 }

module/call_sSV/create_YAML_call_sSV.nf

@@ -19,7 +19,7 @@ process create_YAML_call_sSV {
         )
 
     output:
-        path(input_yaml)
+        tuple val(tumor_id), path(input_yaml)
 
     exec:
     input_yaml = 'call_sSV_input.yaml'

module/call_sSV/identify_outputs.nf

@@ -0,0 +1,42 @@
+include { identify_file } from '../common'
+include { sanitize_string } from '../../external/pipeline-Nextflow-module/modules/common/generate_standardized_filename/main.nf'
+
+workflow identify_call_ssv_outputs {
+    take:
+    och_call_ssv
+
+    main:
+    och_call_ssv.map{ call_ssv_out ->
+        def raw_sample_id = call_ssv_out[0];
+        def sample_id = sanitize_string(raw_sample_id);
+        def ssv_output_dir = new File(call_ssv_out[1].toString());
+        def ssv_output_pattern = /(.*)-([\d\.]*)$/;
+
+        def output_info = [
+            'Manta': ['Manta-sSV', "Manta-*${sample_id}*candidateSV.vcf.gz"],
+            // 'DELLY': ['Delly2-sSV', "DELLY-*${sample_id}.vcf.gz"]
+        ]
+
+        def outputs_to_check = [];
+        def match = null;
+
+        ssv_output_dir.eachFile { file ->
+            match = (file.name =~ ssv_output_pattern);
+            if (match) {
+                outputs_to_check << [match[0][1], file.name];
+            }
+        }
+
+        outputs_to_check.each { output_tool, output_dir_name ->
+            params.sample_data[raw_sample_id]['call-sSV'][output_info[output_tool][0]] = identify_file("${ssv_output_dir}/${output_dir_name}/output/${output_info[output_tool][1]}");
+        }
+
+        return 'done';
+    }
+    .collect()
+    .map{ 'done' }
+    .set{ och_call_ssv_identified }
+
+    emit:
+    och_call_ssv_identified = och_call_ssv_identified
+}

module/call_sSV/run_call_sSV.nf

@@ -25,15 +25,17 @@ process run_call_sSV {
         saveAs: { "${task.process.replace(':', '/')}-${task.index}/log${file(it).getName()}" }
 
     input:
-        path(input_yaml)
+        tuple val(sample_id), path(input_yaml)
 
     output:
+        tuple val(sample_id), path(output_directory), emit: identify_call_ssv_out, optional: true
         path "call-sSV-*/*", optional: true
         path ".command.*"
         val('done'), emit: complete
         env EXIT_CODE, emit: exit_code
 
     script:
+    output_directory = "call-sSV-*/${sample_id}"
     String params_to_dump = combine_input_with_params(params.call_sSV.metapipeline_arg_map, new File(input_yaml.toRealPath().toString()))
     String setup_commands = generate_graceful_error_controller(task.ext)
     String weblog_args = generate_weblog_args()