Snakemake优化写法咨询：如何提前检测inputs.txt是否存在

需求与问题 #

需要将存储真实输入路径的inputs.txt作为某条rule的输入之一，当前写法在inputs.txt不存在时，会在get_inputs函数的open()调用时才抛出异常，希望在进入该函数前就触发报错。

原代码如下：

rule all:
    input: "test.txt"

def get_inputs(wildcards):
    fh = open("inputs.txt")
    return fh.read().splitlines()

rule test:
    input:
        "inputs.txt",  # 无法提前触发异常
        additional=get_inputs
    output:
        "test.txt"
    shell:
        "echo {input.additional} > {output}"

解决方案 #

方法1：通过命名输入+函数参数传递让Snakemake提前检查 #

核心是让inputs.txt作为rule的命名输入，让函数通过input参数获取文件路径，这样Snakemake会先验证该文件的存在性，再执行函数：

rule all:
    input: "test.txt"

def get_inputs(wildcards, input):
    # 通过input参数获取已被Snakemake验证过的文件路径
    with open(input.input_list, 'r') as fh:
        return fh.read().splitlines()

rule test:
    input:
        # 命名输入，让Snakemake提前检查文件存在性
        input_list="inputs.txt",
        additional=get_inputs
    output:
        "test.txt"
    shell:
        "echo {input.additional} > {output}"

方法2：全局提前检查（适用于整个流程都依赖该文件的场景） #

在snakefile开头直接检查文件是否存在，加载阶段就触发报错：

import os

# 全局检查，加载snakefile时就验证
if not os.path.exists("inputs.txt"):
    raise FileNotFoundError("Error: inputs.txt 文件不存在！")

rule all:
    input: "test.txt"

def get_inputs(wildcards):
    with open("inputs.txt") as fh:
        return fh.read().splitlines()

rule test:
    input:
        "inputs.txt",
        additional=get_inputs
    output:
        "test.txt"
    shell:
        "echo {input.additional} > {output}"

原理说明 #

原代码中，Snakemake的执行顺序是先调用get_inputs生成动态输入，再检查所有输入文件的存在性，因此异常会在函数内部抛出。修改后的方法1通过命名输入让Snakemake优先验证inputs.txt的存在，再执行函数；方法2则在流程加载阶段直接做检查，都能实现“进入函数前触发报错”的需求。

内容的提问来源于stack exchange，提问作者Jindra Helcl