Scripts to project 3D and 4D maps to streamline points and perform math operations in space of streamline points

.python-version

@@ -1 +1 @@
-3.10
+3.10

scilpy/tractograms/dps_and_dpp_management.py

@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+import numpy as np
+
+
+def project_map_to_streamlines(sft, map, endpoints_only=False):
+    """
+    Projects a map onto the points of streamlines.
+
+    Parameters
+    ----------
+    sft: StatefulTractogram
+        Input tractogram.
+    map: DataVolume
+        Input map.
+
+    Optional:
+    ---------
+    endpoints_only: bool
+        If True, will only project the map onto the endpoints of the
+        streamlines (all values along streamlines set to zero). If False,
+        will project the map onto all points of the streamlines.
+
+    Returns
+    -------
+    streamline_data:
+        map projected to each point of the streamlines.
+    """
+    if len(map.data.shape) == 4:
+        dimension = map.data.shape[3]
+    else:
+        dimension = 1
+
+    streamline_data = []
+    if endpoints_only:
+        for s in sft.streamlines:
+            p1_data = map.get_value_at_coordinate(
+                s[0][0], s[0][1], s[0][2],
+                space=sft.space, origin=sft.origin)
+            p2_data = map.get_value_at_coordinate(
+                s[-1][0], s[-1][1], s[-1][2],
+                space=sft.space, origin=sft.origin)
+            thisstreamline_data = []
+            if dimension == 1:
+                thisstreamline_data = np.ones((len(s), 1)) * np.nan
+            else:
+                thisstreamline_data = np.ones(
+                    (len(s), p1_data.shape[0])) * np.nan
+
+            thisstreamline_data[0] = p1_data
+            thisstreamline_data[-1] = p2_data
+            thisstreamline_data = np.asarray(thisstreamline_data)
+
+            streamline_data.append(
+                np.reshape(thisstreamline_data,
+                           (len(thisstreamline_data), dimension)))
+    else:
+        for s in sft.streamlines:
+            thisstreamline_data = []
+            for p in s:
+                thisstreamline_data.append(map.get_value_at_coordinate(
+                    p[0], p[1], p[2], space=sft.space, origin=sft.origin))
+
+            streamline_data.append(
+                np.reshape(thisstreamline_data,
+                           (len(thisstreamline_data), dimension)))
+
+    return streamline_data

scilpy/tractograms/streamline_operations.py

@@ -84,7 +84,6 @@ def _get_point_on_line(first_point, second_point, vox_lower_corner):
 
     return first_point + ray * (t0 + t1) / 2.
 
