<jupyter_start><jupyter_text>Nuisance regression<jupyter_code>import pynipt as pn
# This pipeline requires to install UNCCH_CAMRI pipeline plugin

pipe = pn.Pipeline('../', verbose=False)
pipe.set_package(0)
# Load library and refernce data
from slfmri import ui as slui
from slfmri import io as slio
from tqdm.notebook import tqdm

epitmp_path = '../Template/Rat_CAMRI_400um_EPI_v2-1.nii.gz'
mask_path = '../Template/Rat_CAMRI_400um_MASK_v2-1.nii.gz'
def calc_deriv(y, dt):
    x = np.linspace(0, (y.shape[0]-1)*dt, y.shape[0])
    dydx = np.gradient(y, dt, axis=0)
    return dydx

def nuisance_regression(input_path, output_path, ort_path, mask_path,
                        dt=2, lowcut=0.01, highcut=0.2, fwhm=0.5, 
                        use_deriv=False):
    
    step_pm = slio.PathMan(output_path)
    mask_nib = nib.load(mask_path)
    mparam_df = pipe.get_dset(ort_path, ext='1D').df
    dset = pipe.get_dset(input_path)
    
    for i, finfo in tqdm(dset):
        subj = finfo.Subject
        subj_pm = slio.PathMan(step_pm(subj))
        if os.path.exists(subj_pm(finfo.Filename)):
            pass
        else:
            fname = f'{finfo.Filename[:-7]}.1D'
            mp_path = mparam_df.query('Filename==@fname')['Abspath'].tolist()[0]
            volreg = slio.load_volreg(mp_path).values
            if use_deriv:
                deriv = calc_deriv(volreg, dt=2)
                ort = np.c_[deriv, volreg]
            else:
                ort = volreg
                
            nibobj = nib.load(finfo.Abspath)
            nr_nibobj = slui.rsfc.nuisance_regression(nibobj, mask=mask_nib, ort=ort, dt=dt, 
                                                      lowcut=lowcut, highcut=highcut, fwhm=fwhm)
            nr_nibobj.to_filename(subj_pm(finfo.Filename))
pipe
input_path = '040'
ort_path = '020'
lowcut = 0.01
highcut = 0.1

step_id = '05A'
lc_str = ''.join(str(lowcut).split('.')).ljust(3, '0')
hc_str = ''.join(str(highcut).split('.')).ljust(3, '0')
output_path = f'../Processing/UNCCH_CAMRI/{step_id}_NuisanceRegression-bp{lc_str}{hc_str}'

nuisance_regression(input_path, output_path, ort_path, mask_path, dt=2,
                    lowcut=lowcut, highcut=highcut, fwhm=0.5, use_deriv=False)
step_id = '05B'
lc_str = ''.join(str(lowcut).split('.')).ljust(3, '0')
hc_str = ''.join(str(highcut).split('.')).ljust(3, '0')
output_path = f'../Processing/UNCCH_CAMRI/{step_id}_NuisanceRegression-bp{lc_str}{hc_str}pHM'

nuisance_regression(input_path, output_path, ort_path, mask_path, dt=2,
                    lowcut=lowcut, highcut=highcut, fwhm=0.5, use_deriv=True)
highcut = 0.2

step_id = '05C'
lc_str = ''.join(str(lowcut).split('.')).ljust(3, '0')
hc_str = ''.join(str(highcut).split('.')).ljust(3, '0')
output_path = f'../Processing/UNCCH_CAMRI/{step_id}_NuisanceRegression-bp{lc_str}{hc_str}'

nuisance_regression(input_path, output_path, ort_path, mask_path, dt=2,
                    lowcut=lowcut, highcut=highcut, fwhm=0.5, use_deriv=False)
step_id = '05D'
lc_str = ''.join(str(lowcut).split('.')).ljust(3, '0')
hc_str = ''.join(str(highcut).split('.')).ljust(3, '0')
output_path = f'../Processing/UNCCH_CAMRI/{step_id}_NuisanceRegression-bp{lc_str}{hc_str}pHM'

