如何在Nextflow中用if-else控制流程步骤并实现交互问询？

一、解决流程执行顺序问题 #

Nextflow的执行逻辑基于通道依赖，而非Python式的线性顺序执行。你直接编写的if-else会在工作流初始化阶段就执行，因此会和其他进程并行运行。要让判断逻辑在所有前置进程完成后触发，必须将逻辑与前置进程的输出通道绑定：

方案1：用subscribe监听通道完成事件 #

将判断逻辑放到FASTQC_check.out的subscribe回调中，只有当FASTQC_check的输出通道完成所有数据处理后，才会执行后续逻辑：

workflow {
    read_pairs_ch = Channel.fromFilePairs(params.reads)
    FASTQC(read_pairs_ch)
    trimmomatic(read_pairs_ch)
    FASTQC_1(trimmomatic.out)
    FASTQC_1.out
        .map { sample, fastqc -> sample }
        .collect()
        .set { ready_ch }
    FASTQC_check(ready_ch) 

    // 监听FASTQC_check输出通道的完成与错误事件
    FASTQC_check.out.subscribe(
        onNext: { is_ok ->
            if (is_ok) {
                Channel
                    .fromPath('/home/dungnguyen/genepanel/bin/data3.txt')
                    .splitText()
                    .view { println it }
            } else {
                println "404 .. something went wrong"
            }
        },
        onError: {
            println "404 .. something went wrong"
        }
    )
}

方案2：将判断逻辑封装为独立进程 #

把后续逻辑写成独立进程，让它依赖FASTQC_check.out，Nextflow会自动等待前置进程完成后再执行这个进程：

process PostProcessing {
    input:
    val is_ok

    script:
    if (is_ok) {
        """
        cat /home/dungnguyen/genepanel/bin/data3.txt
        """
    } else {
        """
        echo "404 .. something went wrong"
        """
    }
}

workflow {
    // 前置流程保持不变
    read_pairs_ch = Channel.fromFilePairs(params.reads)
    FASTQC(read_pairs_ch)
    trimmomatic(read_pairs_ch)
    FASTQC_1(trimmomatic.out)
    FASTQC_1.out
        .map { sample, fastqc -> sample }
        .collect()
        .set { ready_ch }
    FASTQC_check(ready_ch) 

    // 调用后置进程，自动等待FASTQC_check完成
    PostProcessing(FASTQC_check.out)
}

二、实现交互式质量确认问询 #

交互式提示需要在本地工作流进程中执行（不能放到远程/容器进程里），可以在质量检查完成后，通过通道回调读取用户输入：

workflow {
    read_pairs_ch = Channel.fromFilePairs(params.reads)
    FASTQC(read_pairs_ch)
    trimmomatic(read_pairs_ch)
    FASTQC_1(trimmomatic.out)

    // 解析FASTQC结果，判断样本质量是否过低
    FASTQC_1.out.map { sample, fastqc ->
        def quality_low = false
        // 读取FASTQC的summary.txt文件，判断是否有质量不合格项
        def summary_path = fastqc[0].toString().replace('.zip', '_fastqc/summary.txt')
        def summary_file = new File(summary_path)
        if (summary_file.exists()) {
            summary_file.eachLine { line ->
                if (line.contains('FAIL') && line.contains('Per base sequence quality')) {
                    quality_low = true
                }
            }
        }
        return [sample, quality_low]
    }.collect() // 收集所有样本的质量结果
     .subscribe { samples_quality ->
        def has_low_quality = samples_quality.any { it[1] }
        if (has_low_quality) {
            print "检测到部分样本质量较低，是否继续执行变异检测？(y/n): "
            def user_input = System.in.readLine().trim().toLowerCase()
            if (user_input == 'y') {
                println "继续执行变异检测..."
                // 在这里触发后续变异检测流程，比如 variant_calling(trimmomatic.out)
            } else {
                println "用户取消操作，工作流终止。"
                workflow.stop() // 终止工作流
            }
        } else {
            println "所有样本质量合格，继续执行..."
            // variant_calling(trimmomatic.out)
        }
    }

    // 其他流程逻辑...
}

注意事项 #

• 交互式输入仅在本地运行工作流时生效，集群或容器环境下无法获取用户输入，此时可改用参数预配置（如params.force_continue = false）替代。
• 解析FASTQC结果时，需根据实际输出文件路径调整代码，确保能正确读取质量指标。

内容的提问来源于stack exchange，提问作者Nguyễn Tiến Dũng