pybids_test_gb.py

import json
import tempfile
import os
import os.path as op
import numpy as np
import nibabel as nib
from bids.layout import BIDSLayout


def touch(fname, times=None):
    with open(fname, 'a'):
        os.utime(fname, times)


def to_bids_description(path, fname='dataset_description.json',
                        BIDSVersion="1.4.0", **kwargs):
    """Dumps a dict into a bids description at the given location"""
    kwargs.update({"BIDSVersion": BIDSVersion})
    if (("GeneratedBy" in kwargs or "PipelineDescription" in kwargs) and
         "DatasetType" not in kwargs):
        kwargs["DatasetType"] = "derivative"
    desc_file = op.join(path, fname)
    with open(desc_file, 'w') as outfile:
        json.dump(kwargs, outfile)


def create_dummy_data(dir, subject, session=None):
    aff = np.eye(4)
    data_dwi = np.ones((100, 100, 100, 6))
    data_t1 = np.ones((100, 100, 100))
    bvecs = np.vstack([np.eye(3), np.eye(3)])
    bvecs[0] = 0
    bvecs = bvecs.T
    bvals = np.ones((1, 6)) * 1000.
    bvals[0, 0] = 0

    if session is None:
        data_dir = op.join(dir, subject)
    else:
        data_dir = op.join(dir, subject, session)

    sub_ses = f"{subject}_{session}_"

    np.savetxt(op.join(data_dir, 'dwi', f'{sub_ses}dwi.bval'), bvals)
    np.savetxt(op.join(data_dir, 'dwi', f'{sub_ses}dwi.bvec'), bvecs)
    nib.save(nib.Nifti1Image(data_dwi, aff),
             op.join(data_dir, 'dwi', f'{sub_ses}dwi.nii.gz'))
    nib.save(nib.Nifti1Image(data_t1, aff),
             op.join(data_dir, 'anat', f'{sub_ses}T1w.nii.gz'))


def create_dummy_bids_path(path, n_subjects, n_sessions, gb=True):
    os.makedirs(path)
    subjects = ['sub-0%s' % (d + 1) for d in range(n_subjects)]
    sessions = ['ses-0%s' % (d + 1) for d in range(n_sessions)]
    to_bids_description(
        path,
        **{"Name": "Dummy",
           "Subjects": subjects,
            "Sessions": sessions})
    if gb == True:
        pipeline_description = {"Name": "Dummy",
                                "GeneratedBy": [{"Name": "synthetic"}]}
    else:
        pipeline_description = {"Name": "Dummy",
                                "PipelineDescription": {"Name": "synthetic"}}

    deriv_dir = op.join(path, "derivatives", "synthetic")
    os.makedirs(deriv_dir)
    to_bids_description(
                deriv_dir,
                **pipeline_description)

    for subject in subjects:
        for session in sessions:
            for modality in ['anat', 'dwi']:
                os.makedirs(
                    op.join(deriv_dir, subject, session, modality))
                os.makedirs(
                    op.join(path, subject, session, modality))
            # Make some dummy data:
            create_dummy_data(deriv_dir, subject, session)
            create_dummy_data(path, subject, session)


n_sessions = 2
n_subjects = 3
bids_path_gb = "./gb"
create_dummy_bids_path(bids_path_gb, n_subjects, n_sessions, gb=True)
layout_gb = BIDSLayout(bids_path_gb, derivatives=True)
get_gb = layout_gb.get(scope="synthetic")

bids_path_pd = "./pd"
create_dummy_bids_path("./pd", n_subjects, n_sessions, gb=False)
layout_pd = BIDSLayout(bids_path_pd, derivatives=True)
get_pd = layout_pd.get(scope="synthetic")

assert len(get_pd) == len(get_gb)