merging template changes with precomp_busco

Author: tkchafin

bin/generate_config.py

@@ -51,7 +51,7 @@ def parse_args(args=None):
     parser.add_argument("--blastx", help="Path to the blastx database", required=True)
     parser.add_argument("--blastn", help="Path to the blastn database", required=True)
     parser.add_argument("--taxdump", help="Path to the taxonomy database", required=True)
-    parser.add_argument("--busco_output", action="append", help="Path to BUSCO output directory", required=False)
+    parser.add_argument("--precomputed_busco", action="append", help="Path to precomputed BUSCO outputs", required=False)
     parser.add_argument("--version", action="version", version="%(prog)s 2.0")
     return parser.parse_args(args)
 
@@ -353,7 +353,7 @@ def main(args=None):
     taxon_info = fetch_taxon_info(args.taxon_query)
     classification = get_classification(taxon_info)
 
-    precomputed_busco = [os.path.basename(path).replace("run_", "") for path in (args.busco_output or [])]
+    precomputed_busco = [os.path.basename(path).replace("run_", "") for path in (args.precomputed_busco or [])]
     odb_arr = get_odb(taxon_info, args.lineage_tax_ids, args.busco, precomputed_busco)
     taxon_id = adjust_taxon_id(args.nt, taxon_info)
 

conf/test_nobusco.config

@@ -10,15 +10,18 @@
 ----------------------------------------------------------------------------------------
 */
 
