# Copyright (C) 2018-2023 C-PAC Developers
# This file is part of C-PAC.
# C-PAC is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
# C-PAC is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details.
# You should have received a copy of the GNU Lesser General Public
# License along with C-PAC. If not, see <https://www.gnu.org/licenses/>.
from nipype.interfaces import fsl
from CPAC.network_centrality.network_centrality import create_centrality_wf
from CPAC.network_centrality.utils import check_centrality_params, create_merge_node
from CPAC.pipeline import nipype_pipeline_engine as pe
from CPAC.pipeline.nodeblock import nodeblock
from CPAC.pipeline.schema import valid_options
[docs]
def connect_centrality_workflow(
workflow,
c,
resample_functional_to_template,
template_node,
template_out,
merge_node,
method_option,
pipe_num,
):
"""
.. exec::
from nipype.interfaces.utility import IdentityInterface
from CPAC.network_centrality.utils import create_merge_node
from CPAC.network_centrality.pipeline import \
connect_centrality_workflow
from CPAC.pipeline import nipype_pipeline_engine as pe
from CPAC.pipeline.schema import valid_options
from CPAC.utils.configuration import Configuration
from CPAC.utils.interfaces.function import Function
wf = pe.Workflow(name='centrality')
# Sphinx didn't like a comprehension here
_d = {'network_centrality': {}}
for method in ['degree_centrality',
'eigenvector_centrality',
'local_functional_connectivity_density']:
_d['network_centrality'][method] = {
'weight_options': valid_options['centrality']['weight_options'],
'correlation_threshold_option': 'Significance threshold',
'correlation_threshold': 0.001}
cfg = Configuration(_d)
pipe_num = 0
resample_functional_to_template = pe.Node(
IdentityInterface(fields=['in_file', 'reference', 'out_file']),
name='resample_functional_to_template')
template_node = pe.Node(
IdentityInterface(fields=['template_node']),
name='template_node')
merge_node = create_merge_node(pipe_num)
for option in valid_options['centrality']['method_options']:
if cfg.network_centrality[option]['weight_options']:
connect_centrality_workflow(
wf, cfg, resample_functional_to_template, template_node,
'template_node', merge_node, option, 0)
wf.write_graph(
graph2use='orig',
dotfilename='./images/generated/network_centrality.dot')
wf.write_graph(
graph2use='hierarchical',
dotfilename='./images/generated/network_centrality.dot')
High Level Workflow Graph:
.. image:: ../../images/generated/network_centrality.png
:width: 500
Detailed Workflow Graph:
.. image:: ../../images/generated/network_centrality_detailed.png
:width: 500
"""
# Set method_options variables
if method_option == "degree_centrality":
out_list = "deg_list"
elif method_option == "eigenvector_centrality":
out_list = "eig_list"
elif method_option == "local_functional_connectivity_density":
out_list = "lfcd_list"
threshold_option = c.network_centrality[method_option][
"correlation_threshold_option"
]
threshold = c.network_centrality[method_option]["correlation_threshold"]
# Init workflow name and resource limits
wf_name = f"afni_centrality_{method_option}_{pipe_num}"
num_threads = c.pipeline_setup["system_config"]["max_cores_per_participant"]
memory = c.network_centrality["memory_allocation"]
# Format method and threshold options properly and check for
# errors
method_option, threshold_option = check_centrality_params(
method_option, threshold_option, threshold
)
# Change sparsity thresholding to % to work with afni
if threshold_option == "Sparsity threshold":
threshold = threshold * 100
afni_centrality_wf = create_centrality_wf(
wf_name,
method_option,
c.network_centrality[method_option]["weight_options"],
threshold_option,
threshold,
num_threads,
memory,
)
workflow.connect(
resample_functional_to_template,
"out_file",
afni_centrality_wf,
"inputspec.in_file",
)
workflow.connect(
template_node, template_out, afni_centrality_wf, "inputspec.template"
)
if "degree" in method_option:
out_list = "deg_list"
elif "eigen" in method_option:
out_list = "eig_list"
elif "lfcd" in method_option:
out_list = "lfcd_list"
workflow.connect(
afni_centrality_wf, "outputspec.outfile_list", merge_node, out_list
)
[docs]
@nodeblock(
name="network_centrality",
config=["network_centrality"],
switch=["run"],
inputs=[
("space-template_desc-preproc_bold", "T1w-brain-template-funcreg"),
"template-specification-file",
],
outputs={
"space-template_dcw": {"Template": "T1w-brain-template-funcreg"},
"space-template_dcb": {"Template": "T1w-brain-template-funcreg"},
"space-template_ecw": {"Template": "T1w-brain-template-funcreg"},
"space-template_ecb": {"Template": "T1w-brain-template-funcreg"},
"space-template_lfcdw": {"Template": "T1w-brain-template-funcreg"},
"space-template_lfcdb": {"Template": "T1w-brain-template-funcreg"},
},
)
def network_centrality(wf, cfg, strat_pool, pipe_num, opt=None):
"""Run Network Centrality."""
# Resample the functional mni to the centrality mask resolution
resample_functional_to_template = pe.Node(
interface=fsl.FLIRT(),
name=f"resample_functional_to_template_{pipe_num}",
mem_gb=4.0,
)
resample_functional_to_template.inputs.set(
interp="trilinear",
in_matrix_file=cfg.registration_workflows["functional_registration"][
"func_registration_to_template"
]["FNIRT_pipelines"]["identity_matrix"],
apply_xfm=True,
)
node, out = strat_pool.get_data("space-template_desc-preproc_bold")
wf.connect(node, out, resample_functional_to_template, "in_file")
node, out = strat_pool.get_data("template-specification-file")
wf.connect(node, out, resample_functional_to_template, "reference")
merge_node = create_merge_node(pipe_num)
outputs = {}
for option in valid_options["centrality"]["method_options"]:
if cfg.network_centrality[option]["weight_options"]:
connect_centrality_workflow(
wf,
cfg,
resample_functional_to_template,
node,
out,
merge_node,
option,
pipe_num,
)
for weight in cfg.network_centrality[option]["weight_options"]:
_option = option.lower()
_weight = weight.lower()
if "degree" in _option:
if "weight" in _weight:
outputs["space-template_dcw"] = (merge_node, "degree_weighted")
elif "binarize" in _weight:
outputs["space-template_dcb"] = (merge_node, "degree_binarized")
elif "eigen" in _option:
if "weight" in _weight:
outputs["space-template_ecw"] = (merge_node, "eigen_weighted")
elif "binarize" in _weight:
outputs["space-template_ecb"] = (merge_node, "eigen_binarized")
elif "lfcd" in _option or "local_functional" in _option:
if "weight" in _weight:
outputs["space-template_lfcdw"] = (merge_node, "lfcd_weighted")
elif "binarize" in _weight:
outputs["space-template_lfcdb"] = (merge_node, "lfcd_binarized")
return (wf, outputs)