juaml/junifer
Watch
3
0
Fork
You've already forked junifer
 0
Code Issues 14 Pull requests 2 Packages Activity Actions

[ENH]: Use junifer-data datalad dataset for fetching data #418

Merged
synchon merged 22 commits from enh/junifer-data into main 2025-02-13 08:59:08 +00:00
Conversation 5 Commits 22 Files changed 12 +715 -1151
12 changed files with 715 additions and 1151 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/changes/newsfragments/418.enh Normal file
View file

@ -0,0 +1 @@
Adapt usage of ``junifer-data`` DataLad dataset to fetch parcellations, masks, coordinates and xfms by `Synchon Mandal`_

1
docs/changes/newsfragments/418.misc Normal file
View file

@ -0,0 +1 @@
Remove ``httpx`` as a dependency by `Synchon Mandal`_

2
junifer/cli/tests/test_cli_utils.py
View file

@ -43,7 +43,6 @@ def test_get_dependency_information_short() -> None:
"nilearn",
"sqlalchemy",
"ruamel.yaml",
"httpx",
"tqdm",
"templateflow",
"lapy",
@ -73,7 +72,6 @@ def test_get_dependency_information_long() -> None:
"nilearn",
"sqlalchemy",
"ruamel.yaml",
"httpx",
"tqdm",
"templateflow",
"lapy",

231
junifer/data/coordinates/_coordinates.py
View file

@ -4,7 +4,6 @@
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from pathlib import Path
from typing import Any, Optional
import numpy as np
@ -14,7 +13,7 @@ from numpy.typing import ArrayLike
from ...utils import logger, raise_error
from ...utils.singleton import Singleton
from ..pipeline_data_registry_base import BasePipelineDataRegistry
from ..utils import get_native_warper
from ..utils import check_dataset, fetch_file_via_datalad, get_native_warper
from ._ants_coordinates_warper import ANTsCoordinatesWarper
from ._fsl_coordinates_warper import FSLCoordinatesWarper
@ -32,104 +31,104 @@ class CoordinatesRegistry(BasePipelineDataRegistry, metaclass=Singleton):
def __init__(self) -> None:
"""Initialize the class."""
super().__init__()
# Each entry in registry is a dictionary that must contain at least
# the following keys:
# * 'space': the coordinates' space (e.g., 'MNI')
# The built-in coordinates are files that are shipped with the package
# in the data/VOIs directory. The user can also register their own
# The built-in coordinates are files that are shipped with the
# junifer-data dataset. The user can also register their own
# coordinates, which will be stored as numpy arrays in the dictionary.
# Make built-in and external dictionaries for validation later
self._builtin = {}
self._external = {}
# Path to the metadata of the VOIs
_vois_meta_path = Path(__file__).parent / "VOIs" / "meta"
self._builtin.update(
{
"CogAC": {
"file_path_suffix": "CogAC_VOIs.txt",
"space": "MNI",
},
"CogAR": {
"file_path_suffix": "CogAR_VOIs.txt",
"space": "MNI",
},
"DMNBuckner": {
"file_path_suffix": "DMNBuckner_VOIs.txt",
"space": "MNI",
},
"eMDN": {
"file_path_suffix": "eMDN_VOIs.txt",
"space": "MNI",
},
"Empathy": {
"file_path_suffix": "Empathy_VOIs.txt",
"space": "MNI",
},
"eSAD": {
"file_path_suffix": "eSAD_VOIs.txt",
"space": "MNI",
},
"extDMN": {
"file_path_suffix": "extDMN_VOIs.txt",
"space": "MNI",
},
"Motor": {
"file_path_suffix": "Motor_VOIs.txt",
"space": "MNI",
},
"MultiTask": {
"file_path_suffix": "MultiTask_VOIs.txt",
"space": "MNI",
},
"PhysioStress": {
"file_path_suffix": "PhysioStress_VOIs.txt",
"space": "MNI",
},
"Rew": {
"file_path_suffix": "Rew_VOIs.txt",
"space": "MNI",
},
"Somatosensory": {
"file_path_suffix": "Somatosensory_VOIs.txt",
"space": "MNI",
},
"ToM": {
"file_path_suffix": "ToM_VOIs.txt",
"space": "MNI",
},
"VigAtt": {
"file_path_suffix": "VigAtt_VOIs.txt",
"space": "MNI",
},
"WM": {
"file_path_suffix": "WM_VOIs.txt",
"space": "MNI",
},
"Power": {
"file_path_suffix": "Power2011_MNI_VOIs.txt",
"space": "MNI",
},
"Power2011": {
"file_path_suffix": "Power2011_MNI_VOIs.txt",
"space": "MNI",
},
"Dosenbach": {
"file_path_suffix": "Dosenbach2010_MNI_VOIs.txt",
"space": "MNI",
},
"Power2013": {
"file_path_suffix": "Power2013_MNI_VOIs.tsv",
"space": "MNI",
},
"AutobiographicalMemory": {
"file_path_suffix": "AutobiographicalMemory_VOIs.txt",
"space": "MNI",
},
}
)
self._builtin = {
"CogAC": {
"path": _vois_meta_path / "CogAC_VOIs.txt",
"space": "MNI",
},
"CogAR": {
"path": _vois_meta_path / "CogAR_VOIs.txt",
"space": "MNI",
},
"DMNBuckner": {
"path": _vois_meta_path / "DMNBuckner_VOIs.txt",
"space": "MNI",
},
"eMDN": {
"path": _vois_meta_path / "eMDN_VOIs.txt",
"space": "MNI",
},
"Empathy": {
"path": _vois_meta_path / "Empathy_VOIs.txt",
"space": "MNI",
},
"eSAD": {
"path": _vois_meta_path / "eSAD_VOIs.txt",
"space": "MNI",
},
"extDMN": {
"path": _vois_meta_path / "extDMN_VOIs.txt",
"space": "MNI",
},
"Motor": {
"path": _vois_meta_path / "Motor_VOIs.txt",
"space": "MNI",
},
"MultiTask": {
"path": _vois_meta_path / "MultiTask_VOIs.txt",
"space": "MNI",
},
"PhysioStress": {
"path": _vois_meta_path / "PhysioStress_VOIs.txt",
"space": "MNI",
},
"Rew": {
"path": _vois_meta_path / "Rew_VOIs.txt",
"space": "MNI",
},
"Somatosensory": {
"path": _vois_meta_path / "Somatosensory_VOIs.txt",
"space": "MNI",
},
"ToM": {
"path": _vois_meta_path / "ToM_VOIs.txt",
"space": "MNI",
},
"VigAtt": {
"path": _vois_meta_path / "VigAtt_VOIs.txt",
"space": "MNI",
},
"WM": {
"path": _vois_meta_path / "WM_VOIs.txt",
"space": "MNI",
},
"Power": {
"path": _vois_meta_path / "Power2011_MNI_VOIs.txt",
"space": "MNI",
},
"Power2011": {
"path": _vois_meta_path / "Power2011_MNI_VOIs.txt",
"space": "MNI",
},
"Dosenbach": {
"path": _vois_meta_path / "Dosenbach2010_MNI_VOIs.txt",
"space": "MNI",
},
"Power2013": {
"path": _vois_meta_path / "Power2013_MNI_VOIs.tsv",
"space": "MNI",
},
"AutobiographicalMemory": {
"path": _vois_meta_path / "AutobiographicalMemory_VOIs.txt",
"space": "MNI",
},
}
# Set built-in to registry
self._registry = self._builtin
# Update registry with built-in ones
self._registry.update(self._builtin)
def register(
self,
@ -161,9 +160,9 @@ class CoordinatesRegistry(BasePipelineDataRegistry, metaclass=Singleton):
Raises
------
ValueError
If the coordinates ``name`` is already registered and
If the coordinates ``name`` is a built-in coordinates or
if the coordinates ``name`` is already registered and
``overwrite=False`` or
if the coordinates ``name`` is a built-in coordinates or
if the ``coordinates`` is not a 2D array or
if coordinate value does not have 3 components or
if the ``voi_names`` shape does not match the
@ -174,11 +173,12 @@ class CoordinatesRegistry(BasePipelineDataRegistry, metaclass=Singleton):
"""
# Check for attempt of overwriting built-in coordinates
if name in self._builtin:
if isinstance(self._registry[name].get("path"), Path):
raise_error(
f"Coordinates: {name} already registered as built-in "
"coordinates."
)
raise_error(
f"Coordinates: {name} already registered as built-in "
"coordinates."
)
# Check for attempt of overwriting external coordinates
if name in self._external:
if overwrite:
logger.info(f"Overwriting coordinates: {name}")
else:
@ -186,7 +186,7 @@ class CoordinatesRegistry(BasePipelineDataRegistry, metaclass=Singleton):
f"Coordinates: {name} already registered. "
"Set `overwrite=True` to update its value."
)
# Further checks
if not isinstance(coordinates, np.ndarray):
raise_error(
"Coordinates must be a `numpy.ndarray`, "
@ -207,6 +207,7 @@ class CoordinatesRegistry(BasePipelineDataRegistry, metaclass=Singleton):
f"Length of `voi_names` ({len(voi_names)}) does not match the "
f"number of `coordinates` ({coordinates.shape[0]})."
)
# Registration
logger.info(f"Registering coordinates: {name}")
# Add coordinates info
self._external[name] = {
@ -257,6 +258,8 @@ class CoordinatesRegistry(BasePipelineDataRegistry, metaclass=Singleton):
------
ValueError
If ``name`` is invalid.
RuntimeError
If there is a problem fetching the coordinates file.
"""
# Check for valid coordinates name
@ -265,17 +268,37 @@ class CoordinatesRegistry(BasePipelineDataRegistry, metaclass=Singleton):
f"Coordinates: {name} not found. "
f"Valid options are: {self.list}"
)
# Load coordinates
# Load coordinates info
t_coord = self._registry[name]
# Load data
if isinstance(t_coord.get("path"), Path):
logger.debug(f"Loading coordinates {t_coord['path'].absolute()!s}")
# Load data for in-built ones
if t_coord.get("file_path_suffix") is not None:
# Get dataset
dataset = check_dataset()
# Set file path to retrieve
coords_file_path = (
dataset.pathobj
/ "coordinates"
/ name
/ t_coord["file_path_suffix"]
)
logger.debug(
f"Loading coordinates `{name}` from: "
f"{coords_file_path.absolute()!s}"
)
# Load via pandas
df_coords = pd.read_csv(t_coord["path"], sep="\t", header=None)
df_coords = pd.read_csv(
fetch_file_via_datalad(
dataset=dataset, file_path=coords_file_path
),
sep="\t",
header=None,
)
# Convert dataframe to numpy ndarray
coords = df_coords.iloc[:, [0, 1, 2]].to_numpy()
# Get label names
names = list(df_coords.iloc[:, [3]].values[:, 0])
# Load data for external ones
else:
coords = t_coord["coords"]
names = t_coord["voi_names"]

