|
| 1 | +"""Unit tests for ``rbc.bids.longitudinal.anatomical``.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from typing import TYPE_CHECKING |
| 6 | + |
| 7 | +import polars as pl |
| 8 | + |
| 9 | +from rbc.bids.longitudinal.anatomical import ( |
| 10 | + export_longitudinal_anat, |
| 11 | + resolve_longitudinal_anat, |
| 12 | +) |
| 13 | +from rbc.context import RunContext |
| 14 | +from rbc.workflows.longitudinal.anatomical import AnatomicalLongOutputs |
| 15 | + |
| 16 | +if TYPE_CHECKING: |
| 17 | + from pathlib import Path |
| 18 | + |
| 19 | + |
| 20 | +def _make_long_outputs(workdir: Path) -> AnatomicalLongOutputs: |
| 21 | + """Build a populated AnatomicalLongOutputs pointing at dummy files.""" |
| 22 | + |
| 23 | + def _dummy(name: str) -> Path: |
| 24 | + p = workdir / name |
| 25 | + p.write_bytes(b"\x00") |
| 26 | + return p |
| 27 | + |
| 28 | + return AnatomicalLongOutputs( |
| 29 | + brain=_dummy("brain.nii.gz"), |
| 30 | + brain_mask=_dummy("brain_mask.nii.gz"), |
| 31 | + long_to_template_xfm=_dummy("long_to_tpl.nii.gz"), |
| 32 | + template_to_long_xfm=_dummy("tpl_to_long.nii.gz"), |
| 33 | + ) |
| 34 | + |
| 35 | + |
| 36 | +class TestExportLongitudinalAnat: |
| 37 | + """Tests for :func:`export_longitudinal_anat`.""" |
| 38 | + |
| 39 | + def test_writes_expected_files(self, tmp_path: Path) -> None: |
| 40 | + """Exports brain, brain_mask, and the two xfms under space-longitudinal.""" |
| 41 | + workdir = tmp_path / "work" |
| 42 | + workdir.mkdir() |
| 43 | + out_dir = tmp_path / "out" |
| 44 | + |
| 45 | + ctx = RunContext(sub="01", ses="baseline", output_dir=out_dir) |
| 46 | + aex = ctx.bids(datatype="anat").derive(space="longitudinal") |
| 47 | + |
| 48 | + export_longitudinal_anat(aex, _make_long_outputs(workdir)) |
| 49 | + |
| 50 | + saved = sorted(p.name for p in out_dir.rglob("*.*")) |
| 51 | + # Xfm filenames currently carry both ``space-longitudinal`` and the |
| 52 | + # ``from-…_to-…`` pair, which is redundant (Jason handover #18 flags |
| 53 | + # the ``from-`` workaround more broadly). We assert the current shape |
| 54 | + # so regressions are caught, but if that cleanup lands the expected |
| 55 | + # xfm names here should drop ``space-longitudinal_``. |
| 56 | + assert saved == [ |
| 57 | + "sub-01_ses-baseline_space-longitudinal_desc-T1w_mask.nii.gz", |
| 58 | + "sub-01_ses-baseline_space-longitudinal_desc-brain_T1w.nii.gz", |
| 59 | + "sub-01_ses-baseline_space-longitudinal_from-MNI152NLin6Asym" |
| 60 | + "_to-longitudinal_mode-image_xfm.nii.gz", |
| 61 | + "sub-01_ses-baseline_space-longitudinal_from-longitudinal" |
| 62 | + "_to-MNI152NLin6Asym_mode-image_xfm.nii.gz", |
| 63 | + ] |
| 64 | + |
| 65 | + def test_tissue_masks_not_produced(self, tmp_path: Path) -> None: |
| 66 | + """No csf/gm/wm masks are written under space-longitudinal (#5).""" |
| 67 | + workdir = tmp_path / "work" |
| 68 | + workdir.mkdir() |
| 69 | + out_dir = tmp_path / "out" |
| 70 | + |
| 71 | + ctx = RunContext(sub="01", ses="baseline", output_dir=out_dir) |
| 72 | + aex = ctx.bids(datatype="anat").derive(space="longitudinal") |
| 73 | + |
| 74 | + export_longitudinal_anat(aex, _make_long_outputs(workdir)) |
| 75 | + |
| 76 | + names = [p.name for p in out_dir.rglob("*.*")] |
| 77 | + for tissue in ("csf", "gm", "wm"): |
| 78 | + assert not any(f"desc-{tissue}_mask" in n for n in names), ( |
| 79 | + f"Expected no space-longitudinal {tissue}_mask; got {names}" |
| 80 | + ) |
| 81 | + |
| 82 | + |
| 83 | +def _anat_row( |
| 84 | + *, |
| 85 | + sub: str, |
| 86 | + ses: str, |
| 87 | + suffix: str, |
| 88 | + desc: str | None, |
| 89 | + ext: str = ".nii.gz", |
| 90 | + extra: list[dict[str, str]] | None = None, |
| 91 | +) -> dict[str, object]: |
| 92 | + """Build a single BIDS-like row, including an ``extra_entities`` list.""" |
| 93 | + desc_part = f"_desc-{desc}" if desc else "" |
| 94 | + path = f"sub-{sub}/ses-{ses}/anat/sub-{sub}_ses-{ses}{desc_part}_{suffix}{ext}" |
| 95 | + return { |
| 96 | + "datatype": "anat", |
| 97 | + "suffix": suffix, |
| 98 | + "ext": ext, |
| 99 | + "sub": sub, |
| 100 | + "ses": ses, |
| 101 | + "desc": desc, |
| 102 | + "root": "/data", |
| 103 | + "path": path, |
| 104 | + "extra_entities": extra or [], |
| 105 | + } |
| 106 | + |
| 107 | + |
| 108 | +def _df(*rows: dict[str, object]) -> pl.DataFrame: |
| 109 | + return pl.DataFrame(list(rows)) |
| 110 | + |
| 111 | + |
| 112 | +class TestResolveLongitudinalAnat: |
| 113 | + """Tests for :func:`resolve_longitudinal_anat`.""" |
| 114 | + |
| 115 | + def test_resolves_all_inputs(self, tmp_path: Path) -> None: |
| 116 | + """Happy path: all four keys resolve to the expected paths.""" |
| 117 | + anat_df = _df( |
| 118 | + _anat_row(sub="01", ses="baseline", suffix="T1w", desc="brain"), |
| 119 | + _anat_row(sub="01", ses="baseline", suffix="mask", desc="T1w"), |
| 120 | + ) |
| 121 | + tpl_df = _df( |
| 122 | + _anat_row(sub="01", ses="longitudinal", suffix="T1w", desc=None), |
| 123 | + _anat_row( |
| 124 | + sub="01", |
| 125 | + ses="longitudinal", |
| 126 | + suffix="xfm", |
| 127 | + desc=None, |
| 128 | + ext=".txt", |
| 129 | + extra=[ |
| 130 | + {"key": "from", "value": "baseline"}, |
| 131 | + {"key": "to", "value": "longitudinal"}, |
| 132 | + ], |
| 133 | + ), |
| 134 | + ) |
| 135 | + |
| 136 | + ctx = RunContext(sub="01", ses="baseline", output_dir=tmp_path) |
| 137 | + anat_q = ctx.bids(datatype="anat") |
| 138 | + tpl_q = anat_q.derive(ses="longitudinal") |
| 139 | + |
| 140 | + resolved = resolve_longitudinal_anat( |
| 141 | + anat_q, tpl_q, anat_df, tpl_df, ses="baseline" |
| 142 | + ) |
| 143 | + |
| 144 | + assert set(resolved) == { |
| 145 | + "template", |
| 146 | + "subj_to_template_xfm", |
| 147 | + "brain", |
| 148 | + "brain_mask", |
| 149 | + } |
| 150 | + assert str(resolved["template"]).endswith("sub-01_ses-longitudinal_T1w.nii.gz") |
| 151 | + assert str(resolved["subj_to_template_xfm"]).endswith( |
| 152 | + "sub-01_ses-longitudinal_xfm.txt" |
| 153 | + ) |
| 154 | + assert str(resolved["brain"]).endswith( |
| 155 | + "sub-01_ses-baseline_desc-brain_T1w.nii.gz" |
| 156 | + ) |
| 157 | + assert str(resolved["brain_mask"]).endswith( |
| 158 | + "sub-01_ses-baseline_desc-T1w_mask.nii.gz" |
| 159 | + ) |
| 160 | + |
| 161 | + def test_missing_brain_mask_is_none(self, tmp_path: Path) -> None: |
| 162 | + """brain_mask is optional (``.find``) — returns None when absent.""" |
| 163 | + anat_df = _df( |
| 164 | + _anat_row(sub="01", ses="baseline", suffix="T1w", desc="brain"), |
| 165 | + ) |
| 166 | + tpl_df = _df( |
| 167 | + _anat_row(sub="01", ses="longitudinal", suffix="T1w", desc=None), |
| 168 | + _anat_row( |
| 169 | + sub="01", |
| 170 | + ses="longitudinal", |
| 171 | + suffix="xfm", |
| 172 | + desc=None, |
| 173 | + ext=".txt", |
| 174 | + extra=[ |
| 175 | + {"key": "from", "value": "baseline"}, |
| 176 | + {"key": "to", "value": "longitudinal"}, |
| 177 | + ], |
| 178 | + ), |
| 179 | + ) |
| 180 | + |
| 181 | + ctx = RunContext(sub="01", ses="baseline", output_dir=tmp_path) |
| 182 | + anat_q = ctx.bids(datatype="anat") |
| 183 | + tpl_q = anat_q.derive(ses="longitudinal") |
| 184 | + |
| 185 | + resolved = resolve_longitudinal_anat( |
| 186 | + anat_q, tpl_q, anat_df, tpl_df, ses="baseline" |
| 187 | + ) |
| 188 | + |
| 189 | + assert resolved["brain_mask"] is None |
| 190 | + assert resolved["brain"] is not None |
| 191 | + |
| 192 | + def test_xfm_query_uses_bids_safe_label(self, tmp_path: Path) -> None: |
| 193 | + """Session labels with non-alphanumerics resolve through ``bids_safe_label``. |
| 194 | +
|
| 195 | + ``pre-op`` is stored as ``from-preop`` in the xfm entity; the resolver |
| 196 | + must sanitize the ses arg before querying or the xfm won't be found. |
| 197 | + """ |
| 198 | + anat_df = _df( |
| 199 | + _anat_row(sub="01", ses="pre-op", suffix="T1w", desc="brain"), |
| 200 | + ) |
| 201 | + tpl_df = _df( |
| 202 | + _anat_row(sub="01", ses="longitudinal", suffix="T1w", desc=None), |
| 203 | + _anat_row( |
| 204 | + sub="01", |
| 205 | + ses="longitudinal", |
| 206 | + suffix="xfm", |
| 207 | + desc=None, |
| 208 | + ext=".txt", |
| 209 | + extra=[ |
| 210 | + {"key": "from", "value": "preop"}, |
| 211 | + {"key": "to", "value": "longitudinal"}, |
| 212 | + ], |
| 213 | + ), |
| 214 | + ) |
| 215 | + |
| 216 | + ctx = RunContext(sub="01", ses="pre-op", output_dir=tmp_path) |
| 217 | + anat_q = ctx.bids(datatype="anat") |
| 218 | + tpl_q = anat_q.derive(ses="longitudinal") |
| 219 | + |
| 220 | + resolved = resolve_longitudinal_anat( |
| 221 | + anat_q, tpl_q, anat_df, tpl_df, ses="pre-op" |
| 222 | + ) |
| 223 | + |
| 224 | + assert resolved["subj_to_template_xfm"] is not None |
| 225 | + assert str(resolved["subj_to_template_xfm"]).endswith( |
| 226 | + "sub-01_ses-longitudinal_xfm.txt" |
| 227 | + ) |
0 commit comments