如何将Nextflow进程输出的多文件对转为独立队列条目？

现有问题 #

现有Nextflow的TileImages进程调用Python脚本生成数量不固定的wsi/mask .npy文件对，但当前输出仅返回元数据，无法将每个文件对与对应的run_id、group_id、sample_id绑定为独立输出条目。

解决方案 #

需要同时修改Python脚本和Nextflow进程配置，通过生成配对清单文件让Nextflow识别每个独立的文件对，并绑定元数据。

1. 修改Python脚本 #

• 添加--prefix参数支持，用于生成统一前缀的文件和配对清单：

import argparse
import numpy as np

# 添加参数解析
parser = argparse.ArgumentParser()
parser.add_argument('--image', required=True, help='输入图像路径')
parser.add_argument('--masks', required=True, help='输入掩码路径')
parser.add_argument('--prefix', required=True, help='输出文件前缀')
args = parser.parse_args()

# 原有的wsi_patch_extractor和mask_patch_extractor逻辑保持不变
# ...（此处省略原有图像/掩码提取代码）

# 保存文件对并记录路径
i = 0
patch_pairs = []
for wsi_patch, mask_patch in zip(wsi_patch_extractor, mask_patch_extractor):
    wsi_path = f"{args.prefix}_wsi_{i}.npy"
    mask_path = f"{args.prefix}_mask_{i}.npy"
    np.save(wsi_path, wsi_patch)
    np.save(mask_path, mask_patch)
    patch_pairs.append( (wsi_path, mask_path) )
    i += 1

# 生成配对清单文件，每行记录一对wsi和mask的路径
with open(f"{args.prefix}_pairs.tsv", "w") as f:
    for wsi, mask in patch_pairs:
        f.write(f"{wsi}\t{mask}\n")

2. 修改Nextflow进程 #

• 调整输出为tuple类型，通过from指令读取配对清单文件，将每行拆分为独立条目并绑定元数据：

process TileImages {

    input:
        val run_id
        val group_id
        val sample_id
        path image
        path masks

    output:
        tuple val(run_id), val(group_id), val(sample_id), path(wsi), path(mask),
            emit: tile_pairs
            from "${sample_id}_pairs.tsv"

    script:
        """
        python ${projectDir}/bin/run-tile-images.py \
            --image ${image} \
            --masks ${masks} \
            --prefix ${sample_id}
        """
}

说明 #

• Python脚本新增的配对清单文件（如{sample_id}_pairs.tsv）会记录所有生成的文件对路径，Nextflow通过from指令读取该文件，将每行解析为一个独立的输出条目。
• 输出的tuple类型将run_id、group_id、sample_id与对应的wsi/mask文件绑定，后续流程可以直接接收这些带元数据的独立条目进行处理。

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