Snakemake生命游戏工作流报错：gridEmpty()参数不匹配求助

问题场景 #

基于Snakemake构建生命游戏工作流，目标实现条件迭代：每完成一个时间步的网格计算后，检查网格是否为空，若为空则停止后续时间步计算，用已生成的所有时间步画面渲染视频。当前通过checkpoint和自定义input函数gridEmpty()实现逻辑，但运行时触发参数错误。

错误信息 #

Error: TypeError: gridEmpty() takes 0 positional arguments but 1 was given Wildcards:

Traceback:

现有Snakefile代码 #

configfile: "config_exercise_2.yaml"
workdir: "../"

import numpy as np

Time = 1
def gridEmpty():
    global Time
    with open(checkpoints.CalculateNextTimeStep.get(time=Time).output[0],"rb") as f:
        if np.all(np.load(f) == 0): #grid is empty, no more timesteps required
            return [f"pictures/picture_t-{i}.jpg" for i in range(0,Time+1,1)]
        else: #grid is not empty, more timesteps required
            Time += 1
            checkpoints.CalculateNextTimeStep.get(time=Time)
                                                   
rule RenderVideo:
    input:
        gridEmpty
    output:
        "video/clip.mp4"
    params:
        framerate = config["FrameRate"]
    shell:
        "cat {input} | ffmpeg -framerate {params.framerate} -f image2pipe -i - {output}"
        
rule Displaytimestep:
    input:
        "arrays/array_t-{time}.npy"
    output:
        "pictures/picture_t-{time}.jpg"
    conda:
        "../environment/environment.yaml"
    params:
        timeStep = lambda wildcards: wildcards.time
    script:
        "../scripts/DisplayTimeStep.py"
    
checkpoint CalculateNextTimeStep:
    input:
        lambda wildcards: f"arrays/array_t-{int(wildcards.time)-1}.npy"
    output:
        "arrays/array_t-{time}.npy"
    params:
        GridSize = config["GridSize"]
    conda:
        "../environment/environment.yaml"
    script:
        "../scripts/CalculateNextTimeStep.py"

解决思路 #

Snakemake的规则input中指定的自定义函数，无论当前规则是否包含通配符，都会自动传入一个wildcards参数。你的gridEmpty()函数没有定义参数，导致触发类型错误。

只需修改函数定义，添加wildcards参数（即使函数内部不需要使用这个参数）即可解决：

def gridEmpty(wildcards):  # 添加wildcards参数
    global Time
    with open(checkpoints.CalculateNextTimeStep.get(time=Time).output[0],"rb") as f:
        if np.all(np.load(f) == 0): #grid is empty, no more timesteps required
            return [f"pictures/picture_t-{i}.jpg" for i in range(0,Time+1,1)]
        else: #grid is not empty, more timesteps required
            Time += 1
            checkpoints.CalculateNextTimeStep.get(time=Time)

这个修改让函数符合Snakemake的调用规范，就能正常执行空网格检测的逻辑了。

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