diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index c06256a..950eab3 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -37,7 +37,7 @@ jobs: pytest trx/tests --cov=trx --cov-report=xml --cov-report=term-missing - name: Upload coverage to Codecov - uses: codecov/codecov-action@v4 + uses: codecov/codecov-action@v5 with: files: ./coverage.xml flags: unittests diff --git a/trx/io.py b/trx/io.py index ebf657f..1fdc787 100644 --- a/trx/io.py +++ b/trx/io.py @@ -46,7 +46,7 @@ def get_trx_tmp_dir(): return tempfile.TemporaryDirectory(dir=trx_tmp_dir, prefix="trx_") -def load_sft_with_reference(filepath, reference=None, bbox_check=True, **kwargs): +def load_sft_with_reference(filepath, reference=None, bbox_check=True, from_space=None): """Load a tractogram as a StatefulTractogram with an explicit reference. Parameters @@ -59,8 +59,8 @@ def load_sft_with_reference(filepath, reference=None, bbox_check=True, **kwargs) bbox_check : bool, optional If True, validate that streamlines lie within the reference bounding box. Defaults to True. - **kwargs - Additional keyword arguments passed to dipy's load_tractogram. + from_space : dipy.io.stateful_tractogram.Space, optional + Space to which the tractogram was transformed before saving. Returns ------- @@ -72,7 +72,7 @@ def load_sft_with_reference(filepath, reference=None, bbox_check=True, **kwargs) IOError If the file format is unsupported or a required reference is missing. """ - if not dipy_available: # pragma: no cover + if not dipy_available: logging.error( "Dipy library is missing, cannot use functions related " "to the StatefulTractogram." @@ -85,13 +85,15 @@ def load_sft_with_reference(filepath, reference=None, bbox_check=True, **kwargs) if ext == ".trk": if reference is not None and reference != "same": logging.warning(f"Reference is discarded for this file format {filepath}.") - sft = load_tractogram(filepath, "same", bbox_valid_check=bbox_check, **kwargs) + sft = load_tractogram( + filepath, "same", bbox_valid_check=bbox_check, from_space=from_space + ) elif ext in [".tck", ".fib", ".vtk", ".dpy"]: if reference is None or reference == "same": raise IOError(f"--reference is required for this file format {filepath}.") else: sft = load_tractogram( - filepath, reference, bbox_valid_check=bbox_check, **kwargs + filepath, reference, bbox_valid_check=bbox_check, from_space=from_space ) else: @@ -100,17 +102,17 @@ def load_sft_with_reference(filepath, reference=None, bbox_check=True, **kwargs) return sft -def load(tractogram_filename, reference, **kwargs): +def load(tractogram_filename, reference=None, from_space=None): """Load a tractogram from disk and return a TRX or StatefulTractogram. Parameters ---------- tractogram_filename : str Path to the input tractogram. TRX directories are supported. - reference : str or nibabel.Nifti1Image + reference : str or nibabel.Nifti1Image, optional Reference image used for formats without embedded affine information. - **kwargs - Additional keyword arguments passed to dipy's load_tractogram. + from_space : dipy.io.stateful_tractogram.Space, optional + Space to which the tractogram was transformed before saving. Returns ------- @@ -122,7 +124,7 @@ def load(tractogram_filename, reference, **kwargs): in_ext = split_name_with_gz(tractogram_filename)[1] if in_ext != ".trx" and not os.path.isdir(tractogram_filename): tractogram_obj = load_sft_with_reference( - tractogram_filename, reference, bbox_check=False, **kwargs + tractogram_filename, reference, bbox_check=False, from_space=from_space ) else: tractogram_obj = tmm.load(tractogram_filename) @@ -151,7 +153,7 @@ def save(tractogram_obj, tractogram_filename, bbox_valid_check=False): The function writes to disk and returns ``None``. Returns ``None`` immediately when ``dipy`` is unavailable. """ - if not dipy_available: # pragma: no cover + if not dipy_available: logging.error( "Dipy library is missing, cannot use functions related " "to the StatefulTractogram." diff --git a/trx/tests/test_io.py b/trx/tests/test_io.py index 11a24fa..8b577b5 100644 --- a/trx/tests/test_io.py +++ b/trx/tests/test_io.py @@ -14,16 +14,9 @@ from dipy.io.streamline import load_tractogram, save_tractogram dipy_available = True -except ImportError: # pragma: no cover +except ImportError: dipy_available = False -try: - import fury # noqa: F401 - - fury_available = True -except ImportError: # pragma: no cover - fury_available = False - from trx.fetcher import fetch_data, get_home, get_testing_files_dict from trx.io import load, save import trx.trx_file_memmap as tmm @@ -35,8 +28,6 @@ @pytest.mark.parametrize("path", [("gs.trk"), ("gs.tck"), ("gs.vtk")]) @pytest.mark.skipif(not dipy_available, reason="Dipy is not installed.") def test_seq_ops_sft(path): - if path.endswith(".vtk") and not fury_available: - pytest.skip("fury is not installed") with TemporaryDirectory() as tmp_dir: gs_dir = os.path.join(get_home(), "gold_standard") path = os.path.join(tmp_dir, path) @@ -65,17 +56,13 @@ def test_seq_ops_trx(): @pytest.mark.parametrize("path", [("gs.trx"), ("gs.trk"), ("gs.tck"), ("gs.vtk")]) @pytest.mark.skipif(not dipy_available, reason="Dipy is not installed.") def test_load_vox(path): - if path.endswith(".vtk") and not fury_available: - pytest.skip("fury is not installed") + from dipy.io.stateful_tractogram import Space + gs_dir = os.path.join(get_home(), "gold_standard") path = os.path.join(gs_dir, path) coord = np.loadtxt(os.path.join(get_home(), "gold_standard", "gs_vox_space.txt")) - if path.endswith(".vtk"): - from dipy.io.stateful_tractogram import Space - - obj = load(path, os.path.join(gs_dir, "gs.nii"), from_space=Space.LPSMM) - else: - obj = load(path, os.path.join(gs_dir, "gs.nii")) + from_space = Space.LPSMM if path.endswith("gs.vtk") else None + obj = load(path, os.path.join(gs_dir, "gs.nii"), from_space=from_space) sft = obj.to_sft() if isinstance(obj, TrxFile) else obj sft.to_vox() @@ -88,17 +75,13 @@ def test_load_vox(path): @pytest.mark.parametrize("path", [("gs.trx"), ("gs.trk"), ("gs.tck"), ("gs.vtk")]) @pytest.mark.skipif(not dipy_available, reason="Dipy is not installed.") def test_load_voxmm(path): - if path.endswith(".vtk") and not fury_available: - pytest.skip("fury is not installed") + from dipy.io.stateful_tractogram import Space + gs_dir = os.path.join(get_home(), "gold_standard") path = os.path.join(gs_dir, path) coord = np.loadtxt(os.path.join(get_home(), "gold_standard", "gs_voxmm_space.txt")) - if path.endswith(".vtk"): - from dipy.io.stateful_tractogram import Space - - obj = load(path, os.path.join(gs_dir, "gs.nii"), from_space=Space.LPSMM) - else: - obj = load(path, os.path.join(gs_dir, "gs.nii")) + from_space = Space.LPSMM if path.endswith("gs.vtk") else None + obj = load(path, os.path.join(gs_dir, "gs.nii"), from_space=from_space) sft = obj.to_sft() if isinstance(obj, TrxFile) else obj sft.to_voxmm() diff --git a/trx/tests/test_memmap.py b/trx/tests/test_memmap.py index 9ef2879..2cc33d7 100644 --- a/trx/tests/test_memmap.py +++ b/trx/tests/test_memmap.py @@ -387,7 +387,32 @@ def test_trxfile_getgroup(): def test_trxfile_select(): - pass + path = os.path.join(get_home(), "memmap_test_data", "small.trx") + trx = tmm.load(path) + + assert len(trx.select([]).streamlines) == 0 + assert len(trx.select([0]).streamlines) == 1 + + idx = list(range(10)) + sub = trx.select(idx) + assert len(sub.streamlines) == len(idx) + assert not sub._copy_safe + + trx.close() + + +def test_save_after_select(): + path = os.path.join(get_home(), "memmap_test_data", "small.trx") + trx = tmm.load(path) + sub = trx.select(list(range(5))) + with tempfile.TemporaryDirectory() as tmp_dir: + out = os.path.join(tmp_dir, "sub.trx") + tmm.save(sub, out) + loaded = tmm.load(out) + assert len(loaded.streamlines) == 5 + assert len(loaded.streamlines._data) == len(sub.streamlines.copy()._data) + loaded.close() + trx.close() def test_trxfile_to_memory(): diff --git a/trx/trx_file_memmap.py b/trx/trx_file_memmap.py index 412923d..665ae7b 100644 --- a/trx/trx_file_memmap.py +++ b/trx/trx_file_memmap.py @@ -1109,14 +1109,14 @@ def deepcopy(self) -> Type["TrxFile"]: # noqa: C901 if not self._copy_safe: to_dump = _append_last_offsets( - self.streamlines.copy()._offsets, self.header["NB_VERTICES"] + self.streamlines.copy()._offsets, tmp_header["NB_VERTICES"] ) else: to_dump = _append_last_offsets( self.streamlines._offsets, self.header["NB_VERTICES"] ) offsets_filename = _generate_filename_from_data( - self.streamlines._offsets, os.path.join(tmp_dir.name, "offsets") + to_dump, os.path.join(tmp_dir.name, "offsets") ) _ensure_little_endian(to_dump).tofile(offsets_filename)