Fix sentinel header select

.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

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."

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()

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():

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)