Source code for cmp.stages.connectome.connectome

# Copyright (C) 2009-2022, Ecole Polytechnique Federale de Lausanne (EPFL) and
# Hospital Center and University of Lausanne (UNIL-CHUV), Switzerland, and CMP3 contributors
# All rights reserved.
#
#  This software is distributed under the open-source license Modified BSD.

"""Definition of config and stage classes for building structural connectivity matrices."""

# Global imports
import os

from traits.api import *

import networkx as nx

# Nipype imports
import nipype.interfaces.utility as util
import nipype.pipeline.engine as pe

# Own imports
from cmp.stages.common import Stage
import cmtklib.connectome
from cmtklib.util import get_pipeline_dictionary_outputs


[docs]class ConnectomeConfig(HasTraits): """Class used to store configuration parameters of a :class:`~cmp.stages.connectome.connectome.ConnectomeStage` instance. Attributes ---------- compute_curvature : traits.Bool Compute fiber curvature (Default: False) output_types : ['gpickle', 'mat', 'graphml'] Output connectome format connectivity_metrics : ['Fiber number', 'Fiber length', 'Fiber density', 'Fiber proportion', 'Normalized fiber density', 'ADC', 'gFA'] Set of connectome maps to compute log_visualization : traits.Bool Log visualization that might be obsolete as this has been detached after creation of the bidsappmanager (Default: True) circular_layout : traits.Bool Visualization of the connectivity matrix using a circular layout that might be obsolete as this has been detached after creation of the bidsappmanager (Default: False) subject : traits.Str BIDS subject ID (in the form ``sub-XX``) See Also -------- cmp.stages.connectome.connectome.ConnectomeStage """ # modality = List(['Deterministic','Probabilistic']) compute_curvature = Bool(False) output_types = List(["gpickle", "mat", "graphml"]) connectivity_metrics = List( [ "Fiber number", "Fiber length", "Fiber density", "Fiber proportion", "Normalized fiber density", "ADC", "gFA", ] ) log_visualization = Bool(True) circular_layout = Bool(False) subject = Str
[docs]class ConnectomeStage(Stage): """Class that represents the connectome building stage of a :class:`~cmp.pipelines.diffusion.diffusion.DiffusionPipeline`. Methods ------- create_workflow() Create the workflow of the diffusion `ConnectomeStage` See Also -------- cmp.pipelines.diffusion.diffusion.DiffusionPipeline cmp.stages.connectome.connectome.ConnectomeConfig """ def __init__(self, bids_dir, output_dir): """Constructor of a :class:`~cmp.stages.connectome.connectome.Connectome` instance.""" self.name = "connectome_stage" self.bids_dir = bids_dir self.output_dir = output_dir self.config = ConnectomeConfig() self.inputs = [ "roi_volumes_registered", "roi_graphMLs", "track_file", "parcellation_scheme", "atlas_info", "FA", "ADC", "AD", "RD", "skewness", "kurtosis", "P0", "shore_maps", "mapmri_maps", ] self.outputs = [ "endpoints_file", "endpoints_mm_file", "final_fiberslength_files", "filtered_fiberslabel_files", "final_fiberlabels_files", "streamline_final_file", "connectivity_matrices", ]
[docs] def create_workflow(self, flow, inputnode, outputnode): """Create the stage workflow. Parameters ---------- flow : nipype.pipeline.engine.Workflow The nipype.pipeline.engine.Workflow instance of the Diffusion pipeline inputnode : nipype.interfaces.utility.IdentityInterface Identity interface describing the inputs of the stage outputnode : nipype.interfaces.utility.IdentityInterface Identity interface describing the outputs of the stage """ cmtk_cmat = pe.Node( interface=cmtklib.connectome.DmriCmat(), name="compute_matrice" ) cmtk_cmat.inputs.compute_curvature = self.config.compute_curvature cmtk_cmat.inputs.output_types = self.config.output_types # Additional maps map_merge = pe.Node(interface=util.Merge(9), name="merge_additional_maps") # fmt: off flow.connect( [ (inputnode, map_merge, [("FA", "in1"), ("ADC", "in2"), ("AD", "in3"), ("RD", "in4"), ("skewness", "in5"), ("kurtosis", "in6"), ("P0", "in7"), ("shore_maps", "in8"), ("mapmri_maps", "in9")]), (map_merge, cmtk_cmat, [("out", "additional_maps")]), (inputnode, cmtk_cmat, [("track_file", "track_file"), ("roi_graphMLs", "roi_graphmls"), ("parcellation_scheme", "parcellation_scheme"), ("atlas_info", "atlas_info"), ("roi_volumes_registered", "roi_volumes")]), (cmtk_cmat, outputnode, [("endpoints_file", "endpoints_file"), ("endpoints_mm_file", "endpoints_mm_file"), ("final_fiberslength_files", "final_fiberslength_files"), ("filtered_fiberslabel_files", "filtered_fiberslabel_files"), ("final_fiberlabels_files", "final_fiberlabels_files"), ("streamline_final_file", "streamline_final_file"), ("connectivity_matrices", "connectivity_matrices")]), ] )
# fmt: on
[docs] def define_inspect_outputs(self): # pragma: no cover """Update the `inspect_outputs` class attribute. It contains a dictionary of stage outputs with corresponding commands for visual inspection. """ # print('inspect outputs connectome stage') dwi_sinker_dir = os.path.join( os.path.dirname(self.stage_dir), "dwi_datasinker" ) dwi_sinker_report = os.path.join(dwi_sinker_dir, "_report", "report.rst") if os.path.exists(dwi_sinker_report): dwi_outputs = get_pipeline_dictionary_outputs( dwi_sinker_report, self.output_dir ) tracto = dwi_outputs["dwi.@streamline_final_file"] if os.path.exists(tracto): self.inspect_outputs_dict["Final tractogram"] = ["trackvis", tracto] mat = dwi_outputs["dwi.@connectivity_matrices"] map_scale = "default" if self.config.log_visualization: map_scale = "log" if self.config.circular_layout: layout = "circular" else: layout = "matrix" if isinstance(mat, str): # print("is str") if "gpickle" in mat: # 'Fiber number','Fiber length','Fiber density','ADC','gFA' con_name = os.path.basename(mat).split(".")[0].split("_")[-1] # print("con_name:"+con_name) # Load the connectivity matrix and extract the attributes (weights) # con_mat = pickle.load(mat, encoding="latin1") con_mat = nx.read_gpickle(mat) con_metrics = list(list(con_mat.edges(data=True))[0][2].keys()) # Create dynamically the list of output connectivity metrics for inspection for con_metric in con_metrics: metric_str = " ".join(con_metric.split("_")) self.inspect_outputs_dict[con_name + " - " + metric_str] = [ "showmatrix_gpickle", layout, mat, con_metric, "False", self.config.subject + " - " + con_name + " - " + metric_str, map_scale, ] else: # print("is list") for mat in dwi_outputs["dwi.@connectivity_matrices"]: # print("mat : %s" % mat) if "gpickle" in mat: con_name = " ".join( os.path.basename(mat).split(".")[0].split("_") ) # print("con_name:"+con_name) # Load the connectivity matrix and extract the attributes (weights) # con_mat = pickle.load(mat, encoding="latin1") con_mat = nx.read_gpickle(mat) con_metrics = list(list(con_mat.edges(data=True))[0][2].keys()) # Create dynamically the list of output connectivity metrics for inspection for con_metric in con_metrics: metric_str = " ".join(con_metric.split("_")) self.inspect_outputs_dict[con_name + " - " + metric_str] = [ "showmatrix_gpickle", layout, mat, con_metric, "False", self.config.subject + " - " + con_name + " - " + metric_str, map_scale, ] self.inspect_outputs = sorted( [key for key in list(self.inspect_outputs_dict.keys())], key=str.lower )
# print(self.inspect_outputs)