+process {
+    resourceLimits = [
+        cpus: 2,
+        memory: '6.GB',
+        time: '6.h'
+    ]
+}
+
 params {
     config_profile_name        = 'Test profile'
     config_profile_description = 'Minimal aligned test dataset to check pipeline function'
 
-    // Limit resources so that this can run on GitHub Actions
-    max_cpus   = 2
-    max_memory = '6.GB'
-    max_time   = '6.h'
-
     // Input test data
     // Specify the paths to your test data
     // Give any required params for the test so that command line flags are not needed
@@ -39,8 +42,8 @@ params {
     // Precomputed BUSCO outputs
     // busco_output_noArchaea.tar.gz deliberately leaves out archaea_odb10 to test the pipeline's detection and filling of missing lineages
     // Switch to *_busco_output.tar.gz for fully precomputed BUSCOs
-    busco_output = "https://tolit.cog.sanger.ac.uk/test-data/Meles_meles/resources/GCA_922984935.2_busco_output_noArchaea.tar.gz"
-    //busco_output = "https://tolit.cog.sanger.ac.uk/test-data/Meles_meles/resources/GCA_922984935.2_busco_output.tar.gz"
+    precomputed_busco = "https://tolit.cog.sanger.ac.uk/test-data/Meles_meles/resources/GCA_922984935.2_busco_output_noArchaea.tar.gz"
+    //precomputed_busco = "https://tolit.cog.sanger.ac.uk/test-data/Meles_meles/resources/GCA_922984935.2_busco_output.tar.gz"
 
     // Need to be set to avoid overfilling /tmp
     use_work_dir_as_temp = true

nextflow.config

@@ -16,7 +16,7 @@ params {
     mask                       = false
     fetchngs_samplesheet       = false
     busco_lineages             = null
-    busco_output               = null
+    precomputed_busco          = null
 
     // Reference options
     fasta                      = null

nextflow_schema.json

@@ -334,7 +334,7 @@
         }
     ],
     "properties": {
-        "busco_output": {
+        "precomputed_busco": {
             "type": "string"
         }
     }

subworkflows/local/busco_diamond_blastp.nf

@@ -15,13 +15,14 @@ workflow BUSCO_DIAMOND {
     busco_db     // channel: path(busco_db)
     blastp       // channel: path(blastp_db)
     taxon_id     // channel: val(taxon_id)
-    busco_output // channel: [ val(meta), path(fasta) ] optional precomputed busco outputs
-
+    precomputed_busco // channel: [ val(meta}, path(busco_run_dir) ] optional precomputed busco outputs
 
     main:
     ch_versions = Channel.empty()
 
 
+    precomputed_busco.view()
+
     //
     // Prepare the BUSCO lineages
     //
@@ -44,16 +45,16 @@ workflow BUSCO_DIAMOND {
     //
     // Format pre-computed outputs
     //
-    if (params.busco_output){
-        ch_busco_output = busco_output
+    if (params.precomputed_busco){
+        ch_precomputed_busco = precomputed_busco
             .map { meta, dir -> [meta.lineage, [meta, dir]] }
     }else{
-        ch_busco_output = Channel.empty()
+        ch_precomputed_busco = Channel.empty()
     }
 
     ch_combined = ch_fasta_with_lineage
         .map { meta, fasta -> [meta.lineage_name, [meta, fasta]] }
-        .join(ch_busco_output, by: 0, remainder: true)
+        .join(ch_precomputed_busco, by: 0, remainder: true)
         .map { lineage, fasta_data, busco_data ->
             def (meta, fasta) = fasta_data
             def (busco_meta, busco_dir) = busco_data ?: [null, null]
@@ -96,12 +97,12 @@ workflow BUSCO_DIAMOND {
     //
     // Run BUSCO search
     //
-    BUSCO (
-        ch_fasta_with_lineage,
-        "genome",
-        ch_fasta_with_lineage.map { it[0].lineage_name },
+    BUSCO_BUSCO(
+        ch_busco_to_run.to_compute,
+        'genome',
+        ch_busco_to_run.to_compute.map { it[0].lineage_name },
         busco_db,
-        [],
+        []
     )
     ch_versions = ch_versions.mix ( BUSCO_BUSCO.out.versions.first() )
 

subworkflows/local/input_check.nf

@@ -56,10 +56,11 @@ workflow INPUT_CHECK {
             blastn: db_meta.type == "blastn"
             blastp: db_meta.type == "blastp"
             blastx: db_meta.type == "blastx"
-            busco_output: db_meta.type == "busco_output"
+            precomputed_busco: db_meta.type == "precomputed_busco"
             busco: db_meta.type == "busco"
             taxdump: db_meta.type == "taxdump"
         }
+    ch_databases.precomputed_busco.view()
 
     //
     // SUBWORKFLOW: Process samplesheet
@@ -105,7 +106,7 @@ workflow INPUT_CHECK {
 
     // Get the source paths of all the databases, except Busco which is not recorded in the blobDir meta.json
     databases
-    | filter { meta, file -> meta.type != "busco" }
+    | filter { meta, file -> meta.type != "busco" && meta.type != "precomputed_busco" }
     | map {meta, file -> [meta, file.toUriString()]}
     | set { db_paths }
 
@@ -153,7 +154,22 @@ workflow INPUT_CHECK {
     | collect
     | set { ch_busco_lineages }
 
-    // Remove any invalid lineages from busco_outputs
+    // Format pre-computed BUSCOs (if provided)
+    // Parse the BUSCO output directories
+    if (ch_databases.precomputed_busco) {
+        ch_parsed_busco = ch_databases.precomputed_busco
+            .flatMap { meta, dir ->
+                def subdirs = file(dir).listFiles().findAll { it.isDirectory() }
+                subdirs.collect { subdir ->
+                    def lineage = subdir.name.split('_')[1..-1].join('_')
+                    [[type: 'precomputed_busco', id: subdir.name, lineage: lineage], subdir]
+                }
+            }
+    } else {
+        ch_parsed_busco = Channel.empty()
+    }
+
+    // Remove any invalid lineages from precomputed_busco
     ch_busco_lineages_list = ch_busco_lineages.flatten()
     ch_parsed_busco_filtered = ch_parsed_busco
         .filter { meta, path ->
@@ -180,6 +196,7 @@ workflow INPUT_CHECK {
     blastn = ch_databases.blastn.first()    // channel: [ val(meta), path(blastn_db) ]
     blastp = ch_databases.blastp.first()    // channel: [ val(meta), path(blastp_db) ]
     blastx = ch_databases.blastx.first()    // channel: [ val(meta), path(blastx_db) ]
+    precomputed_busco = ch_parsed_busco    // channel: [ val(meta), path(busco_run_dir) ]
     busco_db = ch_busco_db                  // channel: [ path(busco_db) ]
     taxdump = ch_databases.taxdump.map { _, db_path -> db_path }          // channel: [ path(taxdump) ]
     versions = ch_versions                  // channel: [ versions.yml ]

subworkflows/local/utils_nfcore_blobtoolkit_pipeline/main.nf

@@ -71,6 +71,7 @@ workflow PIPELINE_INITIALISATION {
         .concat( Channel.fromPath(params.blastn).map { tuple(["type": "blastn"], it) } )
         .concat( Channel.fromPath(params.blastx).map { tuple(["type": "blastx"], it) } )
         .concat( Channel.fromPath(params.blastp).map { tuple(["type": "blastp"], it) } )
+        .concat( params.precomputed_busco ? Channel.fromPath(params.precomputed_busco).map { tuple([ "type": "precomputed_busco"], it ) } : Channel.empty() )
         .concat( params.busco ? Channel.fromPath(params.busco).map { tuple([ "type": "busco"], it ) } : Channel.empty() )
         .concat( Channel.fromPath(params.taxdump).map { tuple(["type": "taxdump"], it) } )
         .set { ch_databases }

workflows/blobtoolkit.nf

@@ -90,7 +90,7 @@ workflow BLOBTOOLKIT {
         INPUT_CHECK.out.busco_db.first(),
         INPUT_CHECK.out.blastp.first(),
         INPUT_CHECK.out.taxon_id,
-        INPUT_CHECK.out.busco_output,
+        INPUT_CHECK.out.precomputed_busco,
     )
     ch_versions = ch_versions.mix ( BUSCO_DIAMOND.out.versions )
 
@@ -204,7 +204,7 @@ workflow BLOBTOOLKIT {
         )
     )
 
-    ch_multiqc_files                      = ch_multiqc_files.mix(BUSCO_DIAMOND.out.multiqc.collect{it[1]}.ifEmpty([]))
+//    ch_multiqc_files                      = ch_multiqc_files.mix(BUSCO_DIAMOND.out.multiqc.collect{it[1]}.ifEmpty([]))
 
     MULTIQC (
         ch_multiqc_files.collect(),