From 0ad64d4eb286847cb8058d9ba1b6b37180337a48 Mon Sep 17 00:00:00 2001 From: alejandrogzi Date: Thu, 3 Sep 2026 19:41:00 +0200 Subject: [PATCH 1/2] feat(annevo): add prediction, decoding, and scatter workflow Wrap the ANNEVO container as atomic modules, reuse FXSPLIT -H for chromosome and weighted scatter, and add a standalone workflow. --- assets/catalog/catalog.md | 10 + modules/nextflow/annevo/annotation/main.nf | 67 ++++ modules/nextflow/annevo/decoding/main.nf | 60 +++ modules/nextflow/annevo/prediction/main.nf | 66 +++ modules/nextflow/fxsplit/main.nf | 18 +- subworkflows/annevo/main.nf | 444 +++++++++++++++++++++ workflows/annevo/main.nf | 72 ++++ 7 files changed, 732 insertions(+), 5 deletions(-) create mode 100644 modules/nextflow/annevo/annotation/main.nf create mode 100644 modules/nextflow/annevo/decoding/main.nf create mode 100644 modules/nextflow/annevo/prediction/main.nf create mode 100644 subworkflows/annevo/main.nf create mode 100644 workflows/annevo/main.nf diff --git a/assets/catalog/catalog.md b/assets/catalog/catalog.md index 4abccc9..212851e 100644 --- a/assets/catalog/catalog.md +++ b/assets/catalog/catalog.md @@ -109,6 +109,16 @@ --- +## Gene annotation + +- [annevo/annotation](../../modules/nextflow/annevo/annotation/main.nf) — One-step ANNEVO annotation (prediction + decoding). Prefer the split modules for production. +- [annevo/prediction](../../modules/nextflow/annevo/prediction/main.nf) — ANNEVO nucleotide-level inference (`process_gpu`; CPU if no GPU profile). +- [annevo/decoding](../../modules/nextflow/annevo/decoding/main.nf) — Decode ANNEVO H5 predictions into GFF3 on CPU. +- [subworkflows/annevo](../../subworkflows/annevo/main.nf) — Scatter (none / chromosome / weighted), predict, decode, and gather GFF3. +- [workflows/annevo](../../workflows/annevo/main.nf) — Standalone ANNEVO pipeline. Bundled ANNEVO is non-commercial. + +--- + ## Transcript assembly & quantification - [aletsch](../../modules/nextflow/aletsch/main.nf) — Assemble RNA-seq transcripts using Aletsch. diff --git a/modules/nextflow/annevo/annotation/main.nf b/modules/nextflow/annevo/annotation/main.nf new file mode 100644 index 0000000..70d93ad --- /dev/null +++ b/modules/nextflow/annevo/annotation/main.nf @@ -0,0 +1,67 @@ +/* +Copyright (c) 2026 The Hiller Lab at the Senckenberg Gessellschaft für Naturforschung +Distributed under the terms of the Apache License, Version 2.0. +*/ + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + ANNEVO_ANNOTATION — One-step ANNEVO annotation (prediction + decoding). + + Useful for small genomes and for checking upstream behaviour. Prefer + ANNEVO_PREDICTION → ANNEVO_DECODING for scalable runs so the GPU is not + held during CPU decoding. +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +process ANNEVO_ANNOTATION { + tag "$meta.id" + label 'process_high' + label 'process_gpu' + + container "${ workflow.containerEngine == 'singularity' && !task.ext.singularity_pull_docker_container ? + '' : + 'ghcr.io/hillerlab/annevo:latest' }" + + input: + tuple val(meta), path(fasta) + val lineage + + output: + tuple val(meta), path("*.gff3"), emit: gff + path "versions.yml", emit: versions + + when: + task.ext.when == null || task.ext.when + + script: + def args = [task.ext.args ?: '', meta.annevo_args ?: ''].findAll { it }.join(' ') + def prefix = task.ext.prefix ?: "${meta.id}" + def model = task.ext.model ?: (meta.model ?: "/opt/annevo/saved_model/ANNEVO_${lineage}.pt") + def workers = Math.max(1, (task.cpus as int) - 1) + """ + annevo annotation \\ + -g ${fasta} \\ + -m ${model} \\ + -l ${lineage} \\ + -o ${prefix}.gff3 \\ + -t ${task.cpus} \\ + --num_workers ${workers} \\ + ${args} + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ + + stub: + def prefix = task.ext.prefix ?: "${meta.id}" + """ + touch ${prefix}.gff3 + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ +} diff --git a/modules/nextflow/annevo/decoding/main.nf b/modules/nextflow/annevo/decoding/main.nf new file mode 100644 index 0000000..cd84a66 --- /dev/null +++ b/modules/nextflow/annevo/decoding/main.nf @@ -0,0 +1,60 @@ +/* +Copyright (c) 2026 The Hiller Lab at the Senckenberg Gessellschaft für Naturforschung +Distributed under the terms of the Apache License, Version 2.0. +*/ + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + ANNEVO_DECODING — Convert ANNEVO nucleotide predictions into GFF3. + + CPU-only. Keep this process separate from prediction so GPU allocations + can be released as soon as the H5 file is written. +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +process ANNEVO_DECODING { + tag "$meta.id" + label 'process_high' + + container "${ workflow.containerEngine == 'singularity' && !task.ext.singularity_pull_docker_container ? + '' : + 'ghcr.io/hillerlab/annevo:latest' }" + + input: + tuple val(meta), path(fasta), path(predictions) + + output: + tuple val(meta), path("*.gff3"), emit: gff + path "versions.yml", emit: versions + + when: + task.ext.when == null || task.ext.when + + script: + def args = [task.ext.args ?: '', meta.annevo_decoding_args ?: ''].findAll { it }.join(' ') + def prefix = task.ext.prefix ?: "${meta.id}" + """ + annevo decoding \\ + -g ${fasta} \\ + -p ${predictions} \\ + -o ${prefix}.gff3 \\ + -t ${task.cpus} \\ + ${args} + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ + + stub: + def prefix = task.ext.prefix ?: "${meta.id}" + """ + touch ${prefix}.gff3 + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ +} diff --git a/modules/nextflow/annevo/prediction/main.nf b/modules/nextflow/annevo/prediction/main.nf new file mode 100644 index 0000000..41952db --- /dev/null +++ b/modules/nextflow/annevo/prediction/main.nf @@ -0,0 +1,66 @@ +/* +Copyright (c) 2026 The Hiller Lab at the Senckenberg Gessellschaft für Naturforschung +Distributed under the terms of the Apache License, Version 2.0. +*/ + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + ANNEVO_PREDICTION — Nucleotide-level ANNEVO inference. + + Thin wrapper around `annevo prediction`. GPU is used when `-profile gpu` + exposes a device; otherwise PyTorch falls back to CPU. Does not insert + biological defaults such as `--overlap_pred`. +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +process ANNEVO_PREDICTION { + tag "$meta.id" + label 'process_high' + label 'process_gpu' + + container "${ workflow.containerEngine == 'singularity' && !task.ext.singularity_pull_docker_container ? + '' : + 'ghcr.io/hillerlab/annevo:latest' }" + + input: + tuple val(meta), path(fasta) + val lineage + + output: + tuple val(meta), path("*.h5"), emit: predictions + path "versions.yml", emit: versions + + when: + task.ext.when == null || task.ext.when + + script: + def args = [task.ext.args ?: '', meta.annevo_args ?: ''].findAll { it }.join(' ') + def prefix = task.ext.prefix ?: "${meta.id}" + def model = task.ext.model ?: (meta.model ?: "/opt/annevo/saved_model/ANNEVO_${lineage}.pt") + def workers = Math.max(1, (task.cpus as int) - 1) + """ + annevo prediction \\ + -g ${fasta} \\ + -m ${model} \\ + -l ${lineage} \\ + -p ${prefix}.h5 \\ + --num_workers ${workers} \\ + ${args} + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ + + stub: + def prefix = task.ext.prefix ?: "${meta.id}" + """ + touch ${prefix}.h5 + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ +} diff --git a/modules/nextflow/fxsplit/main.nf b/modules/nextflow/fxsplit/main.nf index c50b985..ca0c781 100644 --- a/modules/nextflow/fxsplit/main.nf +++ b/modules/nextflow/fxsplit/main.nf @@ -7,7 +7,8 @@ Distributed under the terms of the Apache License, Version 2.0. ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ FXSPLIT — Split FASTX/FASTQ reads into chunks for parallel processing. Partitions read files into smaller chunks to enable parallel processing - across multiple CPUs. + across multiple CPUs. Set meta.headers or task.ext.headers to use + fxsplit -H (one output file per FASTA record). ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ */ @@ -35,15 +36,18 @@ process FXSPLIT { def args = task.ext.args ?: '' def prefix = task.ext.prefix ?: "${meta.id}" def chunks = task.ext.chunks ?: 400000 + def header_mode = (meta.headers == true) || (task.ext.headers ? true : false) + def mode_flag = header_mode ? '-H' : "-c ${chunks}" + def suffix_flag = header_mode ? '' : "--suffix ${prefix}" def gzip = reads.name.endsWith('.gz') ? true : false """ fxsplit \\ $args \\ -f $reads \\ - -c $chunks \\ + ${mode_flag} \\ -t $task.cpus \\ -C \\ - --suffix ${prefix} + ${suffix_flag} if [ ${meta.singleton} == true ]; then for f in chunks/*fasta.gz; do @@ -73,10 +77,14 @@ process FXSPLIT { if [ $gzip == true ]; then mkdir chunks/gz - mv chunks/*fast*.gz chunks/gz + mv chunks/*fast*.gz chunks/gz 2>/dev/null || mv chunks/*.gz chunks/gz else mkdir chunks/fx - mv chunks/*fast* chunks/fx + if [ "${header_mode}" = "true" ]; then + find chunks -maxdepth 1 -type f -exec mv {} chunks/fx/ \\; + else + mv chunks/*fast* chunks/fx + fi fi cat <<-END_VERSIONS > versions.yml diff --git a/subworkflows/annevo/main.nf b/subworkflows/annevo/main.nf new file mode 100644 index 0000000..0e51c91 --- /dev/null +++ b/subworkflows/annevo/main.nf @@ -0,0 +1,444 @@ +/* +Copyright (c) 2026 The Hiller Lab at the Senckenberg Gessellschaft für Naturforschung +Distributed under the terms of the Apache License, Version 2.0. +*/ + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + IMPORT LOCAL MODULES/SUBWORKFLOWS +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +include { FXSPLIT } from '../../modules/nextflow/fxsplit/main.nf' +include { ANNEVO_PREDICTION } from '../../modules/nextflow/annevo/prediction/main.nf' +include { ANNEVO_DECODING } from '../../modules/nextflow/annevo/decoding/main.nf' + +def ANNEVO_LINEAGES = ['Mammalia', 'Insecta', 'Aves', 'Actinopteri', 'Magnoliopsida', 'Fungi'] as Set +def ANNEVO_OVERLAP_LINEAGES = ['Mammalia', 'Actinopteri'] as Set +def ANNEVO_SCATTER_MODES = ['none', 'chromosome', 'weighted'] as Set + +def annevoBool(value) { + if (value == null) { + return null + } + if (value instanceof Boolean) { + return value + } + def text = value.toString().trim().toLowerCase() + if (text in ['true', '1', 'yes', 'y', 'on']) { + return true + } + if (text in ['false', '0', 'no', 'n', 'off', '']) { + return false + } + error "Cannot parse boolean value: ${value}" +} + +def annevoPackBins(records, nBins) { + def n = Math.max(1, nBins as int) + n = Math.min(n, records.size()) + def bins = (0.. + def target = bins.min { a, b -> a.total <=> b.total ?: a.order <=> b.order } + target.records << rec + target.total += rec.length as long + } + bins.each { bin -> bin.records = bin.records.toSorted { it.order } } + return bins.findAll { !it.records.isEmpty() } +} + +def annevoResolveRecord(seqid, files) { + def exact = files.find { it.baseName == seqid } + if (exact) { + return exact + } + def prefixed = files.find { it.baseName.startsWith(seqid + '.') || it.baseName.startsWith(seqid + '_') } + if (prefixed) { + return prefixed + } + def safe = seqid.replaceAll(/[^A-Za-z0-9._-]/, '_') + def sanitized = files.find { it.baseName == safe || it.baseName.startsWith(safe + '.') } + if (sanitized) { + return sanitized + } + error "No FXSPLIT output for FASTA record '${seqid}' among ${files*.name}" +} + +def annevoParseManifest(path) { + def records = [] + path.withReader { reader -> + def header = reader.readLine() + reader.eachLine { line -> + if (!line) { + return + } + def parts = line.split('\t', 3) + records << [ + order : parts[0] as int, + seqid : parts[1], + length: parts[2] as long, + ] + } + } + return records +} + +process ANNEVO_MANIFEST { + tag "$meta.id" + label 'process_single' + + container "${ workflow.containerEngine == 'singularity' && !task.ext.singularity_pull_docker_container ? + '' : + 'ghcr.io/hillerlab/annevo:latest' }" + + input: + tuple val(meta), path(fasta) + + output: + tuple val(meta), path("records.tsv"), emit: manifest + path "versions.yml", emit: versions + + script: + """ + awk 'BEGIN { OFS="\\t"; print "order", "seqid", "length" } + /^>/ { + if (seqid != "") print order, seqid, len + if (seqid != "") order++ + seqid = substr(\$1, 2) + len = 0 + next + } + { + gsub(/\\r/, "") + len += length(\$0) + } + END { + if (seqid != "") print order, seqid, len + }' ${fasta} > records.tsv + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + awk: \$( { awk --version || awk -W version || echo awk; } 2>/dev/null | head -n 1 ) + END_VERSIONS + """ + + stub: + """ + echo -e "order\\tseqid\\tlength\\n0\\tchr1\\t1" > records.tsv + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + awk: \$( { awk --version || awk -W version || echo awk; } 2>/dev/null | head -n 1 ) + END_VERSIONS + """ +} + +process ANNEVO_CAT { + tag "$meta.id" + label 'process_single' + + container "${ workflow.containerEngine == 'singularity' && !task.ext.singularity_pull_docker_container ? + '' : + 'ghcr.io/hillerlab/annevo:latest' }" + + input: + tuple val(meta), path(fastas) + + output: + tuple val(meta), path("*.fa"), emit: fasta + path "versions.yml", emit: versions + + script: + prefix = task.ext.prefix ?: "${meta.id}" + """ + cat ${fastas} > ${prefix}.fa + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + cat: \$( cat --version | head -n 1 | sed 's/.* //' ) + END_VERSIONS + """ + + stub: + prefix = task.ext.prefix ?: "${meta.id}" + """ + touch ${prefix}.fa + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + cat: na + END_VERSIONS + """ +} + +process ANNEVO_GATHER { + tag "$meta.id" + label 'process_single' + + container "${ workflow.containerEngine == 'singularity' && !task.ext.singularity_pull_docker_container ? + '' : + 'ghcr.io/hillerlab/annevo:latest' }" + + input: + tuple val(meta), path(gffs), path(manifest) + + output: + tuple val(meta), path("*.annevo.gff3"), emit: gff + path "versions.yml", emit: versions + + script: + prefix = task.ext.prefix ?: "${meta.id}" + """ + python - "${manifest}" "${prefix}.annevo.gff3" ${gffs} <<'PY' +import sys +from pathlib import Path + +manifest = Path(sys.argv[1]) +output = Path(sys.argv[2]) +gff_paths = [Path(path) for path in sys.argv[3:]] + +order = [] +with manifest.open() as handle: + next(handle, None) + for line in handle: + line = line.rstrip("\\n") + if not line: + continue + order.append(line.split("\\t", 2)[1]) + +header = [] +blocks = {} +banner = "# ----- prediction on sequence number" +name_tag = "name = " + +def consume(path): + current_id = None + current = [] + + def flush(): + nonlocal current_id, current + if current_id is not None: + blocks[current_id] = "".join(current) + current_id = None + current = [] + + with path.open() as handle: + for line in handle: + if ( + line.startswith("# This output was generated with ANNEVO") + or line.startswith("# ANNEVO is an ab initio") + or line.startswith("# Citation:") + ): + if line not in header: + header.append(line) + continue + if line.startswith(banner): + flush() + start = line.rfind(name_tag) + if start < 0: + raise SystemExit(f"cannot parse sequence banner: {line!r}") + current_id = line[start + len(name_tag):].split(")", 1)[0].strip() + current = [line] + continue + if current_id is None: + continue + current.append(line) + flush() + +for path in gff_paths: + consume(path) + +written = set() +with output.open("w") as handle: + handle.writelines(header) + for seqid in order: + text = blocks.get(seqid) + if text: + handle.write(text) + written.add(seqid) + for seqid, text in blocks.items(): + if seqid not in written: + handle.write(text) +PY + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ + + stub: + prefix = task.ext.prefix ?: "${meta.id}" + """ + touch ${prefix}.annevo.gff3 + + cat <<-END_VERSIONS > versions.yml + "${task.process}": + annevo: \$( annevo --version | sed 's/annevo //g' ) + END_VERSIONS + """ +} + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + ANNEVO subworkflow +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +workflow ANNEVO { + take: + genome // channel: [ val(meta), path(fasta) ] + lineage // value: Mammalia|Insecta|Aves|Actinopteri|Magnoliopsida|Fungi + scatter_mode // value: none|chromosome|weighted + scatter_bins // value: integer, used when scatter_mode == weighted + overlap_pred // value: boolean or null (null → lineage default) + + main: + ch_versions = Channel.empty() + + def lin = lineage.toString() + if (!ANNEVO_LINEAGES.contains(lin)) { + error "Unsupported ANNEVO lineage '${lin}'. Choose one of: ${ANNEVO_LINEAGES.join(', ')}" + } + + def mode = scatter_mode.toString() + if (!ANNEVO_SCATTER_MODES.contains(mode)) { + error "Unsupported ANNEVO scatter mode '${mode}'. Choose one of: ${ANNEVO_SCATTER_MODES.join(', ')}" + } + + def parsed_overlap = annevoBool(overlap_pred) + def use_overlap = parsed_overlap == null ? ANNEVO_OVERLAP_LINEAGES.contains(lin) : parsed_overlap + + def n_bins = 8 + if (scatter_bins != null && scatter_bins.toString().trim()) { + n_bins = scatter_bins as int + } + if (mode == 'weighted' && n_bins < 1) { + error "ANNEVO weighted scatter requires --annevo_bins >= 1" + } + + genome + .map { meta, fasta -> + def sample = meta.id ?: fasta.baseName + def flags = [] + if (use_overlap) { + flags << '--overlap_pred' + } + if (meta.annevo_args) { + flags << meta.annevo_args + } + [ + meta + [ + sample : sample, + lineage : lin, + annevo_args: flags.join(' '), + ], + fasta, + ] + } + .set { ch_prepared } + + ANNEVO_MANIFEST(ch_prepared) + ch_versions = ch_versions.mix(ANNEVO_MANIFEST.out.versions) + + if (mode == 'none') { + ch_prepared + .map { meta, fasta -> [meta + [id: meta.sample, chunk: 'all', order: 0], fasta] } + .set { ch_pieces } + } else { + FXSPLIT( + ch_prepared.map { meta, fasta -> [meta + [headers: true], fasta] } + ) + ch_versions = ch_versions.mix(FXSPLIT.out.versions) + + FXSPLIT.out.fastx + .mix(FXSPLIT.out.fastx_gz) + .map { meta, files -> tuple(meta.sample, meta, files instanceof List ? files : [files]) } + .join( + ANNEVO_MANIFEST.out.manifest.map { meta, man -> tuple(meta.sample, man) } + ) + .flatMap { sample, meta, files, man -> + def records = annevoParseManifest(man) + if (records.isEmpty()) { + error "No FASTA records found for ${sample}" + } + if (mode == 'weighted') { + return annevoPackBins(records, n_bins).collect { bin -> + def chunk = sprintf('bin%03d', bin.order + 1) + [ + meta + [ + id : "${meta.sample}.${chunk}", + chunk : chunk, + order : bin.order, + records: bin.records.collect { it.seqid }, + ], + bin.records.collect { rec -> annevoResolveRecord(rec.seqid, files) }, + ] + } + } + records.collect { rec -> + [ + meta + [ + id : "${meta.sample}.${rec.seqid}", + chunk : rec.seqid, + order : rec.order, + records: [rec.seqid], + ], + annevoResolveRecord(rec.seqid, files), + ] + } + } + .set { ch_split } + + if (mode == 'weighted') { + ANNEVO_CAT(ch_split) + ch_versions = ch_versions.mix(ANNEVO_CAT.out.versions) + ANNEVO_CAT.out.fasta.set { ch_pieces } + } else { + ch_split.set { ch_pieces } + } + } + + ANNEVO_PREDICTION(ch_pieces, lin) + ch_versions = ch_versions.mix(ANNEVO_PREDICTION.out.versions) + + ch_pieces + .join(ANNEVO_PREDICTION.out.predictions) + .set { ch_decode } + + ANNEVO_DECODING(ch_decode) + ch_versions = ch_versions.mix(ANNEVO_DECODING.out.versions) + + ANNEVO_DECODING.out.gff + .map { meta, gff -> tuple(meta.sample, meta, gff) } + .groupTuple() + .map { sample, metas, gffs -> + def paired = [metas, gffs].transpose().sort { it[0].order } + def first = paired[0][0] + [ + [ + id : sample, + sample : sample, + lineage: first.lineage, + ], + paired.collect { it[1] }, + ] + } + .set { ch_gathered } + + ch_gathered + .map { meta, gffs -> tuple(meta.id, meta, gffs) } + .join( + ANNEVO_MANIFEST.out.manifest.map { meta, man -> tuple(meta.sample, man) } + ) + .map { sample, meta, gffs, man -> [meta, gffs, man] } + .set { ch_gather_input } + + ANNEVO_GATHER(ch_gather_input) + ch_versions = ch_versions.mix(ANNEVO_GATHER.out.versions) + + emit: + gff = ANNEVO_GATHER.out.gff + predictions = ANNEVO_PREDICTION.out.predictions + versions = ch_versions +} diff --git a/workflows/annevo/main.nf b/workflows/annevo/main.nf new file mode 100644 index 0000000..08a653e --- /dev/null +++ b/workflows/annevo/main.nf @@ -0,0 +1,72 @@ +/* +Copyright (c) 2026 The Hiller Lab at the Senckenberg Gessellschaft für Naturforschung +Distributed under the terms of the Apache License, Version 2.0. +*/ + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + IMPORT LOCAL MODULES/SUBWORKFLOWS +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +include { ANNEVO as RUN } from '../../subworkflows/annevo/main.nf' +include { GENOME } from '../../subworkflows/genome/main.nf' + +params.genome = null +params.lineage = null +params.annevo_scatter = 'chromosome' +params.annevo_bins = 8 +params.annevo_overlap_pred = null +params.annevo_prediction_args = '' +params.annevo_decoding_args = '' + +/* +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + WORKFLOWS +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +*/ + +workflow ANNEVO { + main: + if (!params.genome) { + error "Missing required --genome" + } + if (!params.lineage) { + error "Missing required --lineage" + } + + def scatter = (params.annevo_scatter ?: 'chromosome').toString() + def bins = params.annevo_bins ?: 8 + def overlap = params.annevo_overlap_pred + + GENOME(params.genome) + + GENOME.out.genome + .map { fasta -> + [ + [ + id : fasta.baseName, + annevo_args : params.annevo_prediction_args ?: '', + annevo_decoding_args : params.annevo_decoding_args ?: '', + ], + fasta, + ] + } + .set { ch_input } + + RUN( + ch_input, + params.lineage, + scatter, + bins, + overlap + ) + + emit: + gff = RUN.out.gff + predictions = RUN.out.predictions +} + +workflow { + ANNEVO() +} From 2dce02bccf75e326a5a72c7ae53158fdf3899d55 Mon Sep 17 00:00:00 2001 From: alejandrogzi Date: Thu, 3 Sep 2026 20:04:57 +0200 Subject: [PATCH 2/2] fix(annevo): move top-level constants into functions for nextflow lint The linter rejects script-level assignments mixed with includes. Replace sprintf with Groovy padLeft as well. --- subworkflows/annevo/main.nf | 28 +++++++++++++++++++--------- 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/subworkflows/annevo/main.nf b/subworkflows/annevo/main.nf index 0e51c91..18e3dfc 100644 --- a/subworkflows/annevo/main.nf +++ b/subworkflows/annevo/main.nf @@ -13,9 +13,17 @@ include { FXSPLIT } from '../../modules/nextflow/fxsplit/main.nf' include { ANNEVO_PREDICTION } from '../../modules/nextflow/annevo/prediction/main.nf' include { ANNEVO_DECODING } from '../../modules/nextflow/annevo/decoding/main.nf' -def ANNEVO_LINEAGES = ['Mammalia', 'Insecta', 'Aves', 'Actinopteri', 'Magnoliopsida', 'Fungi'] as Set -def ANNEVO_OVERLAP_LINEAGES = ['Mammalia', 'Actinopteri'] as Set -def ANNEVO_SCATTER_MODES = ['none', 'chromosome', 'weighted'] as Set +def annevoLineages() { + return ['Mammalia', 'Insecta', 'Aves', 'Actinopteri', 'Magnoliopsida', 'Fungi'] +} + +def annevoOverlapLineages() { + return ['Mammalia', 'Actinopteri'] +} + +def annevoScatterModes() { + return ['none', 'chromosome', 'weighted'] +} def annevoBool(value) { if (value == null) { @@ -297,17 +305,19 @@ workflow ANNEVO { ch_versions = Channel.empty() def lin = lineage.toString() - if (!ANNEVO_LINEAGES.contains(lin)) { - error "Unsupported ANNEVO lineage '${lin}'. Choose one of: ${ANNEVO_LINEAGES.join(', ')}" + def lineages = annevoLineages() + if (!lineages.contains(lin)) { + error "Unsupported ANNEVO lineage '${lin}'. Choose one of: ${lineages.join(', ')}" } def mode = scatter_mode.toString() - if (!ANNEVO_SCATTER_MODES.contains(mode)) { - error "Unsupported ANNEVO scatter mode '${mode}'. Choose one of: ${ANNEVO_SCATTER_MODES.join(', ')}" + def scatter_modes = annevoScatterModes() + if (!scatter_modes.contains(mode)) { + error "Unsupported ANNEVO scatter mode '${mode}'. Choose one of: ${scatter_modes.join(', ')}" } def parsed_overlap = annevoBool(overlap_pred) - def use_overlap = parsed_overlap == null ? ANNEVO_OVERLAP_LINEAGES.contains(lin) : parsed_overlap + def use_overlap = parsed_overlap == null ? annevoOverlapLineages().contains(lin) : parsed_overlap def n_bins = 8 if (scatter_bins != null && scatter_bins.toString().trim()) { @@ -364,7 +374,7 @@ workflow ANNEVO { } if (mode == 'weighted') { return annevoPackBins(records, n_bins).collect { bin -> - def chunk = sprintf('bin%03d', bin.order + 1) + def chunk = "bin${(bin.order + 1).toString().padLeft(3, '0')}" [ meta + [ id : "${meta.sample}.${chunk}",