rbc/tests/integration/test_functional.py at 5e9083ef1988e144d2187d0df29f3adf3aaa7080 · childmindresearch/rbc · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""Integration tests for functional workflow."""

from types import SimpleNamespace
from typing import cast

import nibabel as nib
import pytest
from niwrap import afni

from rbc.core.common import reorient
from rbc.core.functional import (
    generate_motion_reference,
    motion_correction,
    truncate_trs,
)


def test_truncate_trs(test_subject: SimpleNamespace) -> None:
    """Test truncating initial TRs from BOLD timeseries."""
    original_count = cast("nib.Nifti1Image", nib.load(test_subject.bold)).shape[3]

    start_tr = 4
    reoriented = reorient(
        in_file=test_subject.bold, output_fname="test_reoriented.nii.gz"
    )
    truncated_bold = truncate_trs(
        in_file=reoriented.out_file,
        output_fname="test_truncated.nii.gz",
        start_tr=start_tr,
    )
    # Test truncated BOLD file exists & volume count is reduced
    assert truncated_bold.output_file.exists()
    new_shape = cast("nib.Nifti1Image", nib.load(truncated_bold.output_file)).shape
    assert new_shape[3] == original_count - start_tr


def test_truncate_to_min_volume(test_subject: SimpleNamespace) -> None:
    """Test truncating to minimum volume count of 1."""
    original_count = cast("nib.Nifti1Image", nib.load(test_subject.bold)).shape[3]

    start_tr = original_count - 1
    truncated_bold = truncate_trs(
        in_file=test_subject.bold,
        output_fname="test_truncated_min.nii.gz",
        start_tr=start_tr,
    )
    # Test truncated BOLD file volume count is 1
    new_shape = cast("nib.Nifti1Image", nib.load(truncated_bold.output_file)).shape
    nvols = new_shape[3] if len(new_shape) > 3 else 1
    assert nvols == 1


def test_motion_reference_volume_count(test_subject: SimpleNamespace) -> None:
    """Test motion reference volume count is 1."""
    reference = generate_motion_reference(
        in_file=test_subject.bold, output_fname="test_motion_ref.nii.gz"
    )
    # Test motion reference file has 1 volume
    ref_shape = cast("nib.Nifti1Image", nib.load(reference.output_file)).shape
    nvols = ref_shape[3] if len(ref_shape) > 3 else 1
    assert nvols == 1


@pytest.mark.slow
def test_motion_correction_10vols(test_subject: SimpleNamespace) -> None:
    """Test motion correction on 10 volumes of BOLD timeseries."""
    reoriented = reorient(
        in_file=test_subject.bold, output_fname="test_reoriented.nii.gz"
    )
    truncated_10 = afni.v_3dcalc(
        dataset_a=afni.v_3dcalc_dataset_a_file(
            file=reoriented.out_file, selectors_="[0..9]"
        ),
        expression="a",
        prefix="test_10vols.nii.gz",
    )
    motion_reference = generate_motion_reference(
        in_file=truncated_10.output_file, output_fname="test_ref_10v.nii.gz"
    )
    motion_corrected = motion_correction(
        in_file=truncated_10.output_file,
        ref_file=motion_reference.output_file,
        output_prefix="test_mc_10v",
    )
    assert motion_corrected.bold.with_suffix(".nii.gz").exists()
    mc_shape = cast(
        "nib.Nifti1Image", nib.load(motion_corrected.bold.with_suffix(".nii.gz"))
    ).shape
    assert mc_shape[3] == 10

    assert motion_corrected.par.exists()
    par_data = motion_corrected.par.read_text().splitlines()
    assert len(par_data) == 10


@pytest.mark.slow
def test_motion_correction(test_subject: SimpleNamespace) -> None:
    """Test motion correction on full BOLD timeseries."""
    reoriented = reorient(
        in_file=test_subject.bold, output_fname="test_reoriented.nii.gz"
    )
    truncated = truncate_trs(
        in_file=reoriented.out_file, output_fname="test_truncated.nii.gz", start_tr=2
    )
    motion_reference = generate_motion_reference(
        in_file=truncated.output_file, output_fname="test_full_ref.nii.gz"
    )
    motion_corrected = motion_correction(
        in_file=truncated.output_file,
        ref_file=motion_reference.output_file,
        output_prefix="test_full_mc",
    )
    # Test motion corrected BOLD files exists
    assert motion_corrected.bold.with_suffix(".nii.gz").exists()
    assert motion_corrected.par.exists()
    assert motion_corrected.rms_rel.exists()