Skip to content

Commit

Permalink
read plots, own blast DB
Browse files Browse the repository at this point in the history
  • Loading branch information
replikation committed Dec 20, 2019
1 parent 9b2b5b9 commit 11662e7
Show file tree
Hide file tree
Showing 18 changed files with 174 additions and 43 deletions.
File renamed without changes.
17 changes: 17 additions & 0 deletions modules/databases/phage_references_blastDB.nf
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
process phage_references_blastDB {
if (params.cloudProcess) {
publishDir "${params.cloudDatabase}/", mode: 'copy', pattern: "blast_phage_DB"
}
else {
storeDir "nextflow-autodownload-databases/"
}
label 'metaphinder'
input:
file(references)
output:
file("blast_phage_DB/")
script:
"""
makeblastdb -in ${references} -dbtype nucl -parse_seqids -out blast_phage_DB/phage_db -title phage_db
"""
}
File renamed without changes.
File renamed without changes.
File renamed without changes.
13 changes: 0 additions & 13 deletions modules/filter_metaphinder.nf

This file was deleted.

18 changes: 18 additions & 0 deletions modules/metaphinder.nf
Original file line number Diff line number Diff line change
Expand Up @@ -12,4 +12,22 @@ process metaphinder {
MetaPhinder.py -i ${fasta} -o ${name} -d /MetaPhinder/database/ALL_140821_hr
mv ${name}/output.txt ${name}_\${rnd//0.}.list
"""
}


process metaphinder_own_DB {
publishDir "${params.output}/${name}/metaphinder-own-DB", mode: 'copy', pattern: "${name}_*.list"
label 'metaphinder'
input:
tuple val(name), file(fasta)
file(database)
output:
tuple val(name), file("${name}_*.list")
script:
"""
rnd=${Math.random()}
mkdir ${name}
MetaPhinder.py -i ${fasta} -o ${name} -d ${database}/blast_phage_DB/phage_db
mv ${name}/output.txt ${name}_\${rnd//0.}.list
"""
}
File renamed without changes.
File renamed without changes.
File renamed without changes.
27 changes: 27 additions & 0 deletions modules/parser/filter_metaphinder.nf
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
process filter_metaphinder {
publishDir "${params.output}/${name}/metaphinder", mode: 'copy', pattern: "metaphinder_*.txt"
label 'ubuntu'
input:
tuple val(name), file(results)
output:
tuple val(name), file("metaphinder_*.txt")
script:
"""
rnd=${Math.random()}
cat *.list | sort -g -k4,4 | awk '{if(\$2=="phage"){print \$1}}' | tail -n+2 > metaphinder_\${rnd//0.}.txt
"""
}

process filter_metaphinder_own_DB {
publishDir "${params.output}/${name}/metaphinder-own-DB", mode: 'copy', pattern: "metaphinder-own-DB_*.txt"
label 'ubuntu'
input:
tuple val(name), file(results)
output:
tuple val(name), file("metaphinder-own-DB_*.txt")
script:
"""
rnd=${Math.random()}
cat *.list | sort -g -k4,4 | awk '{if(\$2=="phage"){print \$1}}' | tail -n+2 > metaphinder-own-DB_\${rnd//0.}.txt
"""
}
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
36 changes: 36 additions & 0 deletions modules/shuffle_reads_nts.nf
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
process shuffle_reads_nts {
//publishDir "${params.output}/${name}/sh", mode: 'copy', pattern: "${name}_pos_ctg.fa"
label 'ruby'
input:
tuple val(name), file(reads)
output:
tuple val(name), file("${name}_shuffled.fastq")
script:
"""
#!/usr/bin/env ruby
fastq = File.open("${reads}", 'r')
shuffled = File.open("${name}_shuffled.fastq", 'w')
i = 0
out = ''
fastq.each do |line|
i += 1
if i == 1 || i == 3 || i == 4
out << line
end
if i == 2
out << line.chomp.split("").shuffle.join << "\\n"
end
if i == 4
shuffled << out if out.length > 1
out = ''
end
end
shuffled.close
fastq.close
"""
}

106 changes: 76 additions & 30 deletions phage.nf
Original file line number Diff line number Diff line change
Expand Up @@ -68,33 +68,35 @@ println " "}
*************/

include './modules/PPRmeta' params(output: params.output, cpus: params.cpus)
include './modules/databases/download_references' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/databases/phage_references_blastDB' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/databases/ppr_download_dependencies' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/databases/sourmash_download_DB' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/databases/virsorter_download_DB' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/deepvirfinder' params(output: params.output, cpus: params.cpus)
include './modules/download_references' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/fastqTofasta' params(output: params.output)
include './modules/filter_PPRmeta' params(output: params.output)
include './modules/filter_deepvirfinder' params(output: params.output)
include './modules/filter_marvel' params(output: params.output)
include './modules/filter_metaphinder' params(output: params.output)
include './modules/filter_sourmash' params(output: params.output)
include './modules/filter_tool_names' params(output: params.output)
include './modules/filter_virfinder' params(output: params.output)
include './modules/filter_virsorter' params(output: params.output, cpus: params.cpus)
include './modules/input_suffix_check' params(fastq: params.fastq)
include './modules/marvel' params(output: params.output, cpus: params.cpus)
include './modules/metaphinder' params(output: params.output, cpus: params.cpus)
include './modules/parse_reads.nf' params(output: params.output)
include './modules/ppr_download_dependencies' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/parser/filter_PPRmeta' params(output: params.output)
include './modules/parser/filter_deepvirfinder' params(output: params.output)
include './modules/parser/filter_marvel' params(output: params.output)
include './modules/parser/filter_metaphinder' params(output: params.output)
include './modules/parser/filter_sourmash' params(output: params.output)
include './modules/parser/filter_tool_names' params(output: params.output)
include './modules/parser/filter_virfinder' params(output: params.output)
include './modules/parser/filter_virsorter' params(output: params.output, cpus: params.cpus)
include './modules/parser/parse_reads.nf' params(output: params.output)
include './modules/r_plot.nf' params(output: params.output)
include './modules/r_plot_reads.nf' params(output: params.output)
include './modules/removeSmallReads' params(output: params.output)
include './modules/samtools' params(output: params.output)
include './modules/shuffle_reads_nts' params(output: params.output)
include './modules/sourmash' params(output: params.output)
include './modules/sourmash_download_DB' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)
include './modules/split_multi_fasta' params(output: params.output)
include './modules/upsetr.nf' params(output: params.output)
include './modules/virfinder' params(output: params.output, cpus: params.cpus)
include './modules/virsorter' params(output: params.output, cpus: params.cpus)
include './modules/virsorter_download_DB' params(cloudProcess: params.cloudProcess, cloudDatabase: params.cloudDatabase)

/*************
* DATABASES
Expand Down Expand Up @@ -134,7 +136,7 @@ workflow sourmash_database {
if (params.cloudProcess) {
db_preload = file("${params.cloudDatabase}/sourmash/phages.sbt.json.tar.gz")
if (db_preload.exists()) { db = db_preload }
else { sourmash_download_DB(phage_references()); db = sourmash_download_DB.out }
else { sourmash_download_DB(references); db = sourmash_download_DB.out }
}
emit: db
}
Expand All @@ -152,6 +154,20 @@ workflow phage_references {
emit: db
}

workflow phage_blast_DB {
get: references
main:
// local storage via storeDir
if (!params.cloudProcess) { phage_references_blastDB(references); db = phage_references_blastDB.out }
// cloud storage via db_preload.exists()
if (params.cloudProcess) {
db_preload = file("${params.cloudDatabase}/blast_phage_DB")
if (db_preload.exists()) { db = db_preload }
else { phage_references_blastDB(references); db = phage_references_blastDB.out }
}
emit: db
}

/*************
* SUB WORKFLOWS
*************/
Expand All @@ -165,6 +181,12 @@ workflow read_validation_wf {
get: fastq
main: fastqTofasta(removeSmallReads(fastq.splitFastq(by: 10000, file: true)))
emit: fastqTofasta.out
}

workflow read_shuffling_wf {
get: fastq
main: fastqTofasta(shuffle_reads_nts(removeSmallReads(fastq.splitFastq(by: 10000, file: true))))
emit: fastqTofasta.out
}

workflow sourmash_wf {
Expand Down Expand Up @@ -198,7 +220,7 @@ workflow marvel_wf {
}
else { marvel_results = Channel.from( [ 'deactivated', 'deactivated'] ) }
emit: marvel_results
}
}

workflow metaphinder_wf {
get: fasta
Expand All @@ -210,6 +232,17 @@ workflow metaphinder_wf {
emit: metaphinder_results
}

workflow metaphinder_own_DB_wf {
get: fasta
blast_db
main: if (!params.mp) {
filter_metaphinder_own_DB(metaphinder_own_DB(fasta, blast_db).groupTuple(remainder: true)) ;
metaphinder_results = filter_metaphinder_own_DB.out
}
else { metaphinder_results = Channel.from( [ 'deactivated', 'deactivated'] ) }
emit: metaphinder_results
}

workflow virfinder_wf {
get: fasta
main: if (!params.vf) {
Expand Down Expand Up @@ -251,39 +284,52 @@ workflow {
// input filter
fasta_validation_wf(fasta_input_ch)

// reference phages into DBs creations
phage_references() | ( sourmash_database & phage_blast_DB )

// gather results
results = virsorter_wf(fasta_validation_wf.out, virsorter_database())
.concat(sourmash_wf(fasta_validation_wf.out, sourmash_database(phage_references())))
.concat(marvel_wf(fasta_validation_wf.out))
.concat(sourmash_wf(fasta_validation_wf.out, sourmash_database.out))
.concat(metaphinder_wf(fasta_validation_wf.out))
.concat(metaphinder_own_DB_wf(fasta_validation_wf.out, phage_blast_DB.out))
.concat(deepvirfinder_wf(fasta_validation_wf.out))
.concat(virfinder_wf(fasta_validation_wf.out))
.concat(pprmeta_wf(fasta_validation_wf.out, ppr_dependecies()))
.filter { it != 'deactivated' } // removes deactivated tool channels
.groupTuple()

filter_tool_names(results)
filter_tool_names(results)
//plotting results
r_plot(filter_tool_names.out)
upsetr_plot(filter_tool_names.out)
//samtools
samtools(fasta_validation_wf.out.join((filter_tool_names.out)))




samtools(fasta_validation_wf.out.join((filter_tool_names.out)))
}

if (!params.fasta && params.fastq) {

// input filter
read_validation_wf(fastq_input_ch)

r_plot_reads(parse_reads( metaphinder_wf(read_validation_wf.out)
.concat(virfinder_wf(read_validation_wf.out))
.concat(pprmeta_wf(read_validation_wf.out, ppr_dependecies()))
.filter { it != 'deactivated' } // removes deactivated tool channels
.groupTuple()
) )
// reference phages into DBs creations
phage_references() | ( phage_blast_DB )

// gather results
results = metaphinder_wf(read_validation_wf.out)
.concat(metaphinder_own_DB_wf(read_validation_wf.out, phage_blast_DB.out))
.concat(virfinder_wf(read_validation_wf.out))
.concat(pprmeta_wf(read_validation_wf.out, ppr_dependecies()))
.filter { it != 'deactivated' } // removes deactivated tool channels
.groupTuple()

filter_tool_names(results)

//plotting results
r_plot_reads(parse_reads(results))
upsetr_plot(filter_tool_names.out)

//samtools
samtools(read_validation_wf.out.join(results))
}
}

Expand Down

0 comments on commit 11662e7

Please sign in to comment.