nuisance_regression(input_path, output_path, ort_path, mask_path, dt=2,
                    lowcut=lowcut, highcut=highcut, fwhm=0.5, use_deriv=True)
def data_randseg(dset, step_code, output_path, mask, n_sample=3,
                 size_sample=150, stdout=None, stderr=None):
    """ 
    This item need to be redesigned for interface
    This code takes the mean value of data into account as a regressor to handle the baseline
    """
    import sys
    if stdout is None:
        stdout = sys.stdout
    if stderr is None:
        stderr = sys.stderr

    class_id = 1
    stdout.write('[UNCCH_CAMRI] Simple random segment:\n')
    try:
        if not os.path.exists(output_path):
            os.mkdir(output_path)
        stdout.write(f'Mask: {mask}\n')
        mask_nii = nib.load(mask)
        mask_idx = np.nonzero(mask_nii.dataobj)

        for subj in tqdm(dset.params[class_id].subjects):
            subj_path = os.path.join(output_path, subj)
            if not os.path.exists(subj_path):
                os.mkdir(subj_path)
            stdout.write(f'  Load {subj}...\n')
            list_of_files = dset(class_id, steps=[step_code], subjects=subj).df.Abspath

            stdout.write(f'    Total number of files: {len(list_of_files)}...\n')
            if len(list_of_files) > 1:
                # concat files
                input_data = []
                onsets = []
                for oid, f in enumerate(list_of_files):
                    loaded_data = np.asarray(nib.load(f).dataobj)[mask_idx]
                    input_data.append(loaded_data)
                    if oid == 0:
                        onsets.append(0)
                    else:
                        onsets.append(input_data[oid-1].shape[-1] + onsets[-1])
                input_data = np.concatenate(input_data, -1)
                max_size = input_data.shape[1] - size_sample
                onsets = [o for o in onsets if o <= max_size]
            else:
                input_nii = nib.load(list_of_files[0])
                input_data = np.asarray(input_nii.dataobj)[mask_idx]
                onsets = None
            n_frame = input_data.shape[-1]
            stdout.write(f'    Total number of frames: {n_frame}\n')

            # random sampling
            if len(list_of_files) > 1:
                sampled_onsets = np.random.choice(onsets, n_sample, replace=False)
            else:
                sampled_onsets = np.random.choice(n_frame - size_sample - 1, n_sample, 
                                                  replace=False)
            for time, onset in enumerate(sampled_onsets):
                fname = os.path.join(subj_path, f'{subj}_run-{time + 1}_cap')
                if not os.path.exists(f'{fname}.nii.gz'):
                    trial_data = input_data[..., onset:onset+size_sample]
                    output_data = np.zeros(list(mask_nii.shape) + [trial_data.shape[-1]])
                    output_data[mask_idx] = trial_data
                    output_nii = nib.Nifti1Image(output_data, affine=input_nii.affine)
                    output_nii.header['qform_code'] = 1
                    output_nii.header['sform_code'] = 0
                    output_nii.to_filename(f'{fname}.nii.gz')
                    stdout.write(f'{time+1}..')
                else:
                    pass
            stdout.write('\n')
        stdout.write('Done..\n')
    except:
        import traceback
        stderr.write('[ERROR] Failed.\n')
        traceback.print_exception(*sys.exc_info(), file=stderr)
        return 1
    return 0
import nibabel as nib
import numpy as np

lowcut = 0.01
highcut = 0.1
lc_str = ''.join(str(lowcut).split('.')).ljust(3, '0')
hc_str = ''.join(str(highcut).split('.')).ljust(3, '0')

dcode_rsfmri = f'05B'
output_dr = f'../Processing/UNCCH_CAMRI/20B_DataRandSeg-bp{lc_str}{hc_str}pHM'
dset_rsfmri = pipe.get_dset(dcode_rsfmri)
data_randseg(dset_rsfmri, dcode_rsfmri, output_dr, mask_path, size_sample=150)<jupyter_output>[UNCCH_CAMRI] Simple random segment:
Mask: ../Template/Rat_CAMRI_400um_MASK_v2-1.nii.gz
StarcoderdataJupyterStructuredCleanDedup