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", "nilearn",
"sqlalchemy", "sqlalchemy",
"ruamel.yaml", "ruamel.yaml",
"httpx",
"tqdm", "tqdm",
"templateflow", "templateflow",
"lapy", "lapy",
@ -73,7 +72,6 @@ def test_get_dependency_information_long() -> None:
"nilearn", "nilearn",
"sqlalchemy", "sqlalchemy",
"ruamel.yaml", "ruamel.yaml",
"httpx",
"tqdm", "tqdm",
"templateflow", "templateflow",
"lapy", "lapy",

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

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

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

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

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

102
junifer/data/template_spaces.py
View file

@ -6,22 +6,18 @@
from pathlib import Path from pathlib import Path
from typing import Any, Optional, Union from typing import Any, Optional, Union
import datalad.api as dl
import nibabel as nib import nibabel as nib
import numpy as np import numpy as np
from datalad.support.exceptions import IncompleteResultsError
from templateflow import api as tflow from templateflow import api as tflow
from ..utils import logger, raise_error 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"] __all__ = ["get_template", "get_xfm"]
def get_xfm( def get_xfm(src: str, dst: str) -> Path: # pragma: no cover
src: str, dst: str, xfms_dir: Union[str, Path, None] = None
) -> Path: # pragma: no cover
"""Fetch warp files to convert from ``src`` to ``dst``. """Fetch warp files to convert from ``src`` to ``dst``.
Parameters Parameters
@ -30,94 +26,24 @@ def get_xfm(
The template space to transform from. The template space to transform from.
dst : str dst : str
The template space to transform to. 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 Returns
------- -------
pathlib.Path pathlib.Path
The path to the transformation file. 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 # Get dataset
if xfms_dir is None: dataset = check_dataset()
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()}"
)
# Set file path to retrieve # Set file path to retrieve
xfm_file_path = ( 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 # Retrieve file
try: return fetch_file_via_datalad(dataset=dataset, file_path=xfm_file_path)
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()}"
)
def get_template( def get_template(
@ -190,7 +116,7 @@ def get_template(
logger.info( logger.info(
f"Downloading template {space} ({template_type} in " f"Downloading template {space} ({template_type} in "
f"resolution {resolution}" f"resolution {resolution})"
) )
# Retrieve template # Retrieve template
try: try:
@ -224,9 +150,11 @@ def get_template(
) )
except Exception: # noqa: BLE001 except Exception: # noqa: BLE001
raise_error( raise_error(
f"Template {space} ({template_type}) with resolution {resolution} " msg=(
"not found", f"Template {space} ({template_type}) with resolution "
f"{resolution}) not found"
),
klass=RuntimeError, klass=RuntimeError,
) )
else: 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 # License: AGPL
from collections.abc import MutableMapping from collections.abc import MutableMapping
from pathlib import Path
from typing import Optional, Union from typing import Optional, Union
import datalad.api as dl
import numpy as np 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( 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] 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,
)

2
pyproject.toml
View file

@ -46,7 +46,6 @@ dependencies = [
"sqlalchemy>=2.0.25,<=2.1.0", "sqlalchemy>=2.0.25,<=2.1.0",
"ruamel.yaml>=0.17,<0.19", "ruamel.yaml>=0.17,<0.19",
"h5py>=3.10", "h5py>=3.10",
"httpx[http2]>=0.26.0,<0.28.0",
"tqdm>=4.66.1,<4.67.0", "tqdm>=4.66.1,<4.67.0",
"templateflow>=23.0.0", "templateflow>=23.0.0",
"lapy>=1.0.0,<2.0.0", "lapy>=1.0.0,<2.0.0",
@ -206,7 +205,6 @@ known-third-party =[
"sqlalchemy", "sqlalchemy",
"yaml", "yaml",
"importlib_metadata", "importlib_metadata",
"httpx",
"tqdm", "tqdm",
"templateflow", "templateflow",
"bct", "bct",