如何在完成并行通配符规则时在 Snakefile 条件中制作规则 "all"

Question

我有一些包含音频 SPH 样本的 TGZ 文件，我在 snakemake 中将其解压如下：

tgz_files = ["a.tgz", "b.tgz"]
tgz_dirs = ["a", "b"]
rule untar_tgz_files:
     input:
        tgz_files
     output:
        directory(tgz_dirs)
     shell:
        tar -xzvf {input}

直到解压后我才知道 SPH 示例文件的名称。然后我有一个将 SPH 文件转换为 WAV 文件的规则，如下所示：

rule sph_to_wav:
     input:
        "{root}/{filename}.sph"
     output:
        "{root}_wav/{filename}.wav"
     shell:
        sox -t sph {input} -b 16  -t wav {output}

我希望我的 Snakefile 运行这两个步骤（解压和转换），事先不知道 TGZ 档案中 SPH 文件的确切名称。我需要这样的东西来标记 sph_to_wav 规则的完成：

rule sph_to_wav_finished:
     input:
        "{root}_wav/{filename}.wav"
     output:
        "sph_to_wav_finished.txt"

然后我想在这两个过程中设置条件 rule all：

rule all:
     input:
         tgz_dirs, "sph_to_wav_finished.txt"

但是，我收到错误消息：

Building DAG of jobs...
MissingInputException in Snakefile:
Missing input files for rule all:
sph_to_wav_finished.txt

我该如何写才能

Snakemake 没有抱怨，运行s 解压和 sph 到 wav
解压后将 sph 运行到 wav

?

Answer 1

这听起来像是 use-case 对应 checkpoint。由于规则 untar_tgz_files 会生成事先不知道的文件，因此您可以将其转换为 checkpoint:

checkpoint untar_tgz_files:
    ... # everything defined as in a regular rule

这将告诉 snakemake 一旦此 checkpoint 完成，DAG 需要 re-evaluated 以考虑创建的新文件。

下游规则需要找出新文件，因此通常您会执行某种 glob.glob 来获取新文件列表。这是一个粗略的想法，但您可能需要对其进行微调：

def list_new_files(wildcards):
    output_dir = checkpoints. untar_tgz_files.get(sample=wildcards.sample).output
    # you will also want to parse "root" here, skipping it for simplicity
    filenames, _ = glob_wildcards(output_dir+"/{filename}.sph")
    new_files = expand("{filename}.wav", filename=filenames)
    return new_files

最后，收集所有已翻译的文件：

rule sph_to_wav_finished:
     input:
        list_new_files,
     output:
        "sph_to_wav_finished.txt"

如何在完成并行通配符规则时在 Snakefile 条件中制作规则 "all"

How to make rule "all" in Snakefile condition on completion of parallel wildcard rule

python

glob

snakemake

directed-acyclic-graphs