|
| 1 | +import os |
| 2 | +from pathlib import Path |
| 3 | +import subprocess |
| 4 | +import yaml |
| 5 | +import inquirer |
| 6 | + |
| 7 | +def prompt_input_directory(): |
| 8 | + return inquirer.prompt([ |
| 9 | + inquirer.Path( |
| 10 | + "path", |
| 11 | + message="📂 Select an input directory containing DICOM image files:", |
| 12 | + path_type=inquirer.Path.DIRECTORY, |
| 13 | + exists=True, |
| 14 | + ) |
| 15 | + ])["path"] |
| 16 | + |
| 17 | + |
| 18 | +def prompt_output_directory(input_dir): |
| 19 | + # List subfolders of input_dir |
| 20 | + subdirs = [ |
| 21 | + name for name in os.listdir(input_dir) |
| 22 | + if os.path.isdir(os.path.join(input_dir, name)) |
| 23 | + ] |
| 24 | + |
| 25 | + choices = [f"./{name}" for name in subdirs] |
| 26 | + choices.append("📥 Enter a custom output path...") |
| 27 | + |
| 28 | + answer = inquirer.prompt([ |
| 29 | + inquirer.List( |
| 30 | + "choice", |
| 31 | + message=f"📁 Choose an output directory for NIfTI files from:\n → {input_dir}", |
| 32 | + choices=choices |
| 33 | + ) |
| 34 | + ])["choice"] |
| 35 | + |
| 36 | + if answer == "📥 Enter a custom output path...": |
| 37 | + return inquirer.prompt([ |
| 38 | + inquirer.Path( |
| 39 | + "custom_path", |
| 40 | + message="📥 Enter custom output directory path:", |
| 41 | + path_type=inquirer.Path.DIRECTORY, |
| 42 | + exists=True |
| 43 | + ) |
| 44 | + ])["custom_path"] |
| 45 | + else: |
| 46 | + return os.path.abspath(os.path.join(input_dir, answer.strip("./"))) |
| 47 | + |
| 48 | + |
| 49 | +def generate_batch_config( |
| 50 | + input_dirs, |
| 51 | + output_dirs, |
| 52 | + config_path="batch_config.yaml" |
| 53 | +): |
| 54 | + """ |
| 55 | + input_dirs: List of input directory paths |
| 56 | + output_dirs: Single output directory (string) OR list of output directories |
| 57 | + """ |
| 58 | + |
| 59 | + # Normalize inputs |
| 60 | + if isinstance(output_dirs, str): |
| 61 | + output_dirs = [output_dirs] * len(input_dirs) |
| 62 | + elif isinstance(output_dirs, list): |
| 63 | + if len(input_dirs) != 1 and len(output_dirs) != len(input_dirs): |
| 64 | + raise ValueError("Number of output directories must match number of input directories, unless only one output directory is provided.") |
| 65 | + |
| 66 | + config = { |
| 67 | + "Options": { |
| 68 | + "isGz": True, # compressed nii.gz |
| 69 | + "isFlipY": False, # flip Y-axis of images |
| 70 | + "isVerbose": False, # by default is verbose; this value does not change anything |
| 71 | + "isCreateBIDS": True, # create BIDS-compatible NIfTI and JSON files |
| 72 | + "isOnlySingleFile": False |
| 73 | + }, |
| 74 | + "Files": [] |
| 75 | + } |
| 76 | + |
| 77 | + for in_dir, out_dir in zip(input_dirs, output_dirs): |
| 78 | + if not os.path.isdir(in_dir): |
| 79 | + print(f"Warning: {in_dir} is not a valid directory.") |
| 80 | + continue |
| 81 | + |
| 82 | + files = os.listdir(in_dir) |
| 83 | + if not files: |
| 84 | + print(f"No files found in {in_dir}") |
| 85 | + continue |
| 86 | + |
| 87 | + files = [f for f in os.listdir(in_dir) if os.path.isfile(os.path.join(in_dir, f))] |
| 88 | + if not files: |
| 89 | + print(f"No files found in {in_dir}") |
| 90 | + continue |
| 91 | + |
| 92 | + for f in files: |
| 93 | + filename = os.path.splitext(f)[0].replace(" ", "_").lower() |
| 94 | + |
| 95 | + config["Files"].append({ |
| 96 | + "in_dir": os.path.abspath(in_dir), |
| 97 | + "out_dir": os.path.abspath(out_dir), |
| 98 | + "filename": filename |
| 99 | + }) |
| 100 | + |
| 101 | + with open(config_path, 'w') as f: |
| 102 | + yaml.dump(config, f, sort_keys=False, default_flow_style=False) |
| 103 | + |
| 104 | + print(f"Config written to {config_path}") |
| 105 | + |
| 106 | +def dicom_to_niix(vol_dir: Path, out_dir: Path, merge_2d: bool = False): |
| 107 | + """ |
| 108 | + For converting DICOM images to a (compresssed) 4d nifti image |
| 109 | + """ |
| 110 | + os.makedirs(out_dir, exist_ok=True) |
| 111 | + |
| 112 | + try: |
| 113 | + res = subprocess.run( |
| 114 | + [ |
| 115 | + "dcm2niix", |
| 116 | + "-f", "%s_%p", # dcm2niix attempts to provide a sensible file naming scheme |
| 117 | + "-o", out_dir, # output directory |
| 118 | + "-z", "y", #specifying compressed nii.gz file |
| 119 | + "-m", "y" if merge_2d else "n", # Add merge option |
| 120 | + # https://www.nitrc.org/plugins/mwiki/index.php/dcm2nii:MainPage |
| 121 | + # for further configuration for general usage see page above |
| 122 | + vol_dir # input directory |
| 123 | + ], |
| 124 | + capture_output=True, |
| 125 | + text=True, |
| 126 | + check=True |
| 127 | + ) |
| 128 | + |
| 129 | + nifti_files = list(Path(out_dir).glob("*.nii.gz")) |
| 130 | + if not nifti_files: |
| 131 | + raise RuntimeError("No NIfTI (.nii.gz) files were generated.") |
| 132 | + |
| 133 | + bval_files = list(out_dir.glob("*.bval")) |
| 134 | + bvec_files = list(out_dir.glob("*.bvec")) |
| 135 | + bval_path = str(bval_files[0]) if bval_files else None |
| 136 | + bvec_path = str(bvec_files[0]) if bvec_files else None |
| 137 | + |
| 138 | + |
| 139 | + return nifti_files[0], bval_path, bvec_path |
| 140 | + |
| 141 | + except subprocess.CalledProcessError as e: |
| 142 | + raise RuntimeError(f"dcm2niix failed: {e.stderr}") |
| 143 | + |
| 144 | +if __name__ == "__main__": |
| 145 | + input_dirs = [] |
| 146 | + output_dirs = [] |
| 147 | + |
| 148 | + print("Inquiring to convert your DICOM Images to NIfTI Image(s) \n") |
| 149 | + |
| 150 | + while True: |
| 151 | + input_dir = prompt_input_directory() |
| 152 | + output_dir = prompt_output_directory(input_dir) |
| 153 | + |
| 154 | + input_dirs.append(input_dir) |
| 155 | + output_dirs.append(output_dir) |
| 156 | + |
| 157 | + add_more = inquirer.prompt([ |
| 158 | + inquirer.Confirm( |
| 159 | + "more", |
| 160 | + message="➕ Do you want to add another input/output pair?", |
| 161 | + default=False |
| 162 | + ) |
| 163 | + ])["more"] |
| 164 | + |
| 165 | + if not add_more: |
| 166 | + break |
| 167 | + |
| 168 | + if len(input_dirs) == 1: |
| 169 | + print("📥 Single input/output pair detected — using `dcm2niix`...") |
| 170 | + vol_dir = Path(input_dirs[0]) |
| 171 | + out_dir = Path(output_dirs[0]) |
| 172 | + |
| 173 | + merge_answer = inquirer.prompt([ |
| 174 | + inquirer.Confirm( |
| 175 | + "merge", |
| 176 | + message="🧩 Do you want to merge 2D slices into a single NIfTI (-m y)?", |
| 177 | + default=True |
| 178 | + ) |
| 179 | + ]) |
| 180 | + merge_2d = merge_answer["merge"] |
| 181 | + |
| 182 | + try: |
| 183 | + nifti, bval, bvec = dicom_to_niix(vol_dir, out_dir, merge_2d=merge_2d) |
| 184 | + print(f"\n✅ NIfTI file created: {nifti}") |
| 185 | + if bval: |
| 186 | + print(f" 📈 bval: {bval}") |
| 187 | + if bvec: |
| 188 | + print(f" 📊 bvec: {bvec}") |
| 189 | + |
| 190 | + post_process = inquirer.prompt([ |
| 191 | + inquirer.Confirm( |
| 192 | + "run_post", |
| 193 | + message="🧩 Do you want to run IVIM fit algorithm on the NIfTI file now?", |
| 194 | + default=True |
| 195 | + ) |
| 196 | + ])["run_post"] |
| 197 | + |
| 198 | + if bvec and bval and post_process: |
| 199 | + print("\n🧩 Running post-processing: OSIPI IVIM fitting...\n") |
| 200 | + subprocess.run([ |
| 201 | + "python3", "-m", "WrapImage.nifti_wrapper", |
| 202 | + str(nifti), |
| 203 | + str(bvec), |
| 204 | + str(bval), |
| 205 | + ], check=True) |
| 206 | + print("✅ NIfTI post-processing completed.") |
| 207 | + except RuntimeError as err: |
| 208 | + print(f"❌ Conversion failed: {err}") |
| 209 | + |
| 210 | + else: |
| 211 | + print("📥📥📥 Multiple inputs detected — generating batch config and using `dcm2niibatch`...") |
| 212 | + config_path = "batch_config.yaml" |
| 213 | + generate_batch_config(input_dirs, output_dirs, config_path=config_path) |
| 214 | + |
| 215 | + try: |
| 216 | + with subprocess.Popen( |
| 217 | + ["dcm2niibatch", config_path], |
| 218 | + stdout=subprocess.PIPE, |
| 219 | + stderr=subprocess.STDOUT, |
| 220 | + text=True, |
| 221 | + bufsize=1 |
| 222 | + ) as proc: |
| 223 | + # for line in proc.stdout: # uncomment this to see verbose logs |
| 224 | + # print(line.strip()) |
| 225 | + proc.wait() |
| 226 | + if proc.returncode != 0: |
| 227 | + raise subprocess.CalledProcessError(proc.returncode, proc.args) |
| 228 | + |
| 229 | + print("✅ Batch conversion completed successfully.") |
| 230 | + |
| 231 | + except subprocess.CalledProcessError as err: |
| 232 | + print(f"❌ Batch conversion failed:\n{err}") |
| 233 | + |
0 commit comments