"""SynthSeg ROI extraction, validation, and resampling node builders."""
from pathlib import Path
from typing import Optional
from nipype import Node
from nipype.interfaces.utility import Function
from thesis.core.config import PipelineConfig
from thesis.core.context import ProcessingContext
from ..common import resolve_with_fallback
from ..operations import (
extract_rois_task,
resample_label_map_task,
resample_rois_to_dwi_task,
validate_warped_rois_passthrough_task,
verify_waypoint_avoid_overlap_task,
)
[docs]
def prepare_synthseg_seg_resampler(t1_image: str, out_dir: Path) -> Node:
"""
Create a node that resamples the SynthSeg segmentation to the T1w voxel grid.
By resampling the integer label map **before** label extraction, every
binary mask produced by ``synthseg_roi_extractor`` (seed, waypoint, stop,
avoid, target) is already at the correct resolution and world-coordinate
frame. This removes the need for a separate per-mask resampling step and
ensures SynthSeg-derived masks never pass through the ANTs transformer.
Uses nearest-neighbour interpolation to preserve integer label values.
The ``input_image`` input is left unset here — it must be supplied either
by connecting the live SynthSeg output (meta-workflow) or by setting
``node.inputs.input_image`` to a pre-computed segmentation file.
Args:
t1_image: Absolute path to the T1w image (defines the target voxel
grid, same image used as SynthSeg input).
out_dir: Output directory; resampled file goes in
``<out_dir>/synthseg_resampled``.
Returns:
Configured Nipype Node wrapping :func:`resample_label_map_task`.
"""
resampler = Node(
Function(
input_names=["input_image", "reference", "output_dir"],
output_names=["resampled_segmentation"],
function=resample_label_map_task,
),
name="synthseg_seg_resampler",
)
resampler.inputs.reference = t1_image
resampler.inputs.output_dir = str(out_dir / "synthseg_resampled")
return resampler
[docs]
def prepare_roi_validator(
reference_image: str,
name: str = "roi_validator",
min_voxels: int = 10,
singularity_threshold: float = 1e-6,
volume_ratio_min: float = 0.5,
volume_ratio_max: float = 1.5,
) -> Node:
"""Create a validation node that checks ROI masks before tractography.
Transparent passthrough: accepts the five ROI paths, validates each mask
(voxel count + affine singularity + centroid plausibility), and re-emits
the same paths unchanged for the next downstream node.
Args:
reference_image: Absolute path to a reference image used for centroid
plausibility checks (e.g. the subject-space brain mask for atlas ROIs
or the T1w image for SynthSeg ROIs).
name: Nipype node name (default ``"roi_validator"``; use
``"synthseg_roi_validator"`` for the SynthSeg source).
min_voxels: Minimum non-zero voxel count per warped mask.
singularity_threshold: Minimum absolute determinant of affine 3x3 block.
volume_ratio_min: Lower bound for warped-vs-original voxel ratio.
volume_ratio_max: Upper bound for warped-vs-original voxel ratio.
Returns:
Configured Nipype Node wrapping
:func:`~thesis.workflows.hcp.operations.validation\
.validate_warped_rois_passthrough_task`.
"""
validator = Node(
Function(
input_names=[
"seed",
"waypoints_file",
"stop_mask",
"avoid_mask",
"target_mask",
"reference_image",
"min_voxels",
"singularity_threshold",
"volume_ratio_min",
"volume_ratio_max",
],
output_names=["seed", "waypoints_file", "stop_mask", "avoid_mask", "target_mask"],
function=validate_warped_rois_passthrough_task,
),
name=name,
)
validator.inputs.seed = ""
validator.inputs.waypoints_file = ""
validator.inputs.stop_mask = ""
validator.inputs.avoid_mask = ""
validator.inputs.target_mask = ""
validator.inputs.reference_image = reference_image
validator.inputs.min_voxels = min_voxels
validator.inputs.singularity_threshold = singularity_threshold
validator.inputs.volume_ratio_min = volume_ratio_min
validator.inputs.volume_ratio_max = volume_ratio_max
return validator
[docs]
def prepare_waypoint_avoid_overlap_verifier(
name: str = "waypoint_avoid_verifier",
max_overlap_fraction: float = 0.9,
) -> Node:
"""Pre-ProbTrackX2 guard that fails if any waypoint sits inside the avoid mask.
The five ROI inputs are left unset here — they must be wired from the
upstream DWI resampler. Pure passthrough on success.
Args:
name: Nipype node name.
max_overlap_fraction: Maximum allowed waypoint-vs-avoid overlap before
raising. Default 0.9 (90%) — targets the catastrophic
wrong-hemisphere case (~100% overlap) without tripping on
anatomically inherent overlap (e.g. peduncle near brainstem
boundary often shows ~30-40%).
Returns:
Configured Nipype Node wrapping :func:`verify_waypoint_avoid_overlap_task`.
"""
node = Node(
Function(
input_names=[
"seed",
"waypoints_file",
"stop_mask",
"avoid_mask",
"target_mask",
"max_overlap_fraction",
],
output_names=["seed", "waypoints_file", "stop_mask", "avoid_mask", "target_mask"],
function=verify_waypoint_avoid_overlap_task,
),
name=name,
)
node.inputs.seed = ""
node.inputs.waypoints_file = ""
node.inputs.stop_mask = ""
node.inputs.avoid_mask = ""
node.inputs.target_mask = ""
node.inputs.max_overlap_fraction = float(max_overlap_fraction)
return node
[docs]
def prepare_roi_dwi_resampler(
dwi_mask_path: str, out_dir: Path, name: str = "roi_dwi_resampler"
) -> Node:
"""Create a node that brings ROI masks from T1 world into the DWI grid.
The node has two modes determined at runtime by whether its
``t1_to_dwi_transform`` input is connected:
* When connected to ``preprocess.dwi_to_t1_registration.composite_transform``
(i.e. meta-workflows that include a preprocess stage), the masks are
warped with ``antsApplyTransforms`` using the inverted DWI→T1 transform.
This bridges the world-coordinate gap between the atlas-warped masks
(T1 world) and the DWI grid; without it, ProbTrackX2 silently produces
anatomically-displaced tracks and tckgen+ACT rejects every streamline.
* When unconnected (standalone runs on HCP-preprocessed data where T1 and
DWI already share a world), the legacy ``nilearn.image.resample_to_img``
regrid is used.
The five ROI inputs and the ``t1_to_dwi_transform`` input are left unset
here and must be connected from upstream nodes via ``workflow.connect``.
Args:
dwi_mask_path: Absolute path to the DWI brain mask (e.g.
``nodif_brain_mask.nii.gz``) that defines the target voxel grid.
out_dir: Output directory; resampled files go in
``<out_dir>/rois_dwi_resampled``.
name: Nipype node name (default ``"roi_dwi_resampler"``).
Returns:
Configured Nipype Node wrapping
:func:`~thesis.workflows.hcp.operations.resampling.resample_rois_to_dwi_task`.
"""
resampler = Node(
Function(
input_names=[
"seed",
"waypoints_file",
"stop_mask",
"avoid_mask",
"target_mask",
"reference",
"output_dir",
"t1_to_dwi_transform",
"hemisphere",
],
output_names=["seed", "waypoints_file", "stop_mask", "avoid_mask", "target_mask"],
function=resample_rois_to_dwi_task,
),
name=name,
)
resampler.inputs.seed = ""
resampler.inputs.waypoints_file = ""
resampler.inputs.stop_mask = ""
resampler.inputs.avoid_mask = ""
resampler.inputs.target_mask = ""
resampler.inputs.t1_to_dwi_transform = ""
if Path(dwi_mask_path).exists():
resampler.inputs.reference = dwi_mask_path
resampler.inputs.output_dir = str(out_dir / "rois_dwi_resampled")
resampler.inputs.hemisphere = "both"
return resampler
