diff --git a/src/rbc/metadata.py b/src/rbc/metadata.py index fae76464..e3708671 100644 --- a/src/rbc/metadata.py +++ b/src/rbc/metadata.py @@ -117,6 +117,18 @@ def _validate_slice_timing(slice_timing: list[float], tr: float) -> None: raise ValueError(msg) +def _header_slice_timing(hdr: nib.Nifti1Header) -> list[float] | None: + """Extract per-slice acquisition times from the NIfTI header, if present. + + Returns *None* when the header lacks the necessary fields + (slice_code, slice_duration, dim_info). + """ + try: + return list(hdr.get_slice_times()) + except nib.spatialimages.HeaderDataError: + return None + + @dataclass(frozen=True) class FunctionalMetadata: """Validated, immutable metadata for a single BOLD run. @@ -166,6 +178,23 @@ def load( slice_timing: list[float] | None = sidecar.get("SliceTiming") if slice_timing is not None: + _logger.info( + "SliceTiming: from BIDS sidecar (%d slices)", len(slice_timing) + ) _validate_slice_timing(slice_timing, tr) + else: + slice_timing = _header_slice_timing(hdr) + if slice_timing is not None: + _logger.warning( + "No SliceTiming in BIDS sidecar; " + "falling back to NIfTI header (%d slices)", + len(slice_timing), + ) + _validate_slice_timing(slice_timing, tr) + else: + _logger.warning( + "No SliceTiming in BIDS sidecar or NIfTI header; " + "slice timing correction will be skipped" + ) return cls(tr=tr, slice_timing=slice_timing) diff --git a/src/rbc/workflows/functional.py b/src/rbc/workflows/functional.py index ce854aef..c2e86195 100644 --- a/src/rbc/workflows/functional.py +++ b/src/rbc/workflows/functional.py @@ -254,12 +254,18 @@ def single_session_preprocess( mc = fsl_motion_correction(in_file=despiked, ref_file=effective_ref) # 7. Slice timing correction - _logger.info("Slice timing correction") - st_corrected = slice_timing_correction( - in_file=despiked, - tr=metadata.tr, - tpattern=metadata.slice_timing, - ) + if metadata.slice_timing is not None: + _logger.info("Slice timing correction") + st_corrected = slice_timing_correction( + in_file=despiked, + tr=metadata.tr, + tpattern=metadata.slice_timing, + ) + else: + _logger.info( + "Skipping slice timing correction (no SliceTiming in sidecar or header)" + ) + st_corrected = despiked # 8. Apply pre-STC motion transforms to STC BOLD # native-space MC + STC BOLD used in step 12 diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index cae353da..353bef7a 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch +import nibabel as nib import pytest from rbc.metadata import ( @@ -146,7 +147,7 @@ def test_load_with_override(self, tmp_path: Path) -> None: assert meta.tr == pytest.approx(1.5) def test_load_no_slice_timing(self, tmp_path: Path) -> None: - """load() sets slice_timing to None when absent from sidecar.""" + """load() sets slice_timing to None when absent from sidecar and header.""" bold = tmp_path / "bold.nii.gz" bold.touch() @@ -154,6 +155,7 @@ def test_load_no_slice_timing(self, tmp_path: Path) -> None: mock_hdr.__getitem__ = lambda _, key: ( [0, 1, 1, 1, 2.0] if key == "pixdim" else None ) + mock_hdr.get_slice_times.side_effect = nib.spatialimages.HeaderDataError() mock_img = MagicMock() mock_img.header = mock_hdr @@ -168,6 +170,32 @@ def test_load_no_slice_timing(self, tmp_path: Path) -> None: assert meta.slice_timing is None + def test_load_slice_timing_from_header(self, tmp_path: Path) -> None: + """load() falls back to NIfTI header when sidecar lacks SliceTiming.""" + bold = tmp_path / "bold.nii.gz" + bold.touch() + + header_times = [0.0, 0.5, 1.0, 1.5] + + mock_hdr = MagicMock() + mock_hdr.__getitem__ = lambda _, key: ( + [0, 1, 1, 1, 2.0] if key == "pixdim" else None + ) + mock_hdr.get_slice_times.return_value = header_times + mock_img = MagicMock() + mock_img.header = mock_hdr + + with ( + patch( + "rbc.metadata.load_bids_metadata", + return_value={"RepetitionTime": 2.0}, + ), + patch("rbc.metadata.nib.nifti1.load", return_value=mock_img), + ): + meta = FunctionalMetadata.load(bold) + + assert meta.slice_timing == header_times + def test_load_missing_tr_raises(self, tmp_path: Path) -> None: """load() raises when TR cannot be determined.""" bold = tmp_path / "bold.nii.gz"