Python NeuroImage Pipeline Tool
The key features of this module are
Dependency:
Compatibility:
ChangeLog:
from PyPI
$ pip install pynipt
from Github repository (nightly build) if you have any issues with PyPI version, please try this instead
$ pip install git+https://github.com/pynipt/pynipt
Project_Root/
├── JupyterNotes/
│ └── fMRI_Data_Preprocessing.ipynb
├── Data/
│ ├── dataset_description.json
│ ├── README
│ ├── sub-01/
│ │ ├── anat/
│ │ │ ├── sub-01_T2w.json
│ │ │ └── sub-01_T2w.nii.gz
│ │ ├── fmap/
│ │ │ ├── sub-01_fieldmap.json
│ │ │ ├── sub-01_fieldmap.nii.gz
│ │ │ └── sub-01_magnitude.nii.gz
│ │ └── func/
│ │ ├── sub-01_task-rest_bold.json
│ │ ├── sub-01_task-rest_bold.nii.gz
│ │ ├── sub-01_task-active_bold.json
│ │ └── sub-01_task-active_bold.nii.gz
│ └── sub-02/
│ ├── anat/
│ │ ├── sub-02_T2w.json
│ │ └── sub-02_T2w.nii.gz
│ ├── fmap/
│ │ ├── sub-02_fieldmap.json
│ │ ├── sub-02_fieldmap.nii.gz
│ │ └── sub-02_magnitude.nii.gz
│ └── func/
│ ├── sub-02_task-rest_bold.json
│ ├── sub-02_task-rest_bold.nii.gz
│ ├── sub-02_task-active_bold.json
│ └── sub-02_task-active_bold.nii.gz
├── Mask/
│ ├── 02A_BrainMasks-func/
│ │ ├── sub-01/
│ │ │ ├── sub-01_task-rest_bold.nii.gz
│ │ │ └── sub-01_task-active_bold_mask.nii.gz
│ │ └── sub-02/
│ │ ├── sub-02_task-rest_bold.nii.gz
│ │ └── sub-02_task-active_bold_mask.nii.gz
│ └── 02B_BrainMasks-anat/
│ ├── sub-01/
│ │ ├── sub-01_T2w.nii.gz
│ │ └── sub-01_T2w_mask.nii.gz
│ └── sub-02/
│ ├── sub-02_T2w.nii.gz
│ └── sub-02_T2w_mask.nii.gz
├── Processing/
│ └── MyPipeline/
│ ├── 01A_ProcessingStep1A-func/
│ │ ├── sub-01/
│ │ │ ├── sub-01_task-rest_bold.nii.gz
│ │ │ └── sub-01_task-active_bold.nii.gz
│ │ ├── sub-02/
│ │ │ ├── sub-02_task-rest_bold.nii.gz
│ │ │ └── sub-02_task-active_bold.nii.gz
│ └── 01B_ProcessingStep1B-func/
│ ├── sub-01/
│ │ ├── sub-01_task-rest_bold.nii.gz
│ │ └── sub-01_task-active_bold.nii.gz
│ └── sub-02/
│ ├── sub-02_task-rest_bold.nii.gz
│ └── sub-02_task-active_bold.nii.gz
├── Results/
│ └── MyPipeline/
│ └── 030_2ndLevelStatistic-func/
│ ├── TTest.nii.gz
│ └── TTest_report.html
├── Temp/
├── Logs/
│ ├── DEBUG.log
│ ├── STDERR.log
│ └── STDOUT.log
└── Templates/
└── BrainTemplate.nii.gz
import pynipt as pn
pipe = pn.Pipeline()
** Dataset summary
Path of Dataset: /absolute/path/to/
Name of Dataset:
Selected DataClass: Data
Subject(s): [‘sub-01’, ‘sub-02’]
Datatype(s): [‘anat’, ‘fmap’, ‘func’]
List of installed pipeline packages:
pipe.set_scratch_package(‘MyPipeline’)
The scratch package [MyPipeline] is initiated.
```
Execute linux shell command ‘mycommand’ for the all file in Datatype ‘func’ and output files to Processing/01A_ProcessingStep1A-func.
>> itb = pipe.get_builder(n_threads=1) # in case mycommand take huge computing resources
>> itb.init_step(title='ProcessingStep1A', suffix='func',
>> idx=1, subcode='A', mode='processing', type='cmd')
>> itb.set_input(label='input', input_path='func')
>> itb.set_var(label='param', value=10)
>> itb.set_output(label='output')
>> itb.set_cmd('mycommand -i *[input] -o *[output] -o *[param]')
>> itb.set_output_checker('output')
>> itb.run()
Execute linux shell command ‘getmask’ for the first file in Datatype ‘01A’ to generate brain mask and output to Mask/02A_BrainMasks-func.
additionally copy the original data to output folder
>> itb = pipe.get_builder(n_threads=4) # multi threasing
>> itb.init_step(title='BrainMasks', suffix='func',
>> idx=2, subcode='A', mode='masking', type='cmd')
>> itb.set_input(label='input', input_path='01A', idx=0,
>> filter_dict=filter_dict)
>> itb.set_output(label='mask', suffix='_mask')
>> itb.set_output(label='copy')
>> itb.set_cmd('getmask *[input] *[mask]')
>> itb.set_cmd('cp *[input] *[copy]')
>> itb.set_output_checker('mask')
>> itb.run()
Execute python function ‘myfunction’ for the all file in StepCode ‘01A’ and output files to Processing/01B_ProcessingStep1B-func.
```python
def myfunction(input, output, param, stdout=None, stderr=None):
## import modules here
import sys
import numpy as np
import nibabel as nib
if stdout == None: # for handling output/error messages
stdout = sys.stdout
stderr = sys.stderr
try:
# put your code here
stdout.write(f'Running MyFunction for input: {input}\n')
img = nib.load(input)
img_data = np.asarray(img.dataobj)
result_data = img_data * param
stdout.write(f'Multiply image by {param}\n')
nii = nib.Nifti1Image(result_data, affine=img._affine, header=img._header)
stdout.write(f'Save to {output}..\n')
nii.to_filename(output)
stdout.write('Done\n')
# until here
except: # for handling error
stderr.write('[ERROR] Failed.\n')
import traceback
traceback.print_exception(*sys.exc_info(), file=stderr)
return 1
return 0
itb = pipe.get_builder()
itb.init_step(title=’ProcessingStep1B’, suffix=’func’,
idx=1, subcode=’B’, mode=’processing’, type=’python’)
itb.set_input(label=’input’, input_path=’01A’) # the data from ‘01A’ will be assigned as ‘input’ argument of ‘myfunction’
itb.set_var(label=’param’, value=10) # 10 will be assigned ‘param’ argument of ‘myfunction’
itb.set_output(label=’output’) # no modification on filename, so output filename will be same as input
itb.set_func(myfunction)
itb.set_output_checker(‘output’)
itb.run()
```
Simple example of 2nd level stats (TTest), ‘rest’ vs ‘active’
>> def myttestfuncion(group_a, group_b, output, stdout=None, stderr=None):
>> import sys
>> if stdout is None:
>> stdout = sys.stdout
>> if stderr is None:
>> stderr = sys.stderr
>>
>> import nibabel as nib
>> import numpy as np
>> import scipy.stats as stats
>> affine = None
>> try:
>> groups = dict()
>> for group_id, subj_list in dict(group_a=group_a, group_b=group_b).items():
>> stack = []
>> for i, img_path in enumerate(subj_list):
>> stdout.write(f'{img_path} is loaded as group {group_id}')
>> img = nib.load(img_path)
>> if i == 0:
>> affine = img.affine
>> stack.append(np.asarray(img._dataobj))
>> groups[group_id] = np.concatenate(stack, axis=-1)
>> stdout.write(f'{group_id} has been stacked.')
>> t, p = stats.ttest_ind(groups['group_a'], groups['group_b'], axis=-1)
>> imgobj = np.concatenate([t[..., np.newaxis], p[..., np.newaxis]], axis=-1)
>> ttest_result = nib.Nifti1Image(imgobj, affine)
>> ttest_result.to_filename(output)
>> stdout.write('{} is created'.format(output))
>> open (f'{output}_report.html', 'w') as f:
>> f.write('<html>Hello world.</html>\n')
>> except Exception:
>> stderr.write('[ERROR] Failed.\n')
>> import traceback
>> traceback.print_exception(*sys.exc_info(), file=stderr)
>> return 1
>> return 0
>>
>> itb = pipe.get_builder()
>> itb.init_step('2ndLevelStatistic', suffix='func', idx=3, subcode=0,
>> mode='reporting', type='python') # for reporting, as a default, output is directory without extension
>> itb.set_input(label='group_a', input_path='01B', group_input=True,
>> filter_dict=dict(regex=r'sub_\d{2}_task-rest.*'), join_modifier=False) # if this is False, input will return
>> # 'list obj' so can run loop within python function
>> itb.set_input(label='group_b', input_path='01B', group_input=True,
>> filter_dict=dict(regex=r'sub_\d{2}_task-active.*'), join_modifier=False)
>> itb.set_output(label='output',
>> modifier='TTest', ext='nii.gz') # for peers to one output
>> itb.set_func(myttestfunc)
>> itb.set_output_checker(label='output') # this will check if TTest.nii.gz is generated but not html file
>> itb.run()
Check progression with using progressbar
>> pipe.check_progression()
MyPipeline 50%|████████████████ | 1/2
Access a image file absolute path in dataset ‘01A’
```python
pipe.get_dset(‘01A’).df # get_dset metrics will return dataset object
…will print out the data structure…
file_1 = pipe.get_dset(‘01A’)[0].Abspath # get absolute path of first indexed file
```
Filter key
Output filename specification
Output checker
The tutorial does not ready yet, will be provided soon
PyNIPT is licensed under the term of the GNU GENERAL PUBLIC LICENSE Version 3
If you interest in contributing this project, please contact shlee@unc.edu.
Lee, SungHo, Ban, Woomi, & Shih, Yen-Yu Ian. (2020, May 25). PyNIPT/pynipt: PyNIPT v0.2.1 (Version 0.2.1). Zenodo. http://doi.org/10.5281/zenodo.3842192
@software{lee_sungho_2020_3842192,
author = {Lee, SungHo and
Ban, Woomi and
Shih, Yen-Yu Ian},
title = {PyNIPT/pynipt: PyNIPT v0.2.1},
month = may,
year = 2020,
publisher = {Zenodo},
version = {0.2.1},
doi = {10.5281/zenodo.3842192},
url = {https://doi.org/10.5281/zenodo.3842192}
}