Batching assemblies throughout workflow

main.nf

@@ -62,10 +62,10 @@ workflow {
     // instead of the output of the DOWNLOAD_BAKTA_DB process
     // and set publishDir move to `move` in that process
     if ( file(params.bakta_db).exists() ) {
-        bakta_db = Channel.of(file(params.bakta_db))
+        bakta_db = channel.value(file(params.bakta_db))
     } else {
         println "Downloading bakta db to ${params.bakta_db}"
-        bakta_db = DOWNLOAD_BAKTA_DB(params.bakta_db_type)
+        bakta_db = DOWNLOAD_BAKTA_DB(params.bakta_db_type).first() // forces a value channel
     }
 
     infiles = Channel.fromPath("${params.indir}/*${params.infile_extension}")
@@ -129,9 +129,9 @@ workflow {
             all_seqs_ch,
             bulk_annotations
         )
-        bulk_ann_final_ch = EXTEND_ANNOTATIONS.out.bulk_annotations_extended
+        bulk_ann_final_ch = EXTEND_ANNOTATIONS.out.bulk_annotations_extended.first() // forces a value channel
     } else {
-        bulk_ann_final_ch = bulk_annotations
+        bulk_ann_final_ch = bulk_annotations.first() // forces a value channel
     }
 
     //-----------------------------
@@ -163,8 +163,7 @@ workflow {
     ch_sorf_in = ch_cds_keyed
         .join(ch_rna_keyed)
         .join(ch_asm_keyed)
-        .map { batchtag, cds_meta, cds_pkl, rna_pkl, asm, bakta_db -> tuple(cds_meta, cds_pkl, asm, rna_pkl, bakta_db) }
-        .combine(bakta_db)
+        .map { batchtag, cds_meta, cds_pkl, rna_pkl, asm, bakta_db -> tuple(cds_meta, asm, cds_pkl, rna_pkl, bakta_db) }
 
     SORF_EXTRA(ch_sorf_in)
 

modules/find_sorf_extra.nf

@@ -1,60 +1,57 @@
 process SORF_EXTRA {
-    tag { sample_id }
+    tag { meta.tag }
     label "sorf_extra_search"
     label 'bakta'
 
-    // TODO: allow specifying output path for the gff3 file in Bakta to avoid doing this
     publishDir (
-        "${params.outdir}/final_annotations", 
-        pattern: "*.sorf-extra.gff3",
-        mode: 'copy', 
-        saveAs: { filename -> 
-            def name = file(filename).name
-            if (name.endsWith(".sorf-extra.gff3")) {
-                def base = name.replaceFirst(/\.sorf-extra\.gff3$/, '')
-                return "${base}.gff3"
-            }
-            return null
-        }, enabled: { ${params.bundle_gff3} != true }
+        "${params.outdir}/final_annotations",
+        pattern: params.bundle_gff3 ? "*.gff3.tar.gz" : "*.gff3",
+        mode: 'copy'
     )
     publishDir (
-        "${params.outdir}/final_annotations", 
-        pattern: "*.gff3.tar.gz",
-        mode: 'copy', 
-        enabled: params.bundle_gff3
-    )
-    publishDir (
-        "${params.outdir}/final_annotations", 
+        "${params.outdir}/final_annotations",
         pattern: "*.sorf-extra.pkl",
-        mode: 'copy', 
-        enabled: params.save_intermediate
+        mode: 'copy',
+        enabled: { params.save_intermediate }
     )
 
     input:
     tuple val(meta), path(assemblies), path(cds_pkl), path(rna_pkl), path(bakta_db)
 
     output:
-    tuple val(meta), path("${meta.tag}.gff3.tar.gz"), emit: tar_gff3_annotations, optional: true
-    tuple val(meta), path("${meta.tag}.gff3"), emit: gff3_annotations, optional: true
+    tuple val(meta), path("${meta.tag}.gff3.tar.gz"),    emit: tar_gff3_annotations, optional: true // bundled
+    tuple val(meta), path("*.gff3"),                     emit: gff3_annotations,     optional: true // unbundled
     tuple val(meta), path("${meta.tag}.sorf-extra.pkl"), emit: pkl_annotations
 
 
     script:
     def compliant = params.compliant ? "--compliant" : ""
-    def individual_pickles = meta.asm_ids.collect { asm_id -> "SORFS_bakta/${asm_id}.sorf-extra.pkl" }.join(' ')
-    def individual_gff3s = meta.asm_ids.collect { asm_id -> "SORFS_bakta/${asm_id}.sorf-extra.gff3" }.join(' ')
+    def individual_pickles = meta.asm_ids.collect { asm_id -> "SORFs_bakta/${asm_id}.sorf-extra.pkl" }.join(' ')
+    def individual_gff3s = meta.asm_ids.collect { asm_id -> "${asm_id}.gff3" }.join(' ')
     """
+    manage_pkls.py unbatch \\
+        --input ${cds_pkl} \\
+        --out-dir cds_unbatched \\
+        --suffix .cds.pkl
+
+    manage_pkls.py unbatch \\
+        --input ${rna_pkl} \\
+        --out-dir rna_unbatched \\
+        --suffix .rna.pkl
+
     # Loop through assemblies in batch running bakta on each
     for asm in ${assemblies}; do
         id=\$(basename "\$asm" | sed -E 's/(\\.[^.]+)+\$//')
-        bakta --db ${bakta_db} --sorf-extra  \
-            --cds-pickle ${cds_pkl} \
-            --rna-pickle ${rna_pkl} \
-            --output SORFS_bakta/  \
-            --prefix ${id} \
-            --threads ${task.cpus} \
-            ${compliant} \
-            ${asm}
+        bakta --db ${bakta_db} --sorf-extra  \\
+            --cds-pickle cds_unbatched/\${id}.cds.pkl \\
+            --rna-pickle rna_unbatched/\${id}.rna.pkl \\
+            --output SORFs_bakta/  \\
+            --prefix \${id} \\
+            --threads ${task.cpus} \\
+            --force \\
+            ${compliant} \\
+            \${asm}
+        mv SORFs_bakta/\${id}.sorf-extra.gff3 \${id}.gff3
     done
 
     # Concatenate to batch-level files
@@ -65,7 +62,6 @@ process SORF_EXTRA {
 
     if ( "${params.bundle_gff3}" == "true" ) ; then
         tar -czf ${meta.tag}.gff3.tar.gz ${individual_gff3s} \
-            --transform "s|SORFS_bakta/||" \
             && rm -rf ${individual_gff3s}
     fi
     """

nextflow.config

@@ -145,7 +145,7 @@ process {
     // containers used by docker/singularity profiles
     withLabel:bakta {
         // container = 'quay.io/d_goryslavets/bakta_pannotator:1.12.0-pannotator.5'
-        container = 'quay.io/sangerpathogens/bakta:v1.12.0-pannotator.5_test_bulk_mode'
+        container = 'quay.io/d_goryslavets/bakta_pannotator:1.12.0-pannotator.5_test_bulk_mode2'
     }
     withLabel:clustering {
         container = 'quay.io/d_goryslavets/mmseqs2_pannotator:17-b804f-pannotator.1'