snakemake - как составить список входных файлов на основе предыдущего правила, которое выдает переменное количество файлов

Question

Скажем, я начинаю с таких файлов:

group_1_in.txt, group_2_in.txt, group_3_in.txt

Я обрабатываю их, используя правило, которое генерирует структуру каталогов, показанную ниже.

rule process_group_files:
    input: 'group_{num}_in.txt'
    output: directory('group_{num}')
    shell: "some_command {input} {output}'

## directory structure produced: 
group_1
    sample1_content.txt
    sample2_content.txt
    sample3_content.txt
group_2
    sample2_content.txt
    sample3_content.txt
    sample4_content.txt
group_3
    sample1_content.txt
    sample2_content.txt
    sample5_content.txt 

Затем у меня есть правило, которое обрабатывает их для агрегирования файлов по образцу:

rule aggregate_by_sample:
    input: expand('{group}/{sample}_content.txt')
    output: '{sample}_allcontent.txt'
    shell: "cat {input} | some_command > {output}"

Я ожидаю, что входные данные для этого правила будут:

group_1/sample1_content.txt, group_3/sample1_content.txt
group_1/sample2_content.txt, group_2/sample2_content.txt, group_3/sample2_content.txt
group_1/sample3_content.txt, group_2/sample3_content.txt
group_2/sample4_content.txt 
group_3/sample5_content.txt

и создайте следующие выходные файлы:

sample1_allcontent.txt
sample2_allcontent.txt
sample3_allcontent.txt
sample4_allcontent.txt
sample5_allcontent.txt

На данный момент я хочу работать с этими выходными файлами. Итак, правило для этого может быть примерно таким:

rule process_by_sample:
    input: <list of all sample_allcontent files>
    output: final_output.txt 
    shell: "cat {input} | some_other_command > {output}"

Мой вопрос таков: как я могу сказать snakemake ждать, пока он не завершит обработку всех файлов в правиле aggregate_by_sample, затем используйте этот набор выходных файлов для правила process_by_sample? Я изучил идею контрольных точек, сделав aggregate_by_sample контрольной точкой, но я должен использовать «каталог» в качестве вывода, так как я не знаю apriori сколько выходных файлов будет создано. Но я не могу этого сделать, потому что в именах моих выходных файлов используются символы подстановки, а snakemake жалуется, что Wildcards in input files cannot be determined from output files.

РЕДАКТИРОВАТЬ - Увидев ответ @ troy-comi, я понял, что упростил свою проблему. Я обновил свой вопрос, включив в него первое правило process_group_files. Все, что я знаю в начале конвейера, это сколько у меня групп и какой список подстановочных знаков «число».

Answer 1

Поскольку файлы уже существуют, вы можете использовать glob_wildcards для получения списка группы / семплов в файловой системе. Используя это, вы можете создать свои входные файлы с некоторой дополнительной обработкой.

Вот моя (непроверенная) идея:

wc =  glob_wildcards('{group}/{sample}_content.txt')
samples_to_group = {}
for samp, group in zip(wc.group, wc.sample):
    if samp not in samples_to_group:
        samples_to_group[samp] = []
    samples_to_group.append(group)

# now samples_to_group is a map of which groups are present for each sample

rule all:
    input: "final_output.txt"

rule aggregate_by_sample:
    input: expand('{group}/{sample}_content.txt', 
                  group=samples_to_group[wildcards.sample],
                  allow_missing=True)
    output: '{sample}_allcontent.txt'
    shell: "cat {input} | some_command > {output}"

rule process_by_sample:
    input: expand('{sample}_allcontent.txt', sample=samples_to_group.keys())
    output: final_output.txt 
    shell: "cat {input} | some_other_command > {output}"

Если другое правило создает файлы, вы должны использовать контрольные точки.

- РЕДАКТИРОВАТЬ, чтобы ответить на уточненный вопрос -

Я могу заставить это работать, только если вы знаете образцы заранее, не нужно групповое сопоставление образцов, просто у вас всего 5 образцов. ..

Настройка каталога со следующими файлами:

$ tail data/group_*.txt
==> data/group_1.txt <==
1
2
3

==> data/group_2.txt <==
2
3
4

==> data/group_3.txt <==
1
2
5

Тогда Snakefile с

wildcard_constraints:
    num="\d+"

groups = glob_wildcards('data/group_{num}.txt').num
samples = range(1, 6)

rule all:
    input: "final_output.txt"

checkpoint process_group_files:
    input: 'data/group_{num}.txt'
    output: directory('data/group_{num}')
    shell:
        'mkdir {output} \n'
        'for line in $(cat {input}) ; do echo "$line {input}" '
            '> {output}/${{line}}_content.txt ; '
        'done \n'
        'sleep 1'

def aggregate_input(wildcards):
    for num in groups:
        checkpoints.process_group_files.get(num=num).output

    grps = glob_wildcards(f'data/group_{{group}}/{wildcards.sample}_content.txt').group
    return expand('data/group_{group}/{sample}_content.txt',
            group=grps,
            sample=wildcards.sample)


rule aggregate_by_sample:
    input: aggregate_input
    output: 'data/agg/{sample}_allcontent.txt'
    shell: 'cat {input} > {output}'

rule process_by_sample:
    input: expand('data/agg/{sample}_allcontent.txt', sample=samples)
    output: 'final_output.txt'
    shell: 'cat {input} > {output}'

даст окончательный результат:

$ cat final_output.txt
1 data/group_1.txt
1 data/group_3.txt
2 data/group_1.txt
2 data/group_2.txt
2 data/group_3.txt
3 data/group_1.txt
3 data/group_2.txt
4 data/group_2.txt
5 data/group_3.txt

'magi c' вызывает контрольные точки с a для l oop, который является необходимой блокировкой. Опять же, это требует знания образцов раньше. Вы можете попробовать второй уровень контрольных точек, но это обычно не получается. Я также помню, что у кого-то еще были проблемы с контрольными точками в for l oop, поэтому он может сломаться в не игрушечном примере. КСТАТИ это змейка 5.10

Может оказаться, что проще будет честно разделить на два рабочих процесса (snakemake -s Snakefile1 && snakemake -s Snakefile2)!

Удачи!