-
 def filter_streamlines_by_length(sft, min_length=0., max_length=np.inf):
     """
     Filter streamlines using minimum and max length.
@@ -389,3 +388,144 @@ def smooth_line_spline(streamline, smoothing_parameter, nb_ctrl_points):
     smoothed_streamline[-1] = streamline[-1]
 
     return smoothed_streamline
+
+
+def perform_streamline_operation_per_point(op_name, sft, dpp_name='metric',
+                                           endpoints_only=False):
+    """Peforms an operation per point for all streamlines.
+
+    Parameters
+    ----------
+    op_name: str
+        A callable that takes a list of streamline data per point (4D) and
+        returns a list of streamline data per point.
+    sft: StatefulTractogram
+        The streamlines used in the operation.
+    dpp_name: str
+        The name of the data per point to be used in the operation.
+    endpoints_only: bool
+        If True, will only perform operation on endpoints
+
+    Returns
+    -------
+    new_sft: StatefulTractogram
+        sft with data per streamline resulting from the operation.
+    """
+
+    # Performing operation
+    call_op = OPERATIONS[op_name]
+    if endpoints_only:
+        new_data_per_point = []
+        for s in sft.data_per_point[dpp_name]:
+            this_data_per_point = np.nan * np.ones((len(s), 1))
+            this_data_per_point[0] = call_op(s[0])
+            this_data_per_point[-1] = call_op(s[-1])
+            new_data_per_point.append(
+                np.reshape(this_data_per_point, (len(this_data_per_point), 1)))
+    else:
+        new_data_per_point = []
+        for s in sft.data_per_point[dpp_name]:
+            this_data_per_point = []
+            for p in s:
+                this_data_per_point.append(call_op(p))
+            new_data_per_point.append(
+                np.reshape(this_data_per_point, (len(this_data_per_point), 1)))
+
+    # Extracting streamlines
+    return new_data_per_point
+
+
+def perform_operation_per_streamline(op_name, sft, dpp_name='metric',
+                                     endpoints_only=False):
+    """Performs an operation across all data points for each streamline.
+
+    Parameters
+    ----------
+    op_name: str
+        A callable that takes a list of streamline data per streamline and
+        returns a list of data per streamline.
+    sft: StatefulTractogram
+        The streamlines used in the operation.
+    dpp_name: str
+        The name of the data per point to be used in the operation.
+    endpoints_only: bool
+        If True, will only perform operation on endpoints
+
+    Returns
+    -------
+    new_sft: StatefulTractogram
+        sft with data per streamline resulting from the operation.
+    """
+    # Performing operation
+    call_op = OPERATIONS[op_name]
+    if endpoints_only:
+        new_data_per_streamline = []
+        for s in sft.data_per_point[dpp_name]:
+            start = s[0]
+            end = s[-1]
+            concat = np.concatenate((start[:], end[:]))
+            new_data_per_streamline.append(call_op(concat))
+    else:
+        new_data_per_streamline = []
+        for s in sft.data_per_point[dpp_name]:
+            s_np = np.asarray(s)
+            new_data_per_streamline.append(call_op(s_np))
+
+    return new_data_per_streamline
+
+
+def perform_pairwise_streamline_operation_on_endpoints(op_name, sft,
+                                                       dpp_name='metric'):
+    """Peforms an operation across endpoints for each streamline.
+
+    Parameters
+    ----------
+    op_name: str
+        A callable that takes a list of streamline data per streamline and
+        returns a list of data per streamline.
+    sft: StatefulTractogram
+        The streamlines used in the operation.
+    dpp_name: str
+        The name of the data per point to be used in the operation.
+
+    Returns
+    -------
+    new_sft: StatefulTractogram
+        sft with data per streamline resulting from the operation.
+    """
+    # Performing operation
+    call_op = OPERATIONS[op_name]
+    new_data_per_streamline = []
+    for s in sft.data_per_point[dpp_name]:
+        new_data_per_streamline.append(call_op(s[0], s[-1])[0, 1])
+
+    return new_data_per_streamline
+
+
+def stream_mean(array):
+    return np.squeeze(np.mean(array, axis=0))
+
+
+def stream_sum(array):
+    return np.squeeze(np.sum(array, axis=0))
+
+
+def stream_min(array):
+    return np.squeeze(np.min(array, axis=0))
+
+
+def stream_max(array):
+    return np.squeeze(np.max(array, axis=0))
+
+
+def stream_correlation(array1, array2):
+    return np.corrcoef(array1, array2)
+
+
+OPERATIONS = {
+    'mean': stream_mean,
+    'sum': stream_sum,
+    'min': stream_min,
+    'max': stream_max,
+    'correlation': stream_correlation,
+}

scilpy/tractograms/tests/test_dps_and_dpp_management.py

@@ -0,0 +1,3 @@
+def test_project_map_to_streamlines():
+    # toDo
+    pass

scilpy/tractograms/tests/test_streamline_operations.py

@@ -367,3 +367,18 @@ def test_smooth_line_spline():
     dist_2 = np.linalg.norm(noisy_streamline - smoothed_streamline)
 
     assert dist_1 < dist_2
+
+
+def test_perform_streamline_operation_per_point():
+    # toDo
+    pass
+
+
+def test_perform_operation_per_streamline():
+    # toDo
+    pass
+
+
+def test_perform_streamline_operation_on_endpoints():
+    # toDo
+    pass