diff --git a/.coveragerc b/.coveragerc index 2cd7255d0..6cdbc7a3c 100644 --- a/.coveragerc +++ b/.coveragerc @@ -6,7 +6,7 @@ omit = */setup.py */tests/* junifer/configs/juseless.py - junifer/testing/* + junifer/testing.py [report] exclude_lines = diff --git a/.gitignore b/.gitignore index 94ac529c9..09d12056a 100644 --- a/.gitignore +++ b/.gitignore @@ -132,6 +132,4 @@ cython_debug/ # OS Stuff .DS_store -junifer/_version.py -scratch/ -junifer_jobs/ \ No newline at end of file +junifer/_version.py \ No newline at end of file diff --git a/Makefile b/Makefile index d1b4e3a6c..f34d0536d 100644 --- a/Makefile +++ b/Makefile @@ -12,4 +12,4 @@ spellcheck: codespell junifer/ docs/ examples/ test: - pytest -vv --cov=junifer --cov-report html --cov-report term \ No newline at end of file + pytest -v \ No newline at end of file diff --git a/dev-requirements.txt b/dev-requirements.txt index 2e4e347ff..aad120b7c 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -1,3 +1,2 @@ flake8 -pytest -click \ No newline at end of file +pytest \ No newline at end of file diff --git a/examples/norun_hcpfc_pearson.py b/examples/norun_hcpfc_pearson.py index fe2632e92..d2d17640d 100644 --- a/examples/norun_hcpfc_pearson.py +++ b/examples/norun_hcpfc_pearson.py @@ -6,14 +6,7 @@ License: BSD 3 clause """ -from junifer.api import run - -datagrabber = { - 'kind': 'HCPOpenAccess', - 'modality': 'fMRI', - 'preprocessed': 'ICA+FIX', - 'space': 'volumetric', -} +from junifer.api import run_pipeline custom_confound_strategy = { 'filter': 'butterworth', @@ -55,15 +48,18 @@ markers = [ 'confound_strategy': custom_confound_strategy} ] -storage = { - 'kind': 'SQLiteFeatureStorage', - 'uri': '/data/project/juniferexample' +dg_params = { + 'modality': 'fMRI', + 'preprocessed': 'ICA+FIX', + 'space': 'volumetric', } -run( +run_pipeline( workdir='/tmp', - datagrabber=datagrabber, - elements=[('100408', 'REST1', "LR")], + datagrabber='HCPOpenAccess', + datagrabber_params=dg_params, + element=('100408', 'REST1', "LR"), markers=markers, - storage=storage, + storage='SQLDataFrameStorage', + storage_params={'outpath': '/data/project/juniferexample'}, ) diff --git a/examples/run_run_gmd_mean.py b/examples/run_run_gmd_mean.py index 6259530b7..db461246d 100644 --- a/examples/run_run_gmd_mean.py +++ b/examples/run_run_gmd_mean.py @@ -9,11 +9,10 @@ License: BSD 3 clause import tempfile from junifer.api import run -import junifer.testing.registry # noqa: F401 +from junifer.testing import register_testing -datagrabber = { - 'kind': 'OasisVBMTestingDatagrabber', -} + +register_testing() markers = [ {'name': 'Schaefer1000x7_TrimMean80', @@ -31,16 +30,12 @@ markers = [ 'method': 'std'} ] -storage = { - 'kind': 'SQLiteFeatureStorage', -} - with tempfile.TemporaryDirectory() as tmpdir: uri = f'{tmpdir}/test.db' - storage['uri'] = uri run( workdir='/tmp', - datagrabber=datagrabber, + datagrabber='OasisVBMTestingDatagrabber', markers=markers, - storage=storage, + storage='SQLiteFeatureStorage', + storage_params={'uri': uri}, ) diff --git a/examples/yamls/gmd_mean.yaml b/examples/yamls/gmd_mean.yaml deleted file mode 100644 index ce50d5d5c..000000000 --- a/examples/yamls/gmd_mean.yaml +++ /dev/null @@ -1,25 +0,0 @@ -with: junifer.testing.registry -workdir: /tmp - -datagrabber: - kind: OasisVBMTestingDatagrabber -elements: -markers: - - name: Schaefer1000x7_TrimMean80 - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: trim_mean - method_params: - proportiontocut: 0.2 - - name: Schaefer1000x7_Mean - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: mean - - name: Schaefer1000x7_Std - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: std -storage: - kind: SQLiteFeatureStorage - uri: /Users/fraimondo/dev/tbox/junifer/scratch/db/test.db - diff --git a/examples/yamls/gmd_mean_htcondor.yaml b/examples/yamls/gmd_mean_htcondor.yaml deleted file mode 100644 index fde41b733..000000000 --- a/examples/yamls/gmd_mean_htcondor.yaml +++ /dev/null @@ -1,20 +0,0 @@ -with: junifer.testing.registry -workdir: /tmp - -datagrabber: - kind: OasisVBMTestingDatagrabber -markers: - - name: Schaefer1000x7_Mean - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: mean -storage: - kind: SQLiteFeatureStorage - uri: /data/group/appliedml/fraimondo/junifer_test/test.db -queue: - jobname: TestHTCondorQueue - kind: HTCondor - env: - kind: conda - name: junifer - mem: 8G \ No newline at end of file diff --git a/examples/yamls/ukb_gmd_mean.yaml b/examples/yamls/ukb_gmd_mean.yaml deleted file mode 100644 index 7d2303d7e..000000000 --- a/examples/yamls/ukb_gmd_mean.yaml +++ /dev/null @@ -1,25 +0,0 @@ -with: junifer.configs.juseless -workdir: /tmp - -datagrabber: - kind: JuselessUKBVBM -elements: -markers: - - name: Schaefer1000x7_TrimMean80 - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: trim_mean - method_params: - proportiontocut: 0.2 - - name: Schaefer1000x7_Mean - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: mean - - name: Schaefer1000x7_Std - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: std -storage: - kind: SQLiteFeatureStorage - uri: /data/project/ukb_motor/junifer_test/test.db - diff --git a/junifer/api/__init__.py b/junifer/api/__init__.py index ecd3d2c69..4bc42cc57 100644 --- a/junifer/api/__init__.py +++ b/junifer/api/__init__.py @@ -1,2 +1 @@ -from .functions import run, collect -from .import cli \ No newline at end of file +from . run import run \ No newline at end of file diff --git a/junifer/api/cli.py b/junifer/api/cli.py deleted file mode 100644 index 6991ae713..000000000 --- a/junifer/api/cli.py +++ /dev/null @@ -1,89 +0,0 @@ -import click - -from .parser import parse_yaml -from .functions import run as api_run -from .functions import collect as api_collect -from .functions import queue as api_queue -from ..utils.logging import configure_logging, logger, warn - - -def _parse_elements(element, config): - logger.debug(f'Parsing elements: {element}') - if len(element) == 0: - return None - # TODO: If len == 1, check if its a file, then parse elements from file - elements = [x.split(',') if ',' in x else x for x in element] - logger.debug(f'Parsed elements: {elements}') - if elements is not None and 'elements' in config: - warn('One or more elements have been specified in both the command ' - 'line and in the config file. The command line has precedence ' - 'over the configuration file. That is, the elements specified ' - 'in the command line will be used. The elements specified in ' - 'the configuration file will be ignored. To remove this warning, ' - 'please remove the "elements" item from the configuration file.') - elif elements is None: - elements = config.get('elements', None) - return elements - - -@click.group() -def cli(): - pass - - -@cli.command() -@click.argument( - 'filepath', - type=click.Path(exists=True, readable=True, dir_okay=False)) -@click.option('-v', '--verbose', - type=click.Choice(['warning', 'info', 'debug'], - case_sensitive=False), - default='info') -@click.option('--element', type=str, multiple=True) -def run(filepath, element, verbose): - configure_logging(level=verbose.upper()) - config = parse_yaml(filepath) - workdir = config['workdir'] - datagrabber = config['datagrabber'] - markers = config['markers'] - storage = config['storage'] - elements = _parse_elements(element, config) - api_run( - workdir=workdir, datagrabber=datagrabber, markers=markers, - storage=storage, elements=elements) - - -@cli.command() -@click.argument( - 'filepath', - type=click.Path(exists=True, readable=True, dir_okay=False)) -@click.option('-v', '--verbose', - type=click.Choice(['warning', 'info', 'debug'], - case_sensitive=False), - default='info') -def collect(filepath, verbose): - configure_logging(level=verbose.upper()) - config = parse_yaml(filepath) - storage = config['storage'] - api_collect(storage) - - -@cli.command() -@click.argument( - 'filepath', - type=click.Path(exists=True, readable=True, dir_okay=False)) -@click.option('-v', '--verbose', - type=click.Choice(['warning', 'info', 'debug'], - case_sensitive=False), - default='info') -@click.option('--overwrite', is_flag=True) -@click.option('--submit', is_flag=True) -@click.option('--element', type=str, multiple=True) -def queue(filepath, element, overwrite, submit, verbose): - configure_logging(level=verbose.upper()) - config = parse_yaml(filepath) - elements = _parse_elements(element, config) - queue_config = config.pop('queue') - kind = queue_config.pop('kind') - api_queue(config, kind=kind, overwrite=overwrite, submit=submit, - elements=elements, **queue_config) diff --git a/junifer/api/functions.py b/junifer/api/functions.py deleted file mode 100644 index e3fb8b80d..000000000 --- a/junifer/api/functions.py +++ /dev/null @@ -1,269 +0,0 @@ -# Authors: Federico Raimondo -# Leonard Sasse -# License: AGPL - -from pathlib import Path -import shutil -import yaml -import subprocess - -from .registry import build -from ..utils import logger, raise_error -from ..utils.fs import make_executable -from ..datagrabber.base import BaseDataGrabber -from ..markers.base import BaseMarker -from ..storage.base import BaseFeatureStorage -from ..markers.collection import MarkerCollection - - -def _get_datagrabber(datagrabber_config): - datagrabber_params = datagrabber_config.copy() - datagrabber_kind = datagrabber_params.pop('kind') - datagrabber = build( - 'datagrabber', datagrabber_kind, BaseDataGrabber, - init_params=datagrabber_params) - return datagrabber - - -def run( - workdir, datagrabber, markers, storage, elements=None): - """Run the pipeline on the selected element - - Parameters - ---------- - workdir : str or path-like object - Directory where the pipeline will be executed - datagrabber : dict - Datagrabber to use. Must have a key 'kind' with the kind of - datagrabber to use. All other keys are passed to the datagrabber - init function. - elements : str, tuple or list[str or tuple] - Element(s) to process. Will be used to index the datagrabber. - markers : list of dict - List of markers to extract. Each marker is a dict with at least two - keys: 'name' and 'kind'. The 'name' key is used to name the output - marker. The 'kind' key is used to specify the kind of marker to - extract. The rest of the keys are used to pass parameters to the - marker calculation. - storage : dict - Storage to use. Must have a key 'kind' with the kind of - storage to use. All other keys are passed to the storage - init function. - """ - storage_params = storage.copy() - storage_kind = storage_params.pop('kind') - - if isinstance(workdir, str): - workdir = Path(workdir) - - datagrabber = _get_datagrabber(datagrabber) - # Copy to avoid changing the original dict - _markers = [x.copy() for x in markers] - built_markers = [] - for t_marker in _markers: - kind = t_marker.pop('kind') - t_m = build('marker', kind, BaseMarker, init_params=t_marker) - built_markers.append(t_m) - - storage = build( - 'storage', storage_kind, BaseFeatureStorage, - init_params=storage_params) - - mc = MarkerCollection(built_markers, storage=storage) - - with datagrabber: - if elements is not None: - for t_element in elements: - mc.fit(datagrabber[t_element]) - else: - for t_element in datagrabber: - mc.fit(datagrabber[t_element]) - - -def collect(storage): - storage_params = storage.copy() - storage_kind = storage_params.pop('kind') - logger.info(f'Collecting data using {storage_kind}') - logger.debug(f'\tStorage params: {storage_params}') - storage = build( - 'storage', storage_kind, BaseFeatureStorage, - init_params=storage_params) - logger.debug('Running storage.collect()') - storage.collect() - logger.info('Collect done') - - -def queue(config, kind, jobname='junifer_job', overwrite=False, elements=None, - **kwargs): - """Queue a job to be executed later - - Parameters - ---------- - kind : str - The kind of job to queue. - **kwargs : dict - The parameters to pass to the job. - """ - # Create a folder within the CWD to store the job files / config - cwd = Path.cwd() - jobdir = cwd / 'junifer_jobs' / jobname - logger.info(f'Creating job in {jobdir.as_posix()}') - if jobdir.exists(): - if overwrite is not True: - raise_error(f'Job folder for {jobname} already exists. ' - 'This error is raise to prevent overwriting files ' - 'of jobs that might be scheduled but yet not ' - 'executed. Either delete the directory ' - f'{jobdir.as_posix()} or set overwrite to True.') - else: - logger.info( - f'Deleting previous job directory {jobdir.as_posix()}') - shutil.rmtree(jobdir) - jobdir.mkdir(exist_ok=True, parents=True) - - yaml_config = jobdir / 'config.yaml' - logger.info(f'Writing YAML config to {yaml_config}') - with open(yaml_config, 'w') as f: - f.write(yaml.dump(config)) - - # Get list of elements - if elements is None: - if 'elements' in config: - elements = config['elements'] - else: - # If no elements are specified, use all elements from the - # datagrabber - datagrabber = _get_datagrabber(config['datagrabber']) - with datagrabber as dg: - elements = dg.get_elements() - if kind == 'HTCondor': - _queue_condor(jobname, jobdir, yaml_config, elements, **kwargs) - elif kind == 'SLURM': - _queue_slurm(jobname, jobdir, yaml_config, elements, **kwargs) - else: - raise ValueError(f'Unknown queue kind: {kind}') - - logger.info('Queue done') - - -def _queue_condor(jobname, jobdir, yaml_config, elements, env, mem='8G', - cpus=1, disk='1G', extra_preamble='', verbose='info', - collect=True, submit=False): - logger.debug('Creating HTCondor job') - run_junifer_args = (f'run {yaml_config.as_posix()} ' - f'--verbose {verbose} --element $(element)') - collect_junifer_args = \ - f'collect {yaml_config.as_posix()} --verbose {verbose} ' - - # Set up the env_name, executable and arguments according to the - # environment type - if env is None: - env = {'kind': 'local'} - if env['kind'] == 'conda': - env_name = env['name'] - executable = 'run_conda.sh' - arguments = f'{env_name} junifer' - # TODO: Copy run_conda.sh to jobdir - exec_path = jobdir / executable - shutil.copy(Path(__file__).parent / 'res' / executable, exec_path) - make_executable(exec_path) - elif env['kind'] == 'venv': - env_name = env['name'] - executable = 'run_venv.sh' - arguments = f'{env_name} junifer' - # TODO: Copy run_venv.sh to jobdir - elif env['kind'] == 'local': - executable = 'junifer' - arguments = '' - else: - raise ValueError(f'Unknown env kind: {env["kind"]}') - log_dir = jobdir / 'logs' - log_dir.mkdir(exist_ok=True, parents=True) - run_preamble = f""" - # The environment - universe = vanilla - getenv = True - - # Resources - request_cpus = {cpus} - request_memory = {mem} - request_disk = {disk} - - # Executable - initial_dir = {jobdir.as_posix()} - executable = $(initial_dir)/{executable} - transfer_executable = False - - arguments = {arguments} {run_junifer_args} - - {extra_preamble} - - # Logs - log = {log_dir.as_posix()}/junifer_run_$(element).log - output = {log_dir.as_posix()}/junifer_run_$(element).out - error = {log_dir.as_posix()}/junifer_run_$(element).err - """ - - submit_run_fname = jobdir / f'run_{jobname}.submit' - submit_collect_fname = jobdir / f'collect_{jobname}.submit' - dag_fname = jobdir / f'{jobname}.dag' - - # Write to run submit files - with open(submit_run_fname, 'w') as submit_file: - submit_file.write(run_preamble) - submit_file.write('queue\n') - - collect_preamble = f""" - # The environment - universe = vanilla - getenv = True - - # Resources - request_cpus = {cpus} - request_memory = {mem} - request_disk = {disk} - - # Executable - initial_dir = {jobdir.as_posix()} - executable = $(initial_dir)/{executable} - transfer_executable = False - - arguments = {arguments} {collect_junifer_args} - - {extra_preamble} - - # Logs - log = {log_dir.as_posix()}/junifer_collect.log - output = {log_dir.as_posix()}/junifer_collect.out - error = {log_dir.as_posix()}/junifer_collect.err - """ - - # Now create the collect submit file - with open(submit_collect_fname, 'w') as submit_file: - submit_file.write(collect_preamble) # Eval preamble here - submit_file.write('queue\n') - - with open(dag_fname, 'w') as dag_file: - # Get all subject and session names from file list - for i_job, t_elem in enumerate(elements): - dag_file.write(f'JOB run{i_job} {submit_run_fname}\n') - dag_file.write(f'VARS run{i_job} element="{t_elem}"\n\n') - if collect is True: - dag_file.write(f'JOB collect {submit_collect_fname}\n') - dag_file.write('PARENT ') - for i_job, t_elem in enumerate(elements): - dag_file.write(f'run{i_job} ') - dag_file.write(f'CHILD collect\n\n') - - if submit is True: - logger.info('Submitting HTCondor job') - subprocess.run(['condor_submit_dag', dag_fname]) - logger.info('HTCondor job submitted') - else: - cmd = f'condor_submit_dag {dag_fname.as_posix()}' - logger.info('HTCondor job files created, to submit the job, ' - f'run "{cmd}"') - - -def _queue_slurm(jobname, jobdir, yaml_config, elements): - pass diff --git a/junifer/api/parser.py b/junifer/api/parser.py deleted file mode 100644 index a5883ad0a..000000000 --- a/junifer/api/parser.py +++ /dev/null @@ -1,25 +0,0 @@ -import yaml -import importlib -from pathlib import Path - -from ..utils.logging import raise_error, logger - - -def parse_yaml(filepath): - if not isinstance(filepath, Path): - filepath = Path(filepath) - logger.info(f'Parsing yaml file: {filepath.as_posix()}') - if not filepath.exists(): - raise_error(f'File does not exist: {filepath.as_posix()}') - with open(filepath, 'r') as f: - contents = yaml.safe_load(f) - - if 'with' in contents: - to_load = contents['with'] - if not isinstance(to_load, list): - to_load = [to_load] - for t_module in to_load: - logger.info(f'Importing module {t_module}') - importlib.import_module(t_module) - - return contents diff --git a/junifer/api/registry.py b/junifer/api/registry.py index 922433f7c..3de096073 100644 --- a/junifer/api/registry.py +++ b/junifer/api/registry.py @@ -93,12 +93,14 @@ def build(step, name, baseclass, init_params=None): If the name is not a string or the object is not an instance of the baseclass parameter. """ + if not isinstance(name, str): + raise_error(f'Invalid name: {name}', ValueError) klass = get(step, name) if init_params is None: init_params = {} object = klass(**init_params) if not isinstance(object, baseclass): raise_error( - f'Invalid {step} ({object.__class__.__name__}). ' - f'Must inherit from {baseclass.__name__}', ValueError) + f'Invalid {step} ({object.__class__.name}). ' + f'Must inherit from {baseclass.name}', ValueError) return object diff --git a/junifer/api/res/run_conda.sh b/junifer/api/res/run_conda.sh deleted file mode 100755 index f07c86c6f..000000000 --- a/junifer/api/res/run_conda.sh +++ /dev/null @@ -1,17 +0,0 @@ -#!/bin/bash - -if [ $# -lt 2 ]; then - echo "This script is meant to run a command within a python environment" - echo "It needs at least 2 parameters." - echo "The first one must be the environment name." - echo "The rest will be the command" - exit 255 -fi - -eval "$(conda shell.bash hook)" -env_name=$1 -echo "Activating ${env_name}" -conda activate "$1" -shift 1 -echo "Running ${*} in virtual environment" -"$@" \ No newline at end of file diff --git a/junifer/api/run.py b/junifer/api/run.py new file mode 100644 index 000000000..f33737a33 --- /dev/null +++ b/junifer/api/run.py @@ -0,0 +1,69 @@ +# Authors: Federico Raimondo +# Leonard Sasse +# License: AGPL + +from pathlib import Path + +from .registry import build +from ..datagrabber.base import BaseDataGrabber +from ..markers.base import BaseMarker +from ..storage.base import BaseFeatureStorage +from ..markers.collection import MarkerCollection + + +def run( + workdir, datagrabber, markers, storage, source_params=None, + storage_params=None, elements=None): + """Run the pipeline on the selected element + + Parameters + ---------- + workdir : str or path-like object + Directory where the pipeline will be executed + datagrabber : str + Name of the datagrabber to use + elements : str, tuple or list[str or tuple] + Element(s) to process. Will be used to index the datagrabber. + markers : list of dict + List of markers to extract. Each marker is a dict with at least two + keys: 'name' and 'kind'. The 'name' key is used to name the output + marker. The 'kind' key is used to specify the kind of marker to + extract. The rest of the keys are used to pass parameters to the + marker calculation. + storage: str + Name of the storage to use. + source_params : dict + Parameters to pass to the datagrabber. + storage_params: dict + Parameters to pass to the storage. + """ + if source_params is None: + source_params = {} + + if storage_params is None: + storage_params = {} + + if isinstance(workdir, str): + workdir = Path(workdir) + + datagrabber = build( + 'datagrabber', datagrabber, BaseDataGrabber, init_params=source_params) + + built_markers = [] + for t_marker in markers: + kind = t_marker.pop('kind') + t_m = build('marker', kind, BaseMarker, init_params=t_marker) + built_markers.append(t_m) + + storage = build( + 'storage', storage, BaseFeatureStorage, init_params=storage_params) + + mc = MarkerCollection(built_markers, storage=storage) + + with datagrabber: + if elements is not None: + for t_element in elements: + mc.fit(datagrabber[t_element]) + else: + for t_element in datagrabber: + mc.fit(datagrabber[t_element]) diff --git a/junifer/api/tests/data/gmd_mean.yaml b/junifer/api/tests/data/gmd_mean.yaml deleted file mode 100644 index c492cc559..000000000 --- a/junifer/api/tests/data/gmd_mean.yaml +++ /dev/null @@ -1,15 +0,0 @@ -with: junifer.testing.registry -workdir: /tmp - -datagrabber: - kind: OasisVBMTestingDatagrabber -elements: [1, 2] -markers: - - name: Schaefer1000x7_Mean - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: mean -storage: - kind: SQLiteFeatureStorage - uri: /Users/fraimondo/dev/tbox/junifer/scratch/db/test.db - diff --git a/junifer/api/tests/data/gmd_mean_htcondor.yaml b/junifer/api/tests/data/gmd_mean_htcondor.yaml deleted file mode 100644 index fddf7aa9b..000000000 --- a/junifer/api/tests/data/gmd_mean_htcondor.yaml +++ /dev/null @@ -1,20 +0,0 @@ -with: junifer.testing.registry -workdir: /tmp - -datagrabber: - kind: OasisVBMTestingDatagrabber -markers: - - name: Schaefer1000x7_Mean - kind: ParcelAggregation - atlas: Schaefer1000x7 - method: mean -storage: - kind: SQLiteFeatureStorage - uri: /Users/fraimondo/dev/tbox/junifer/scratch/db/test.db -queue: - jobname: TestHTCondorQueue - kind: HTCondor - env: - kind: conda - name: junifer - mem: 8G \ No newline at end of file diff --git a/junifer/api/tests/test_cli.py b/junifer/api/tests/test_cli.py deleted file mode 100644 index c29b7262e..000000000 --- a/junifer/api/tests/test_cli.py +++ /dev/null @@ -1,60 +0,0 @@ -from pathlib import Path -import tempfile -import yaml -from junifer.api.cli import run, collect - -from click.testing import CliRunner - - -runner = CliRunner() - - -def _modify_path(tmpdir, in_file): - """Modify the path to use the temporary directory""" - if not isinstance(tmpdir, Path): - tmpdir = Path(tmpdir) - with open(in_file, 'r') as f: - contents = yaml.safe_load(f) - outfile = tmpdir / 'in.yaml' - outdir = tmpdir / 'out' - workdir = tmpdir / 'work' - contents['storage']['uri'] = outdir.as_posix() - contents['workdir'] = workdir.as_posix() - with open(outfile, 'w') as f: - yaml.dump(contents, f) - return outfile - - -def test_run_collect(): - """Test run and collect""" - infile = Path(__file__).parent / 'data' / 'gmd_mean.yaml' - with tempfile.TemporaryDirectory() as _tmpdir: - runfile = _modify_path(_tmpdir, infile) - run_args = [runfile.as_posix(), '--verbose', 'debug', '--element', - 'sub-01', '--element', 'sub-02', '--element', 'sub-03'] - response = runner.invoke(run, run_args) - assert response.exit_code == 0 - - # TODO: Check that there are 3 files in the output directory - - collect_args = [runfile.as_posix(), '--verbose', 'debug'] - response = runner.invoke(collect, collect_args) - assert response.exit_code == 0 - - # TODO: Check that there are 4 files in the output directory - # TODO: Check that the collected file has the correct number of rows - - run_args = [runfile.as_posix(), '--verbose', 'debug', '--element', - 'sub-01', '--element', 'sub-02', '--element', 'sub-04'] - response = runner.invoke(run, run_args) - assert response.exit_code == 0 - - # TODO: Check that there are 5 files in the output directory - # TODO: Check that the collected file has 3 rows (sub-04 is not there) - - collect_args = [runfile.as_posix(), '--verbose', 'debug'] - response = runner.invoke(collect, collect_args) - assert response.exit_code == 0 - - # TODO: Check that there are 5 files in the output directory - # TODO: Check that the collected file has 4 rows (sub-04 is there) diff --git a/junifer/api/tests/test_functions.py b/junifer/api/tests/test_functions.py deleted file mode 100644 index ae44675fc..000000000 --- a/junifer/api/tests/test_functions.py +++ /dev/null @@ -1,91 +0,0 @@ -import tempfile -from pathlib import Path - -from junifer.api.registry import build -from junifer.api.functions import run, collect -from junifer.datagrabber.base import BaseDataGrabber -import junifer.testing.registry # noqa: F401 - -datagrabber = { - 'kind': 'OasisVBMTestingDatagrabber', -} - -markers = [ - {'name': 'Schaefer1000x7_Mean', - 'kind': 'ParcelAggregation', - 'atlas': 'Schaefer1000x7', - 'method': 'mean'}, - {'name': 'Schaefer1000x7_Std', - 'kind': 'ParcelAggregation', - 'atlas': 'Schaefer1000x7', - 'method': 'std'} -] - -storage = { - 'kind': 'SQLiteFeatureStorage', -} - - -def test_run(): - """Test run function""" - with tempfile.TemporaryDirectory() as tmpdir: - tmp_path = Path(tmpdir) - workdir = tmp_path / 'workdir' - workdir.mkdir() - outdir = tmp_path / 'out' - outdir.mkdir() - uri = outdir / 'test.db' - storage['uri'] = uri # type: ignore - run( - workdir=workdir, - datagrabber=datagrabber, - markers=markers, - storage=storage, - elements=['sub-01'] - ) - - files = list(outdir.glob('*.db')) - assert len(files) == 1 - - run( - workdir=workdir.as_posix(), - datagrabber=datagrabber, - markers=markers, - storage=storage, - elements=['sub-01', 'sub-03'] - ) - - files = list(outdir.glob('*.db')) - assert len(files) == 2 - - -def test_collect(): - """Test run and collect functions""" - - with tempfile.TemporaryDirectory() as tmpdir: - tmp_path = Path(tmpdir) - workdir = tmp_path / 'workdir' - workdir.mkdir() - outdir = tmp_path / 'out' - outdir.mkdir() - uri = outdir / 'test.db' - storage['uri'] = uri # type: ignore - run( - workdir=workdir, - datagrabber=datagrabber, - markers=markers, - storage=storage, - ) - dg = build('datagrabber', datagrabber['kind'], BaseDataGrabber) - elements = dg.get_elements() - - # This should create 10 files - files = list(outdir.glob('*.db')) - assert len(files) == len(elements) - - # But the test.db file should not exist - assert not uri.exists() - collect(storage) - - # Now the file exists - assert uri.exists() diff --git a/junifer/api/tests/test_parser.py b/junifer/api/tests/test_parser.py deleted file mode 100644 index 392a90d7f..000000000 --- a/junifer/api/tests/test_parser.py +++ /dev/null @@ -1,42 +0,0 @@ -import sys -from pathlib import Path -import tempfile -import pytest - -from junifer.api.parser import parse_yaml - - -def test_parse_yaml(): - """Test parse yaml""" - with pytest.raises(ValueError, match='does not exist'): - parse_yaml('foo.yaml') - - with tempfile.TemporaryDirectory() as _tmpdir: - fname = Path(_tmpdir) / 'test.yaml' - - with open(fname, 'w') as f: - f.write('foo: bar\n') - contents = parse_yaml(fname) - assert 'foo' in contents - assert 'bar' == contents['foo'] - - with open(fname, 'w') as f: - f.write('foo: bar\n') - f.write('with: numpy\n') - - contents = parse_yaml(fname) - assert 'foo' in contents - assert 'bar' == contents['foo'] - assert 'with' in contents - assert 'numpy' in contents['with'] - - assert 'junifer.configs.wrong_config' not in sys.modules - - with open(fname, 'w') as f: - f.write('foo: bar\n') - f.write('with:\n') - f.write(' - numpy\n') - f.write(' - junifer.testing.wrong_config\n') - - with pytest.raises(ImportError, match='wrong_config'): - contents = parse_yaml(fname.as_posix()) diff --git a/junifer/api/tests/test_registry.py b/junifer/api/tests/test_registry.py deleted file mode 100644 index d8716f541..000000000 --- a/junifer/api/tests/test_registry.py +++ /dev/null @@ -1,56 +0,0 @@ -import pytest -from abc import ABC - -from junifer.api.registry import register, get_step_names, get, build - - -def test_register_error(): - """Test register error""" - with pytest.raises(ValueError, match='Invalid ste'): - register('foo', 'bar', 'baz') - - -def test_gets(): - """Test get""" - with pytest.raises(ValueError, match='Invalid ste'): - get_step_names('foo') - - datagrabbers = get_step_names('datagrabber') - assert 'bar' not in datagrabbers - register('datagrabber', 'bar', 'baz') - datagrabbers = get_step_names('datagrabber') - assert 'bar' in datagrabbers - - with pytest.raises(ValueError, match='Invalid ste'): - get('foo', 'bar') - - with pytest.raises(ValueError, match='Invalid name'): - get('datagrabber', 'foo') - - obj = get('datagrabber', 'bar') - assert obj == 'baz' - - -def test_build(): - """Test building objects from names""" - import numpy as np - - class SuperClass(ABC): - pass - - class ConcreteClass(SuperClass): - def __init__(self, value=1): - self.value = value - - register('datagrabber', 'concrete', ConcreteClass) - - obj = build('datagrabber', 'concrete', SuperClass) - assert isinstance(obj, ConcreteClass) - assert obj.value == 1 - - obj = build('datagrabber', 'concrete', SuperClass, {'value': 2}) - assert isinstance(obj, ConcreteClass) - assert obj.value == 2 - - with pytest.raises(ValueError, match='Must inherit'): - build('datagrabber', 'concrete', np.ndarray) diff --git a/junifer/configs/juseless.py b/junifer/configs/juseless.py index db88315b7..7bebc030b 100644 --- a/junifer/configs/juseless.py +++ b/junifer/configs/juseless.py @@ -1,10 +1,10 @@ -from ..datagrabber import PatternDataladDataGrabber +from ..datagrabber import DataladDataGrabber from ..api.decorators import register_datagrabber @register_datagrabber -class JuselessDataladUKBVBM(PatternDataladDataGrabber): - """Juseless UKB VBM DataGrabber class. +class JuselessUKBVBM(DataladDataGrabber): + """Juseless UKB VMG DataGrabber class. Implements a DataGrabber to access the UKB VBM data in Juseless. @@ -23,10 +23,44 @@ class JuselessDataladUKBVBM(PatternDataladDataGrabber): uri = 'ria+http://ukb.ds.inm7.de#~cat_m0wp1' rootdir = 'm0wp1' types = ['VBM_GM'] - replacements = ['subject', 'session'] - patterns = { - 'VBM_GM': 'm0wp1sub-{subject}_ses-{session}_T1w.nii.gz' - } super().__init__( - types=types, datadir=datadir, uri=uri, rootdir=rootdir, - replacements=replacements, patterns=patterns) + types=types, datadir=datadir, uri=uri, rootdir=rootdir) + + def get_elements(self): + """Get the list of subjects in the dataset. + + Returns + ------- + elements : list[str] + The list of subjects in the dataset. + """ + elems = [] + for x in self.datadir.glob('*._T1w.nii.gz'): + sub, ses = x.name.split('_') + sub = sub.replace('m0wp1', '') + ses = ses[:5] + elems.append((sub, ses)) + return elems + + def __getitem__(self, element): + """Index one element in the dataset. + + Parameters + ---------- + element : tuple[str, str] + The element to be indexed. First element in the tuple is the + subject, second element is the session. + + Returns + ------- + out : dict[str -> Path] + Dictionary of paths for each type of data required for the + specified element. + """ + sub, ses = element + out = {} + + out['VBM_GM'] = self.datadir / f'm0wp1{sub}_{ses}_T1w.nii.gz' + self._dataset_get(out) + out['meta']['element'] = dict(subject=sub, session=ses) + return out diff --git a/junifer/configs/tests/test_juseless.py b/junifer/configs/tests/test_juseless.py index 9a266c19c..7c5131327 100644 --- a/junifer/configs/tests/test_juseless.py +++ b/junifer/configs/tests/test_juseless.py @@ -1,33 +1,15 @@ import socket import pytest -from junifer.configs.juseless import JuselessDataladUKBVBM -from junifer.utils.logging import configure_logging -from junifer.datagrabber.hcp import DataladHCP1200 +from junifer.configs.juseless import JuselessUKBVBM if socket.gethostname() != 'juseless': pytest.skip('This tests are only for juseless', allow_module_level=True) -configure_logging(level='DEBUG') - -def test_juselessdataladukbvbm_datagrabber(): - with JuselessDataladUKBVBM() as dg: - all_elements = dg.get_elements() - test_element = all_elements[0] - out = dg[test_element] +def test_juselessukbvbm_datagrabber(): + with JuselessUKBVBM() as dg: + out = dg[('sub-2670511', 'ses-2')] assert 'VBM_GM' in out - assert out['VBM_GM']['path'].name == \ - f'm0wp1sub-{test_element[0]}_ses-{test_element[1]}_T1w.nii.gz' - assert out['VBM_GM']['path'].exists() - - -def test_juselessdataladhcp_datagrabber(): - with DataladHCP1200() as dg: - all_elements = dg.get_elements() - test_element = all_elements[0] - - out = dg[test_element] - - assert out['BOLD']['path'].exists() - assert out['BOLD']['path'].isfile() + assert out['VBM_GM'].name == 'm0wp1sub-2670511_ses-2_T1w.nii.gz' + assert out['VBM_GM'].exists() diff --git a/junifer/data/atlases.py b/junifer/data/atlases.py index 519cbced5..5d92f2463 100644 --- a/junifer/data/atlases.py +++ b/junifer/data/atlases.py @@ -3,7 +3,10 @@ # License: AGPL from pathlib import Path import io +import tempfile import requests +import shutil +import zipfile import numpy as np import pandas as pd @@ -44,6 +47,29 @@ for n_rois in range(100, 1001, 100): 'yeo_networks': t_net, } +for scale in range(1, 5): + t_name = f'TianxS{scale}x7TxMNI6thgeneration' + _available_atlases[t_name] = { + 'family': 'Tian', + 'scale': scale, + 'magneticfield': '7T', + 'space': 'MNI6thgeneration' + } + t_name = f'TianxS{scale}x3TxMNI6thgeneration' + _available_atlases[t_name] = { + 'family': 'Tian', + 'scale': scale, + 'magneticfield': '3T', + 'space': 'MNI6thgeneration' + } + t_name = f'TianxS{scale}x3TxMNInonlinear2009cAsym' + _available_atlases[t_name] = { + 'family': 'Tian', + 'scale': scale, + 'magneticfield': '3T', + 'space': 'MNInonlinear2009cAsym' + } + def register_atlas(name, atlas_path, atl_labels, overwrite=False): """Register a custom user atlas. @@ -136,7 +162,15 @@ def load_atlas(name, atlas_dir=None, resolution=None, path_only=False): Number of yeo networks to use. Valid values: 7, 17. Defaults to 7. Tian : - # TODO add + scale (required) : int + Scale of atlas between 1 and 4 (defines granularity) + space (optional) : str + Space of atlas can be either 'MNI6thgeneration' or + 'MNInonlinear2009cAsym' (for some cases). + Defaults to 'MNI6thgeneration'. (For more information see + https://github.com/yetianmed/subcortex) + magneticfield (optional) : str + Options are 3T and 7T, defaults to 3T. SUIT : space (optional) : str @@ -207,6 +241,16 @@ def _retrieve_atlas(family, atlas_dir=None, resolution=None, **kwargs): (included) in steps of 100. yeo_network (optional) : int Number of yeo networks to use [7 or 17]. Defaults to 7. + Tian : + scale (required) : int + Scale of atlas between 1 and 4 (defines granularity) + space (optional) : str + Space of atlas can be either 'MNI6thgeneration' or + 'MNInonlinear2009cAsym' (for some cases). + Defaults to 'MNI6thgeneration'. (For more information see + https://github.com/yetianmed/subcortex) + magneticfield (optional) : str + Options are 3T and 7T, defaults to 3T. SUIT : space (optional) : str Space of atlas can be either 'MNI' or 'SUIT' (for more information @@ -235,6 +279,9 @@ def _retrieve_atlas(family, atlas_dir=None, resolution=None, **kwargs): elif family == 'SUIT': atlas_fname, atl_labels = \ _retrieve_suit(atlas_dir, resolution=resolution, **kwargs) + elif family == 'Tian': + atlas_fname, atl_labels = \ + _retrieve_tian(atlas_dir, resolution=resolution, **kwargs) else: raise_error( f"The provided atlas name {family} cannot be retrieved. ") @@ -293,7 +340,7 @@ def _retrieve_schaefer(atlas_dir, resolution, n_rois=None, yeo_networks=7): 'At least one of the atlas files is missing. ' 'Fetching using nilearn.') datasets.fetch_atlas_schaefer_2018( - n_rois=n_rois, # type: ignore + n_rois=n_rois, yeo_networks=yeo_networks, resolution_mm=resolution, data_dir=atlas_dir.as_posix()) @@ -312,6 +359,108 @@ def _retrieve_schaefer(atlas_dir, resolution, n_rois=None, yeo_networks=7): return atlas_fname, labels +def _retrieve_tian( + atlas_dir, resolution, scale=None, space='MNI6thgeneration', + magneticfield='3T'): + # show atlas parameters to user + logger.info('Atlas parameters:') + logger.info(f'\tscale: {scale}') + logger.info(f'\tspace: {space}') + logger.info(f'\tmagneticfield: {magneticfield}') + logger.info(f'\tresolution: {resolution}') + # check validity of atlas parameters + _valid_scales = [1, 2, 3, 4] + _valid_fields = ['3T', '7T'] + if scale not in _valid_scales: + raise_error( + f'The parameter `scale` ({scale}) needs to be one of the ' + f'following: {_valid_scales}') + if magneticfield not in _valid_fields: + raise_error( + f'The parameter `magneticfield` ({magneticfield}) needs to be ' + f'one of the following: {_valid_fields}') + + if magneticfield == '3T': + _valid_spaces = ['MNI6thgeneration', 'MNInonlinear2009cAsym'] + if space == 'MNI6thgeneration': + _valid_resolutions = [1, 2] + else: # space == 'MNInonlinear2009cAsym': + _valid_resolutions = [2] + else: # magneticfield == '7T': + _valid_spaces = ['MNI6thgeneration'] + _valid_resolutions = [1.6] + if space not in _valid_spaces: + raise_error( + f'The parameter `space` ({space}) needs to be one of ' + f'the following: {_valid_spaces}') + + resolution = _closest_resolution(resolution, _valid_resolutions) + + # define file names + if magneticfield == '3T': + atlas_fname_base_3T = ( + atlas_dir / 'Tian2020MSA_v1.1' / '3T' / 'Subcortex-Only') + atlas_lname = atlas_fname_base_3T / ( + f'Tian_Subcortex_S{scale}_3T_label.txt') + if space == 'MNI6thgeneration': + atlas_fname = atlas_fname_base_3T / ( + f'Tian_Subcortex_S{scale}_{magneticfield}.nii.gz') + if resolution == 1: + atlas_fname = atlas_fname_base_3T / \ + f'Tian_Subcortex_S{scale}_{magneticfield}_1mm.nii.gz' + else: # space == 'MNInonlinear2009cAsym': + space = '2009cAsym' + atlas_fname = atlas_fname_base_3T / ( + f'Tian_Subcortex_S{scale}_{magneticfield}_{space}.nii.gz') + else: # magneticfield == '7T': + atlas_fname_base_7T = ( + atlas_dir / 'Tian2020MSA_v1.1' / '7T') + atlas_fname = atlas_dir / 'Tian2020MSA_v1.1' / f'{magneticfield}' / ( + f'Tian_Subcortex_S{scale}_{magneticfield}.nii.gz') + # define 7T labels (b/c currently no labels file available for 7T) + scale7Trois = {1: 16, 2: 34, 3: 54, 4: 62} + labels = [ + ('parcel_' + str(x)) for x in np.arange(1, scale7Trois[scale] + 1)] + atlas_lname = atlas_fname_base_7T / ( + f'Tian_Subcortex_S{scale}_7T_labelnumbering.txt') + with open(atlas_lname, 'w') as filehandle: + for listitem in labels: + filehandle.write('%s\n' % listitem) + logger.info( + 'Currently there are no labels provided for the 7T Tian atlas. A ' + 'simple numbering scheme for distinction was therefore used.') + + # check existance of atlas + if not (atlas_fname.exists() and atlas_lname.exists()): + logger.info( + 'At least one of the atlas files is missing. ' + 'Fetching.') + + url_basis = \ + 'https://www.nitrc.org/frs/download.php/12012/Tian2020MSA_v1.1.zip' + + logger.info(f'Downloading TIAN from {url_basis}') + with tempfile.TemporaryDirectory() as tmpdir: + atlas_download = requests.get(url_basis) + atlas_zip_fname = Path(tmpdir) / 'Tian2020MSA_v1.1.zip' + with open(atlas_zip_fname, 'wb') as f: + f.write(atlas_download.content) + with zipfile.ZipFile(atlas_zip_fname, 'r') as zip_ref: + zip_ref.extractall(atlas_dir.as_posix()) + # clean after unzipping + if (atlas_dir / '__MACOSX').exists(): + shutil.rmtree((atlas_dir / '__MACOSX').as_posix()) + + labels = pd.read_csv(atlas_lname, sep=" ", header=None)[0].to_list() + + if not (atlas_fname.exists() and atlas_lname.exists()): + raise_error('There was a problem fetching the atlases.') + + labels = pd.read_csv(atlas_lname, sep=" ", header=None)[0].to_list() + + return atlas_fname, labels + + def _retrieve_suit(atlas_path, resolution, space='MNI'): logger.info('Atlas parameters:') logger.info(f'\tspace: {space}') diff --git a/junifer/data/tests/test_atlas.py b/junifer/data/tests/test_atlas.py index c02b1b42b..e42c02228 100644 --- a/junifer/data/tests/test_atlas.py +++ b/junifer/data/tests/test_atlas.py @@ -1,11 +1,11 @@ import tempfile import pytest from pathlib import Path -from numpy.testing import assert_array_equal +from numpy.testing import assert_array_equal, assert_array_almost_equal from junifer.data.atlases import (register_atlas, list_atlases, load_atlas, _retrieve_schaefer, _retrieve_suit, - _retrieve_atlas) + _retrieve_atlas, _retrieve_tian) def test_register_atlas(): @@ -150,3 +150,67 @@ def test_suit(): with pytest.raises(ValueError, match=r"The parameter `space`"): _retrieve_suit(tmpdir, 1, space='wrong') + + +def test_tian(): + """Test TIAN atlas""" + + atlases = list_atlases() + assert 'TianxS1x3TxMNI6thgeneration' in atlases + assert 'TianxS2x3TxMNI6thgeneration' in atlases + assert 'TianxS3x3TxMNI6thgeneration' in atlases + assert 'TianxS4x3TxMNI6thgeneration' in atlases + + assert 'TianxS1x3TxMNInonlinear2009cAsym' in atlases + assert 'TianxS2x3TxMNInonlinear2009cAsym' in atlases + assert 'TianxS3x3TxMNInonlinear2009cAsym' in atlases + assert 'TianxS4x3TxMNInonlinear2009cAsym' in atlases + + assert 'TianxS1x7TxMNI6thgeneration' in atlases + assert 'TianxS2x7TxMNI6thgeneration' in atlases + assert 'TianxS3x7TxMNI6thgeneration' in atlases + assert 'TianxS4x7TxMNI6thgeneration' in atlases + + with tempfile.TemporaryDirectory() as tmpdir: + for scale, n_lbl in zip([1, 2, 3, 4], [16, 32, 50, 54]): + img, lbl, fname = load_atlas( + f'TianxS{scale}x3TxMNI6thgeneration', atlas_dir=tmpdir) + fname1 = f'Tian_Subcortex_S{scale}_3T_1mm.nii.gz' + assert img is not None + assert fname.name == fname1 # type: ignore + assert len(lbl) == n_lbl + assert_array_equal(img.header['pixdim'][1:4], [1, 1, 1]) + + img, lbl, fname = load_atlas( + f'TianxS{scale}x3TxMNI6thgeneration', atlas_dir=tmpdir, + resolution=2) + fname1 = f'Tian_Subcortex_S{scale}_3T.nii.gz' + assert img is not None + assert fname.name == fname1 # type: ignore + assert len(lbl) == n_lbl + assert_array_equal(img.header['pixdim'][1:4], [2, 2, 2]) + + img, lbl, fname = load_atlas( + f'TianxS{scale}x3TxMNInonlinear2009cAsym', atlas_dir=tmpdir) + fname1 = f'Tian_Subcortex_S{scale}_3T_2009cAsym.nii.gz' + assert img is not None + assert fname.name == fname1 # type: ignore + assert len(lbl) == n_lbl + assert_array_equal(img.header['pixdim'][1:4], [2, 2, 2]) + + for scale, n_lbl in zip([1, 2, 3, 4], [16, 34, 54, 62]): + img, lbl, fname = load_atlas( + f'TianxS{scale}x7TxMNI6thgeneration', atlas_dir=tmpdir) + fname1 = f'Tian_Subcortex_S{scale}_7T.nii.gz' + assert img is not None + assert fname.name == fname1 # type: ignore + assert len(lbl) == n_lbl + assert_array_almost_equal( + img.header['pixdim'][1:4], [1.6, 1.6, 1.6]) + + with pytest.raises(ValueError, match=r"The parameter `space`"): + _retrieve_tian(tmpdir, resolution=1, scale=1, space='wrong') + + with pytest.raises(ValueError, match=r"The parameter `magneticfield`"): + _retrieve_tian( + tmpdir, resolution=1, scale=1, magneticfield='wrong') diff --git a/junifer/datagrabber/__init__.py b/junifer/datagrabber/__init__.py index ce54e9f35..e9866a1be 100644 --- a/junifer/datagrabber/__init__.py +++ b/junifer/datagrabber/__init__.py @@ -1,5 +1,4 @@ # Authors: Federico Raimondo # Leonard Sasse # License: AGPL -from .base import (DataladDataGrabber, PatternDataGrabber, - PatternDataladDataGrabber) \ No newline at end of file +from .base import BIDSDataladDataGrabber, DataladDataGrabber, BIDSDataGrabber \ No newline at end of file diff --git a/junifer/datagrabber/base.py b/junifer/datagrabber/base.py index 431f7ca4d..7df89e8d2 100644 --- a/junifer/datagrabber/base.py +++ b/junifer/datagrabber/base.py @@ -2,14 +2,13 @@ # Leonard Sasse # License: AGPL from pathlib import Path -import re import tempfile import datalad.api as dl from abc import ABC, abstractmethod from ..api.decorators import register_datagrabber -from ..utils.logging import logger, raise_error, warn +from ..utils.logging import logger, raise_error def _validate_types(types): @@ -23,22 +22,6 @@ def _validate_types(types): "types must be a list of strings", TypeError) # type: ignore -def _validate_replacements(replacements, patterns): - """ - Validate the replacements - """ - if not isinstance(replacements, list): - raise_error("replacements must be a list", TypeError) # type: ignore - if any(not isinstance(x, str) for x in replacements): - raise_error( - "replacements must be a list of strings", - TypeError) # type: ignore - - for x in replacements: - if all(f'{{x}}' not in y for y in patterns.values()): - warn(f"Replacement {x} is not part of any pattern") - - def _validate_patterns(types, patterns): """ Validate the patterns. @@ -90,9 +73,6 @@ class BaseDataGrabber(ABC): _validate_types(types) if not isinstance(datadir, Path): datadir = Path(datadir) - logger.debug('Initializing BaseDataGrabber') - logger.debug(f'\t_datadir = {datadir}') - logger.debug(f'\ttypes = {types}') self._datadir = datadir self.types = types @@ -131,17 +111,13 @@ class BaseDataGrabber(ABC): for elem in self.get_elements(): yield elem + @abstractmethod def __getitem__(self, element): - logger.info(f'Getting element {element}') - out = {} - out['meta'] = dict(datagrabber=self.get_meta()) - return out + raise NotImplementedError('__getitem__ not implemented') @abstractmethod def get_elements(self): - raise_error( - 'get_elements not implemented', - NotImplementedError) # type: ignore + raise NotImplementedError('get_elements not implemented') def __enter__(self): return self @@ -151,37 +127,34 @@ class BaseDataGrabber(ABC): @register_datagrabber -class PatternDataGrabber(BaseDataGrabber): - """Patternd DataGrabber class (abstract). Implements a DataGrabber that - understands patterns to grab data. +class BIDSDataGrabber(BaseDataGrabber): + """BIDS DataGrabber class. Implements a DataGrabber that understands BIDS + database format. Attributes ---------- - datadir: Path - Directory where the data is stored - types : list[str] + datadir + types : list List of data types to be grabbed. patterns : dict[str -> str] Patterns for each type of data. - replacements: list[str] - Replacements in the patterns for each item in the `element` tuple. Methods ------- get_elements: list[str] Returns a list of elements that can be grabbed. Each element is a subject in the BIDS database. - __getitem__(str): dict[str -> dict] + __getitem__(str): dict[str -> Path] Returns a dictionary of paths for each type of data required for the specified element. Each occurrence of the string `{subject}` is replaced by the indexed element """ - def __init__(self, types=None, patterns=None, replacements=None, **kwargs): + def __init__(self, types=None, patterns=None, **kwargs): """Initialize a BaseDataGrabber object. Parameters ---------- - types : list[str] + types : list of str The types of data to be grabbed. patterns : dict[str -> str] Patterns for each type of data. The keys are the types and the @@ -191,151 +164,62 @@ class PatternDataGrabber(BaseDataGrabber): That directory where the data is/will be stored. """ _validate_patterns(types, patterns) - if not isinstance(replacements, list): - replacements = [replacements] - _validate_replacements(replacements, patterns) super().__init__(types=types, **kwargs) - logger.debug('Initializing PatternDataGrabber') - logger.debug(f'\tpatterns = {patterns}') - logger.debug(f'\treplacements = {replacements}') self.patterns = patterns - self.replacements = replacements - - def _replace_patterns_regex(self, pattern): - """Replace the patterns in the pattern with the named groups so the - elements can be obtained from the filesystem. - - Parameters - ---------- - pattern : str - The pattern to be replaced. - - Returns - ------- - re_pattern : str - The regular expression with the named groups. - glob_pattern : str - The search pattern to be used with glob - - """ - re_pattern = pattern - glob_pattern = pattern - for t_r in self.replacements: - # Replace the first of each with a named group definition - re_pattern = re_pattern.replace( - f'{{{t_r}}}', f'(?P<{t_r}>.*)', 1) - - for t_r in self.replacements: - # Replace the second appearance of each with the named group - # back reference - re_pattern = re_pattern.replace(f'{{{t_r}}}', f'(?P={t_r})') - - for t_r in self.replacements: - glob_pattern = glob_pattern.replace(f'{{{t_r}}}', f'*') - return re_pattern, glob_pattern def get_elements(self): - """Get the list of elements in the dataset. It will use regex - to search for `replacements` in the `patterns` and return the - intersection of the results for each type. That is, build a list - of elements that have all the required types. + """Get all the elements in the BIDS database Returns ------- - elements : list - The list of elements in the dataset. + elems : list[str] + List of all the elements in the database root directory """ - elements = None - for t_type in self.types: - types_element = set() - t_pattern = self.patterns[t_type] # get the pattern - re_pattern, glob_pattern = self._replace_patterns_regex(t_pattern) - for fname in self.datadir.glob(glob_pattern): - suffix = fname.relative_to(self.datadir).as_posix() - m = re.match(re_pattern, suffix) - if m is not None: - t_element = tuple(m.group(k) for k in self.replacements) - if len(self.replacements) == 1: - t_element = t_element[0] - types_element.add(t_element) - if elements is None: - elements = types_element - else: - elements = elements.intersection(types_element) - - return list(elements) - - def _replace_patterns_glob(self, element, pattern): - """Replace the patterns in the pattern with the element so it can - be globbed. - - Parameters - ---------- - element : tuple - The element to be used in the replacement. - pattern : str - The pattern to be replaced. - - Returns - ------- - str - The pattern with the element replaced. - """ - if len(element) != len(self.replacements): - raise_error( - f'The element lenght must be {len(self.replacements)}, ' - f'indicating {self.replacements}') - to_replace = dict(zip(self.replacements, element)) - return pattern.format(**to_replace) + elems = [x.name for x in self.datadir.iterdir() if x.is_dir()] + return elems def __getitem__(self, element): - """Index one element in the database. + """Index one element in the BIDS database. - Each occurrence of the strings in `replacements` is replaced by the - corresponding item in the element tuple. + Each occurrence of the string `{subject}` is replaced by the indexed + element. Parameters ---------- element : str or tuple The element to be indexed. If one string is provided, it is - assumed to be a tuple with only one item. If a tuple is provided, - each item in the tuple is the value for the replacement string - specified in `replacements`. - + assumed to be a subject. If a tuple is provided, it is assumed to + be a (subject, session) pair. Returns ------- out : dict[str -> Path] Dictionary of paths for each type of data required for the specified element. """ - out = super().__getitem__(element) + out = {} if not isinstance(element, tuple): element = (element,) for t_type in self.types: t_pattern = self.patterns[t_type] # type: ignore - t_replace = self._replace_patterns_glob(element, t_pattern) - if '*' in t_replace: - t_matches = list(self.datadir.glob(t_replace)) - if len(t_matches) > 1: - raise_error( - f'More than one file matches for {element} / {t_type}:' - f' {t_matches}') - elif len(t_matches) == 0: - raise_error(f'No file matches for {element} / {t_type}') - t_out = t_matches[0] - else: - t_out = self.datadir / t_replace + t_replace = t_pattern.replace('{subject}', element[0]) + if len(element) > 1: + t_replace = t_replace.replace( + '{session}', element[1]) # type: ignore + t_out = self.datadir / element[0] / t_replace out[t_type] = dict(path=t_out) # Meta here is element and types - out['meta']['element'] = dict(zip(self.replacements, element)) + out['meta'] = dict(datagrabber=self.get_meta()) + out['meta']['element'] = {'subject': element[0]} + if len(element) > 1: + out['meta']['element']['session'] = element[1] # type: ignore return out @register_datagrabber class DataladDataGrabber(BaseDataGrabber): """ - Datalad DataGrabber class (abstract). Implements a DataGrabber that gets - data from a datalad sibling. + Datalad DataGrabber class. Implements a DataGrabber that gets data from + a datalad sibling. Attributes ---------- @@ -383,9 +267,6 @@ class DataladDataGrabber(BaseDataGrabber): datadir = tempfile.mkdtemp() logger.info(f'datadir set to {datadir}') super().__init__(datadir=datadir, **kwargs) - logger.debug('Initializing DataladDataGrabber') - logger.debug(f'\turi = {uri}') - logger.debug(f'\t_rootdir = {rootdir}') self.uri = uri self._rootdir = rootdir @@ -399,15 +280,11 @@ class DataladDataGrabber(BaseDataGrabber): def install(self): """Install the datalad dataset into the datadir.""" - logger.debug(f'Installing dataset {self.uri} to {self._datadir}') self.dataset = dl.install( # type: ignore self._datadir, source=self.uri) - logger.debug('Dataset installed') def __exit__(self, exc_type, exc_value, exc_traceback): - logger.debug('Removing dataset') self.remove() - logger.debug('Dataset removed') def remove(self): """Remove the datalad dataset from the datadir.""" @@ -416,9 +293,7 @@ class DataladDataGrabber(BaseDataGrabber): def _dataset_get(self, out): for _, v in out.items(): if 'path' in v: - logger.debug(f'Getting {v["path"]}') self.dataset.get(v['path']) - logger.debug(f'Get done') # append the version of the dataset out['meta']['datagrabber']['dataset_commit_id'] = \ @@ -439,16 +314,15 @@ class DataladDataGrabber(BaseDataGrabber): return out -class PatternDataladDataGrabber(DataladDataGrabber, PatternDataGrabber): - """Pattern-based Datalad DataGrabber class (abstract). - Implements a DataGrabber that gets data from a datalad sibling, - interpreting patterns. - +class BIDSDataladDataGrabber(DataladDataGrabber, BIDSDataGrabber): + """BIDS Datalad DataGrabber class. + Implements a DataGrabber that gets data from a datalad sibling which + follows a BIDS format. See Also -------- DataladDataGrabber - PatternDataGrabber + BIDSDataGrabber """ def __init__(self, types=None, patterns=None, **kwargs): diff --git a/junifer/datagrabber/hcp.py b/junifer/datagrabber/hcp.py deleted file mode 100644 index f3ef35215..000000000 --- a/junifer/datagrabber/hcp.py +++ /dev/null @@ -1,148 +0,0 @@ -from itertools import product - -from junifer.datagrabber.base import DataladDataGrabber - -from ..datagrabber import PatternDataGrabber -from ..api.decorators import register_datagrabber - - -@register_datagrabber -class HCP1200(PatternDataGrabber): - - def __init__( - self, datadir=None, tasks=None, phase_encodings=None - ): - """Initialize a HCP object. - - Parameters - ---------- - datadir : str or Path - That directory where the datalad dataset will be cloned. If None, - (default), the datalad dataset will be cloned into a temporary - directory. - tasks : str or list of strings - HCP task sessions. If 'None' (default), all available task - sessions are selected. Can be 'REST1', 'REST2', 'SOCIAL', 'WM', - 'RELATIONAL', 'EMOTION', 'LANGUAGE', 'GAMBLING', 'MOTOR', or a - list consisting of these names. - phase_encoding : str or list of strings - HCP phase encoding directions. Can be 'LR' or 'RL'. If 'None' - (default) both will be used. - - """ - - types = ['BOLD'] - - replacements = ['subject', 'task', 'phase_encoding'] - patterns = { - 'BOLD': ('{subject}/MNINonLinear/Results/' - '{task}_{phase_encoding}/' - '{task}_{phase_encoding}_hp2000_clean.nii.gz') - } - super().__init__( - types=types, datadir=datadir, patterns=patterns, - replacements=replacements - ) - - self.tasks = tasks - self.phase_encodings = phase_encodings - - if isinstance(self.tasks, str): - self.tasks = [self.tasks] - if isinstance(self.phase_encodings, str): - self.phase_encodings = [self.phase_encodings] - - all_tasks = [ - 'REST1', - 'REST2', - 'SOCIAL', - 'WM', - 'RELATIONAL', - 'EMOTION', - 'LANGUAGE', - 'GAMBLING', - 'MOTOR', - ] - - if self.tasks is None: - self.tasks = all_tasks - - if self.phase_encodings is None: - self.phase_encodings = ['LR', 'RL'] - - for task in self.tasks: - if task not in all_tasks: - raise ValueError( - f'{task} not a valid HCP-YA fMRI task input! \n' - f'task can be any of {all_tasks}' - ) - - for pe in self.phase_encodings: - if pe not in ['LR', 'RL']: - raise ValueError( - f'{pe} not a valid HCP-YA phase encoding. \n' - 'phase_encoding can be LR or RL (or both)!' - ) - - - def get_elements(self): - """Get the list of subjects in the dataset. - - Returns - ------- - elements : list[str] - The list of subjects in the dataset. - """ - - subjects = [x.name for x in self.datadir.iterdir() if x.is_dir()] - elems = [] - for subject, task, phase_encoding in product( - subjects, self.tasks, self.phase_encodings - ): - elems.append((subject, task, phase_encoding)) - - return elems - - def __getitem__(self, element): - """Index one element in the dataset. - - Parameters - ---------- - element : tuple[str, str] - The element to be indexed. First element in the tuple is the - subject, second element is the task, third element is the - phase encoding direction. - - Returns - ------- - out : dict[str -> Path] - Dictionary of paths for each type of data required for the - specified element. - """ - - sub, task, phase_encoding = element - - if "REST" in task: - new_task = f'rfMRI_{task}' - else: - new_task = f'tfMRI_{task}' - - out = super().__getitem__((sub, new_task, phase_encoding)) - out['meta']['element'] = dict( - subject=sub, task=task, phase_encoding=phase_encoding - ) - - return out - - -@register_datagrabber -class DataladHCP1200(DataladDataGrabber, HCP1200): - def __init__(self, datadir=None, tasks=None, phase_encodings=None): - uri = ( - 'https://github.com/datalad-datasets/' - 'human-connectome-project-openaccess.git' - ) - rootdir = 'HCP1200' - super().__init__(datadir=datadir, tasks=tasks, - phase_encodings=phase_encodings, - uri=uri, rootdir=rootdir) # type: ignore diff --git a/junifer/datagrabber/meta.py b/junifer/datagrabber/meta.py deleted file mode 100644 index 1b20949d2..000000000 --- a/junifer/datagrabber/meta.py +++ /dev/null @@ -1,32 +0,0 @@ -from .base import BaseDataGrabber - - -class MultipleDataGrabber(BaseDataGrabber): - def __init__(self, datagrabbers): - # TODO: Check datagrabbers consistency - # - same element keys - # - no overlapping types - self._datagrabbers = datagrabbers - - def get_types(self): - types = [x for dg in self._datagrabbers for x in dg.get_types()] - return types - - def get_meta(self): - t_meta = {} - t_meta['class'] = self.__class__.__name__ - t_meta['datagrabbers'] = [dg.get_meta() for dg in self._datagrabbers] - - def __enter__(self): - for dg in self._datagrabbers: - dg.__enter__() - - def __exit__(self, exc_type, exc_value, exc_traceback): - for dg in self._datagrabbers: - dg.__exit__(exc_type, exc_value, exc_traceback) - - def __getitem__(self, element): - out = {} - for dg in self._datagrabbers: - t_out = dg[element] - out.update(t_out) diff --git a/junifer/datagrabber/tests/test_base_datagrabber.py b/junifer/datagrabber/tests/test_base_datagrabber.py index 689dd24a6..a31fa8677 100644 --- a/junifer/datagrabber/tests/test_base_datagrabber.py +++ b/junifer/datagrabber/tests/test_base_datagrabber.py @@ -4,18 +4,14 @@ import tempfile import pytest from pathlib import Path -from junifer.datagrabber.base import (PatternDataGrabber, BaseDataGrabber, - PatternDataladDataGrabber) +from junifer.datagrabber.base import BIDSDataGrabber, BIDSDataladDataGrabber, \ + BaseDataGrabber _testing_dataset = { 'example_bids': { 'uri': 'https://gin.g-node.org/juaml/datalad-example-bids', 'id': 'e2ce149bd723088769a86c72e57eded009258c6b' - }, - 'example_bids_ses': { - 'uri': 'https://gin.g-node.org/juaml/datalad-example-bids-ses', - 'id': '3d08d55d1faad4f12ab64ac9497544a0d924d47a' } } @@ -33,11 +29,8 @@ def test_BaseDataGrabber(): return super().get_elements() dg = MyDataGrabber(datadir='/tmp', types=['func']) - elem = dg['elem'] - assert 'meta' in elem - assert 'datagrabber' in elem['meta'] - assert 'class' in elem['meta']['datagrabber'] - assert MyDataGrabber.__name__ in elem['meta']['datagrabber']['class'] + with pytest.raises(NotImplementedError): + dg['elem'] with pytest.raises(NotImplementedError): dg.get_elements() @@ -47,93 +40,60 @@ def test_BaseDataGrabber(): assert dg.types == ['func'] -def test_PatternDataGrabber(): - class MyDataGrabber(PatternDataGrabber): - def get_elements(self): - return super().get_elements() - - """Test test_PatternDataGrabber""" +def test_BIDSDataGrabber(): + """Test BIDSDataGrabber""" with pytest.raises(TypeError, match=r"types must be a list"): - MyDataGrabber(datadir='/tmp', types='wrong', - patterns=dict(wrong='pattern'), - replacements='subject') + BIDSDataGrabber(datadir='/tmp', types='wrong', + patterns=dict(wrong='pattern')) with pytest.raises(TypeError, match=r"must be a list of strings"): - MyDataGrabber(datadir='/tmp', types=[1, 2, 3], - patterns={'1': 'pattern', '2': 'pattern', - '3': 'pattern'}, - replacements='subject') + BIDSDataGrabber(datadir='/tmp', types=[1, 2, 3], + patterns={'1': 'pattern', '2': 'pattern', + '3': 'pattern'}) - with pytest.raises(ValueError, match=r"must have the same length"): - MyDataGrabber(datadir='/tmp', types=['func', 'anat'], - patterns={'1': 'pattern', '2': 'pattern', - '3': 'pattern'}, - replacements=1) + datagrabber = BIDSDataGrabber( + datadir='/tmp/data', types=['func', 'anat'], + patterns=dict(func='pattern1', anat='pattern2')) + assert datagrabber.datadir == Path('/tmp/data') + assert datagrabber.types == ['func', 'anat'] + + datagrabber = BIDSDataGrabber( + datadir=Path('/tmp/data'), types=['func', 'anat'], + patterns=dict(func='pattern1', anat='pattern2')) + assert datagrabber.datadir == Path('/tmp/data') + assert datagrabber.types == ['func', 'anat'] with pytest.raises(TypeError, match=r"patterns must be a dict"): - MyDataGrabber(datadir='/tmp', types=['func', 'anat'], - patterns='wrong', replacements='subject') + BIDSDataGrabber(datadir='/tmp', types=['func', 'anat'], + patterns='wrong') with pytest.raises(ValueError, match=r"patterns must have the same length"): - MyDataGrabber(datadir='/tmp', types=['func', 'anat'], - patterns={'wrong': 'pattern'}, replacements='subject') + BIDSDataGrabber(datadir='/tmp', types=['func', 'anat'], + patterns={'wrong': 'pattern'}) with pytest.raises(ValueError, match=r"patterns must contain all types"): - MyDataGrabber(datadir='/tmp', types=['func', 'anat'], - patterns={'wrong': 'pattern', 'func': 'pattern'}, - replacements='subject') - - with pytest.raises(TypeError, match=r"must be a list of strings"): - MyDataGrabber(datadir='/tmp', types=['func', 'anat'], - patterns={'func': 'func/test', 'anat': 'anat/test'}, - replacements=1) - - with pytest.warns(RuntimeWarning, match=r"not part of any pattern"): - MyDataGrabber(datadir='/tmp', types=['func', 'anat'], - patterns={'func': 'func/{subject}.nii', - 'anat': 'anat/{subject}.nii'}, - replacements=['subject', 'wrong']) - - datagrabber = MyDataGrabber( - datadir='/tmp/data', types=['func', 'anat'], - patterns={'func': 'func/{subject}.nii', - 'anat': 'anat/{subject}.nii'}, - replacements='subject') - assert datagrabber.datadir == Path('/tmp/data') - assert datagrabber.types == ['func', 'anat'] - assert datagrabber.replacements == ['subject'] - - datagrabber = MyDataGrabber( - datadir=Path('/tmp/data'), types=['func', 'anat'], - patterns={'func': 'func/{subject}.nii', - 'anat': 'anat/{subject}_{session}.nii'}, - replacements=['subject', 'session']) - assert datagrabber.datadir == Path('/tmp/data') - assert datagrabber.types == ['func', 'anat'] - assert datagrabber.replacements == ['subject', 'session'] + BIDSDataGrabber(datadir='/tmp', types=['func', 'anat'], + patterns={'wrong': 'pattern', 'func': 'pattern'}) -def test_bids_datalad_PatternDataGrabber(): - """Test a subject-based BIDS datalad datagrabber""" +def test_BIDSDataladDataGrabber(): + """Test BIDSDataladDataGrabber""" types = ['T1w', 'bold'] patterns = { - 'T1w': '{subject}/anat/{subject}_T1w.nii.gz', - 'bold': '{subject}/func/{subject}_task-rest_bold.nii.gz' + 'T1w': 'anat/{subject}_T1w.nii.gz', + 'bold': 'func/{subject}_task-rest_bold.nii.gz' } - replacements = ['subject'] with pytest.raises(ValueError, match=r"uri must be provided"): - PatternDataladDataGrabber(datadir=None, types=types, patterns=patterns, - replacements=replacements) + BIDSDataladDataGrabber(datadir=None, types=types, patterns=patterns) repo_uri = _testing_dataset['example_bids']['uri'] rootdir = 'example_bids' repo_commit = _testing_dataset['example_bids']['id'] - with PatternDataladDataGrabber( - rootdir=rootdir, uri=repo_uri, types=types, patterns=patterns, - replacements=replacements) as dg: + with BIDSDataladDataGrabber(rootdir=rootdir, uri=repo_uri, + types=types, patterns=patterns) as dg: subs = [x for x in dg] expected_subs = [f'sub-{i:02d}' for i in range(1, 10)] assert set(subs) == set(expected_subs) @@ -151,7 +111,7 @@ def test_bids_datalad_PatternDataGrabber(): assert 'datagrabber' in t_sub['meta'] dg_meta = t_sub['meta']['datagrabber'] assert 'class' in dg_meta - assert dg_meta['class'] == 'PatternDataladDataGrabber' + assert dg_meta['class'] == 'BIDSDataladDataGrabber' assert 'uri' in dg_meta assert dg_meta['uri'] == repo_uri assert 'dataset_commit_id' in dg_meta @@ -162,60 +122,7 @@ def test_bids_datalad_PatternDataGrabber(): with tempfile.TemporaryDirectory() as tmpdir: datadir = Path(tmpdir) / 'dataset' # Need this for testing - patterns = { - 'T1w': '{subject}/anat/{subject}_T*w.nii.gz', - 'bold': '{subject}/func/{subject}_task-rest_*.nii.gz' - } - with PatternDataladDataGrabber( - rootdir=rootdir, uri=repo_uri, types=types, patterns=patterns, - datadir=datadir, replacements=replacements) as dg: + with BIDSDataladDataGrabber(rootdir=rootdir, uri=repo_uri, + types=types, patterns=patterns, + datadir=datadir) as dg: assert dg.datadir == datadir / rootdir - for elem in dg: - t_sub = dg[elem] - assert 'path' in t_sub['T1w'] - assert t_sub['T1w']['path'] == \ - (dg.datadir / f'{elem}/anat/{elem}_T1w.nii.gz') - assert 'path' in t_sub['bold'] - assert t_sub['bold']['path'] == \ - (dg.datadir / f'{elem}/func/{elem}_task-rest_bold.nii.gz') - - -def test_bids_datalad_PatternDataGrabber_session(): - """Test a subject and session-based BIDS datalad datagrabber""" - types = ['T1w', 'bold'] - patterns = { - 'T1w': '{subject}/{session}/anat/{subject}_{session}_T1w.nii.gz', - 'bold': '{subject}/{session}/func/' - '{subject}_{session}_task-rest_bold.nii.gz' - } - replacements = ['subject', 'session'] - - with pytest.raises(ValueError, match=r"uri must be provided"): - PatternDataladDataGrabber(datadir=None, types=types, patterns=patterns, - replacements=replacements) - - repo_uri = _testing_dataset['example_bids_ses']['uri'] - rootdir = 'example_bids_ses' - # repo_commit = _testing_dataset['example_bids_ses']['id'] - - # With T1W and bold, only 2 sessions are available - with PatternDataladDataGrabber( - rootdir=rootdir, uri=repo_uri, types=types, patterns=patterns, - replacements=replacements) as dg: - subs = [x for x in dg] - expected_subs = [(f'sub-{i:02d}', f'ses-{j:02d}') for j in range(1, 3) - for i in range(1, 10)] - assert set(subs) == set(expected_subs) - - # Test with a different T1w only, it should have 3 sessions - types = ['T1w'] - patterns = { - 'T1w': '{subject}/{session}/anat/{subject}_{session}_T1w.nii.gz', - } - with PatternDataladDataGrabber( - rootdir=rootdir, uri=repo_uri, types=types, patterns=patterns, - replacements=replacements) as dg: - subs = [x for x in dg] - expected_subs = [(f'sub-{i:02d}', f'ses-{j:02d}') for j in range(1, 4) - for i in range(1, 10)] - assert set(subs) == set(expected_subs) diff --git a/junifer/markers/base.py b/junifer/markers/base.py index 0cfec4fc6..9d736aa15 100644 --- a/junifer/markers/base.py +++ b/junifer/markers/base.py @@ -18,9 +18,8 @@ class PipelineStepMixin(): Parameters ---------- - input : list[str] - The input to the pipeline step. The list must contain the - available Junifer Data dictionary keys. + input : Junifer Data dictionary + The input to the pipeline step. Raises ------ @@ -34,15 +33,13 @@ class PipelineStepMixin(): Parameters ---------- - input : list[str] - The input to the pipeline step. The list must contain the - available Junifer Data dictionary keys. + input : Junifer Data dictionary + The input to the pipeline step. Returns ------- - output : list[str] - The updated list of available Junifer Data dictionary keys after - the pipeline step. + output : Junifer Data dictionary + The output of the pipeline step. """ raise NotImplementedError('get_output_kind not implemented') @@ -82,10 +79,8 @@ class BaseMarker(PipelineStepMixin): def get_meta(self, kind): s_meta = super().get_meta() - # same marker can be fit into different kinds, so the name - # is created from the kind and the name of the marker - s_meta['name'] = f'{kind}_{self.name}' - s_meta['kind'] = kind + s_meta['name'] = self.name + s_meta['kind'] = kind # same marker can be fit into different kinds return dict(marker=s_meta) def validate_input(self, input): diff --git a/junifer/markers/parcel.py b/junifer/markers/parcel.py index 6ede62d7e..79e23d405 100644 --- a/junifer/markers/parcel.py +++ b/junifer/markers/parcel.py @@ -22,13 +22,13 @@ class ParcelAggregation(BaseMarker): self.method_params = {} if method_params is None else method_params def get_output_kind(self, input): - if input in ['VBM_GM', 'VBM_WM']: + if input in ['GMD_GM', 'GMD_WM']: return 'table' if input in ['BOLD']: return 'timeseries' def store(self, kind, out, storage): - if kind in ['VBM_GM', 'VBM_WM']: + if kind in ['GMD_GM', 'GMD_WM']: storage.store_table(**out) if kind in ['BOLD']: storage.store_timeseries(**out) diff --git a/junifer/markers/tests/test_base_marker.py b/junifer/markers/tests/test_base_marker.py index a0d43c81c..cf9ed5298 100644 --- a/junifer/markers/tests/test_base_marker.py +++ b/junifer/markers/tests/test_base_marker.py @@ -20,14 +20,14 @@ def test_meta(): base = BaseMarker(on=['bold', 'dwi']) - t_meta = base.get_meta('bold') + t_meta = base.get_meta() assert t_meta['marker']['class'] == 'BaseMarker' - assert t_meta['marker']['name'] == 'bold_BaseMarker' + assert t_meta['marker']['name'] == 'BaseMarker' base = BaseMarker(on=['bold', 'dwi'], name='mymarker') - t_meta = base.get_meta('dwi') - assert t_meta['marker']['name'] == 'dwi_mymarker' + t_meta = base.get_meta() + assert t_meta['marker']['name'] == 'mymarker' def test_BaseMarker(): @@ -50,12 +50,12 @@ def test_BaseMarker(): out = base.fit_transform(input) assert out['bold']['data'] == 1 - assert out['bold']['meta']['marker']['name'] == 'bold_mymarker' + assert out['bold']['meta']['marker']['name'] == 'mymarker' assert out['bold']['meta']['marker']['class'] == 'BaseMarker' base2 = BaseMarker(on='bold', name='mymarker') base2.compute = lambda x: dict(data=1) # type: ignore out2 = base2.fit_transform(input) assert out2['bold']['data'] == 1 - assert out2['bold']['meta']['marker']['name'] == 'bold_mymarker' + assert out2['bold']['meta']['marker']['name'] == 'mymarker' assert out2['bold']['meta']['marker']['class'] == 'BaseMarker' diff --git a/junifer/markers/tests/test_collection.py b/junifer/markers/tests/test_collection.py index bb0bdf154..b68d2a7fd 100644 --- a/junifer/markers/tests/test_collection.py +++ b/junifer/markers/tests/test_collection.py @@ -7,7 +7,7 @@ from numpy.testing import assert_array_equal from junifer.datareader.default import DefaultDataReader from junifer.markers import MarkerCollection, ParcelAggregation from junifer.markers.base import PipelineStepMixin -from junifer.testing.datagrabbers import OasisVBMTestingDatagrabber +from junifer.testing import OasisVBMTestingDatagrabber from junifer.storage import SQLiteFeatureStorage @@ -47,7 +47,7 @@ def test_MarkerCollection(): mc.validate(dg) with dg: - input = dg['sub-01'] + input = dg[1] out = mc.fit(input) assert out is not None assert isinstance(out, dict) @@ -74,7 +74,7 @@ def test_MarkerCollection(): datareader=DefaultDataReader()) assert isinstance(mc2._datareader, DefaultDataReader) with dg: - input = dg['sub-01'] + input = dg[1] out2 = mc2.fit(input) for t_marker in markers: t_name = t_marker.name @@ -99,14 +99,14 @@ def test_MarkerCollection_storage(): # Test storage dg = OasisVBMTestingDatagrabber() with tempfile.TemporaryDirectory() as tmpdir: - uri = f'{tmpdir}/test.db' - storage = SQLiteFeatureStorage(uri=uri, single_output=True) + uri = f'sqlite:///{tmpdir}/test.db' + storage = SQLiteFeatureStorage(uri=uri) mc = MarkerCollection( markers=markers, storage=storage, datareader=DefaultDataReader()) mc.validate(dg) assert mc._storage.uri == storage.uri with dg: - input = dg['sub-01'] + input = dg[1] out = mc.fit(input) assert out is None @@ -116,28 +116,11 @@ def test_MarkerCollection_storage(): assert mc2._storage is None with dg: - input = dg['sub-01'] + input = dg[1] out = mc2.fit(input) features = storage.list_features() - assert len(features) == 3 - feature_md5 = list(features.keys())[0] + assert len(features) == 1 + feature_md5 = features.keys()[0] t_feature = storage.read_df(feature_md5=feature_md5) - fname = 'gmd_schaefer100x7_mean' - t_data = out[fname]['VBM_GM']['data'] # type: ignore - cols = out[fname]['VBM_GM']['columns'] # type: ignore - assert_array_equal(t_feature[cols].values, t_data) # type: ignore - - feature_md5 = list(features.keys())[1] - t_feature = storage.read_df(feature_md5=feature_md5) - fname = 'gmd_schaefer100x7_std' - t_data = out[fname]['VBM_GM']['data'] # type: ignore - cols = out[fname]['VBM_GM']['columns'] # type: ignore - assert_array_equal(t_feature[cols].values, t_data) # type: ignore - - feature_md5 = list(features.keys())[2] - t_feature = storage.read_df(feature_md5=feature_md5) - fname = 'gmd_schaefer100x7_trim_mean90' - t_data = out[fname]['VBM_GM']['data'] # type: ignore - cols = out[fname]['VBM_GM']['columns'] # type: ignore - assert_array_equal(t_feature[cols].values, t_data) # type: ignore + diff --git a/junifer/markers/tests/test_parcel.py b/junifer/markers/tests/test_parcel.py index 5098f9b01..7ab06d591 100644 --- a/junifer/markers/tests/test_parcel.py +++ b/junifer/markers/tests/test_parcel.py @@ -64,7 +64,7 @@ def test_ParcelAggregation_3D(): meta = marker.get_meta('VBM_GM')['marker'] assert meta['method'] == 'mean' assert meta['atlas'] == 'Schaefer100x7' - assert meta['name'] == 'VBM_GM_gmd_schaefer100x7_mean' + assert meta['name'] == 'gmd_schaefer100x7_mean' assert meta['class'] == 'ParcelAggregation' assert meta['kind'] == 'VBM_GM' assert meta['method_params'] == {} @@ -88,7 +88,7 @@ def test_ParcelAggregation_3D(): meta = marker.get_meta('VBM_GM')['marker'] assert meta['method'] == 'std' assert meta['atlas'] == 'Schaefer100x7' - assert meta['name'] == 'VBM_GM_ParcelAggregation' + assert meta['name'] == 'ParcelAggregation' assert meta['class'] == 'ParcelAggregation' assert meta['kind'] == 'VBM_GM' assert meta['method_params'] == {} @@ -97,8 +97,7 @@ def test_ParcelAggregation_3D(): manual = [] for t_v in sorted(np.unique(atlas_values)): t_values = trim_mean( - data[:, atlas_values == t_v], proportiontocut=0.1, - axis=None) # type: ignore + data[:, atlas_values == t_v], proportiontocut=0.1, axis=None) manual.append(t_values) manual = np.array(manual)[np.newaxis, :] @@ -116,7 +115,7 @@ def test_ParcelAggregation_3D(): meta = marker.get_meta('VBM_GM')['marker'] assert meta['method'] == 'trim_mean' assert meta['atlas'] == 'Schaefer100x7' - assert meta['name'] == 'VBM_GM_ParcelAggregation' + assert meta['name'] == 'ParcelAggregation' assert meta['class'] == 'ParcelAggregation' assert meta['kind'] == 'VBM_GM' assert meta['method_params'] == {'proportiontocut': 0.1} @@ -147,7 +146,7 @@ def test_ParcelAggregation_4D(): meta = marker.get_meta('BOLD')['marker'] assert meta['method'] == 'mean' assert meta['atlas'] == 'Schaefer100x7' - assert meta['name'] == 'BOLD_ParcelAggregation' + assert meta['name'] == 'ParcelAggregation' assert meta['class'] == 'ParcelAggregation' assert meta['kind'] == 'BOLD' assert meta['method_params'] == {} diff --git a/junifer/preprocess/confounds.py b/junifer/preprocess/confounds.py deleted file mode 100644 index 0d56f818a..000000000 --- a/junifer/preprocess/confounds.py +++ /dev/null @@ -1,423 +0,0 @@ -# Authors: Federico Raimondo -# Leonard Sasse -# License: AGPL - -import numpy as np -import pandas as pd - -from nilearn._utils.niimg_conversions import check_niimg_4d -from nilearn.masking import compute_brain_mask -from nilearn.image import clean_img - -from junifer.markers.base import PipelineStepMixin -from ..utils.logging import logger, raise_error - - -class BaseConfoundRemover(PipelineStepMixin): - """ A base class to read confound files and select columns according to - a pre-defined strategy - - """ - - # lower priority - # TODO: implement more strategies from - # nilearn.interfaces.fmriprep.load_confounds for Felix's confound files, - # in particular scrubbing - # TODO: Implement read_confounds for fmriprep data - - def __init__(self, strategy=None, spike=None, detrend=True, - standardize=True, low_pass=None, high_pass=None, t_r=None, - mask_img=None): - """ Initialise a BaseConfoundReader object - - Confound removal is based on nilearn.image.clean_img - - Parameters - ----------- - strategy : dict[str -> str] - The keys of the dictionary should correspond to names of noise - components to include: - - 'motion' - - 'wm_csf' - - 'global_signal' - The values of dictionary should correspond to types of confounds - extracted from each signal: - - 'basic': only the confounding time series - - 'power2': signal + quadratic term - - 'derivatives': signal + derivatives - - 'full': signal + deriv. + quadratic terms + power2 deriv. - spike : float | None (default) - If None, no spike regressor is added. If spike is a float, it will - add a spike regressor for every point at which FD exceeds the - specified float. - detrend : bool - If True (default), detrending will be applied on timeseries - (before confound removal). - standardize : bool - If True (default), returned signals are set to unit variance. - low_pass : float | None (default) - Low cutoff frequencies, in Hertz. If None, no filtering is applied. - high_pass : float | None (default) - High cutoff frequencies, in Hertz. If None, no filtering is - applied. - t_r : float - Repetition time, in second (sampling period). - If None (default) it will use t_r from nifti header. - mask_img: Niimg-like object - If provided, signal is only cleaned from voxels inside the mask. - If mask is provided, it should have same shape and affine as imgs. - If not provided, a mask is computed using - nilearn.masking.compute_brain_mask - """ - if strategy is None: - strategy = { - 'motion': 'full', - 'wm_csf': 'full', - 'global_signal': 'full' - } - - self.strategy = strategy - self.spike = spike - self.detrend = detrend - self.standardize = standardize - self.low_pass = low_pass - self.high_pass = high_pass - self.t_r = t_r - self.mask_img = mask_img - - self._valid_components = ['motion', 'wm_csf', 'global_signal'] - self._valid_confounds = ['basic', 'power2', 'derivatives', 'full'] - - if any(not isinstance(k, str) for k in strategy.keys()): - raise_error( - 'Strategy keys must be strings', ValueError - ) - - if any(not isinstance(v, str) for v in strategy.values()): - raise_error( - 'Strategy values must be strings', ValueError - ) - - if any(x not in self._valid_components for x in strategy.keys()): - raise_error( - f'Invalid component names {list(strategy.keys())}. ' - f'Valid components are {self._valid_components}.\n' - f'If any of them is a valid parameter in ' - 'nilearn.interfaces.fmriprep.load_confounds we may ' - 'include it in the future', ValueError - ) - - if any(x not in self._valid_confounds for x in strategy.values()): - raise_error( - f'Invalid component names {list(strategy.values())}. ' - f'Valid confound types are {self._valid_confounds}.\n' - f'If any of them is a valid parameter in ' - 'nilearn.interfaces.fmriprep.load_confounds we may ' - 'include it in the future', ValueError - ) - - def validate_input(self, input): - """Validate the input to the pipeline step. - - Parameters - ---------- - input : list[str] - The input to the pipeline step. The list must contain the - available Junifer Data dictionary keys. - - Raises - ------ - ValueError: - If the input does not have the required data. - """ - _required_inputs = ['BOLD', 'confounds'] - if any(x not in input for x in _required_inputs): - raise_error( - 'Input does not have the required data. \n' - f'Input: {input} \n' - f'Required (all off): {_required_inputs} \n', ValueError - ) - - def get_output_kind(self, input): - """Get the kind of the pipeline step. - - Parameters - ---------- - input : list[str] - The input to the pipeline step. The list must contain the - available Junifer Data dictionary keys. - - Returns - ------- - output : list[str] - The updated list of available Junifer Data dictionary keys after - the pipeline step. - """ - # Does not add any new keys - return input - - def _pick_confounds(self, input): - """ Select relevant confounds from the specified file """ - to_select = [] - confounds_df = input['data'] - confounds_spec = input['names']['spec'] - derivatives_to_compute = input['names'].get('derivatives', {}) - spike_name = input['names']['spike'] - - # Get all the column names according to the strategy - for comp, param in self.strategy.items(): - to_select.extend(confounds_spec[comp][param]) - - # Add derivatives if needed - to_compute = [x in derivatives_to_compute.keys() for x in to_select] - out_df = confounds_df.copy() - if any(to_compute): - for t_dst, t_src in derivatives_to_compute.items(): - out_df[t_dst] = np.append( # type: ignore - np.diff(out_df[t_src]), 0) # type: ignore - out_df = out_df[to_select] - # add binary spike regressor if needed at given threshold - if self.spike is not None: - fd = confounds_df[spike_name].copy() - fd.loc[fd > self.spike] = 1 - fd.loc[fd != 1] = 0 - out_df['spike'] = fd - - return out_df - - def _remove_confounds(self, bold_img, confounds_df): - """ Remove confounds from the BOLD image - - bold_img : Niimg-like object - 4D image. The signals in the last dimension are filtered - (see http://nilearn.github.io/manipulating_images/input_output.html - for a detailed description of the valid input types). - confounds_df : pd.DataFrame - Dataframe containing confounds to remove. Number of rows should - correspond to number of volumes in the BOLD image. - - returns - -------- - clean_bold : Niimg-like object - input image, cleaned. - - """ - - confounds_array = confounds_df.values - - t_r = self.t_r - if t_r is None: - logger.info('No t_r specified, using t_r from nifti header!') - zooms = bold_img.header.get_zooms() - t_r = zooms[3] - logger.info(f'Read t_r from nifti header: {t_r}', ) - - mask_img = self.mask_img - if mask_img is None: - logger.info('Computing brain mask from image') - mask_img = compute_brain_mask(bold_img) - - clean_bold = clean_img( - imgs=bold_img, - detrend=self.detrend, - standardize=self.standardize, - confounds=confounds_array, - low_pass=self.low_pass, - high_pass=self.high_pass, - t_r=t_r, - mask_img=mask_img - ) - - return clean_bold - - def _validate_data(self, input): - # Bold must be 4D niimg - check_niimg_4d(input['BOLD']['data']) - - # Confounds must be a dataframe - if not isinstance(input['confounds']['data'], pd.DataFrame): - raise_error( - 'confounds data must be a pandas dataframe', ValueError - ) - - confound_df = input['confounds']['data'] - bold_img = input['BOLD']['data'] - if bold_img.get_fdata().shape[3] != len(confound_df): - raise_error( - 'Image time series and confounds have different length!\n' - f'\tImage time series: { bold_img.get_fdata().shape[3]}\n' - f'\tConfounds: {len(confound_df)}') - - # Check the column names of the dataframe and the spec - # spec must be a dictionary: - # { - # 'motion': { - # 'basic': [(list of columns)] - # 'power2', [(list of columns)] - # 'derivatives', [(list of columns)] - # 'full', [(list of columns)]}, - # 'wm_csf': { - # 'basic': [(list of columns)] - # 'power2', [(list of columns)] - # 'derivatives', [(list of columns)] - # 'full', [(list of columns)]} - # 'global_signal': { - # 'basic': [(list of columns)] - # 'power2', [(list of columns)] - # 'derivatives', [(list of columns)] - # 'full', [(list of columns)]} - # } - - # Check the columns in the dataframe - conf_spec = input['confounds']['names']['spec'] - if any(x not in conf_spec.keys() for x in self._valid_components): - raise_error( - 'All of the component types must be in the confounds data ' - 'object `spec`. Please check your datagrabber.', ValueError) - - if any(x not in v.keys() for x in self._valid_confounds - for v in conf_spec.values()): - raise_error( - 'All of the confound types must be in the confounds data ' - 'object `spec`. Please check your datagrabber.', ValueError) - - spike_name = input['confounds']['names']['spike'] - - derivatives_to_compute = input['confounds']['names'].get( - 'derivatives', {}) - if not(isinstance(derivatives_to_compute, dict)): - raise_error( - 'input["confounds"]["names"]["derivatives"] ' - 'must be a dictionary. Please check your datagrabber', - ValueError) - - if any(not (isinstance(k, str) or isinstance(v, str)) - for k, v in derivatives_to_compute.items()): - raise_error( - 'input["confounds"]["names"]["derivatives"] ' - 'must be a dictionary with string keys and values. ' - 'Please check your datagrabber', - ValueError) - - missing_derivatives = [ - x for x in derivatives_to_compute.values() - if x not in confound_df.columns] - if len(missing_derivatives) > 0: - raise_error( - 'Some of the derivatives to calculate are not in the confounds' - f' dataframe: {missing_derivatives}.' - 'Please check your data ' - f'({input["confounds"]["path"].as_posix()}) ' - 'and the datagrabber.', ValueError) - - t_conf_spec = {k: input['confounds']['names']['spec'][k][v] - for k, v in self.strategy.items()} - - column_names = set([x for y in t_conf_spec.values() for x in y]) - column_names.add(spike_name) - - missing_columns = [ - x for x in column_names - if x not in confound_df.columns and - x not in derivatives_to_compute.keys()] - - if len(missing_columns) > 0: - raise_error( - 'Some of the columns in the confound spec are not in the ' - f'confounds dataframe: {missing_columns}. ' - 'Please check your data ' - f'({input["confounds"]["path"].as_posix()}) ' - 'and the datagrabber.', ValueError) - - def fit_transform(self, input): - self._validate_data(input) - bold_img = input['BOLD']['data'] - confounds_df = self._pick_confounds(input['confounds']) - input['BOLD']['data'] = self._remove_confounds(bold_img, confounds_df) - - # TODO: Update meta - return input - - -# class FelixConfoundRemover(BaseConfoundRemover): -# """ A class to read confounds from confound files generated by Felix's -# pipeline using CAT and some other custom scripts. It is meant to emulate -# the new nilearn.interfaces.fmriprep.load_confounds as closely as -# possible. - -# """ - -# def read_confounds(self, confound_dataframe): - -# confounds_to_select = [] -# # for some confounds we need to manually calculate derivatives using -# # numpy and add them to the output -# derivatives_to_compute = [] - -# for comp, param in self.strategy.items(): -# if comp == 'motion': -# confounds = [] -# # there should be six rigid body parameters -# for i in range(1, 7): -# # select basic -# confounds_to_select.append(f'RP.{i}') - -# # select squares -# if param in ['power2', 'full']: -# confounds_to_select.append(f'RP^2.{i}') - -# # select derivatives -# if param in ['derivatives', 'full']: -# confounds_to_select.append(f'DRP.{i}') - -# # if 'full' we should not forget the derivative -# # of the squares -# if param in ['full']: -# confounds_to_select.append(f'DRP^2.{i}') - -# elif comp == 'wm_csf': -# confounds = ['WM', 'CSF'] -# elif comp == 'global_signal': -# confounds = ['GS'] - -# for conf in confounds: - -# confounds_to_select.append(conf) - -# # select squares -# if param in ['power2', 'full']: -# confounds_to_select.append(f'{conf}^2') - -# # we have to calculate derivatives (not included in felix' -# # confound files) -# if param in ['derivatives', 'full']: -# derivatives_to_compute.append(conf) - -# if param in ['full']: -# derivatives_to_compute.append(f'{conf}^2') - -# confounds_to_remove = confound_dataframe[confounds_to_select] - -# # calc additional derivatives -# for conf in derivatives_to_compute: -# confounds_to_remove[f'D{conf}'] = np.append( -# np.diff(confound_dataframe[conf]), 0 -# ) - -# # add binary spike regressor if needed at given threshold -# if self.spike is not None: -# fd = confound_dataframe["FD"].copy() -# fd.loc[fd > self.spike] = 1 -# fd.loc[fd != 1] = 0 -# confounds_to_remove['spike'] = fd - -# return confounds_to_remove - - -# class FmriprepConfoundRemover(BaseConfoundRemover): -# """ A ConfoundRemover class for fmriprep output utilising -# nilearn's nilearn.interfaces.fmriprep.load_confounds -# """ - -# def read_confounds(self): -# raise NotImplementedError('read_confounds not implemented') diff --git a/junifer/preprocess/tests/test_confounds.py b/junifer/preprocess/tests/test_confounds.py deleted file mode 100644 index ede77c861..000000000 --- a/junifer/preprocess/tests/test_confounds.py +++ /dev/null @@ -1,340 +0,0 @@ -# Authors: Federico Raimondo -# Leonard Sasse -# License: AGPL - -from pathlib import Path -import numpy as np -import pandas as pd -from nibabel import Nifti1Image -from nilearn._utils.niimg_conversions import check_niimg_4d -import random -import string -from junifer.preprocess.confounds import BaseConfoundRemover - -np.random.seed(1234567) - - -def generate_conf_name(size=6, chars=string.ascii_uppercase + string.digits): - return ''.join(random.choice(chars) for _ in range(size)) - - -def _simu_img(): - - # Random 4D volume with 100 time points - vol = 100 + 10 * np.random.randn(5, 5, 2, 100) - img = Nifti1Image(vol, np.eye(4)) - # Create an nifti image with the data, and corresponding mask - mask = Nifti1Image(np.ones([5, 5, 2]), np.eye(4)) - return img, mask - - -def test_baseconfoundremover(): - # Generate a simulated BOLD img - siimg, simsk = _simu_img() - - # generate random confound dataframe with Felix's column naming - - motion_basic = [f'RP.{i}' for i in range(1, 7)] - motion_power2 = [f'RP^2.{i}' for i in range(1, 7)] - motion_derivatives = [f'DRP.{i}' for i in range(1, 7)] - motion_full = [f'DRP^2.{i}' for i in range(1, 7)] - - wm_csf_basic = ['WM', 'CSF'] - wm_csf_power2 = ['WM^2', 'CSF^2'] - wm_csf_derivatives = ['DWM', 'DCSF'] - wm_csf_full = ['DWM^2', 'DCSF^2'] - - gs_basic = ['GS'] - gs_power2 = ['GS^2'] - gs_derivatives = ['DGS'] - gs_full = ['DGS^2'] - - confound_column_names = [] - - confound_column_names.append('FD') # spike - - confound_column_names.extend(motion_basic) - confound_column_names.extend(motion_power2) - confound_column_names.extend(motion_derivatives) - confound_column_names.extend(motion_full) - - confound_column_names.extend(wm_csf_basic) - confound_column_names.extend(wm_csf_power2) - confound_column_names.extend(wm_csf_derivatives) - confound_column_names.extend(wm_csf_full) - - confound_column_names.extend(gs_basic) - confound_column_names.extend(gs_power2) - confound_column_names.extend(gs_derivatives) - confound_column_names.extend(gs_full) - - # add some random irrelevant confounds - for i in range(10): - confound_column_names.append(generate_conf_name()) - - np.random.shuffle(confound_column_names) - n_cols = len(confound_column_names) - confounds_df = pd.DataFrame( - np.random.randint(0, 100, size=(100, n_cols)), - columns=confound_column_names - ) - - # Generate spec from Felix's column naming - spec = { - 'motion': { - 'basic': motion_basic, - 'power2': motion_basic + motion_power2, - 'derivatives': motion_basic + motion_derivatives, - 'full': - motion_basic + motion_derivatives + motion_power2 + motion_full - }, - 'wm_csf': { - 'basic': wm_csf_basic, - 'power2': wm_csf_basic + wm_csf_power2, - 'derivatives': wm_csf_basic + wm_csf_derivatives, - 'full': - wm_csf_basic + wm_csf_derivatives + wm_csf_power2 + wm_csf_full - }, - 'global_signal': { - 'basic': gs_basic, - 'power2': gs_basic + gs_power2, - 'derivatives': gs_basic + gs_derivatives, - 'full': gs_basic + gs_derivatives + gs_power2 + gs_full - } - } - - # generate a junifer pipeline data object dictionary - input_data_obj = {} - input_data_obj['meta'] = {} - input_data_obj['BOLD'] = {} - input_data_obj['BOLD']['data'] = siimg - input_data_obj['confounds'] = {} - input_data_obj['confounds']['path'] = Path('/test.df') - input_data_obj['confounds']['data'] = confounds_df - input_data_obj['confounds']['names'] = {} - input_data_obj['confounds']['names']['spec'] = spec - input_data_obj['confounds']['names']['spike'] = 'FD' - - # generate confound removal strategies with varying numbers of parameters - - # Test #1: 36 params, no derivatives to compute, no spike - # 36 params - strat1 = { - 'motion': 'full', - 'wm_csf': 'full', - 'global_signal': 'full' - } - - cr = BaseConfoundRemover( - strategy=strat1, spike=None, mask_img=simsk, t_r=0.75 - ) - cr.validate_input(input_data_obj.keys()) - out_type = cr.get_output_kind(input_data_obj.keys()) - - assert 'BOLD' in out_type - - # Check if the input data is valid - cr._validate_data(input_data_obj) - - # Check that the confounds are picked correctly: - t_df = cr._pick_confounds(input_data_obj['confounds']) - assert len(t_df.columns) == 36 - assert all(x in t_df.columns for x in motion_basic) - assert all(x in t_df.columns for x in motion_power2) - assert all(x in t_df.columns for x in motion_derivatives) - assert all(x in t_df.columns for x in motion_full) - assert all(x in t_df.columns for x in wm_csf_basic) - assert all(x in t_df.columns for x in wm_csf_power2) - assert all(x in t_df.columns for x in wm_csf_derivatives) - assert all(x in t_df.columns for x in wm_csf_full) - assert all(x in t_df.columns for x in gs_basic) - assert all(x in t_df.columns for x in gs_power2) - assert all(x in t_df.columns for x in gs_derivatives) - assert all(x in t_df.columns for x in gs_full) - assert 'FD' not in t_df.columns - assert 'spike' not in t_df.columns - - # Test #2: 24 params, no derivatives to compute, no spike - # 24 params - strat2 = { - 'motion': 'full', - } - - cr = BaseConfoundRemover( - strategy=strat2, spike=None, mask_img=simsk, t_r=0.75 - ) - cr.validate_input(input_data_obj.keys()) - out_type = cr.get_output_kind(input_data_obj.keys()) - - assert 'BOLD' in out_type - - # Check if the input data is valid - cr._validate_data(input_data_obj) - - # Check that the confounds are picked correctly: - t_df = cr._pick_confounds(input_data_obj['confounds']) - assert len(t_df.columns) == 24 - assert all(x in t_df.columns for x in motion_basic) - assert all(x in t_df.columns for x in motion_power2) - assert all(x in t_df.columns for x in motion_derivatives) - assert all(x in t_df.columns for x in motion_full) - assert all(x not in t_df.columns for x in wm_csf_basic) - assert all(x not in t_df.columns for x in wm_csf_power2) - assert all(x not in t_df.columns for x in wm_csf_derivatives) - assert all(x not in t_df.columns for x in wm_csf_full) - assert all(x not in t_df.columns for x in gs_basic) - assert all(x not in t_df.columns for x in gs_power2) - assert all(x not in t_df.columns for x in gs_derivatives) - assert all(x not in t_df.columns for x in gs_full) - assert 'FD' not in t_df.columns - assert 'spike' not in t_df.columns - - # Test #3: 9 params, no derivatives to compute, no spike - strat3 = { - 'motion': 'basic', - 'wm_csf': 'basic', - 'global_signal': 'basic' - } - - cr = BaseConfoundRemover( - strategy=strat3, spike=None, mask_img=simsk, t_r=0.75 - ) - cr.validate_input(input_data_obj.keys()) - out_type = cr.get_output_kind(input_data_obj.keys()) - - assert 'BOLD' in out_type - - # Check if the input data is valid - cr._validate_data(input_data_obj) - - # Check that the confounds are picked correctly: - t_df = cr._pick_confounds(input_data_obj['confounds']) - assert len(t_df.columns) == 9 - assert all(x in t_df.columns for x in motion_basic) - assert all(x not in t_df.columns for x in motion_power2) - assert all(x not in t_df.columns for x in motion_derivatives) - assert all(x not in t_df.columns for x in motion_full) - assert all(x in t_df.columns for x in wm_csf_basic) - assert all(x not in t_df.columns for x in wm_csf_power2) - assert all(x not in t_df.columns for x in wm_csf_derivatives) - assert all(x not in t_df.columns for x in wm_csf_full) - assert all(x in t_df.columns for x in gs_basic) - assert all(x not in t_df.columns for x in gs_power2) - assert all(x not in t_df.columns for x in gs_derivatives) - assert all(x not in t_df.columns for x in gs_full) - assert 'FD' not in t_df.columns - assert 'spike' not in t_df.columns - - # Test #4: 6 params, no derivatives to compute, no spike - strat4 = { - 'motion': 'basic', - } - cr = BaseConfoundRemover( - strategy=strat4, spike=None, mask_img=simsk, t_r=0.75 - ) - cr.validate_input(input_data_obj.keys()) - out_type = cr.get_output_kind(input_data_obj.keys()) - - assert 'BOLD' in out_type - - # Check if the input data is valid - cr._validate_data(input_data_obj) - - # Check that the confounds are picked correctly: - t_df = cr._pick_confounds(input_data_obj['confounds']) - assert len(t_df.columns) == 6 - assert all(x in t_df.columns for x in motion_basic) - assert all(x not in t_df.columns for x in motion_power2) - assert all(x not in t_df.columns for x in motion_derivatives) - assert all(x not in t_df.columns for x in motion_full) - assert all(x not in t_df.columns for x in wm_csf_basic) - assert all(x not in t_df.columns for x in wm_csf_power2) - assert all(x not in t_df.columns for x in wm_csf_derivatives) - assert all(x not in t_df.columns for x in wm_csf_full) - assert all(x not in t_df.columns for x in gs_basic) - assert all(x not in t_df.columns for x in gs_power2) - assert all(x not in t_df.columns for x in gs_derivatives) - assert all(x not in t_df.columns for x in gs_full) - assert 'FD' not in t_df.columns - assert 'spike' not in t_df.columns - - # Test #5: 2 params, no derivatives to compute, no spike - strat5 = { - 'wm_csf': 'basic', - } - cr = BaseConfoundRemover( - strategy=strat5, spike=None, mask_img=simsk, t_r=0.75 - ) - cr.validate_input(input_data_obj.keys()) - out_type = cr.get_output_kind(input_data_obj.keys()) - - assert 'BOLD' in out_type - - # Check if the input data is valid - cr._validate_data(input_data_obj) - - # Check that the confounds are picked correctly: - t_df = cr._pick_confounds(input_data_obj['confounds']) - assert len(t_df.columns) == 2 - assert all(x not in t_df.columns for x in motion_basic) - assert all(x not in t_df.columns for x in motion_power2) - assert all(x not in t_df.columns for x in motion_derivatives) - assert all(x not in t_df.columns for x in motion_full) - assert all(x in t_df.columns for x in wm_csf_basic) - assert all(x not in t_df.columns for x in wm_csf_power2) - assert all(x not in t_df.columns for x in wm_csf_derivatives) - assert all(x not in t_df.columns for x in wm_csf_full) - assert all(x not in t_df.columns for x in gs_basic) - assert all(x not in t_df.columns for x in gs_power2) - assert all(x not in t_df.columns for x in gs_derivatives) - assert all(x not in t_df.columns for x in gs_full) - assert 'FD' not in t_df.columns - assert 'spike' not in t_df.columns - - out = cr.fit_transform(input_data_obj) - check_niimg_4d(out['BOLD']['data']) - # TODO: check meta - - # Test #6: 12 params, derivatives to compute, spike - to_select = [x for x in confounds_df.columns - if x not in motion_derivatives] - no_d_df = confounds_df[to_select] - input_data_obj['confounds']['data'] = no_d_df - - derivatives = { - f'D{x}': x for x in motion_basic - } - - input_data_obj['confounds']['names']['derivatives'] = derivatives - - strat6 = { - 'motion': 'derivatives', - } - cr = BaseConfoundRemover( - strategy=strat6, spike=0.75, mask_img=simsk, t_r=0.75 - ) - cr.validate_input(input_data_obj.keys()) - out_type = cr.get_output_kind(input_data_obj.keys()) - - assert 'BOLD' in out_type - - # Check if the input data is valid - cr._validate_data(input_data_obj) - - # Check that the confounds are picked correctly: - t_df = cr._pick_confounds(input_data_obj['confounds']) - assert len(t_df.columns) == 13 - assert all(x in t_df.columns for x in motion_basic) - assert all(x not in t_df.columns for x in motion_power2) - assert all(x in t_df.columns for x in motion_derivatives) - assert all(x not in t_df.columns for x in motion_full) - assert all(x not in t_df.columns for x in wm_csf_basic) - assert all(x not in t_df.columns for x in wm_csf_power2) - assert all(x not in t_df.columns for x in wm_csf_derivatives) - assert all(x not in t_df.columns for x in wm_csf_full) - assert all(x not in t_df.columns for x in gs_basic) - assert all(x not in t_df.columns for x in gs_power2) - assert all(x not in t_df.columns for x in gs_derivatives) - assert all(x not in t_df.columns for x in gs_full) - assert 'FD' not in t_df.columns - assert 'spike' in t_df.columns diff --git a/junifer/storage/base.py b/junifer/storage/base.py index dbdf0f204..c9a326d6e 100644 --- a/junifer/storage/base.py +++ b/junifer/storage/base.py @@ -10,7 +10,7 @@ from ..utils import logger from .. import __version__ -def process_meta(meta): +def process_meta(meta, return_idx=False, n_rows=1, rows_col_name=None): """Process the metadata for storage. It removes the "element" key and adds the "_element_keys" with the keys used to index the element. @@ -38,6 +38,10 @@ def process_meta(meta): if meta is None: raise ValueError('Meta must be a dict (currently is None)') t_meta = meta.copy() + idx = None + if return_idx is True: + idx = _element_to_index( + meta, n_rows=n_rows, rows_col_name=rows_col_name) element = t_meta.pop('element', None) if element is None: if '_element_keys' not in t_meta: @@ -48,8 +52,14 @@ def process_meta(meta): t_meta['_element_keys'] = list(element.keys()) else: t_meta['_element_keys'] = ['element'] + logger.debug(f'Hasing meta {t_meta}') md5_hash = _meta_hash(t_meta) - return md5_hash, t_meta + logger.debug(f'Hash computed: {md5_hash}') + if return_idx is True: + out = md5_hash, t_meta, idx + else: + out = md5_hash, t_meta + return out def _meta_hash(meta): @@ -65,14 +75,12 @@ def _meta_hash(meta): md5: str The md5 hash of the meta """ - logger.debug(f'Hashing meta {meta}') meta_md5 = hashlib.md5( json.dumps(meta, sort_keys=True).encode('utf-8')).hexdigest() - logger.debug(f'Hash computed: {meta_md5}') return meta_md5 -def element_to_index(meta, n_rows=1, rows_col_name=None): +def _element_to_index(meta, n_rows=1, rows_col_name=None): """Convert the element meta to index Parameters @@ -80,7 +88,7 @@ def element_to_index(meta, n_rows=1, rows_col_name=None): meta: dict The metadata. Must contain the key 'element' n_rows: int - Number of rows to create. Defaults to 1. + Number of rows to create rows_col_name: str The column name to use in case n_rows > 1. If None (default) and n_rows > 1, the name will be 'index'. @@ -101,30 +109,32 @@ def element_to_index(meta, n_rows=1, rows_col_name=None): element = meta['element'] if not isinstance(element, dict): element = dict(element=element) - if rows_col_name is None: - rows_col_name = 'idx' - elem_idx = { - k: [v] * n_rows for k, v in element.items() - } - elem_idx[rows_col_name] = np.arange(n_rows) # type: ignore + if n_rows > 1: + if rows_col_name is None: + rows_col_name = 'index' + elem_idx = { + k: [v] * n_rows for k, v in element.items() + } + elem_idx[rows_col_name] = np.arange(n_rows) # type: ignore + else: + elem_idx = element index = pd.MultiIndex.from_frame( pd.DataFrame(elem_idx, index=range(n_rows))) + return index def element_to_prefix(element): - logger.debug(f'Converting element {element} to prefix') prefix = 'element' - if isinstance(element, tuple): - prefix = f"{prefix}_{'_'.join([f'{x}' for x in element])}" - elif isinstance(element, dict): - prefix = f"{prefix}_{'_'.join([f'{x}' for x in element.values()])}" - elif isinstance(element, (str, int)): + if isinstance(element, str): prefix = f'{prefix}_{element}' + elif isinstance(element, tuple): + prefix = f"{prefix}_{'_'.join(element)}" + elif isinstance(element, dict): + prefix = f"{prefix}_{'_'.join(element.values())}" else: raise ValueError(f'Cannot convert element {element} to prefix. ' 'Must be a str, tuple or dict') - logger.debug(f'Converted prefix {prefix}') return f'{prefix}_' @@ -161,19 +171,15 @@ class BaseFeatureStorage(ABC): raise NotImplementedError('validate_input not implemented') @abstractmethod - def list_features(self, return_df=False): + def list_features(self): """List the features in the storage - Parameters - ---------- - return_df : bool - If True, return a dataframe. If False, (default) return a - dictionary + Returns ------- - features: dict(str, dict) | pd.DataFrame - List of features in the storage. If dictionarly, the keys are the - feature names to be used in read_features. The values are the - metadata of each feature. + features: dict(str, dict) + List of features in the storage. The keys are the feature names + to be used in read_features. The values are the metadata of each + feature """ raise NotImplementedError('list_features not implemented') @@ -208,10 +214,6 @@ class BaseFeatureStorage(ABC): def store_timeseries(self, data, meta): raise NotImplementedError('store_timeseries not implemented') - @abstractmethod - def collect(self): - raise NotImplementedError('collect not implemented') - class PandasFeatureStoreage(BaseFeatureStorage): diff --git a/junifer/storage/sqlite.py b/junifer/storage/sqlite.py index 82d5b0590..d872d9978 100644 --- a/junifer/storage/sqlite.py +++ b/junifer/storage/sqlite.py @@ -7,9 +7,8 @@ from pandas.io.sql import pandasSQL_builder from sqlalchemy import create_engine, inspect from ..api.decorators import register_storage -from .base import (PandasFeatureStoreage, process_meta, element_to_prefix, - element_to_index) -from ..utils.logging import warn, logger +from .base import PandasFeatureStoreage, process_meta, element_to_prefix +from ..utils.logging import warn @register_storage @@ -23,15 +22,14 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): Parameters ---------- - uri : str or Path (must be a file) - The Path to the file to be used. - single_output : bool - If False (default), will create one file per element. The name - of the file will be prefixed with the respective element. - If True, will create only one file, the specified in the URI and - store all the elements in the same file. This behaviour is only - suitable for non-parallel executions. SQLite does not support - concurrency. + uri : str + The connection URI. + Easy options: + 'sqlite://' for an in memory sqlite database + 'sqlite:///' to save in a file + + Check https://docs.sqlalchemy.org/en/14/core/engines.html for more + options upsert : str Upsert mode. Options are 'ignore' and 'update' (default). If 'ignore', the existing elements are ignored. If update, the @@ -42,10 +40,6 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): raise ValueError('upsert must be either "update" or "ignore"') if not isinstance(uri, Path): uri = Path(uri) - if not uri.parent.exists(): - logger.info(f'Output directory ({uri.parent.as_posix()}) ' - 'does not exist, creating') - uri.parent.mkdir(parents=True, exist_ok=True) super().__init__(uri, single_output=single_output) self._upsert = upsert self._valid_inputs = ['table', 'timeseries'] @@ -69,13 +63,10 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): uri = f'sqlite:///{self.uri.parent}/{prefix}{self.uri.name}' return create_engine(uri, echo=False) - def list_features(self, return_df=False): + def list_features(self): meta_df = pd.read_sql( 'meta', con=self.get_engine(), index_col='meta_md5') - out = meta_df - if return_df is False: - out = meta_df.to_dict(orient='index') - return out + return meta_df.to_dict(orient='index') def read_df(self, feature_name=None, feature_md5=None): """Read features from the storage. @@ -102,7 +93,7 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): raise ValueError('At least one of feature_name or feature_md5 ' 'must be specified') elif feature_md5 is not None: - table_name = f'meta_{feature_md5}' + feature_md5 = f'meta_{feature_md5}' else: meta_df = pd.read_sql( 'meta', con=engine, index_col='meta_md5') @@ -114,22 +105,14 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): f'More than one feature with name {feature_name} found', 'This file is invalid. You can bypass this issue by ' 'specifying a feature_md5') - table_name = f'meta_{t_df.index[0]}' - df = pd.read_sql(table_name, con=engine) - # Read the index: - query = ("SELECT ii.name FROM sqlite_master AS m, " - "pragma_index_list(m.name) AS il, " - "pragma_index_info(il.name) AS ii " - f"WHERE tbl_name='{table_name}' " - "ORDER BY cid;") - index_names = pd.read_sql(query, con=engine).values.squeeze().tolist() - df = df.set_index(index_names) - return df + feature_md5 = f'meta_{t_df.index[0]}' + return pd.read_sql(feature_md5, con=engine) def store_metadata(self, meta): t_meta = meta.copy() t_meta.update(self.get_meta()) - meta_md5, t_meta_row = process_meta(t_meta) + meta_md5, t_meta_row = process_meta( # type: ignore + t_meta, return_idx=False) engine = self.get_engine(t_meta) if meta_md5 not in inspect(engine).get_table_names(): meta_df = self._meta_row(t_meta_row, meta_md5) @@ -146,35 +129,22 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): def store_2d(self, data, meta, columns=None, rows_col_name=None): n_rows = len(data) - idx = element_to_index( - meta, n_rows=n_rows, rows_col_name=rows_col_name) + _, _, idx = process_meta( # type: ignore + meta, return_idx=True, n_rows=n_rows, rows_col_name=rows_col_name) data_df = pd.DataFrame(data, columns=columns, index=idx) self.store_df(data_df, meta) def store_df(self, df, meta): - # TODO: Test this function - # Check that the index generated by meta matches the one in - # the dataframe. - idx = element_to_index(meta) - # Given the meta, we might not know if there is an extra column added - # when storing a timeseries or 2d elements. We need to check if the - # extra element is only one. - extra = [x for x in df.index.names if x not in idx.names] - if len(extra) > 1: - raise ValueError( - 'The index of the dataframe has extra items that are not ' - 'in the index generated from the meta data.') - elif len(extra) == 1: - # The df has one extra index item, this should be the new name - # of the missing element in the index - idx = element_to_index(meta, rows_col_name=extra[0]) - - if any(x not in df.index.names for x in idx.names): - raise ValueError( - 'The index of the dataframe is missing index items that are ' - 'generated from the meta data.') - table_name = self.store_metadata(meta) + # Check that the index generated by meta is in the dataframe, at least + # for one row + _, _, idx = process_meta( # type: ignore + meta, return_idx=True, n_rows=1) + if any(x not in df.index.names for x in idx.names): # type: ignore + raise ValueError( + 'The index of the dataframe does not match the ' + 'index of the meta data. This happens when the element ' + 'is not part of the index.') # Save engine = self.get_engine(meta) self._save_upsert(df, table_name, engine) @@ -182,31 +152,6 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): def store_timeseries(self, data, meta): raise NotImplementedError('store_timeseries not implemented') - def collect(self): - if self.single_output is True: - raise ValueError('collect is not implemented for single output') - logger.info( - f'Collecting data from {self.uri.parent}/*{self.uri.name}') - - out_storage = SQLiteFeatureStorage( - uri=self.uri, single_output=True, upsert='ignore') - - for elem in self.uri.parent.glob(f'*{self.uri.name}'): - logger.debug(f'Reading from {elem.as_posix()}') - in_storage = SQLiteFeatureStorage(uri=elem, single_output=True) - in_engine = in_storage.get_engine() - # Open "meta" table - t_meta_df = pd.read_sql( - 'meta', con=in_engine, index_col='meta_md5') - out_storage._save_upsert(t_meta_df, 'meta') - for meta_md5 in t_meta_df.index: - logger.debug(f'Collecting feature {meta_md5}') - # TODO: Fix this, needs that read_feature sets the index - # properly - table_name = f'meta_{meta_md5}' - t_df = in_storage.read_df(feature_md5=meta_md5) - out_storage._save_upsert(t_df, table_name) - def _save_upsert(self, df, name, engine=None, if_exist='append'): """ Implementation of UPSERT functionality. @@ -218,7 +163,7 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): Name of the table to save if_exist : str If the table exists, the behavior is controlled by this parameter. - Options are 'append' (default) and 'fail'. If 'fail' and the table + Options are 'append' (default) and 'fail'. If 'fail' and th table exists, it will raise an error. If 'append', the data will be appended to the existing table (following the upsert mode). @@ -272,7 +217,7 @@ class SQLiteFeatureStorage(PandasFeatureStoreage): def _get_existing_pk(con, table_name, index_col): - pk_cols = ', '.join(index_col) + pk_cols = ','.join(index_col) query = f'SELECT {pk_cols} FROM {table_name};' pk_indb = pd.read_sql(query, con=con) return pk_indb diff --git a/junifer/storage/tests/test_base.py b/junifer/storage/tests/test_base.py index 532907250..01f2156f8 100644 --- a/junifer/storage/tests/test_base.py +++ b/junifer/storage/tests/test_base.py @@ -1,8 +1,7 @@ import pytest -from junifer.storage.base import (process_meta, element_to_index, - BaseFeatureStorage, - element_to_prefix) +from junifer.storage.base import (process_meta, _element_to_index, + BaseFeatureStorage) def test_process_meta_hash(): @@ -13,14 +12,14 @@ def test_process_meta_hash(): process_meta(meta) meta = {'element': 'foo', 'A': 1, 'B': [2, 3, 4, 5, 6]} - hash1, _ = process_meta(meta) + hash1, _ = process_meta(meta, return_idx=False) # type: ignore meta = {'element': 'foo', 'B': [2, 3, 4, 5, 6], 'A': 1} - hash2, _ = process_meta(meta) + hash2, _ = process_meta(meta, return_idx=False) # type: ignore assert hash1 == hash2 meta = {'element': 'foo', 'A': 1, 'B': [2, 3, 1, 5, 6]} - hash3, _ = process_meta(meta) + hash3, _ = process_meta(meta, return_idx=False) # type: ignore assert hash1 != hash3 meta1 = { @@ -42,8 +41,8 @@ def test_process_meta_hash(): 'element': 'foo' } - hash4, _ = process_meta(meta1) - hash5, _ = process_meta(meta2) + hash4, _ = process_meta(meta1, return_idx=False) # type: ignore + hash5, _ = process_meta(meta2, return_idx=False) # type: ignore assert hash4 == hash5 @@ -52,10 +51,10 @@ def test_process_meta_element(): meta = {} with pytest.raises(ValueError, match=r"_element_keys"): - process_meta(meta) + process_meta(meta, return_idx=False) meta = {'element': 'foo', 'A': 1, 'B': [2, 3, 4, 5, 6]} - _, new_meta = process_meta(meta) + _, new_meta = process_meta(meta, return_idx=False) # type: ignore assert '_element_keys' in new_meta assert new_meta['_element_keys'] == ['element'] assert 'A' in new_meta @@ -64,7 +63,7 @@ def test_process_meta_element(): meta = { 'element': {'subject': 'foo', 'session': 'bar'}, 'B': [2, 3, 4, 5, 6], 'A': 1} - _, new_meta = process_meta(meta) + _, new_meta = process_meta(meta, return_idx=False) # type: ignore assert '_element_keys' in new_meta assert new_meta['_element_keys'] == ['subject', 'session'] assert 'A' in new_meta @@ -72,44 +71,35 @@ def test_process_meta_element(): def test_process_meta_index(): - """Test element_to_index""" + """Test _element_to_index""" meta = {'noelement': 'foo'} with pytest.raises(ValueError, match=r'meta must contain the key'): - element_to_index(meta) + _element_to_index(meta) meta = {'element': 'foo', 'A': 1, 'B': [2, 3, 4, 5, 6]} - index = element_to_index(meta) - assert index.names == ['element', 'idx'] + index = _element_to_index(meta) + assert index.names == ['element'] assert index.levels[0].name == 'element' assert index.levels[0].values[0] == 'foo' - assert all(x == 'foo' for x in index.levels[0].values) - assert index.levels[0].values.shape == (1,) - assert index.levels[1].name == 'idx' - assert all(x == i for i, x in enumerate(index.levels[1].values)) - assert index.levels[1].values.shape == (1,) - index = element_to_index(meta, n_rows=10) - assert index.names == ['element', 'idx'] + index = _element_to_index(meta, n_rows=10) + assert index.names == ['element', 'index'] assert index.levels[0].name == 'element' assert all(x == 'foo' for x in index.levels[0].values) assert index.levels[0].values.shape == (1,) - assert index.levels[1].name == 'idx' + assert index.levels[1].name == 'index' assert all(x == i for i, x in enumerate(index.levels[1].values)) assert index.levels[1].values.shape == (10,) - index = element_to_index(meta, n_rows=1, rows_col_name='scan') - assert index.names == ['element', 'scan'] + index = _element_to_index(meta, n_rows=1, rows_col_name='scan') + assert index.names == ['element'] assert index.levels[0].name == 'element' - assert index.levels[0].values[0] == 'foo' assert all(x == 'foo' for x in index.levels[0].values) assert index.levels[0].values.shape == (1,) - assert index.levels[1].name == 'scan' - assert all(x == i for i, x in enumerate(index.levels[1].values)) - assert index.levels[1].values.shape == (1,) - index = element_to_index(meta, n_rows=7, rows_col_name='scan') + index = _element_to_index(meta, n_rows=7, rows_col_name='scan') assert index.names == ['element', 'scan'] assert index.levels[0].name == 'element' assert all(x == 'foo' for x in index.levels[0].values) @@ -122,7 +112,7 @@ def test_process_meta_index(): meta = { 'element': {'subject': 'sub-01', 'session': 'ses-01'}, 'A': 1, 'B': [2, 3, 4, 5, 6]} - index = element_to_index(meta, n_rows=10) + index = _element_to_index(meta, n_rows=10) assert index.levels[0].name == 'subject' assert all(x == 'sub-01' for x in index.levels[0].values) @@ -132,7 +122,7 @@ def test_process_meta_index(): assert all(x == 'ses-01' for x in index.levels[1].values) assert index.levels[1].values.shape == (1,) - assert index.levels[2].name == 'idx' + assert index.levels[2].name == 'index' assert all(x == i for i, x in enumerate(index.levels[2].values)) assert index.levels[2].values.shape == (10,) @@ -171,9 +161,6 @@ def test_BaseFeatureStorage(): def store_timeseries(self, timeseries, meta): super().store_timeseries(timeseries, meta) - def collect(self): - return super().collect() - st = MyFeatureStorage(uri='/tmp') assert st.single_output is False @@ -204,43 +191,4 @@ def test_BaseFeatureStorage(): with pytest.raises(NotImplementedError): st.store_timeseries(None, None) - with pytest.raises(NotImplementedError): - st.collect() - assert st.uri == '/tmp' - - -def test_element_to_prefix(): - """Test converting element to prefix (for file naming)""" - - element = 'sub-01' - prefix = element_to_prefix(element) - assert prefix == 'element_sub-01_' - - element = 1 - prefix = element_to_prefix(element) - assert prefix == 'element_1_' - - element = {'subject': 'sub-01'} - prefix = element_to_prefix(element) - assert prefix == 'element_sub-01_' - - element = {'subject': 1} - prefix = element_to_prefix(element) - assert prefix == 'element_1_' - - element = {'subject': 'sub-01', 'session': 'ses-02'} - prefix = element_to_prefix(element) - assert prefix == 'element_sub-01_ses-02_' - - element = {'subject': 1, 'session': 2} - prefix = element_to_prefix(element) - assert prefix == 'element_1_2_' - - element = ('sub-01', 'ses-02') - prefix = element_to_prefix(element) - assert prefix == 'element_sub-01_ses-02_' - - element = (1, 2) - prefix = element_to_prefix(element) - assert prefix == 'element_1_2_' diff --git a/junifer/storage/tests/test_sqlite.py b/junifer/storage/tests/test_sqlite.py index a4688497d..31abd542d 100644 --- a/junifer/storage/tests/test_sqlite.py +++ b/junifer/storage/tests/test_sqlite.py @@ -7,8 +7,7 @@ from sqlalchemy import create_engine import pytest from junifer.storage.sqlite import SQLiteFeatureStorage -from junifer.storage.base import (process_meta, element_to_prefix, - element_to_index) +from junifer.storage.base import process_meta, element_to_prefix df1 = pd.DataFrame({ @@ -58,18 +57,6 @@ def test_get_engine(): assert engine.url.drivername == 'sqlite' assert f'{engine.url.database}' == uri - storage = SQLiteFeatureStorage( - uri=uri, single_output=False, upsert='ignore') - with pytest.raises(ValueError, match='element must be specified'): - storage.get_engine() - - tocreate = Path(_tmpdir) / 'tocreate' - assert not tocreate.exists() - uri = f'{tocreate.as_posix()}/test.db' - storage = SQLiteFeatureStorage( - uri=uri, single_output=True, upsert='ignore') - assert tocreate.exists() - def test_store_metadata(): """Test store_metadata""" @@ -127,8 +114,7 @@ def test_upsert_ignore(): c_df1 = _read_sql(table_name, uri=uri, index_col=['element', 'pk2']) assert_frame_equal(df1, c_df1) - with pytest.warns(RuntimeWarning, match='are already present'): - storage.store_df(df2, meta) + storage.store_df(df2, meta) c_dfignore = _read_sql( table_name, uri=uri, index_col=['element', 'pk2']) @@ -174,14 +160,11 @@ def test_store_read_df(): to_store = df1[['col1', 'col2']] # Save to SQL - with pytest.raises(ValueError, match=r"missing index items"): + with pytest.raises(ValueError, match=r"index of the dataframe"): storage.store_df(to_store.set_index('col1'), meta) - to_store = df1.reset_index().set_index(['element', 'pk2', 'col1']) - with pytest.raises(ValueError, match=r"extra items"): - storage.store_df(to_store, meta) - - idx = element_to_index(meta, n_rows=len(to_store)) + _, _, idx = process_meta( # type: ignore + meta, return_idx=True, n_rows=len(to_store)) to_store = to_store.set_index(idx) storage.store_df(to_store, meta) @@ -207,7 +190,7 @@ def test_store_read_df(): read_df1 = storage.read_df(feature_md5=feature_md5) read_df2 = storage.read_df(feature_name='fcname') assert_frame_equal(read_df1, read_df2) - assert_frame_equal(read_df1, to_store) + assert_frame_equal(read_df1, to_store.reset_index()) def test_store_table(): @@ -223,8 +206,8 @@ def test_store_table(): [4, 40], [5, 50], ] - - idx = element_to_index(meta, n_rows=5, rows_col_name='scan') + _, _, idx = process_meta( # type: ignore + meta, return_idx=True, n_rows=5, rows_col_name='scan') df1 = pd.DataFrame(data, columns=['f1', 'f2'], index=idx) storage.store_table( @@ -243,7 +226,8 @@ def test_store_table(): [6, 600] ] - idx = element_to_index(meta, n_rows=6, rows_col_name='scan') + _, _, idx = process_meta( # type: ignore + meta, return_idx=True, n_rows=6, rows_col_name='scan') df2 = pd.DataFrame(data2, columns=['f1', 'f2'], index=idx) with pytest.warns(RuntimeWarning, match=r"Some rows"): @@ -277,16 +261,16 @@ def test_store_multiple_output(): data2 = data1 * 10 data3 = data1 * 20 - hash1, _ = process_meta(meta1) - idx1 = element_to_index(meta1, n_rows=5, rows_col_name='scan') + hash1, _, idx1 = process_meta( # type: ignore + meta1, return_idx=True, n_rows=5, rows_col_name='scan') df1 = pd.DataFrame(data1, columns=['f1', 'f2'], index=idx1) - hash2, _ = process_meta(meta2) - idx2 = element_to_index(meta2, n_rows=5, rows_col_name='scan') + hash2, _, idx2 = process_meta( # type: ignore + meta2, return_idx=True, n_rows=5, rows_col_name='scan') df2 = pd.DataFrame(data2, columns=['f1', 'f2'], index=idx2) - hash3, _ = process_meta(meta3) - idx3 = element_to_index(meta3, n_rows=5, rows_col_name='scan') + hash3, _, idx3 = process_meta( # type: ignore + meta3, return_idx=True, n_rows=5, rows_col_name='scan') df3 = pd.DataFrame(data3, columns=['f1', 'f2'], index=idx3) assert hash1 == hash2 @@ -326,66 +310,3 @@ def test_store_multiple_output(): assert_frame_equal(df1, cdf1) assert_frame_equal(df2, cdf2) assert_frame_equal(df3, cdf3) - - -def test_collect(): - meta1 = {'element': {'subject': 'test-01', 'session': 'ses-01'}, - 'version': '0.0.1', 'marker': {'name': 'fc'}} - meta2 = {'element': {'subject': 'test-02', 'session': 'ses-01'}, - 'version': '0.0.1', 'marker': {'name': 'fc'}} - meta3 = {'element': {'subject': 'test-01', 'session': 'ses-02'}, - 'version': '0.0.1', 'marker': {'name': 'fc'}} - with tempfile.TemporaryDirectory() as _tmpdir: - uri = Path(f'{_tmpdir}/test.db') - storage = SQLiteFeatureStorage(uri=uri) - - data1 = np.array([ - [1, 10], - [2, 20], - [3, 30], - [4, 40], - [5, 50], - ]) - - data2 = data1 * 10 - data3 = data1 * 20 - storage.store_table( - data1, meta1, columns=['f1', 'f2'], rows_col_name='scan') - - storage.store_table( - data2, meta2, columns=['f1', 'f2'], rows_col_name='scan') - - storage.store_table( - data3, meta3, columns=['f1', 'f2'], rows_col_name='scan') - - prefix1 = element_to_prefix(meta1['element']) - prefix2 = element_to_prefix(meta2['element']) - prefix3 = element_to_prefix(meta3['element']) - - uri1 = uri.parent / f'{prefix1}{uri.name}' - uri2 = uri.parent / f'{prefix2}{uri.name}' - uri3 = uri.parent / f'{prefix3}{uri.name}' - - assert uri1.exists() - assert uri2.exists() - assert uri3.exists() - - assert not uri.exists() - - storage.collect() - - assert uri.exists() - - cols = ['subject', 'session', 'scan'] - table_name = storage.store_metadata(meta1) - all_df = _read_sql(table_name, uri, index_col=cols) - - cdf1 = _read_sql(table_name, uri1, index_col=cols) - cdf2 = _read_sql(table_name, uri2, index_col=cols) - cdf3 = _read_sql(table_name, uri3, index_col=cols) - - all_cdf = pd.concat([cdf1, cdf2, cdf3]) - all_df.sort_index(level=cols, inplace=True) - all_cdf.sort_index(level=cols, inplace=True) - - assert_frame_equal(all_df, all_cdf) diff --git a/junifer/testing/datagrabbers.py b/junifer/testing.py similarity index 58% rename from junifer/testing/datagrabbers.py rename to junifer/testing.py index db8003a6f..1a20c8695 100644 --- a/junifer/testing/datagrabbers.py +++ b/junifer/testing.py @@ -3,7 +3,15 @@ import tempfile from nilearn import datasets -from ..datagrabber.base import BaseDataGrabber +from .datagrabber.base import BaseDataGrabber +from .api.registry import register + + +def register_testing(): + """Register testing datagrabber""" + register( + 'datagrabber', 'OasisVBMTestingDatagrabber', + OasisVBMTestingDatagrabber) class OasisVBMTestingDatagrabber(BaseDataGrabber): @@ -16,14 +24,12 @@ class OasisVBMTestingDatagrabber(BaseDataGrabber): super().__init__(types=types, datadir=datadir) def get_elements(self): - return [f'sub-{x:02d}' for x in list(range(1, 11))] + return list(range(1, 11)) def __getitem__(self, element): - out = super().__getitem__(element) - i_sub = int(element.split('-')[1]) - 1 - out['VBM_GM'] = self._dataset.gray_matter_maps[i_sub] - # Set the element accordingly - out['meta']['element'] = {'subject': element} + out = {} + out['VBM_GM'] = self._dataset.gray_matter_maps[element - 1] + out['meta'] = {'subject': element} return out def __enter__(self): diff --git a/junifer/testing/__init__.py b/junifer/testing/__init__.py deleted file mode 100644 index 4b2ba56f9..000000000 --- a/junifer/testing/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -# Authors: Federico Raimondo -# License: AGPL -from . import datagrabbers \ No newline at end of file diff --git a/junifer/testing/registry.py b/junifer/testing/registry.py deleted file mode 100644 index 556005b53..000000000 --- a/junifer/testing/registry.py +++ /dev/null @@ -1,6 +0,0 @@ -from .datagrabbers import OasisVBMTestingDatagrabber -from ..api.registry import register - -register( - 'datagrabber', 'OasisVBMTestingDatagrabber', - OasisVBMTestingDatagrabber) diff --git a/junifer/utils/fs.py b/junifer/utils/fs.py deleted file mode 100644 index 667485bb6..000000000 --- a/junifer/utils/fs.py +++ /dev/null @@ -1,7 +0,0 @@ -import os -import stat - - -def make_executable(path): - st = os.stat(path) - os.chmod(path, st.st_mode | stat.S_IEXEC) diff --git a/junifer/utils/logging.py b/junifer/utils/logging.py index 343ae4954..a5558aad8 100644 --- a/junifer/utils/logging.py +++ b/junifer/utils/logging.py @@ -126,8 +126,7 @@ def configure_logging(level='WARNING', fname=None, overwrite=None, """ _close_handlers(logger) if output_format is None: - output_format = ('%(asctime)s [%(levelname)8s] %(message)s ' - '(%(filename)s:%(lineno)s)') + output_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' formatter = logging.Formatter(output_format) if fname is not None: diff --git a/pyproject.toml b/pyproject.toml index 22429039a..9cae268d7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,8 +6,4 @@ build-backend = "setuptools.build_meta" version_scheme = "python-simplified-semver" local_scheme = "no-local-version" write_to = "junifer/_version.py" -write_to_template = "__version__ = '{version}'\n" - -[tool.pytest.ini_options] -log_cli = true -log_cli_level = "WARNING" \ No newline at end of file +write_to_template = "__version__ = '{version}'\n" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 2eb1a7062..eca50d3f0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,4 @@ datalad>=0.15.4, <0.16 pandas>=0.18.0, <1.5 nibabel>=3.2.0, <4.0 nilearn>=0.9.0, <1.0 -sqlalchemy>=1.4.27, <= 1.5.0 -pyyaml>=5.1.2, <7.0 \ No newline at end of file +sqlalchemy>=1.4.27, <= 1.5.0 \ No newline at end of file diff --git a/setup.py b/setup.py index 6c76b7b05..846a52937 100644 --- a/setup.py +++ b/setup.py @@ -25,8 +25,6 @@ def _getversion(): DOWNLOAD_URL = 'https://github.com/juaml/junifer' URL = 'https://juaml.github.io/junifer' -# TODO: Read requirementes from requirements.txt and use them - setuptools.setup( name='junifer', author='Fede Raimondo', @@ -54,13 +52,7 @@ setuptools.setup( 'Source': DOWNLOAD_URL, 'Tracker': f'{DOWNLOAD_URL}issues/', }, - install_requires=['Click'], # TODO: Complete - py_modules=['junifer.api.cli'], - entry_points={ - 'console_scripts': [ - 'junifer=junifer.api.cli:cli', - ] - }, + install_requires=[], # TODO: Complete python_requires='>=3.6', use_scm_version=_getversion, setup_requires=['setuptools_scm'], diff --git a/tools/create_bids_example_dataset_sessions.py b/tools/create_bids_example_dataset_sessions.py deleted file mode 100644 index 3301c6f8a..000000000 --- a/tools/create_bids_example_dataset_sessions.py +++ /dev/null @@ -1,41 +0,0 @@ -# Authors: Federico Raimondo -# License: AGPL -from tempfile import TemporaryDirectory -from pathlib import Path - -import datalad.api as dl - -dst = 'git@gin.g-node.org:/juaml/datalad-example-bids-ses.git' - -with TemporaryDirectory() as tmpdir_name: - tmpdir = Path(tmpdir_name) - ds = dl.create(tmpdir) # type: ignore - - base_dir = tmpdir / 'example_bids_ses' - base_dir.mkdir() - - for i_sub in range(1, 10): - t_sub = f'sub-{i_sub:02d}' - sub_dir = base_dir / t_sub - sub_dir.mkdir() - - for i_ses in range(1, 4): - t_ses = f'ses-{i_ses:02d}' - ses_dir = sub_dir / t_ses - ses_dir.mkdir() - - for dname in ['anat', 'func']: - (ses_dir / dname).mkdir() - - fnames = [f'anat/{t_sub}_{t_ses}_T1w.nii.gz'] - if i_ses != 3: # Session 3 does not have functional data - fnames.extend([ - f'func/{t_sub}_{t_ses}_task-rest_bold.nii.gz', - f'func/{t_sub}_{t_ses}_task-rest_bold.json']) - for fname in fnames: - with open(ses_dir / fname, 'w') as f: - f.write('placeholder') - - ds.save(recursive=True) - ds.siblings('add', name='gin', url=dst) - ds.push(to='gin', force='all')