如何用Python在Ubuntu上自动化批量执行FSL的MRI分割命令？

核心思路 #

将重复命令中的可变部分（图像编号、目标脑区参数、输出路径）抽象为参数对，遍历所有图像编号，自动生成并执行对应命令，同时加入基础的错误检查和日志输出。

具体实现代码 #

import subprocess
import os

# 定义每个任务对应的脑区参数和输出目录后缀
region_tasks = [
    ("L_Puta,R_Puta,BrStem", "Puta"),
    ("L_Caud,R_Caud,BrStem", "Caud"),
    ("L_Accu,R_Accu,BrStem", "Accu"),
    ("L_Pall,R_Pall,BrStem", "Pall"),
    ("L_Hipp,R_Hipp,BrStem", "Hipp"),
    ("L_Amyg,R_Amyg,BrStem", "Amyg"),
    ("L_Thal,R_Thal,BrStem", "Thal"),
]

# 提前创建所有输出子目录，避免命令因目录不存在失败
for _, region_dir in region_tasks:
    output_dir = f"OutputImage/Autism/{region_dir}"
    os.makedirs(output_dir, exist_ok=True)

# 遍历图像编号（示例为01到1000的两位数字格式）
for num in range(1, 1001):
    img_num = f"{num:02d}"  # 生成两位数字编号，如01、04、10
    input_path = f"InputImages/Autism/{img_num}.nii.gz"
    
    # 检查输入文件是否存在，跳过无效编号
    if not os.path.exists(input_path):
        print(f"警告：输入文件 {input_path} 不存在，跳过此编号")
        continue
    
    # 执行当前编号的所有7个脑区分割任务
    for region_s, region_dir in region_tasks:
        output_path = f"OutputImage/Autism/{region_dir}/{img_num}"
        cmd = [
            "run_first_all",
            "-d",
            "-i", input_path,
            "-s", region_s,
            "-o", output_path
        ]
        
        try:
            # 执行命令并捕获结果
            subprocess.run(cmd, check=True, capture_output=True, text=True)
            print(f"编号 {img_num} 的 {region_dir} 任务执行成功")
        except subprocess.CalledProcessError as e:
            print(f"编号 {img_num} 的 {region_dir} 任务执行失败")
            print("错误详情:", e.stderr)

关键调整说明 #

1. 编号格式适配：如果你的图像编号是三位数字（如001），将img_num = f"{num:02d}"改为img_num = f"{num:03d}"；若编号无前置零，直接用img_num = f"{num}"。
2. 并行加速（可选）：如果机器性能充足，可使用concurrent.futures模块实现多任务并行，大幅提升处理速度：

from concurrent.futures import ThreadPoolExecutor

def process_single_task(img_num, region_s, region_dir):
    input_path = f"InputImages/Autism/{img_num}.nii.gz"
    if not os.path.exists(input_path):
        print(f"警告：输入文件 {input_path} 不存在")
        return
    output_path = f"OutputImage/Autism/{region_dir}/{img_num}"
    cmd = ["run_first_all", "-d", "-i", input_path, "-s", region_s, "-o", output_path]
    try:
        subprocess.run(cmd, check=True, capture_output=True, text=True)
        print(f"编号 {img_num} 的 {region_dir} 任务执行成功")
    except subprocess.CalledProcessError as e:
        print(f"编号 {img_num} 的 {region_dir} 任务执行失败")
        print("错误详情:", e.stderr)

# 生成所有任务列表
tasks = []
for num in range(1, 1001):
    img_num = f"{num:02d}"
    for region_s, region_dir in region_tasks:
        tasks.append((img_num, region_s, region_dir))

# 并行执行，max_workers根据CPU核心数调整（建议设为核心数的2倍）
with ThreadPoolExecutor(max_workers=8) as executor:
    executor.map(lambda args: process_single_task(*args), tasks)

3. FSL环境验证：确保运行脚本的终端已加载FSL环境变量（通常通过source /usr/local/fsl/etc/fslconf/fsl.sh配置），若用IDE运行，需在运行配置中手动添加FSL的环境变量。

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