如何动态创建Snakemake规则？批量生成带不同参数的规则

我想创建多个仅输入参数不同的Snakemake规则，请问是否可以编写一个函数，在传入不同输入参数时动态生成这类规则？

预期实现示例代码 #

from snakemake import Rule, expand

def create_rule(rule_name, input_param):
    input_file = f"data/{input_param}.txt"
    output_file = f"results/{rule_name}_{input_param}.txt"
    rule_input = input(input_file)
    rule_output = output(output_file)
    rule_script = "my_script.py"
    rule_shell = f"python {rule_script} {input_file} {output_file}"
    return Rule(
        name=rule_name,
        input=rule_input,
        output=rule_output,
        shell=rule_shell
    )

我自己尝试的代码 #

fit_result_variation = lambda **kwargs: expand(rules.average_results.output, **modified_configuration(default_config, **kwargs))

variations = dict(
    a = np.arange(5,15,1),
    b = np.arange(5,25,1),
    c = [2,3,4,5],
)

for var_name, var_values in variations.items():
    rule_name = f"{var_name}_syst"
    rule_input = fit_result_variation(**{var_name: var_values})
    rule_output = expand(transient(f"{rule_name}_{{value}}.json", value=var_values))
    rule_run = """
        for input_file, output_file in zip(input, output):
            os.symlink(input_file, output_file)
    """
    globals()[rule_name] = rule(rule_name, input=rule_input, output=rule_output, run=rule_run)

当然可以，Snakemake完全支持通过函数或循环动态生成规则，你的思路是对的，但有几个细节需要调整，确保规则能被Snakemake正确识别：

关键调整点 #

1. 修正transient与expand的嵌套方式：transient()是标记临时文件的方法，应包裹在output()内部，而非和expand嵌套使用，否则Snakemake无法正确识别临时文件属性。
2. 补充依赖导入：run代码块中使用了os模块，需要显式导入，否则会报错。
3. 规范路径处理：创建软链接时最好使用绝对路径，避免因工作目录变化导致链接失效。

修正后的代码示例 #

import numpy as np
from snakemake import expand, rule, input, output, transient

# 假设你已定义modified_configuration和default_config
def modified_configuration(base_config, **kwargs):
    config = base_config.copy()
    config.update(kwargs)
    return config

default_config = {"a": 10, "b": 15, "c": 3}

fit_result_variation = lambda **kwargs: expand(rules.average_results.output, **modified_configuration(default_config, **kwargs))

variations = dict(
    a = np.arange(5,15,1),
    b = np.arange(5,25,1),
    c = [2,3,4,5],
)

for var_name, var_values in variations.items():
    rule_name = f"{var_name}_syst"
    rule_input = fit_result_variation(**{var_name: var_values})
    # 修正output的写法：transient放在output内部
    rule_output = output(expand(transient(f"{rule_name}_{{value}}.json"), value=var_values))
    # 添加os导入，并使用绝对路径创建软链接
    rule_run = """
        import os
        for input_file, output_file in zip(input, output):
            os.symlink(os.path.abspath(input_file), os.path.abspath(output_file))
    """
    # 将规则注册到全局命名空间，确保Snakemake能识别
    globals()[rule_name] = rule(
        name=rule_name,
        input=rule_input,
        output=rule_output,
        run=rule_run
    )

封装成函数的写法 #

如果想把规则生成逻辑封装成可复用的函数，也可以这样写：

from snakemake import rule, input, output, expand, transient

def create_symlink_rule(rule_name, input_files, output_pattern):
    rule_output = output(expand(transient(output_pattern)))
    rule_run = """
        import os
        for in_file, out_file in zip(input, output):
            os.symlink(os.path.abspath(in_file), os.path.abspath(out_file))
    """
    return rule(
        name=rule_name,
        input=input_files,
        output=rule_output,
        run=rule_run
    )

# 调用示例
for var_name, var_values in variations.items():
    rule_name = f"{var_name}_syst"
    inputs = fit_result_variation(**{var_name: var_values})
    output_pattern = f"{rule_name}_{{value}}.json"
    globals()[rule_name] = create_symlink_rule(rule_name, inputs, output_pattern)

验证规则 #

生成规则后，可以用以下命令检查是否所有规则都被Snakemake正确识别：

snakemake --list-rules

内容的提问来源于stack exchange，提问作者Ahmed Sameh Wagih