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

[ENH]: Allow users to select types for datagrabbers in order to avoid downloading unnecesary data. #132

Merged
synchon merged 18 commits from update/types-for-aomic-dg into main 2023-07-21 11:28:24 +00:00
Conversation 5 Commits 18 Files changed 13 +584 -340
13 changed files with 584 additions and 340 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/changes/newsfragments/132.change Normal file
View file

@ -0,0 +1 @@
Expose ``types`` parameter for :class:`.DataladAOMICID1000`, :class:`.DataladAOMICPIOP1`, :class:`.DataladAOMICPIOP2` and :class:`.JuselessUCLA` by `Synchon Mandal`_

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

@ -0,0 +1 @@
Change validation of ``types`` against ``patterns`` to allow a subset of ``patterns``'s types to be used for ``DataGrabber`` data fetch by `Synchon Mandal`_

56
junifer/configs/juseless/datagrabbers/tests/test_ucla.py
View file

@ -6,20 +6,17 @@
# License: AGPL
import socket
from typing import Optional
from typing import List, Optional, Union
import pytest
from junifer.configs.juseless.datagrabbers import JuselessUCLA
from junifer.utils.logging import configure_logging
# Check if the test is running on juseless
if socket.gethostname() != "juseless":
pytest.skip("These tests are only for juseless", allow_module_level=True)
configure_logging(level="DEBUG")
def test_JuselessUCLA() -> None:
"""Test JuselessUCLA."""
@ -42,6 +39,57 @@ def test_JuselessUCLA() -> None:
assert out[t]["path"].exists()
@pytest.mark.parametrize(
"types",
[
"BOLD",
"BOLD_confounds",
"T1w",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
["BOLD", "BOLD_confounds"],
["T1w", "probseg_CSF"],
["probseg_GM", "probseg_WM"],
["BOLD", "T1w"],
],
)
def test_JuselessUCLA_partial_data_access(
types: Union[str, List[str]],
) -> None:
"""Test JuselessUCLA DataGrabber partial data access.
Parameters
----------
types : str or list of str
The parametrized types.
"""
dg = JuselessUCLA(types=types)
with dg:
# Get all elements
all_elements = dg.get_elements()
# Get test element
test_element = all_elements[0]
# Get test element data
out = dg[test_element]
# Assert data type
if isinstance(types, list):
for type_ in types:
assert type_ in out
else:
assert types in out
def test_JuselessUCLA_incorrect_data_type() -> None:
"""Test JuselessUCLA DataGrabber incorrect data type."""
with pytest.raises(
ValueError, match="`patterns` must contain all `types`"
):
_ = JuselessUCLA(types="Eunomia")
@pytest.mark.parametrize(
"tasks",
[None, "rest", ["rest", "stopsignal"]],

43
junifer/configs/juseless/datagrabbers/ucla.py
View file

@ -20,9 +20,13 @@ class JuselessUCLA(PatternDataGrabber):
Parameters
----------
datadir : str or pathlib.Path, optional
The directory where the dataset is stored
datadir : str or Path, optional
The directory where the dataset is stored.
(default "/data/project/psychosis_thalamus/data/fmriprep").
types: {"BOLD", "BOLD_confounds", "T1w", "probseg_CSF", "probseg_GM", \
"probseg_WM"} or a list of the options, optional
UCLA data types. If None, all available data types are selected.
(default None).
tasks : {"rest", "bart", "bht", "pamenc", "pamret", \
"scap", "taskswitch", "stopsignal"} or \
list of the options or None, optional
@ -36,20 +40,10 @@ class JuselessUCLA(PatternDataGrabber):
datadir: Union[
str, Path
] = "/data/project/psychosis_thalamus/data/fmriprep",
types: Union[str, List[str], None] = None,
tasks: Union[str, List[str], None] = None,
) -> None:
types = [
"BOLD",
"BOLD_confounds",
"T1w",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
]
if isinstance(tasks, str):
tasks = [tasks]
# Declare all tasks
all_tasks = [
"rest",
"bart",
@ -60,18 +54,21 @@ class JuselessUCLA(PatternDataGrabber):
"taskswitch",
"stopsignal",
]
# Set default tasks
if tasks is None:
tasks = all_tasks
else:
# Convert single task into list
if isinstance(tasks, str):
tasks = [tasks]
# Verify valid tasks
for t in tasks:
if t not in all_tasks:
raise_error(
f"{t} is not a valid task in the UCLA dataset!"
)
self.tasks = tasks
# The patterns
patterns = {
"BOLD": (
"sub-{subject}/func/sub-{subject}_task-{task}_bold_space-"
@ -98,12 +95,18 @@ class JuselessUCLA(PatternDataGrabber):
"-MNI152NLin2009cAsym_class-WM_probtissue.nii.gz"
),
}
# Set default types
if types is None:
types = list(patterns.keys())
# Convert single type into list
else:
if not isinstance(types, list):
types = [types]
# The replacements
replacements = ["subject", "task"]
# the commented out uri leads to new open neuro dataset which does
# NOT have preprocessed data
# uri = "https://github.com/OpenNeuroDatasets/ds000030.git"
replacements = ["subject", "task"]
super().__init__(
types=types,
datadir=datadir,

33
junifer/datagrabber/aomic/id1000.py
View file

@ -4,10 +4,11 @@
# Vera Komeyer <v.komeyer@fz-juelich.de>
# Xuan Li <xu.li@fz-juelich.de>
# Leonard Sasse <l.sasse@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from pathlib import Path
from typing import Dict, Union
from typing import Dict, List, Union
from ...api.decorators import register_datagrabber
from ..pattern_datalad import PatternDataladDataGrabber
@ -23,25 +24,18 @@ class DataladAOMICID1000(PatternDataladDataGrabber):
The directory where the datalad dataset will be cloned. If None,
the datalad dataset will be cloned into a temporary directory
(default None).
types: {"BOLD", "BOLD_confounds", "T1w", "probseg_CSF", "probseg_GM", \
"probseg_WM", "DWI"} or a list of the options, optional
AOMIC data types. If None, all available data types are selected.
(default None).
"""
def __init__(
self,
datadir: Union[str, Path, None] = None,
types: Union[str, List[str], None] = None,
) -> None:
# The types of data
types = [
"BOLD",
"BOLD_confounds",
"BOLD_mask",
"T1w",
"T1w_mask",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
"DWI",
]
# The patterns
patterns = {
"BOLD": (
@ -90,6 +84,13 @@ class DataladAOMICID1000(PatternDataladDataGrabber):
"sub-{subject}_desc-preproc_dwi.nii.gz"
),
}
# Set default types
if types is None:
types = list(patterns.keys())
# Convert single type into list
else:
if not isinstance(types, list):
types = [types]
# The replacements
replacements = ["subject"]
uri = "https://github.com/OpenNeuroDatasets/ds003097.git"
@ -118,6 +119,8 @@ class DataladAOMICID1000(PatternDataladDataGrabber):
"""
out = super().get_item(subject=subject)
out["BOLD"]["mask_item"] = "BOLD_mask"
out["T1w"]["mask_item"] = "T1w_mask"
if out.get("BOLD"):
out["BOLD"]["mask_item"] = "BOLD_mask"
if out.get("T1w"):
out["T1w"]["mask_item"] = "T1w_mask"
return out

48
junifer/datagrabber/aomic/piop1.py
View file

@ -4,6 +4,7 @@
# Vera Komeyer <v.komeyer@fz-juelich.de>
# Xuan Li <xu.li@fz-juelich.de>
# Leonard Sasse <l.sasse@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from itertools import product
@ -25,6 +26,10 @@ class DataladAOMICPIOP1(PatternDataladDataGrabber):
The directory where the datalad dataset will be cloned. If None,
the datalad dataset will be cloned into a temporary directory
(default None).
types: {"BOLD", "BOLD_confounds", "T1w", "probseg_CSF", "probseg_GM", \
"probseg_WM", "DWI"} or a list of the options, optional
AOMIC data types. If None, all available data types are selected.
(default None).
tasks : {"restingstate", "anticipation", "emomatching", "faces", \
"gstroop", "workingmemory"} or list of the options, optional
AOMIC PIOP1 task sessions. If None, all available task sessions are
@ -35,24 +40,10 @@ class DataladAOMICPIOP1(PatternDataladDataGrabber):
def __init__(
self,
datadir: Union[str, Path, None] = None,
types: Union[str, List[str], None] = None,
tasks: Union[str, List[str], None] = None,
) -> None:
# The types of data
types = [
"BOLD",
"BOLD_confounds",
"BOLD_mask",
"T1w",
"T1w_mask",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
"DWI",
]
if isinstance(tasks, str):
tasks = [tasks]
# Declare all tasks
all_tasks = [
"restingstate",
"anticipation",
@ -61,19 +52,22 @@ class DataladAOMICPIOP1(PatternDataladDataGrabber):
"gstroop",
"workingmemory",
]
# Set default tasks
if tasks is None:
tasks = all_tasks
else:
# Convert single task into list
if isinstance(tasks, str):
tasks = [tasks]
# Verify valid tasks
for t in tasks:
if t not in all_tasks:
raise_error(
f"{t} is not a valid task in the AOMIC PIOP1"
" dataset!"
)
self.tasks = tasks
# The patterns
patterns = {
"BOLD": (
"derivatives/fmriprep/sub-{subject}/func/"
@ -120,8 +114,16 @@ class DataladAOMICPIOP1(PatternDataladDataGrabber):
"sub-{subject}_desc-preproc_dwi.nii.gz"
),
}
uri = "https://github.com/OpenNeuroDatasets/ds002785"
# Set default types
if types is None:
types = list(patterns.keys())
# Convert single type into list
else:
if not isinstance(types, list):
types = [types]
# The replacements
replacements = ["subject", "task"]
uri = "https://github.com/OpenNeuroDatasets/ds002785"
super().__init__(
types=types,
datadir=datadir,
@ -162,8 +164,10 @@ class DataladAOMICPIOP1(PatternDataladDataGrabber):
new_task = f"{task}_acq-{acq}"
out = super().get_item(subject=subject, task=new_task)
out["BOLD"]["mask_item"] = "BOLD_mask"
out["T1w"]["mask_item"] = "T1w_mask"
if out.get("BOLD"):
out["BOLD"]["mask_item"] = "BOLD_mask"
if out.get("T1w"):
out["T1w"]["mask_item"] = "T1w_mask"
return out
def get_elements(self) -> List:

76
junifer/datagrabber/aomic/piop2.py
View file

@ -4,8 +4,10 @@
# Vera Komeyer <v.komeyer@fz-juelich.de>
# Xuan Li <xu.li@fz-juelich.de>
# Leonard Sasse <l.sasse@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from itertools import product
from pathlib import Path
from typing import Dict, List, Union
@ -24,7 +26,11 @@ class DataladAOMICPIOP2(PatternDataladDataGrabber):
The directory where the datalad dataset will be cloned. If None,
the datalad dataset will be cloned into a temporary directory
(default None).
tasks : {"restingstate", "stopsignal", "emomatching", "workingmemory"} \
types: {"BOLD", "BOLD_confounds", "T1w", "probseg_CSF", "probseg_GM", \
"probseg_WM", "DWI"} or a list of the options, optional
AOMIC data types. If None, all available data types are selected.
(default None).
tasks : {"restingstate", "stopsignal", "workingmemory"} \
or list of the options, optional
AOMIC PIOP2 task sessions. If None, all available task sessions are
selected (default None).
@ -34,58 +40,46 @@ class DataladAOMICPIOP2(PatternDataladDataGrabber):
def __init__(
self,
datadir: Union[str, Path, None] = None,
types: Union[str, List[str], None] = None,
tasks: Union[str, List[str], None] = None,
) -> None:
# The types of data
types = [
"BOLD",
"BOLD_confounds",
"BOLD_mask",
"T1w",
"T1w_mask",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
"DWI",
]
if isinstance(tasks, str):
tasks = [tasks]
# Declare all tasks
all_tasks = [
"restingstate",
"emomatching",
"workingmemory",
"stopsignal",
"workingmemory",
]
# Set default tasks
if tasks is None:
tasks = all_tasks
else:
# Convert single task into list
if isinstance(tasks, str):
tasks = [tasks]
# Verify valid tasks
for t in tasks:
if t not in all_tasks:
raise_error(
f"{t} is not a valid task in the AOMIC PIOP2"
" dataset!"
)
self.tasks = tasks
# The patterns
patterns = {
"BOLD": (
"derivatives/fmriprep/sub-{subject}/func/"
"sub-{subject}_task-{task}_acq-seq_"
"sub-{subject}_task-{task}_"
"space-MNI152NLin2009cAsym_desc-preproc_bold.nii.gz"
),
"BOLD_confounds": (
"derivatives/fmriprep/sub-{subject}/func/"
"sub-{subject}_task-{task}_acq-seq_"
"sub-{subject}_task-{task}_"
"desc-confounds_regressors.tsv"
),
"BOLD_mask": (
"derivatives/fmriprep/sub-{subject}/func/"
"sub-{subject}_task-{task}_acq-seq_space"
"-MNI152NLin2009cAsym_desc-brain_mask.nii.gz"
"sub-{subject}_task-{task}_"
"space-MNI152NLin2009cAsym_desc-brain_mask.nii.gz"
),
"T1w": (
"derivatives/fmriprep/sub-{subject}/anat/"
@ -117,8 +111,16 @@ class DataladAOMICPIOP2(PatternDataladDataGrabber):
"sub-{subject}_desc-preproc_dwi.nii.gz"
),
}
uri = "https://github.com/OpenNeuroDatasets/ds002790"
# Set default types
if types is None:
types = list(patterns.keys())
# Convert single type into list
else:
if not isinstance(types, list):
types = [types]
# The replacements
replacements = ["subject", "task"]
uri = "https://github.com/OpenNeuroDatasets/ds002790"
super().__init__(
types=types,
datadir=datadir,
@ -138,8 +140,11 @@ class DataladAOMICPIOP2(PatternDataladDataGrabber):
imposing constraints based on specified tasks.
"""
all_elements = super().get_elements()
return [x for x in all_elements if x[1] in self.tasks]
subjects = [f"{x:04d}" for x in range(1, 227)]
elems = []
for subject, task in product(subjects, self.tasks):
elems.append((subject, task))
return elems
def get_item(self, subject: str, task: str) -> Dict:
"""Index one element in the dataset.
@ -148,9 +153,8 @@ class DataladAOMICPIOP2(PatternDataladDataGrabber):
----------
subject : str
The subject ID.
task : str
The task to get. Possible values are:
{"restingstate", "stopsignal", "emomatching", "workingmemory"}
task : {"restingstate", "stopsignal", "workingmemory"}
The task to get.
Returns
-------
@ -159,7 +163,9 @@ class DataladAOMICPIOP2(PatternDataladDataGrabber):
specified element.
"""
out = super().get_item(subject=subject, task=task)
out["BOLD"]["mask_item"] = "BOLD_mask"
out["T1w"]["mask_item"] = "T1w_mask"
out = super().get_item(subject=subject, task=f"{task}_acq-seq")
if out.get("BOLD"):
out["BOLD"]["mask_item"] = "BOLD_mask"
if out.get("T1w"):
out["T1w"]["mask_item"] = "T1w_mask"
return out

74
junifer/datagrabber/aomic/tests/test_id1000.py
View file

@ -4,22 +4,24 @@
# Vera Komeyer <v.komeyer@fz-juelich.de>
# Xuan Li <xu.li@fz-juelich.de>
# Leonard Sasse <l.sasse@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from junifer.datagrabber import DataladAOMICID1000
from junifer.utils import configure_logging
from typing import List, Union
import pytest
from junifer.datagrabber.aomic.id1000 import DataladAOMICID1000
URI = "https://gin.g-node.org/juaml/datalad-example-aomic1000"
def test_DataladAOMICID1000() -> None:
"""Test DataladAOMICID1000 DataGrabber."""
uri_ID1000 = "https://gin.g-node.org/juaml/datalad-example-aomic1000"
configure_logging(level="DEBUG")
dg = DataladAOMICID1000()
# change uri here to use fake data instead of real dataset
dg.uri = uri_ID1000
# Set URI to Gin
dg.uri = URI
with dg:
all_elements = dg.get_elements()
@ -122,3 +124,57 @@ def test_DataladAOMICID1000() -> None:
assert "element" in meta
assert "subject" in meta["element"]
assert test_element == meta["element"]["subject"]
@pytest.mark.parametrize(
"types",
[
"BOLD",
"BOLD_confounds",
"T1w",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
"DWI",
["BOLD", "BOLD_confounds"],
["T1w", "probseg_CSF"],
["probseg_GM", "probseg_WM"],
["DWI", "BOLD"],
],
)
def test_DataladAOMICID1000_partial_data_access(
types: Union[str, List[str]],
) -> None:
"""Test DataladAOMICID1000 DataGrabber partial data access.
Parameters
----------
types : str or list of str
The parametrized types.
"""
dg = DataladAOMICID1000(types=types)
# Set URI to Gin
dg.uri = URI
with dg:
# Get all elements
all_elements = dg.get_elements()
# Get test element
test_element = all_elements[0]
# Get test element data
out = dg[test_element]
# Assert data type
if isinstance(types, list):
for type_ in types:
assert type_ in out
else:
assert types in out
def test_DataladAOMICID1000_incorrect_data_type() -> None:
"""Test DataladAOMICID1000 DataGrabber incorrect data type."""
with pytest.raises(
ValueError, match="`patterns` must contain all `types`"
):
_ = DataladAOMICID1000(types="Scooby-Doo")

253
junifer/datagrabber/aomic/tests/test_piop1.py
View file

@ -4,142 +4,201 @@
# Vera Komeyer <v.komeyer@fz-juelich.de>
# Xuan Li <xu.li@fz-juelich.de>
# Leonard Sasse <l.sasse@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from typing import List, Optional, Union
import pytest
from junifer.datagrabber import DataladAOMICPIOP1
from junifer.utils import configure_logging
def test_DataladAOMICPIOP1() -> None:
"""Test DataladAOMICPIOP1 DataGrabber."""
configure_logging(level="DEBUG")
URI = "https://gin.g-node.org/juaml/datalad-example-aomicpiop1"
uri_PIOP1 = "https://gin.g-node.org/juaml/datalad-example-aomicpiop1"
task_params = [None, "restingstate"]
for task_param in task_params:
dg = DataladAOMICPIOP1(tasks=task_param)
@pytest.mark.parametrize(
"tasks",
[None, "restingstate"],
)
def test_DataladAOMICPIOP1(tasks: Optional[str]) -> None:
"""Test DataladAOMICPIOP1 DataGrabber.
# change uri here to use fake data instead of real dataset
dg.uri = uri_PIOP1
Parameters
----------
tasks : str or None
The parametrized task values.
with dg:
all_elements = dg.get_elements()
test_element = all_elements[0]
sub, task = test_element
"""
dg = DataladAOMICPIOP1(tasks=tasks)
# Set URI to Gin
dg.uri = URI
out = dg[test_element]
with dg:
all_elements = dg.get_elements()
test_element = all_elements[0]
sub, task = test_element
# asserts type "BOLD"
assert "BOLD" in out
out = dg[test_element]
# depending on task 'acquisition is different'
task_acqs = {
"anticipation": "seq",
"emomatching": "seq",
"faces": "mb3",
"gstroop": "seq",
"restingstate": "mb3",
"workingmemory": "seq",
}
acq = task_acqs[task]
new_task = f"{task}_acq-{acq}"
assert (
out["BOLD"]["path"].name == f"sub-{sub}_task-{new_task}_"
"space-MNI152NLin2009cAsym_desc-preproc_bold.nii.gz"
)
# asserts type "BOLD"
assert "BOLD" in out
assert out["BOLD"]["path"].exists()
assert out["BOLD"]["path"].is_file()
# depending on task 'acquisition is different'
task_acqs = {
"anticipation": "seq",
"emomatching": "seq",
"faces": "mb3",
"gstroop": "seq",
"restingstate": "mb3",
"workingmemory": "seq",
}
acq = task_acqs[task]
new_task = f"{task}_acq-{acq}"
assert (
out["BOLD"]["path"].name == f"sub-{sub}_task-{new_task}_"
"space-MNI152NLin2009cAsym_desc-preproc_bold.nii.gz"
)
# asserts type "BOLD_confounds"
assert "BOLD_confounds" in out
assert out["BOLD"]["path"].exists()
assert out["BOLD"]["path"].is_file()
assert (
out["BOLD_confounds"]["path"].name
== f"sub-{sub}_task-{new_task}_"
"desc-confounds_regressors.tsv"
)
# asserts type "BOLD_confounds"
assert "BOLD_confounds" in out
assert out["BOLD_confounds"]["path"].exists()
assert out["BOLD_confounds"]["path"].is_file()
assert (
out["BOLD_confounds"]["path"].name == f"sub-{sub}_task-{new_task}_"
"desc-confounds_regressors.tsv"
)
# assert BOLD_mask
assert out["BOLD_mask"]["path"].exists()
assert out["BOLD_confounds"]["path"].exists()
assert out["BOLD_confounds"]["path"].is_file()
# asserts type "T1w"
assert "T1w" in out
# assert BOLD_mask
assert out["BOLD_mask"]["path"].exists()
assert (
out["T1w"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_"
"desc-preproc_T1w.nii.gz"
)
# asserts type "T1w"
assert "T1w" in out
assert out["T1w"]["path"].exists()
assert out["T1w"]["path"].is_file()
assert (
out["T1w"]["path"].name == f"sub-{sub}_space-MNI152NLin2009cAsym_"
"desc-preproc_T1w.nii.gz"
)
# asserts T1w_mask
assert out["T1w_mask"]["path"].exists()
assert out["T1w"]["path"].exists()
assert out["T1w"]["path"].is_file()
# asserts type "probseg_CSF"
assert "probseg_CSF" in out
# asserts T1w_mask
assert out["T1w_mask"]["path"].exists()
assert (
out["probseg_CSF"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"CSF_probseg.nii.gz"
)
# asserts type "probseg_CSF"
assert "probseg_CSF" in out
assert out["probseg_CSF"]["path"].exists()
assert out["probseg_CSF"]["path"].is_file()
assert (
out["probseg_CSF"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"CSF_probseg.nii.gz"
)
# asserts type "probseg_GM"
assert "probseg_GM" in out
assert out["probseg_CSF"]["path"].exists()
assert out["probseg_CSF"]["path"].is_file()
assert (
out["probseg_GM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"GM_probseg.nii.gz"
)
# asserts type "probseg_GM"
assert "probseg_GM" in out
assert out["probseg_GM"]["path"].exists()
assert out["probseg_GM"]["path"].is_file()
assert (
out["probseg_GM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"GM_probseg.nii.gz"
)
# asserts type "probseg_WM"
assert "probseg_WM" in out
assert out["probseg_GM"]["path"].exists()
assert out["probseg_GM"]["path"].is_file()
assert (
out["probseg_WM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"WM_probseg.nii.gz"
)
# asserts type "probseg_WM"
assert "probseg_WM" in out
assert out["probseg_WM"]["path"].exists()
assert out["probseg_WM"]["path"].is_file()
assert (
out["probseg_WM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"WM_probseg.nii.gz"
)
# asserts type "DWI"
assert "DWI" in out
assert out["probseg_WM"]["path"].exists()
assert out["probseg_WM"]["path"].is_file()
assert (
out["DWI"]["path"].name == f"sub-{sub}_desc-preproc_dwi.nii.gz"
)
# asserts type "DWI"
assert "DWI" in out
assert out["DWI"]["path"].exists()
assert out["DWI"]["path"].is_file()
assert out["DWI"]["path"].name == f"sub-{sub}_desc-preproc_dwi.nii.gz"
# asserts meta
assert "meta" in out["BOLD"]
meta = out["BOLD"]["meta"]
assert "element" in meta
assert "subject" in meta["element"]
assert sub == meta["element"]["subject"]
assert out["DWI"]["path"].exists()
assert out["DWI"]["path"].is_file()
# asserts meta
assert "meta" in out["BOLD"]
meta = out["BOLD"]["meta"]
assert "element" in meta
assert "subject" in meta["element"]
assert sub == meta["element"]["subject"]
@pytest.mark.parametrize(
"types",
[
"BOLD",
"BOLD_confounds",
"T1w",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
"DWI",
["BOLD", "BOLD_confounds"],
["T1w", "probseg_CSF"],
["probseg_GM", "probseg_WM"],
["DWI", "BOLD"],
],
)
def test_DataladAOMICPIOP1_partial_data_access(
types: Union[str, List[str]],
) -> None:
"""Test DataladAOMICPIOP1 DataGrabber partial data access.
Parameters
----------
types : str or list of str
The parametrized types.
"""
dg = DataladAOMICPIOP1(types=types)
# Set URI to Gin
dg.uri = URI
with dg:
# Get all elements
all_elements = dg.get_elements()
# Get test element
test_element = all_elements[0]
# Get test element data
out = dg[test_element]
# Assert data type
if isinstance(types, list):
for type_ in types:
assert type_ in out
else:
assert types in out
def test_DataladAOMICPIOP1_incorrect_data_type() -> None:
"""Test DataladAOMICPIOP1 DataGrabber incorrect data type."""
with pytest.raises(
ValueError, match="`patterns` must contain all `types`"
):
_ = DataladAOMICPIOP1(types="Ceres")
def test_DataladAOMICPIOP1_invalid_tasks():
"""Test whether invalid task fails."""
"""Test DataladAOMICIDPIOP1 DataGrabber invalid tasks."""
with pytest.raises(
ValueError,
match=(

239
junifer/datagrabber/aomic/tests/test_piop2.py
View file

@ -4,136 +4,195 @@
# Vera Komeyer <v.komeyer@fz-juelich.de>
# Xuan Li <xu.li@fz-juelich.de>
# Leonard Sasse <l.sasse@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from typing import List, Optional, Union
import pytest
from junifer.datagrabber import DataladAOMICPIOP2
from junifer.utils import configure_logging
def test_DataladAOMICPIOP2() -> None:
"""Test DataladAOMICPIOP2 DataGrabber."""
configure_logging(level="DEBUG")
URI = "https://gin.g-node.org/juaml/datalad-example-aomicpiop2"
uri_PIOP2 = "https://gin.g-node.org/juaml/datalad-example-aomicpiop2"
task_params = [None, "restingstate"]
for task_param in task_params:
dg = DataladAOMICPIOP2(tasks=task_param)
@pytest.mark.parametrize(
"tasks",
[None, "restingstate"],
)
def test_DataladAOMICPIOP2(tasks: Optional[str]) -> None:
"""Test DataladAOMICPIOP2 DataGrabber.
# change uri here to use fake data instead of real dataset
dg.uri = uri_PIOP2
Parameters
----------
tasks : str or None
The parametrized task values.
with dg:
all_elements = dg.get_elements()
"""
dg = DataladAOMICPIOP2(tasks=tasks)
# Set URI to Gin
dg.uri = URI
if task_param == "restingstate":
for el in all_elements:
assert el[1] == "restingstate"
with dg:
all_elements = dg.get_elements()
test_element = all_elements[0]
sub, task = test_element
out = dg[test_element]
if tasks == "restingstate":
for el in all_elements:
assert el[1] == "restingstate"
# asserts type "BOLD"
assert "BOLD" in out
test_element = all_elements[0]
sub, task = test_element
out = dg[test_element]
new_task = f"{task}_acq-seq"
assert (
out["BOLD"]["path"].name == f"sub-{sub}_task-{new_task}_"
"space-MNI152NLin2009cAsym_desc-preproc_bold.nii.gz"
)
# asserts type "BOLD"
assert "BOLD" in out
assert out["BOLD"]["path"].exists()
assert out["BOLD"]["path"].is_file()
new_task = f"{task}_acq-seq"
assert (
out["BOLD"]["path"].name == f"sub-{sub}_task-{new_task}_"
"space-MNI152NLin2009cAsym_desc-preproc_bold.nii.gz"
)
# asserts type "BOLD_confounds"
assert "BOLD_confounds" in out
assert out["BOLD"]["path"].exists()
assert out["BOLD"]["path"].is_file()
assert (
out["BOLD_confounds"]["path"].name
== f"sub-{sub}_task-{new_task}_"
"desc-confounds_regressors.tsv"
)
# asserts type "BOLD_confounds"
assert "BOLD_confounds" in out
assert out["BOLD_confounds"]["path"].exists()
assert out["BOLD_confounds"]["path"].is_file()
assert (
out["BOLD_confounds"]["path"].name == f"sub-{sub}_task-{new_task}_"
"desc-confounds_regressors.tsv"
)
# assert BOLD_mask
assert out["BOLD_mask"]["path"].exists()
assert out["BOLD_confounds"]["path"].exists()
assert out["BOLD_confounds"]["path"].is_file()
# asserts type "T1w"
assert "T1w" in out
# assert BOLD_mask
assert out["BOLD_mask"]["path"].exists()
assert (
out["T1w"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_"
"desc-preproc_T1w.nii.gz"
)
# asserts type "T1w"
assert "T1w" in out
assert out["T1w"]["path"].exists()
assert out["T1w"]["path"].is_file()
assert (
out["T1w"]["path"].name == f"sub-{sub}_space-MNI152NLin2009cAsym_"
"desc-preproc_T1w.nii.gz"
)
# asserts T1w_mask
assert out["T1w_mask"]["path"].exists()
assert out["T1w"]["path"].exists()
assert out["T1w"]["path"].is_file()
# asserts type "probseg_CSF"
assert "probseg_CSF" in out
# asserts T1w_mask
assert out["T1w_mask"]["path"].exists()
assert (
out["probseg_CSF"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"CSF_probseg.nii.gz"
)
# asserts type "probseg_CSF"
assert "probseg_CSF" in out
assert out["probseg_CSF"]["path"].exists()
assert out["probseg_CSF"]["path"].is_file()
assert (
out["probseg_CSF"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"CSF_probseg.nii.gz"
)
# asserts type "probseg_GM"
assert "probseg_GM" in out
assert out["probseg_CSF"]["path"].exists()
assert out["probseg_CSF"]["path"].is_file()
assert (
out["probseg_GM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"GM_probseg.nii.gz"
)
# asserts type "probseg_GM"
assert "probseg_GM" in out
assert out["probseg_GM"]["path"].exists()
assert out["probseg_GM"]["path"].is_file()
assert (
out["probseg_GM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"GM_probseg.nii.gz"
)
# asserts type "probseg_WM"
assert "probseg_WM" in out
assert out["probseg_GM"]["path"].exists()
assert out["probseg_GM"]["path"].is_file()
assert (
out["probseg_WM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"WM_probseg.nii.gz"
)
# asserts type "probseg_WM"
assert "probseg_WM" in out
assert out["probseg_WM"]["path"].exists()
assert out["probseg_WM"]["path"].is_file()
assert (
out["probseg_WM"]["path"].name
== f"sub-{sub}_space-MNI152NLin2009cAsym_label-"
"WM_probseg.nii.gz"
)
# asserts type "DWI"
assert "DWI" in out
assert out["probseg_WM"]["path"].exists()
assert out["probseg_WM"]["path"].is_file()
assert (
out["DWI"]["path"].name == f"sub-{sub}_desc-preproc_dwi.nii.gz"
)
# asserts type "DWI"
assert "DWI" in out
assert out["DWI"]["path"].exists()
assert out["DWI"]["path"].is_file()
assert out["DWI"]["path"].name == f"sub-{sub}_desc-preproc_dwi.nii.gz"
# asserts meta
assert "meta" in out["BOLD"]
meta = out["BOLD"]["meta"]
assert "element" in meta
assert "subject" in meta["element"]
assert sub == meta["element"]["subject"]
assert out["DWI"]["path"].exists()
assert out["DWI"]["path"].is_file()
# asserts meta
assert "meta" in out["BOLD"]
meta = out["BOLD"]["meta"]
assert "element" in meta
assert "subject" in meta["element"]
assert sub == meta["element"]["subject"]
@pytest.mark.parametrize(
"types",
[
"BOLD",
"BOLD_confounds",
"T1w",
"probseg_CSF",
"probseg_GM",
"probseg_WM",
"DWI",
["BOLD", "BOLD_confounds"],
["T1w", "probseg_CSF"],
["probseg_GM", "probseg_WM"],
["DWI", "BOLD"],
],
)
def test_DataladAOMICPIOP2_partial_data_access(
types: Union[str, List[str]],
) -> None:
"""Test DataladAOMICPIOP2 DataGrabber partial data access.
Parameters
----------
types : str or list of str
The parametrized types.
"""
dg = DataladAOMICPIOP2(types=types)
# Set URI to Gin
dg.uri = URI
with dg:
# Get all elements
all_elements = dg.get_elements()
# Get test element
test_element = all_elements[0]
# Get test element data
out = dg[test_element]
# Assert data type
if isinstance(types, list):
for type_ in types:
assert type_ in out
else:
assert types in out
def test_DataladAOMICPIOP2_incorrect_data_type() -> None:
"""Test DataladAOMICPIOP2 DataGrabber incorrect data type."""
with pytest.raises(
ValueError, match="`patterns` must contain all `types`"
):
_ = DataladAOMICPIOP2(types="Vesta")
def test_DataladAOMICPIOP2_invalid_tasks():
"""Test whether invalid task fails."""
"""Test DataladAOMICIDPIOP2 DataGrabber invalid tasks."""
with pytest.raises(
ValueError,
match=(

4
junifer/datagrabber/tests/test_datagrabber_utils.py
View file

@ -61,7 +61,9 @@ def test_validate_patterns() -> None:
"T1w": "{subject}/anat/{subject}_T1w.nii.gz",
}
with pytest.raises(ValueError, match="same length"):
with pytest.raises(
ValueError, match="Length of `types` more than that of `patterns`."
):
validate_patterns(types, wrongpatterns) # type: ignore
wrongpatterns = {

6
junifer/datagrabber/tests/test_pattern.py
View file

@ -38,7 +38,9 @@ def test_PatternDataGrabber_errors(tmp_path: Path) -> None:
replacements="subject", # type: ignore
)
with pytest.raises(ValueError, match=r"must have the same length"):
with pytest.raises(
ValueError, match=r"`patterns` must contain all `types`"
):
PatternDataGrabber(
datadir="/tmp",
types=["func", "anat"],
@ -55,7 +57,7 @@ def test_PatternDataGrabber_errors(tmp_path: Path) -> None:
)
with pytest.raises(
ValueError, match=r"`patterns` must have the same length"
ValueError, match=r"Length of `types` more than that of `patterns`"
):
PatternDataGrabber(
datadir="/tmp",

8
junifer/datagrabber/utils.py
View file

@ -77,17 +77,17 @@ def validate_patterns(types: List[str], patterns: Dict[str, str]) -> None:
if not isinstance(patterns, dict):
raise_error(msg="`patterns` must be a dict.", klass=TypeError)
# Unequal length of objects
if len(types) != len(patterns):
if len(types) > len(patterns):
raise_error(
msg="`types` and `patterns` must have the same length.",
msg="Length of `types` more than that of `patterns`.",
klass=ValueError,
)
# Missing type in patterns
if any(x not in patterns for x in types):
raise_error(
msg="`patterns` must contain all `types`", klass=ValueError
)
# Wildcard check in patterns
if any("}*" in pattern for pattern in patterns.values()):
raise_error(
msg="`patterns` must not contain `*` following a replacement",