Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
streamlines = sft.get_streamlines_copy()
if not streamlines:
if init_only:
logging.warning('{} is empty'.format(filename))
return None
if os.path.isfile(tmp_density_filename) \
and os.path.isfile(tmp_endpoints_filename) \
and os.path.isfile(tmp_centroids_filename):
# If initilization, loading the data is useless
if init_only:
return None
density = nib.load(tmp_density_filename).get_fdata().astype(np.uint16)
endpoints_density = nib.load(
tmp_endpoints_filename).get_fdata().astype(np.uint16)
sft_centroids = load_tractogram(tmp_centroids_filename, reference)
sft_centroids.to_vox()
sft_centroids.to_corner()
centroids = sft_centroids.get_streamlines_copy()
else:
transformation, dimensions, _, _ = sft.space_attributes
density = compute_tract_counts_map(streamlines, dimensions)
endpoints_density = get_endpoints_density_map(streamlines, dimensions,
point_to_select=3)
thresholds = [32, 24, 12, 6]
if disable_centroids:
centroids = []
else:
centroids = qbx_and_merge(streamlines, thresholds,
rng=RandomState(0),
verbose=False).centroids
def compute_streamlines_measures(args):
bundle_filename, bundle_reference = args[0]
wb_streamlines = args[1]
gs_streamlines_indices = args[2]
if not os.path.isfile(bundle_filename):
logging.info('{} does not exist'.format(bundle_filename))
return None
bundle_sft = load_tractogram(bundle_filename, bundle_reference)
bundle_sft.to_vox()
bundle_sft.to_corner()
bundle_streamlines = bundle_sft.streamlines
_, bundle_dimensions, _, _ = bundle_sft.space_attributes
if not bundle_streamlines:
logging.info('{} is empty'.format(bundle_filename))
return None
_, streamlines_indices = perform_streamlines_operation(intersection,
[wb_streamlines,
bundle_streamlines],
precision=0)
streamlines_binary = binary_classification(streamlines_indices,
gs_streamlines_indices,
filename = args[0]
reference = args[1]
init_only = args[2]
disable_centroids = args[3]
# Since data is often re-use when comparing multiple bundles, anything
# that can be computed once is saved temporarily and simply loaded on demand
hash_tmp = hashlib.md5(filename.encode()).hexdigest()
tmp_density_filename = os.path.join('tmp_measures/',
'{}_density.nii.gz'.format(hash_tmp))
tmp_endpoints_filename = os.path.join('tmp_measures/',
'{}_endpoints.nii.gz'.format(hash_tmp))
tmp_centroids_filename = os.path.join('tmp_measures/',
'{}_centroids.trk'.format(hash_tmp))
sft = load_tractogram(filename, reference)
sft.to_vox()
sft.to_corner()
streamlines = sft.get_streamlines_copy()
if not streamlines:
if init_only:
logging.warning('{} is empty'.format(filename))
return None
if os.path.isfile(tmp_density_filename) \
and os.path.isfile(tmp_endpoints_filename) \
and os.path.isfile(tmp_centroids_filename):
# If initilization, loading the data is useless
if init_only:
return None
density = nib.load(tmp_density_filename).get_fdata().astype(np.uint16)
endpoints_density = nib.load(
def prepare_data_for_actors(bundle_filename, reference_filename,
target_template_filename):
sft = load_tractogram(bundle_filename, reference_filename)
streamlines = sft.streamlines
# Load and prepare the data
reference_img = nib.load(reference_filename)
reference_data = reference_img.get_data()
reference_affine = reference_img.get_affine()
target_template_img = nib.load(target_template_filename)
target_template_data = target_template_img.get_data()
target_template_affine = target_template_img.affine
# Register the DWI data to the template
transformed_reference, transformation = register_image(target_template_data,
target_template_affine,
reference_data,
reference_affine)
def bench_compress_streamlines():
repeat = 10
fname = get_fnames('fornix')
fornix = load_tractogram(fname, 'same',
bbox_valid_check=False).streamlines
streamlines = Streamlines(fornix)
print("Timing compress_streamlines() in Cython"
" ({0} streamlines)".format(len(streamlines)))
cython_time = measure("compress_streamlines(streamlines)", repeat)
print("Cython time: {0:.3}sec".format(cython_time))
del streamlines
streamlines = Streamlines(fornix)
python_time = measure("map(compress_streamlines_python, streamlines)",
repeat)
print("Python time: {0:.2}sec".format(python_time))
print("Speed up of {0}x".format(python_time/cython_time))
del streamlines
def compute_measures(filename_tuple):
sft = load_tractogram(filename_tuple[0], filename_tuple[1])
_, dimensions, voxel_size, _ = sft.space_attributes
nbr_streamlines = len(sft)
if not nbr_streamlines:
logging.warning('{} is empty'.format(filename_tuple[0]))
return dict(zip(['volume', 'volume_endpoints', 'streamlines_count',
'avg_length', 'std_length', 'min_length', 'max_length',
'mean_curvature'], [0, 0, 0, 0, 0, 0, 0, 0]))
length_list = list(length(list(sft.streamlines)))
length_avg = float(np.average(length_list))
length_std = float(np.std(length_list))
length_min = float(np.min(length_list))
length_max = float(np.max(length_list))
sft.to_vox()
def get_streamlines():
from dipy.data import get_fnames
from dipy.io.streamline import load_tractogram
from dipy.tracking.streamline import Streamlines
fname = get_fnames('fornix')
fornix = load_tractogram(fname, 'same',
bbox_valid_check=False).streamlines
streamlines = Streamlines(fornix)
return streamlines
def get_streamlines():
from dipy.data import get_fnames
from dipy.io.streamline import load_tractogram
from dipy.tracking.streamline import Streamlines
fname = get_fnames('fornix')
fornix = load_tractogram(fname, 'same',
bbox_valid_check=False).streamlines
streamlines = Streamlines(fornix)
return streamlines