3
junifer/data/coordinates/tests/test_coordinates.py
View file

@ -21,7 +21,6 @@ def test_register_built_in_check() -> None:
coordinates=np.zeros(2),
voi_names=["1", "2"],
space="MNI",
overwrite=True,
)
@ -32,7 +31,6 @@ def test_register_overwrite() -> None:
coordinates=np.zeros((2, 3)),
voi_names=["roi1", "roi2"],
space="MNI",
overwrite=True,
)
with pytest.raises(ValueError, match=r"already registered"):
CoordinatesRegistry().register(
@ -40,6 +38,7 @@ def test_register_overwrite() -> None:
coordinates=np.ones((2, 3)),
voi_names=["roi2", "roi3"],
space="MNI",
overwrite=False,
)
CoordinatesRegistry().register(

140
junifer/data/masks/_masks.py
View file

@ -26,22 +26,24 @@ from ...utils import logger, raise_error
from ...utils.singleton import Singleton
from ..pipeline_data_registry_base import BasePipelineDataRegistry
from ..template_spaces import get_template
from ..utils import closest_resolution, get_native_warper
from ..utils import (
check_dataset,
closest_resolution,
fetch_file_via_datalad,
get_native_warper,
)
from ._ants_mask_warper import ANTsMaskWarper
from ._fsl_mask_warper import FSLMaskWarper
if TYPE_CHECKING:
from datalad.api import Dataset
from nibabel.nifti1 import Nifti1Image
__all__ = ["MaskRegistry", "compute_brain_mask"]
# Path to the masks
_masks_path = Path(__file__).parent
def compute_brain_mask(
target_data: dict[str, Any],
warp_data: Optional[dict[str, Any]] = None,
@ -224,6 +226,7 @@ class MaskRegistry(BasePipelineDataRegistry, metaclass=Singleton):
def __init__(self) -> None:
"""Initialize the class."""
super().__init__()
# Each entry in registry is a dictionary that must contain at least
# the following keys:
# * 'family': the mask's family name
@ -240,38 +243,40 @@ class MaskRegistry(BasePipelineDataRegistry, metaclass=Singleton):
self._builtin = {}
self._external = {}
self._builtin = {
"GM_prob0.2": {
"family": "Vickery-Patil",
"space": "IXI549Space",
},
"GM_prob0.2_cortex": {
"family": "Vickery-Patil",
"space": "IXI549Space",
},
"compute_brain_mask": {
"family": "Callable",
"func": compute_brain_mask,
"space": "inherit",
},
"compute_background_mask": {
"family": "Callable",
"func": compute_background_mask,
"space": "inherit",
},
"compute_epi_mask": {
"family": "Callable",
"func": compute_epi_mask,
"space": "inherit",
},
"UKB_15K_GM": {
"family": "UKB",
"space": "MNI152NLin6Asym",
},
}
self._builtin.update(
{
"GM_prob0.2": {
"family": "Vickery-Patil",
"space": "IXI549Space",
},
"GM_prob0.2_cortex": {
"family": "Vickery-Patil",
"space": "IXI549Space",
},
"compute_brain_mask": {
"family": "Callable",
"func": compute_brain_mask,
"space": "inherit",
},
"compute_background_mask": {
"family": "Callable",
"func": compute_background_mask,
"space": "inherit",
},
"compute_epi_mask": {
"family": "Callable",
"func": compute_epi_mask,
"space": "inherit",
},
"UKB_15K_GM": {
"family": "UKB",
"space": "MNI152NLin6Asym",
},
}
)
# Set built-in to registry
self._registry = self._builtin
# Update registry with built-in ones
self._registry.update(self._builtin)
def register(
self,
@ -297,19 +302,18 @@ class MaskRegistry(BasePipelineDataRegistry, metaclass=Singleton):
Raises
------
ValueError
If the mask ``name`` is already registered and
``overwrite=False`` or
if the mask ``name`` is a built-in mask.
If the mask ``name`` is a built-in mask or
if the mask ``name`` is already registered and
``overwrite=False``.
"""
# Check for attempt of overwriting built-in mask
if name in self._builtin:
raise_error(f"Mask: {name} already registered as built-in mask.")
# Check for attempt of overwriting external masks
if name in self._external:
if overwrite:
logger.info(f"Overwriting mask: {name}")
if self._registry[name]["family"] != "CustomUserMask":
raise_error(
f"Mask: {name} already registered as built-in mask."
)
else:
raise_error(
f"Mask: {name} already registered. Set `overwrite=True` "
@ -318,16 +322,17 @@ class MaskRegistry(BasePipelineDataRegistry, metaclass=Singleton):
# Convert str to Path
if not isinstance(mask_path, Path):
mask_path = Path(mask_path)
# Registration
logger.info(f"Registering mask: {name}")
# Add mask info
self._external[name] = {
"path": str(mask_path.absolute()),
"path": mask_path,
"family": "CustomUserMask",
"space": space,
}
# Update registry
self._registry[name] = {
"path": str(mask_path.absolute()),
"path": mask_path,
"family": "CustomUserMask",
"space": space,
}
@ -396,14 +401,22 @@ class MaskRegistry(BasePipelineDataRegistry, metaclass=Singleton):
# Check if the mask family is custom or built-in
mask_img = None
if t_family == "CustomUserMask":
mask_fname = Path(mask_definition["path"])
elif t_family == "Vickery-Patil":
mask_fname = _load_vickery_patil_mask(name, resolution)
mask_fname = mask_definition["path"]
elif t_family == "Callable":
mask_img = mask_definition["func"]
mask_fname = None
elif t_family == "UKB":
mask_fname = _load_ukb_mask(name)
elif t_family in ["Vickery-Patil", "UKB"]:
# Get dataset
dataset = check_dataset()
# Load mask
if t_family == "Vickery-Patil":
mask_fname = _load_vickery_patil_mask(
dataset=dataset,
name=name,
resolution=resolution,
)
elif t_family == "UKB":
mask_fname = _load_ukb_mask(dataset=dataset, name=name)
else:
raise_error(f"Unknown mask family: {t_family}")
@ -685,6 +698,7 @@ class MaskRegistry(BasePipelineDataRegistry, metaclass=Singleton):
def _load_vickery_patil_mask(
dataset: "Dataset",
name: str,
resolution: Optional[float] = None,
) -> Path:
@ -692,6 +706,8 @@ def _load_vickery_patil_mask(
Parameters
----------
dataset : datalad.api.Dataset
The datalad dataset to fetch mask from.
name : {"GM_prob0.2", "GM_prob0.2_cortex"}
The name of the mask.
resolution : float, optional
@ -712,6 +728,7 @@ def _load_vickery_patil_mask(
``name = "GM_prob0.2"``.
"""
# Check name
if name == "GM_prob0.2":
available_resolutions = [1.5, 3.0]
to_load = closest_resolution(resolution, available_resolutions)
@ -730,17 +747,20 @@ def _load_vickery_patil_mask(
else:
raise_error(f"Cannot find a Vickery-Patil mask called {name}")
# Set path for masks
mask_fname = _masks_path / "vickery-patil" / mask_fname
return mask_fname
# Fetch file
return fetch_file_via_datalad(
dataset=dataset,
file_path=dataset.pathobj / "masks" / "Vickery-Patil" / mask_fname,
)
def _load_ukb_mask(name: str) -> Path:
def _load_ukb_mask(dataset: "Dataset", name: str) -> Path:
"""Load UKB mask.
Parameters
----------
dataset : datalad.api.Dataset
The datalad dataset to fetch mask from.
name : {"UKB_15K_GM"}
The name of the mask.
@ -755,15 +775,17 @@ def _load_ukb_mask(name: str) -> Path:
If ``name`` is invalid.
"""
# Check name
if name == "UKB_15K_GM":
mask_fname = "UKB_15K_GM_template.nii.gz"
else:
raise_error(f"Cannot find a UKB mask called {name}")
# Set path for masks
mask_fname = _masks_path / "ukb" / mask_fname
return mask_fname
# Fetch file
return fetch_file_via_datalad(
dataset=dataset,
file_path=dataset.pathobj / "masks" / "UKB" / mask_fname,
)
def _get_interpolation_method(img: "Nifti1Image") -> str:

7
junifer/data/masks/tests/test_masks.py
View file

@ -26,6 +26,7 @@ from junifer.data.masks._masks import (
_load_ukb_mask,
_load_vickery_patil_mask,
)
from junifer.data.utils import check_dataset
from junifer.datagrabber import DMCC13Benchmark
from junifer.datareader import DefaultDataReader
from junifer.testing.datagrabbers import (
@ -282,7 +283,9 @@ def test_vickery_patil(
def test_vickery_patil_error() -> None:
"""Test error for Vickery-Patil mask."""
with pytest.raises(ValueError, match=r"find a Vickery-Patil mask "):
_load_vickery_patil_mask(name="wrong", resolution=2.0)
_load_vickery_patil_mask(
dataset=check_dataset(), name="wrong", resolution=2.0
)
def test_ukb() -> None:
@ -297,7 +300,7 @@ def test_ukb() -> None:
def test_ukb_error() -> None:
"""Test error for UKB mask."""
with pytest.raises(ValueError, match=r"find a UKB mask "):
_load_ukb_mask(name="wrong")
_load_ukb_mask(dataset=check_dataset(), name="wrong")
def test_get() -> None:

975
junifer/data/parcellations/_parcellations.py
View file

File diff suppressed because it is too large Load diff

295
junifer/data/parcellations/tests/test_parcellations.py
View file

@ -18,13 +18,13 @@ from junifer.data.parcellations import merge_parcellations
from junifer.data.parcellations._parcellations import (
_retrieve_aicha,
_retrieve_brainnetome,
_retrieve_parcellation,
_retrieve_schaefer,
_retrieve_shen,
_retrieve_suit,
_retrieve_tian,
_retrieve_yan,
)
from junifer.data.utils import check_dataset
from junifer.datareader import DefaultDataReader
from junifer.pipeline.utils import _check_ants
from junifer.testing.datagrabbers import (
@ -247,12 +247,6 @@ def test_load_incorrect() -> None:
ParcellationRegistry().load("wrongparcellation", "MNI152NLin6Asym")
def test_retrieve_parcellation_incorrect() -> None:
"""Test retrieval of invalid parcellations."""
with pytest.raises(ValueError, match=r"provided parcellation name"):
_retrieve_parcellation("wrongparcellation")
@pytest.mark.parametrize(
"resolution, n_rois, yeo_networks",
[
@ -299,7 +293,6 @@ def test_retrieve_parcellation_incorrect() -> None:
],
)
def test_schaefer(
tmp_path: Path,
resolution: float,
n_rois: int,
yeo_networks: int,
@ -308,8 +301,6 @@ def test_schaefer(
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
resolution : float
The parametrized resolution values.
n_rois : int
@ -329,7 +320,6 @@ def test_schaefer(
img, label, img_path, space = ParcellationRegistry().load(
name=parcellation_name,
target_space="MNI152NLin6Asym",
parcellations_dir=tmp_path,
resolution=resolution,
)
assert img is not None
@ -341,36 +331,22 @@ def test_schaefer(
)
def test_retrieve_schaefer_incorrect_n_rois(tmp_path: Path) -> None:
"""Test retrieve Schaefer with incorrect ROIs.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_schaefer_incorrect_n_rois() -> None:
"""Test retrieve Schaefer with incorrect ROIs."""
with pytest.raises(ValueError, match=r"The parameter `n_rois`"):
_retrieve_schaefer(
parcellations_dir=tmp_path,
dataset=check_dataset(),
resolution=1,
n_rois=101,
yeo_networks=7,
)
def test_retrieve_schaefer_incorrect_yeo_networks(tmp_path: Path) -> None:
"""Test retrieve Schaefer with incorrect Yeo networks.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_schaefer_incorrect_yeo_networks() -> None:
"""Test retrieve Schaefer with incorrect Yeo networks."""
with pytest.raises(ValueError, match=r"The parameter `yeo_networks`"):
_retrieve_schaefer(
parcellations_dir=tmp_path,
dataset=check_dataset(),
resolution=1,
n_rois=100,
yeo_networks=8,
@ -381,13 +357,11 @@ def test_retrieve_schaefer_incorrect_yeo_networks(tmp_path: Path) -> None:
"space_key, space",
[("SUIT", "SUIT"), ("MNI", "MNI152NLin6Asym")],
)
def test_suit(tmp_path: Path, space_key: str, space: str) -> None:
def test_suit(space_key: str, space: str) -> None:
"""Test SUIT parcellation.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
space_key : str
The parametrized space values for the key.
space : str
@ -399,7 +373,6 @@ def test_suit(tmp_path: Path, space_key: str, space: str) -> None:
img, label, img_path, parcellation_space = ParcellationRegistry().load(
name=f"SUITx{space_key}",
target_space=space,
parcellations_dir=tmp_path,
)
assert img is not None
assert img_path.name == f"SUIT_{space_key}Space_1mm.nii"
@ -408,33 +381,20 @@ def test_suit(tmp_path: Path, space_key: str, space: str) -> None:
assert_array_equal(img.header["pixdim"][1:4], [1, 1, 1]) # type: ignore
def test_retrieve_suit_incorrect_space(tmp_path: Path) -> None:
"""Test retrieve SUIT with incorrect space.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_suit_incorrect_space() -> None:
"""Test retrieve SUIT with incorrect space."""
with pytest.raises(ValueError, match=r"The parameter `space`"):
_retrieve_suit(
parcellations_dir=tmp_path, resolution=1.0, space="wrong"
)
_retrieve_suit(dataset=check_dataset(), resolution=1.0, space="wrong")
@pytest.mark.parametrize(
"scale, n_label", [(1, 16), (2, 32), (3, 50), (4, 54)]
)
def test_tian_3T_6thgeneration(
tmp_path: Path, scale: int, n_label: int
) -> None:
def test_tian_3T_6thgeneration(scale: int, n_label: int) -> None:
"""Test Tian parcellation.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
scale : int
The parametrized scale values.
n_label : int
@ -447,44 +407,38 @@ def test_tian_3T_6thgeneration(
assert "TianxS3x3TxMNI6thgeneration" in parcellations
assert "TianxS4x3TxMNI6thgeneration" in parcellations
# Load parcellation
img, lbl, fname, parcellation_space_1 = ParcellationRegistry().load(
img, lbl, fname, space = ParcellationRegistry().load(
name=f"TianxS{scale}x3TxMNI6thgeneration",
parcellations_dir=tmp_path,
target_space="MNI152NLin2009cAsym",
target_space="MNI152NLin2009cAsym", # force highest resolution
)
fname1 = f"Tian_Subcortex_S{scale}_3T_1mm.nii.gz"
expected_fname = f"Tian_Subcortex_S{scale}_3T_1mm.nii.gz"
assert img is not None
assert fname.name == fname1
assert parcellation_space_1 == "MNI152NLin6Asym"
assert fname.name == expected_fname
assert space == "MNI152NLin6Asym"
assert len(lbl) == n_label
assert_array_equal(img.header["pixdim"][1:4], [1, 1, 1]) # type: ignore
assert_array_equal(img.header["pixdim"][1:4], [1, 1, 1])
# Load parcellation
img, lbl, fname, parcellation_space_2 = ParcellationRegistry().load(
img, lbl, fname, space = ParcellationRegistry().load(
name=f"TianxS{scale}x3TxMNI6thgeneration",
target_space="MNI152NLin6Asym",
parcellations_dir=tmp_path,
resolution=2,
)
fname1 = f"Tian_Subcortex_S{scale}_3T.nii.gz"
expected_fname = f"Tian_Subcortex_S{scale}_3T.nii.gz"
assert img is not None
assert fname.name == fname1
assert parcellation_space_2 == "MNI152NLin6Asym"
assert fname.name == expected_fname
assert space == "MNI152NLin6Asym"
assert len(lbl) == n_label
assert_array_equal(img.header["pixdim"][1:4], [2, 2, 2]) # type: ignore
assert_array_equal(img.header["pixdim"][1:4], [2, 2, 2])
@pytest.mark.parametrize(
"scale, n_label", [(1, 16), (2, 32), (3, 50), (4, 54)]
)
def test_tian_3T_nonlinear2009cAsym(
tmp_path: Path, scale: int, n_label: int
) -> None:
def test_tian_3T_nonlinear2009cAsym(scale: int, n_label: int) -> None:
"""Test Tian parcellation.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
scale : int
The parametrized scale values.
n_label : int
@ -499,29 +453,36 @@ def test_tian_3T_nonlinear2009cAsym(
# Load parcellation
img, lbl, fname, space = ParcellationRegistry().load(
name=f"TianxS{scale}x3TxMNInonlinear2009cAsym",
target_space="MNI152NLin2009cAsym",
parcellations_dir=tmp_path,
target_space="MNI152NLin6Asym", # force highest resolution
)
fname1 = f"Tian_Subcortex_S{scale}_3T_2009cAsym.nii.gz"
expected_fname = f"Tian_Subcortex_S{scale}_3T_2009cAsym_1mm.nii.gz"
assert img is not None
assert fname.name == fname1
assert fname.name == expected_fname
assert space == "MNI152NLin2009cAsym"
assert len(lbl) == n_label
assert_array_equal(img.header["pixdim"][1:4], [2, 2, 2]) # type: ignore
assert_array_equal(img.header["pixdim"][1:4], [1, 1, 1])
# Load parcellation
img, lbl, fname, space = ParcellationRegistry().load(
name=f"TianxS{scale}x3TxMNInonlinear2009cAsym",
target_space="MNI152NLin2009cAsym",
resolution=2,
)
expected_fname = f"Tian_Subcortex_S{scale}_3T_2009cAsym.nii.gz"
assert img is not None
assert fname.name == expected_fname
assert space == "MNI152NLin2009cAsym"
assert len(lbl) == n_label
assert_array_equal(img.header["pixdim"][1:4], [2, 2, 2])
@pytest.mark.parametrize(
"scale, n_label", [(1, 16), (2, 34), (3, 54), (4, 62)]
)
def test_tian_7T_6thgeneration(
tmp_path: Path, scale: int, n_label: int
) -> None:
def test_tian_7T_6thgeneration(scale: int, n_label: int) -> None:
"""Test Tian parcellation.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
scale : int
The parametrized scale values.
n_label : int
@ -537,7 +498,6 @@ def test_tian_7T_6thgeneration(
img, lbl, fname, space = ParcellationRegistry().load(
name=f"TianxS{scale}x7TxMNI6thgeneration",
target_space="MNI152NLin6Asym",
parcellations_dir=tmp_path,
)
fname1 = f"Tian_Subcortex_S{scale}_7T.nii.gz"
assert img is not None
@ -549,23 +509,16 @@ def test_tian_7T_6thgeneration(
)
def test_retrieve_tian_incorrect_space(tmp_path: Path) -> None:
"""Test retrieve tian with incorrect space.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_tian_incorrect_space() -> None:
"""Test retrieve tian with incorrect space."""
with pytest.raises(ValueError, match=r"The parameter `space`"):
_retrieve_tian(
parcellations_dir=tmp_path, resolution=1, scale=1, space="wrong"
dataset=check_dataset(), resolution=1, scale=1, space="wrong"
)
with pytest.raises(ValueError, match=r"MNI152NLin6Asym"):
_retrieve_tian(
parcellations_dir=tmp_path,
dataset=check_dataset(),
resolution=1,
scale=1,
magneticfield="7T",
@ -573,18 +526,11 @@ def test_retrieve_tian_incorrect_space(tmp_path: Path) -> None:
)
def test_retrieve_tian_incorrect_magneticfield(tmp_path: Path) -> None:
"""Test retrieve tian with incorrect magneticfield.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_tian_incorrect_magneticfield() -> None:
"""Test retrieve tian with incorrect magneticfield."""
with pytest.raises(ValueError, match=r"The parameter `magneticfield`"):
_retrieve_tian(
parcellations_dir=tmp_path,
dataset=check_dataset(),
resolution=1,
scale=1,
magneticfield="wrong",
@ -592,17 +538,10 @@ def test_retrieve_tian_incorrect_magneticfield(tmp_path: Path) -> None:
def test_retrieve_tian_incorrect_scale(tmp_path: Path) -> None:
"""Test retrieve tian with incorrect scale.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
"""Test retrieve tian with incorrect scale."""
with pytest.raises(ValueError, match=r"The parameter `scale`"):
_retrieve_tian(
parcellations_dir=tmp_path,
dataset=check_dataset(),
resolution=1,
scale=5,
space="MNI152NLin6Asym",
@ -610,7 +549,7 @@ def test_retrieve_tian_incorrect_scale(tmp_path: Path) -> None:
@pytest.mark.parametrize("version", [1, 2])
def test_aicha(tmp_path: Path, version: int) -> None:
def test_aicha(version: int) -> None:
"""Test AICHA parcellation.
Parameters
@ -626,7 +565,6 @@ def test_aicha(tmp_path: Path, version: int) -> None:
img, label, img_path, space = ParcellationRegistry().load(
name=f"AICHA_v{version}",
target_space="IXI549Space",
parcellations_dir=tmp_path,
)
assert img is not None
assert img_path.name == "AICHA.nii"
@ -635,18 +573,11 @@ def test_aicha(tmp_path: Path, version: int) -> None:
assert_array_equal(img.header["pixdim"][1:4], [2, 2, 2]) # type: ignore
def test_retrieve_aicha_incorrect_version(tmp_path: Path) -> None:
"""Test retrieve AICHA with incorrect version.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_aicha_incorrect_version() -> None:
"""Test retrieve AICHA with incorrect version."""
with pytest.raises(ValueError, match="The parameter `version`"):
_retrieve_aicha(
parcellations_dir=tmp_path,
dataset=check_dataset(),
version=100,
)
@ -666,7 +597,6 @@ def test_retrieve_aicha_incorrect_version(tmp_path: Path) -> None:
],
)
def test_shen(
tmp_path: Path,
resolution: float,
year: int,
n_rois: int,
@ -677,8 +607,6 @@ def test_shen(
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
resolution : float
The parametrized resolution values.
year : int
@ -696,7 +624,6 @@ def test_shen(
img, label, img_path, space = ParcellationRegistry().load(
name=f"Shen_{year}_{n_rois}",
target_space="MNI152NLin2009cAsym",
parcellations_dir=tmp_path,
resolution=resolution,
)
assert img is not None
@ -708,34 +635,20 @@ def test_shen(
)
def test_retrieve_shen_incorrect_year(tmp_path: Path) -> None:
"""Test retrieve Shen with incorrect year.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_shen_incorrect_year() -> None:
"""Test retrieve Shen with incorrect year."""
with pytest.raises(ValueError, match="The parameter `year`"):
_retrieve_shen(
parcellations_dir=tmp_path,
dataset=check_dataset(),
year=1969,
)
def test_retrieve_shen_incorrect_n_rois(tmp_path: Path) -> None:
"""Test retrieve Shen with incorrect ROIs.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_shen_incorrect_n_rois() -> None:
"""Test retrieve Shen with incorrect ROIs."""
with pytest.raises(ValueError, match="The parameter `n_rois`"):
_retrieve_shen(
parcellations_dir=tmp_path,
dataset=check_dataset(),
year=2015,
n_rois=10,
)
@ -758,7 +671,6 @@ def test_retrieve_shen_incorrect_n_rois(tmp_path: Path) -> None:
],
)
def test_retrieve_shen_incorrect_param_combo(
tmp_path: Path,
resolution: float,
year: int,
n_rois: int,
@ -779,7 +691,7 @@ def test_retrieve_shen_incorrect_param_combo(
"""
with pytest.raises(ValueError, match="The parameter combination"):
_retrieve_shen(
parcellations_dir=tmp_path,
dataset=check_dataset(),
resolution=resolution,
year=year,
n_rois=n_rois,
@ -852,7 +764,6 @@ def test_retrieve_shen_incorrect_param_combo(
],
)
def test_yan(
tmp_path: Path,
resolution: float,
n_rois: int,
yeo_networks: int,
@ -862,8 +773,6 @@ def test_yan(
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
resolution : float
The parametrized resolution values.
n_rois : int
@ -893,7 +802,6 @@ def test_yan(
img, label, img_path, space = ParcellationRegistry().load(
name=parcellation_name,
target_space="MNI152NLin6Asym",
parcellations_dir=tmp_path,
resolution=resolution,
)
assert img is not None
@ -905,20 +813,13 @@ def test_yan(
)
def test_retrieve_yan_incorrect_networks(tmp_path: Path) -> None:
"""Test retrieve Yan with incorrect networks.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_yan_incorrect_networks() -> None:
"""Test retrieve Yan with incorrect networks."""
with pytest.raises(
ValueError, match="Either one of `yeo_networks` or `kong_networks`"
):
_retrieve_yan(
parcellations_dir=tmp_path,
dataset=check_dataset(),
n_rois=31418,
yeo_networks=100,
kong_networks=100,
@ -928,59 +829,38 @@ def test_retrieve_yan_incorrect_networks(tmp_path: Path) -> None:
ValueError, match="Either one of `yeo_networks` or `kong_networks`"
):
_retrieve_yan(
parcellations_dir=tmp_path,
dataset=check_dataset(),
n_rois=31418,
yeo_networks=None,
kong_networks=None,
)
def test_retrieve_yan_incorrect_n_rois(tmp_path: Path) -> None:
"""Test retrieve Yan with incorrect ROIs.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_yan_incorrect_n_rois() -> None:
"""Test retrieve Yan with incorrect ROIs."""
with pytest.raises(ValueError, match="The parameter `n_rois`"):
_retrieve_yan(
parcellations_dir=tmp_path,
dataset=check_dataset(),
n_rois=31418,
yeo_networks=7,
)
def test_retrieve_yan_incorrect_yeo_networks(tmp_path: Path) -> None:
"""Test retrieve Yan with incorrect Yeo networks.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_yan_incorrect_yeo_networks() -> None:
"""Test retrieve Yan with incorrect Yeo networks."""
with pytest.raises(ValueError, match="The parameter `yeo_networks`"):
_retrieve_yan(
parcellations_dir=tmp_path,
dataset=check_dataset(),
n_rois=100,
yeo_networks=27,
)
def test_retrieve_yan_incorrect_kong_networks(tmp_path: Path) -> None:
"""Test retrieve Yan with incorrect Kong networks.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_yan_incorrect_kong_networks() -> None:
"""Test retrieve Yan with incorrect Kong networks."""
with pytest.raises(ValueError, match="The parameter `kong_networks`"):
_retrieve_yan(
parcellations_dir=tmp_path,
dataset=check_dataset(),
n_rois=100,
kong_networks=27,
)
@ -1001,7 +881,6 @@ def test_retrieve_yan_incorrect_kong_networks(tmp_path: Path) -> None:
],
)
def test_brainnetome(
tmp_path: Path,
resolution: float,
threshold: int,
) -> None:
@ -1009,8 +888,6 @@ def test_brainnetome(
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
resolution : float
The parametrized resolution values.
threshold : int
@ -1030,7 +907,6 @@ def test_brainnetome(
img, label, img_path, space = ParcellationRegistry().load(
name=parcellation_name,
target_space="MNI152NLin6Asym",
parcellations_dir=tmp_path,
resolution=resolution,
)
assert img is not None
@ -1042,18 +918,11 @@ def test_brainnetome(
)
def test_retrieve_brainnetome_incorrect_threshold(tmp_path: Path) -> None:
"""Test retrieve Brainnetome with incorrect threshold.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
def test_retrieve_brainnetome_incorrect_threshold() -> None:
"""Test retrieve Brainnetome with incorrect threshold."""
with pytest.raises(ValueError, match="The parameter `threshold`"):
_retrieve_brainnetome(
parcellations_dir=tmp_path,
dataset=check_dataset(),
threshold=100,
)
@ -1214,7 +1083,7 @@ def test_get_single() -> None:
bold_img = bold["data"]
# Get tailored parcellation
tailored_parcellation, tailored_labels = ParcellationRegistry().get(
parcellations=["TianxS1x3TxMNInonlinear2009cAsym"],
parcellations=["Shen_2015_268"],
target_data=bold,
)
# Check shape and affine with original element data
@ -1222,9 +1091,9 @@ def test_get_single() -> None:
assert_array_equal(tailored_parcellation.affine, bold_img.affine)
# Get raw parcellation
raw_parcellation, raw_labels, _, _ = ParcellationRegistry().load(
"TianxS1x3TxMNInonlinear2009cAsym",
name="Shen_2015_268",
target_space="MNI152NLin2009cAsym",
resolution=1.5,
resolution=4,
)
resampled_raw_parcellation = resample_to_img(
source_img=raw_parcellation,
@ -1266,7 +1135,9 @@ def test_get_multi_same_space() -> None:
]
for name in parcellations_names:
img, labels, _, _ = ParcellationRegistry().load(
name=name, target_space="MNI152NLin2009cAsym", resolution=1.5
name=name,
target_space="MNI152NLin2009cAsym",
resolution=4,
)
# Resample raw parcellations
resampled_img = resample_to_img(

102
junifer/data/template_spaces.py
View file

@ -6,22 +6,18 @@
from pathlib import Path
from typing import Any, Optional, Union
import datalad.api as dl
import nibabel as nib
import numpy as np
from datalad.support.exceptions import IncompleteResultsError
from templateflow import api as tflow
from ..utils import logger, raise_error
from .utils import closest_resolution
from .utils import check_dataset, closest_resolution, fetch_file_via_datalad
__all__ = ["get_template", "get_xfm"]
def get_xfm(
src: str, dst: str, xfms_dir: Union[str, Path, None] = None
) -> Path: # pragma: no cover
def get_xfm(src: str, dst: str) -> Path: # pragma: no cover
"""Fetch warp files to convert from ``src`` to ``dst``.
Parameters
@ -30,94 +26,24 @@ def get_xfm(
The template space to transform from.
dst : str
The template space to transform to.
xfms_dir : str or pathlib.Path, optional
Path where the retrieved transformation files are stored.
The default location is "$HOME/junifer/data/xfms" (default None).
Returns
-------
pathlib.Path
The path to the transformation file.
Raises
------
RuntimeError
If there is a problem cloning the xfm dataset or
if there is a problem fetching the xfm file.
"""
# Set default path for storage
if xfms_dir is None:
xfms_dir = Path().home() / "junifer" / "data" / "xfms"
# Convert str to Path
if not isinstance(xfms_dir, Path):
xfms_dir = Path(xfms_dir)
# Check if the template xfms dataset is installed at storage path
is_installed = dl.Dataset(xfms_dir).is_installed()
# Use existing dataset
if is_installed:
logger.debug(
f"Found existing template xfms dataset at: {xfms_dir.resolve()}"
)
# Set dataset
dataset = dl.Dataset(xfms_dir)
# Clone a fresh copy
else:
logger.debug(f"Cloning template xfms dataset to: {xfms_dir.resolve()}")
# Clone dataset
try:
dataset = dl.clone(
"https://github.com/juaml/human-template-xfms.git",
path=xfms_dir,
result_renderer="disabled",
)
except IncompleteResultsError as e:
raise_error(
msg=f"Failed to clone dataset: {e.failed}",
klass=RuntimeError,
)
else:
logger.debug(
f"Successfully cloned template xfms dataset to: "
f"{xfms_dir.resolve()}"
)
# Get dataset
dataset = check_dataset()
# Set file path to retrieve
xfm_file_path = (
xfms_dir / "xfms" / f"{src}_to_{dst}" / f"{src}_to_{dst}_Composite.h5"
dataset.pathobj
/ "xfms"
/ f"{src}_to_{dst}"
/ f"{src}_to_{dst}_Composite.h5"
)
# Retrieve file
try:
got = dataset.get(xfm_file_path, result_renderer="disabled")
except IncompleteResultsError as e:
raise_error(
msg=f"Failed to get file from dataset: {e.failed}",
klass=RuntimeError,
)
else:
file_path = Path(got[0]["path"])
# Conditional logging based on file fetch
status = got[0]["status"]
if status == "ok":
logger.info(
f"Successfully fetched xfm file for {src} to {dst} at "
f"{file_path.resolve()}"
)
return file_path
elif status == "notneeded":
logger.info(
f"Found existing xfm file for {src} to {dst} at "
f"{file_path.resolve()}"
)
return file_path
else:
raise_error(
f"Failed to fetch xfm file for {src} to {dst} at "
f"{file_path.resolve()}"
)
return fetch_file_via_datalad(dataset=dataset, file_path=xfm_file_path)
def get_template(
@ -190,7 +116,7 @@ def get_template(
logger.info(
f"Downloading template {space} ({template_type} in "
f"resolution {resolution}"
f"resolution {resolution})"
)
# Retrieve template
try:
@ -224,9 +150,11 @@ def get_template(
)
except Exception: # noqa: BLE001
raise_error(
f"Template {space} ({template_type}) with resolution {resolution} "
"not found",
msg=(
f"Template {space} ({template_type}) with resolution "
f"{resolution}) not found"
),
klass=RuntimeError,
)
else:
return nib.load(template_path) # type: ignore
return nib.load(template_path)

105
junifer/data/utils.py
View file

@ -5,14 +5,22 @@
# License: AGPL
from collections.abc import MutableMapping
from pathlib import Path
from typing import Optional, Union
import datalad.api as dl
import numpy as np
from datalad.support.exceptions import IncompleteResultsError
from ..utils import logger, raise_error
from ..utils import config, logger, raise_error
__all__ = ["closest_resolution", "get_native_warper"]
__all__ = [
"check_dataset",
"closest_resolution",
"fetch_file_via_datalad",
"get_native_warper",
]
def closest_resolution(
@ -114,3 +122,96 @@ def get_native_warper(
)
fraimondo commented 2025-01-10 06:27:51 +00:00 (Migrated from github.com)
Copy link

This should be DEBUG info.

This should be DEBUG info.
return possible_warpers[0]
def check_dataset() -> dl.Dataset:
"""Get or install junifer-data dataset.
Returns
-------
datalad.api.Dataset
The junifer-data dataset.
Raises
------
RuntimeError
If there is a problem cloning the dataset.
"""
# Check config and set default if not passed
data_dir = config.get("data.location")
if data_dir is not None:
data_dir = Path(data_dir)
else:
data_dir = Path().home() / "junifer_data"
# Check if the dataset is installed at storage path;
# else clone a fresh copy
if dl.Dataset(data_dir).is_installed():
logger.debug(f"Found existing junifer-data at: {data_dir.resolve()}")
return dl.Dataset(data_dir)
fraimondo commented 2025-01-10 06:27:29 +00:00 (Migrated from github.com)
Copy link

This function needs to check that we are also using/cloning the exact junifer data version that we want.

In our last discussion, we said we will have a subdirectory inside the junifer_data with the corresponding version to be used with the currently running junifer version.

So somewhere in junifer code, we will have a JUNIFER_DATA_VERSION variable with a string/number/id.

Then we will look if the directory junifer_data/{JUNIFER_DATA_VERSION} exists, that its a datalad dataset, it's clean and its tag corresponds to JUNIFER_DATA_VERSION.

This way we can keep junifer data for several installed junifer versions, each one consistently using the data.

This function needs to check that we are also using/cloning the exact junifer data version that we want. In our last discussion, we said we will have a subdirectory inside the `junifer_data` with the corresponding version to be used with the currently running junifer version. So somewhere in junifer code, we will have a JUNIFER_DATA_VERSION variable with a string/number/id. Then we will look if the directory `junifer_data/{JUNIFER_DATA_VERSION}` exists, that its a datalad dataset, it's clean and its tag corresponds to JUNIFER_DATA_VERSION. This way we can keep junifer data for several installed junifer versions, each one consistently using the data.
synchon commented 2025-01-10 13:01:24 +00:00 (Migrated from github.com)
Copy link

This PR is part of that migration and serves to move manual inconsistent asset downloads to using junifer-data via datalad. The idea is to move all dataset validating and data fetching machinery to https://github.com/juaml/junifer-data-client and junifer using its API to fetch data as required using the technique you described. v1.0.0 image of junifer-data is being built and will be published soon.

The next PR will adapt the usage of junifer-data-client API while taking care of versions.

This PR is part of that migration and serves to move manual inconsistent asset downloads to using `junifer-data` via datalad. The idea is to move all dataset validating and data fetching machinery to https://github.com/juaml/junifer-data-client and `junifer` using its API to fetch data as required using the technique you described. v1.0.0 image of `junifer-data` is being built and will be published soon. The next PR will adapt the usage of `junifer-data-client` API while taking care of versions.
else:
logger.debug(f"Cloning junifer-data to: {data_dir.resolve()}")
# Clone dataset
try:
dataset = dl.clone(
"https://github.com/juaml/junifer-data.git",
path=data_dir,
result_renderer="disabled",
)
except IncompleteResultsError as e:
raise_error(
msg=f"Failed to clone junifer-data: {e.failed}",
klass=RuntimeError,
)
else:
logger.debug(
f"Successfully cloned junifer-data to: "
f"{data_dir.resolve()}"
)
return dataset
def fetch_file_via_datalad(dataset: dl.Dataset, file_path: Path) -> Path:
"""Fetch `file_path` from `dataset` via datalad.
Parameters
----------
dataset : datalad.api.Dataset
The datalad dataset to fetch files from.
file_path : pathlib.Path
The file path to fetch.
Returns
-------
pathlib.Path
Resolved fetched file path.
Raises
------
RuntimeError
If there is a problem fetching the file.
"""
try:
got = dataset.get(file_path, result_renderer="disabled")
except IncompleteResultsError as e:
raise_error(
msg=f"Failed to get file from dataset: {e.failed}",
klass=RuntimeError,
)
else:
got_path = Path(got[0]["path"])
# Conditional logging based on file fetch
status = got[0]["status"]
if status == "ok":
logger.info(f"Successfully fetched file: {got_path.resolve()}")
return got_path
elif status == "notneeded":
logger.debug(f"Found existing file: {got_path.resolve()}")
return got_path
else:
raise_error(
msg=f"Failed to fetch file: {got_path.resolve()}",
klass=RuntimeError,
)

4
pyproject.toml
View file

@ -46,7 +46,6 @@ dependencies = [
"sqlalchemy>=2.0.25,<=2.1.0",
"ruamel.yaml>=0.17,<0.19",
"h5py>=3.10",
"httpx[http2]>=0.26.0,<0.28.0",
"tqdm>=4.66.1,<4.67.0",
"templateflow>=23.0.0",
"lapy>=1.0.0,<2.0.0",
@ -195,7 +194,7 @@ ignore = [
[tool.ruff.lint.isort]
lines-after-imports = 2
known-first-party = ["junifer"]
known-third-party =[
known-third-party = [
"click",
"numpy",
"scipy",
@ -206,7 +205,6 @@ known-third-party =[
"sqlalchemy",
"yaml",
"importlib_metadata",
"httpx",
"tqdm",
"templateflow",
"bct",