"""Mask resampling utilities for aligning ROIs to diffusion space.
NOTE: Runs inside a Nipype Function node; the thesis logger is not available at
runtime. ``sys.stderr.write()`` may be used for progress output instead.
"""
[docs]
def resample_label_map_task(input_image: str, reference: str, output_dir: str) -> str:
"""
Resample a label map to match a reference image's voxel grid.
Used to bring the SynthSeg segmentation into the T1w voxel grid **before**
label extraction, so every extracted binary mask (seed, waypoint, stop,
avoid, target) is already at the correct resolution — no per-mask
resampling step is needed afterwards.
Uses nearest-neighbour interpolation to preserve integer label values.
``copy_header=True`` copies the reference image's full NIfTI header
(including sform/qform codes) to the output.
Args:
input_image: Path to the SynthSeg label map NIfTI file, or ``""``
if absent.
reference: Path to a reference NIfTI image that defines the target
voxel grid (e.g. the T1w image used as SynthSeg input).
output_dir: Directory for the resampled output file.
Returns:
Path to the resampled label map, or the original ``input_image`` path
unchanged when the input image is absent.
"""
import sys
from pathlib import Path
import nibabel as nib
from nilearn.image import resample_to_img
if not input_image or not Path(input_image).exists():
return input_image
out_path = Path(output_dir)
out_path.mkdir(parents=True, exist_ok=True)
out_file = str(out_path / "segmentation_resampled.nii.gz")
resampled = resample_to_img(
source_img=input_image,
target_img=reference,
interpolation="nearest",
copy_header=True,
fill_value=0,
)
nib.save(resampled, out_file)
sys.stderr.write(
f"[synthseg_resampler] Resampled segmentation to reference space: {out_file}\n"
)
sys.stderr.flush()
return out_file
[docs]
def resample_rois_to_dwi_task(
seed: str,
waypoints_file: str,
stop_mask: str,
avoid_mask: str,
target_mask: str,
reference: str,
output_dir: str,
t1_to_dwi_transform: str = "",
hemisphere: str = "both",
) -> tuple:
"""Resample all ROI masks to a DWI reference grid before tractography.
ProbTrackX2 / tckgen require that every mask (seed, waypoints, stop,
avoid, target) shares the voxel grid AND world coordinates of the
diffusion data. When the ROI masks come out of the atlas warp in T1
world but the DWI was acquired in a different physical orientation
(typical for non-HCP-preprocessed datasets), a pure grid-only regrid
leaves the data at the wrong DWI voxels and tractography produces 0
streamlines under ACT or anatomically-displaced streamlines under
ProbTrackX2.
Two modes:
* **Transform mode** (``t1_to_dwi_transform`` non-empty and pointing at
a valid file): the path is the ANTs composite transform produced by
preprocess.dwi_to_t1_registration (direction: DWI→T1). Each mask is
warped via ``antsApplyTransforms -t [transform,1]`` (inverted to give
T1→DWI) with NearestNeighbor interpolation, landing on the DWI grid
at the correct anatomical positions.
* **Pass-through mode** (``t1_to_dwi_transform`` empty): falls back to
``nilearn.image.resample_to_img`` with nearest-neighbour
interpolation. This is correct when the input ROIs and reference
already share a world (e.g. HCP-preprocessed data where T1 and DWI
have both been ACPC-aligned upstream).
NOTE: Runs inside a Nipype Function node; use ``print()`` for logging.
Args:
seed: Path to seed ROI mask, or ``""`` if absent.
waypoints_file: Path to a text file listing waypoint mask paths, one
per line, or ``""`` if absent.
stop_mask: Path to stop mask, or ``""`` if absent.
avoid_mask: Path to avoid mask, or ``""`` if absent.
target_mask: Path to target mask, or ``""`` if absent.
reference: Path to the DWI reference image (e.g.
``nodif_brain_mask.nii.gz``) that defines the target voxel grid.
output_dir: Directory where resampled masks are written.
t1_to_dwi_transform: Optional path to an ANTs composite transform
(DWI→T1 direction). When supplied, antsApplyTransforms with the
inverse transform is used; when empty, nilearn regrid only.
hemisphere: ``"left"``, ``"right"``, or ``"both"`` (default). Scopes
``<output_dir>/<hemisphere>/`` for left/right so parallel
hemisphere iterations under ``--hemisphere both-separately`` don't
race on identically-named resampled mask paths.
Returns:
Tuple of ``(seed, waypoints_file, stop_mask, avoid_mask, target_mask)``
paths pointing to the resampled files. Absent inputs (empty string or
non-existent path) are returned unchanged.
"""
import sys
from pathlib import Path
out_path = Path(output_dir)
if hemisphere in ("left", "right"):
out_path = out_path / hemisphere
out_path.mkdir(parents=True, exist_ok=True)
use_transform = bool(t1_to_dwi_transform) and Path(t1_to_dwi_transform).is_file()
if use_transform:
import subprocess
def _resample_one(mask_path: str, out_name: str) -> str:
"""Warp a mask T1→DWI via antsApplyTransforms (inverse of dwi_to_t1)."""
if not mask_path or not Path(mask_path).exists():
return mask_path
out_file = str(out_path / out_name)
cmd = [
"antsApplyTransforms",
"-d",
"3",
"-i",
str(mask_path),
"-r",
str(reference),
"-o",
out_file,
"-t",
f"[{t1_to_dwi_transform},1]",
"-n",
"NearestNeighbor",
]
print(f"[roi_dwi_resampler] {' '.join(cmd)}")
sys.stdout.flush()
subprocess.run(cmd, check=True)
return out_file
else:
import nibabel as nib
from nilearn.image import resample_to_img
def _resample_one(mask_path: str, out_name: str) -> str:
"""Regrid a mask to the reference voxel grid (assumes shared world)."""
if not mask_path or not Path(mask_path).exists():
return mask_path
out_file = str(out_path / out_name)
resampled = resample_to_img(
source_img=mask_path,
target_img=reference,
interpolation="nearest",
copy_header=True,
fill_value=0,
)
nib.save(resampled, out_file)
print(f"[roi_dwi_resampler] {mask_path} -> {out_file}")
sys.stdout.flush()
return out_file
resampled_seed = _resample_one(seed, "seed_dwi.nii.gz")
resampled_waypoints_file = waypoints_file
if waypoints_file and Path(waypoints_file).exists():
wp_out_dir = out_path / "waypoints"
wp_out_dir.mkdir(parents=True, exist_ok=True)
with open(waypoints_file, "r", encoding="utf-8") as fh:
wp_paths = [line.strip() for line in fh if line.strip()]
resampled_wp_paths = []
for wp in wp_paths:
wp_name = Path(wp).name
resampled_wp = _resample_one(wp, str(Path("waypoints") / wp_name))
resampled_wp_paths.append(resampled_wp)
resampled_waypoints_file = str(out_path / "waypoints_dwi.txt")
with open(resampled_waypoints_file, "w", encoding="utf-8") as fh:
for wp in resampled_wp_paths:
fh.write(wp + "\n")
resampled_stop = _resample_one(stop_mask, "stop_dwi.nii.gz")
resampled_avoid = _resample_one(avoid_mask, "avoid_dwi.nii.gz")
resampled_target = _resample_one(target_mask, "target_dwi.nii.gz")
return (
resampled_seed,
resampled_waypoints_file,
resampled_stop,
resampled_avoid,
resampled_target,
)
