"""tckgen, tcksift2, tckmap, and waytotal writer nodes."""
from pathlib import Path
from nipype import Node
from nipype.interfaces.mrtrix3 import ComputeTDI
from nipype.interfaces.utility import Function
from ..config.params import MRtrix3Params
from ..operations import (
adapt_tckgen_inputs_task,
make_per_target_maps_task,
rename_density_map_task,
run_tckgen_task,
run_tcksift2_task,
write_mrtrix3_params_task,
write_waytotal_task,
)
[docs]
def prepare_tckgen_node(
params: MRtrix3Params,
out_dir: Path,
name: str = "tckgen",
) -> Node:
"""Configure ``tckgen`` as a Function node wrapping the binary directly.
Nipype's ``Tractography`` interface declares ``roi_incl`` / ``roi_excl``
as single-value traits, which cannot represent the multiple ``-include``
and ``-exclude`` flags MRtrix3 supports. This builder uses a Function
node that constructs the command line itself so any number of include
and exclude masks can be wired through.
ACT-specific flags (``-backtrack``, ``-crop_at_gmwmi``) are set from the
``params`` dict; the 5TT input is always passed as ``act_file`` and the
workflow controls the wiring.
Args:
params: MRtrix3 parameter dictionary.
out_dir: Output directory; ``tracks.tck`` is written here.
name: Nipype node name.
Returns:
Configured Nipype Function Node with output field ``out_file``.
"""
out_dir.mkdir(parents=True, exist_ok=True)
node = Node(
Function(
input_names=[
"fod_file",
"act_file",
"seed_image",
"seed_gmwmi",
"include_masks",
"exclude_masks",
"mask_stop",
"output_dir",
"algorithm",
"select",
"seeds",
"min_length",
"max_length",
"backtrack",
"crop_at_gmwmi",
"cutoff",
],
output_names=["out_file"],
function=run_tckgen_task,
),
name=name,
)
node.inputs.output_dir = str(out_dir)
# Seed/filter inputs default to empty so an unconnected input (whole-brain
# gmwmi seeding wires only seed_gmwmi; ROI seeding wires only seed_image) is
# passed deterministically rather than as traits.Undefined. The workflow's
# connections override whichever apply.
node.inputs.seed_image = ""
node.inputs.seed_gmwmi = ""
node.inputs.include_masks = []
node.inputs.exclude_masks = []
node.inputs.mask_stop = ""
node.inputs.algorithm = params["tckgen_algorithm"]
node.inputs.select = int(params["tckgen_select"])
node.inputs.seeds = int(params["tckgen_seeds"]) if params["tckgen_seeds"] else 0
node.inputs.min_length = float(params["tckgen_minlength"])
node.inputs.max_length = float(params["tckgen_maxlength"])
node.inputs.backtrack = bool(params["tckgen_backtrack"])
node.inputs.crop_at_gmwmi = bool(params["tckgen_crop_at_gmwmi"])
node.inputs.cutoff = params.get("tckgen_cutoff")
return node
[docs]
def prepare_tcksift2_node(
out_dir: Path,
name: str = "tcksift2",
) -> Node:
"""Configure ``tcksift2`` (per-streamline weighting + mu coefficient).
No Nipype interface exists for tcksift2, so this is a Function node
wrapping a subprocess call. The streamline file, FOD, and 5TT inputs
are wired by the workflow.
Args:
out_dir: Output directory; ``sift2_weights.txt`` and ``sift2_mu.txt``
are written here.
name: Nipype node name.
Returns:
Configured Nipype Node with outputs ``weights_file`` and ``mu_file``.
"""
out_dir.mkdir(parents=True, exist_ok=True)
node = Node(
Function(
input_names=["tracks_file", "fod_file", "fivett_file", "output_dir"],
output_names=["weights_file", "mu_file"],
function=run_tcksift2_task,
),
name=name,
)
node.inputs.output_dir = str(out_dir)
return node
[docs]
def prepare_tckmap_node(
template_image: Path,
out_dir: Path,
name: str = "tckmap",
) -> Node:
"""Configure ``tckmap`` (track-density imaging).
SIFT2 weighting is controlled entirely by the workflow, which decides
whether to wire the SIFT2 weights file into ``tck_weights``.
Args:
template_image: Reference image whose grid the density map adopts
(typically the WM FOD or another DWI-grid image).
out_dir: Output directory; the raw ComputeTDI output lands here
and is later renamed to ``fdt_paths.nii.gz``.
name: Nipype node name.
Returns:
Configured Nipype Node.
"""
out_dir.mkdir(parents=True, exist_ok=True)
node = Node(ComputeTDI(), name=name)
# ``ComputeTDI.reference`` carries an ``exists=True`` trait; skip the
# static set when the template doesn't exist at build time so a parent
# meta-workflow (e.g. full_pipeline with the mrtrix3 backend) can wire it at runtime.
if Path(template_image).exists():
node.inputs.reference = str(template_image)
node.inputs.out_file = str(out_dir / "tdi_raw.nii.gz")
node.inputs.args = "-force"
return node
[docs]
def prepare_density_map_renamer(
out_dir: Path,
name: str = "fdt_paths_writer",
) -> Node:
"""Copy the tckmap output to ``fdt_paths.nii.gz`` for downstream tools."""
node = Node(
Function(
input_names=["in_file", "output_dir"],
output_names=["out_file"],
function=rename_density_map_task,
),
name=name,
)
node.inputs.output_dir = str(out_dir)
return node
[docs]
def prepare_waytotal_writer(
out_dir: Path,
name: str = "waytotal_writer",
) -> Node:
"""Write a single-line ``waytotal`` file for atlas / tract_similarity.
Declares an optional ``weights_file`` input so callers can wire the
SIFT2 weights file when present; the underlying task sums those
weights instead of relying on ``tckinfo -count``.
"""
node = Node(
Function(
input_names=["tracks_file", "output_dir", "weights_file"],
output_names=["waytotal_file"],
function=write_waytotal_task,
),
name=name,
)
node.inputs.output_dir = str(out_dir)
node.inputs.weights_file = None
return node
[docs]
def prepare_per_target_maps_node(
out_dir: Path,
name: str = "per_target_maps",
) -> Node:
"""Configure the per-target ``seeds_to_<target>.nii.gz`` writer.
Wraps ``make_per_target_maps_task`` (tckedit + tckmap) so MRtrix3
matches the ProbTrackX2 per-target filename layout that
``qc.checks.collect_connectivity_map_stats`` already consumes.
Args:
out_dir: Per-hemisphere tractography output directory.
name: Nipype node name.
Returns:
Configured Function node with output ``seeds_files`` (list of
absolute paths).
"""
out_dir.mkdir(parents=True, exist_ok=True)
node = Node(
Function(
input_names=[
"tracks_file",
"reference_image",
"output_dir",
"target_mask",
"target_name",
"weights_file",
],
output_names=["seeds_files"],
function=make_per_target_maps_task,
),
name=name,
)
node.inputs.output_dir = str(out_dir)
node.inputs.target_name = "target_dwi"
node.inputs.weights_file = None
return node
[docs]
def prepare_mrtrix3_params_writer(
out_dir: Path,
select: int,
name: str = "mrtrix3_params_writer",
) -> Node:
"""Write ``tractography_params.json`` (backend, select, mu) for stats.
Args:
out_dir: Per-hemisphere tractography output directory.
select: The ``-select`` count passed to ``tckgen``.
name: Nipype node name.
Returns:
Configured Function node with output ``params_file``.
"""
out_dir.mkdir(parents=True, exist_ok=True)
node = Node(
Function(
input_names=["output_dir", "select", "mu_file"],
output_names=["params_file"],
function=write_mrtrix3_params_task,
),
name=name,
)
node.inputs.output_dir = str(out_dir)
node.inputs.select = int(select)
node.inputs.mu_file = None
return node
