refactor: junifer.pipeline #49

Merged
synchon merged 4 commits from refactor/pipeline into dev 2022-09-12 16:29:06 +00:00
22 changed files with 89 additions and 73 deletions

View file

@ -4,10 +4,10 @@
# Synchon Mandal <s.mandal@fz-juelich.de> # Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL # License: AGPL
import pathlib
from typing import Dict, List, Union from typing import Dict, List, Union
import click import click
import pathlib
from ..utils.logging import configure_logging, logger, warn_with_log from ..utils.logging import configure_logging, logger, warn_with_log
from .functions import collect as api_collect from .functions import collect as api_collect
@ -59,8 +59,10 @@ def cli() -> None: # pragma: no cover
@cli.command() @cli.command()
@click.argument( @click.argument(
"filepath", type=click.Path( "filepath",
exists=True, readable=True, dir_okay=False, path_type=pathlib.Path) type=click.Path(
exists=True, readable=True, dir_okay=False, path_type=pathlib.Path
),
) )
@click.option("--element", type=str, multiple=True) @click.option("--element", type=str, multiple=True)
@click.option( @click.option(
@ -102,8 +104,11 @@ def run(filepath: click.Path, element: str, verbose: click.Choice) -> None:
@cli.command() @cli.command()
@click.argument( @click.argument(
"filepath", type=click.Path( "filepath",
exists=True, readable=True, dir_okay=False, path_type=pathlib.Path)) type=click.Path(
exists=True, readable=True, dir_okay=False, path_type=pathlib.Path
),
)
@click.option( @click.option(
"-v", "-v",
"--verbose", "--verbose",
@ -131,8 +136,10 @@ def collect(filepath: click.Path, verbose: click.Choice) -> None:
@cli.command() @cli.command()
@click.argument( @click.argument(
"filepath", type=click.Path( "filepath",
exists=True, readable=True, dir_okay=False, path_type=pathlib.Path) type=click.Path(
exists=True, readable=True, dir_okay=False, path_type=pathlib.Path
),
) )
@click.option("--element", type=str, multiple=True) @click.option("--element", type=str, multiple=True)
@click.option("--overwrite", is_flag=True) @click.option("--overwrite", is_flag=True)
@ -180,3 +187,9 @@ def queue(
submit=submit, submit=submit,
**queue_config, **queue_config,
) )
@cli.command()
def selftest() -> None:
"""Selftest command for CLI."""
pass

View file

@ -7,9 +7,9 @@
import shutil import shutil
import subprocess import subprocess
from pathlib import Path
from typing import Dict, List, Optional, Union, Tuple
import typing import typing
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union
import yaml import yaml

View file

@ -8,9 +8,11 @@
from typing import TYPE_CHECKING, Dict, List, Optional, Union from typing import TYPE_CHECKING, Dict, List, Optional, Union
from ..utils.logging import logger, raise_error from ..utils.logging import logger, raise_error
if TYPE_CHECKING: if TYPE_CHECKING:
from ..datagrabber.base import BaseDataGrabber from ..datagrabber.base import BaseDataGrabber
from ..markers.base import PipelineStepMixin from ..pipeline import PipelineStepMixin
from ..storage.base import BaseFeatureStorage from ..storage.base import BaseFeatureStorage
# Define valid steps for operation # Define valid steps for operation

View file

@ -4,10 +4,9 @@
# Leonard Sasse <l.sasse@fz-juelich.de> # Leonard Sasse <l.sasse@fz-juelich.de>
# Synchon Mandal <s.mandal@fz-juelich.de> # Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL # License: AGPL
from typing import Type
import logging import logging
from abc import ABC from abc import ABC
from typing import Type
import pytest import pytest

View file

@ -10,7 +10,7 @@ import shutil
import tempfile import tempfile
import zipfile import zipfile
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, List, Optional, Tuple, Union, Any, Dict from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
import nibabel as nib import nibabel as nib
import numpy as np import numpy as np
@ -536,7 +536,8 @@ def _retrieve_tian(
if space != "MNI6thgeneration": if space != "MNI6thgeneration":
raise_error( raise_error(
f"The parameter `space` ({space}) for 7T needs to be " f"The parameter `space` ({space}) for 7T needs to be "
f"MNI6thgeneration") f"MNI6thgeneration"
)
else: else:
raise_error( raise_error(
f"The parameter `magneticfield` ({magneticfield}) needs to be " f"The parameter `magneticfield` ({magneticfield}) needs to be "
@ -592,7 +593,7 @@ def _retrieve_tian(
"A simple numbering scheme for distinction was therefore used." "A simple numbering scheme for distinction was therefore used."
) )
else: # pragma: no cover else: # pragma: no cover
raise_error('This should not happen. Please report this error.') raise_error("This should not happen. Please report this error.")
# check existence of atlas # check existence of atlas
if not (atlas_fname.exists() and atlas_lname.exists()): if not (atlas_fname.exists() and atlas_lname.exists()):
@ -625,9 +626,7 @@ def _retrieve_tian(
def _retrieve_suit( def _retrieve_suit(
atlas_dir: Path, atlas_dir: Path, resolution: Optional[float], space: str = "MNI"
resolution: Optional[float],
space: str = "MNI"
) -> Tuple[Path, List[str]]: ) -> Tuple[Path, List[str]]:
"""Retrieve SUIT atlas. """Retrieve SUIT atlas.

View file

@ -443,7 +443,8 @@ def test_tian_7T_6thgeneration(
assert fname.name == fname1 assert fname.name == fname1
assert len(lbl) == n_label assert len(lbl) == n_label
assert_array_almost_equal( assert_array_almost_equal(
img.header["pixdim"][1:4], [1.6, 1.6, 1.6]) # type: ignore img.header["pixdim"][1:4], [1.6, 1.6, 1.6]
) # type: ignore
def test_retrieve_tian_incorrect_space(tmp_path: Path) -> None: def test_retrieve_tian_incorrect_space(tmp_path: Path) -> None:

View file

@ -120,7 +120,8 @@ class DataladDataGrabber(BaseDataGrabber):
"""Install the datalad dataset into the datadir.""" """Install the datalad dataset into the datadir."""
logger.debug(f"Installing dataset {self.uri} to {self._datadir}") logger.debug(f"Installing dataset {self.uri} to {self._datadir}")
self._dataset = dl.install( # type: ignore because of datalad self._dataset = dl.install( # type: ignore because of datalad
self._datadir, source=self.uri) self._datadir, source=self.uri
)
logger.debug("Dataset installed") logger.debug("Dataset installed")
def remove(self): def remove(self):

View file

@ -10,7 +10,7 @@ from typing import Dict, List
import nibabel as nib import nibabel as nib
import pandas as pd import pandas as pd
from ..markers.base import PipelineStepMixin from ..pipeline.pipeline_mixin import PipelineStepMixin
from ..utils.logging import logger, warn_with_log from ..utils.logging import logger, warn_with_log

View file

@ -7,4 +7,3 @@
from .base import BaseMarker from .base import BaseMarker
from .collection import MarkerCollection from .collection import MarkerCollection
from .parcel import ParcelAggregation from .parcel import ParcelAggregation
from .pipeline_mixin import PipelineStepMixin

View file

@ -6,8 +6,8 @@
from typing import Dict, List, Optional, Union from typing import Dict, List, Optional, Union
from ..pipeline.pipeline_mixin import PipelineStepMixin
from ..utils import logger, raise_error from ..utils import logger, raise_error
from .pipeline_mixin import PipelineStepMixin
class BaseMarker(PipelineStepMixin): class BaseMarker(PipelineStepMixin):
@ -23,9 +23,7 @@ class BaseMarker(PipelineStepMixin):
""" """
def __init__( def __init__(
self, self, on: Union[List[str], str], name: Optional[str] = None
on: Union[List[str], str],
name: Optional[str] = None
) -> None: ) -> None:
"""Initialize the class.""" """Initialize the class."""
if not isinstance(on, list): if not isinstance(on, list):

View file

@ -5,14 +5,13 @@
# License: AGPL # License: AGPL
from collections import Counter from collections import Counter
from typing import Dict, Optional, List from typing import Dict, List, Optional
from junifer.markers.pipeline_mixin import PipelineStepMixin from ..datareader.default import DefaultDataReader
from ..markers.base import BaseMarker
from ..datareader import DefaultDataReader from ..pipeline import PipelineStepMixin
from ..utils import logger
from .base import BaseMarker
from ..storage.base import BaseFeatureStorage from ..storage.base import BaseFeatureStorage
from ..utils import logger
class MarkerCollection: class MarkerCollection:
@ -32,7 +31,7 @@ class MarkerCollection:
markers: List[BaseMarker], markers: List[BaseMarker],
datareader: Optional[PipelineStepMixin] = None, datareader: Optional[PipelineStepMixin] = None,
preprocessing: Optional[PipelineStepMixin] = None, preprocessing: Optional[PipelineStepMixin] = None,
storage: Optional[BaseFeatureStorage] = None storage: Optional[BaseFeatureStorage] = None,
): ):
"""Initialize the class.""" """Initialize the class."""
# Check that the markers have different names # Check that the markers have different names

View file

@ -116,7 +116,8 @@ class ParcelAggregation(BaseMarker):
) )
logger.debug("Masking") logger.debug("Masking")
masker = NiftiMasker( masker = NiftiMasker(
atlas_bin, target_affine=t_input.affine) # type: ignore atlas_bin, target_affine=t_input.affine
) # type: ignore
# Mask the input data and the atlas # Mask the input data and the atlas
data = masker.fit_transform(t_input) data = masker.fit_transform(t_input)

View file

@ -9,7 +9,7 @@ from numpy.testing import assert_array_equal
from junifer.datareader.default import DefaultDataReader from junifer.datareader.default import DefaultDataReader
from junifer.markers import MarkerCollection, ParcelAggregation from junifer.markers import MarkerCollection, ParcelAggregation
from junifer.markers.base import PipelineStepMixin from junifer.pipeline import PipelineStepMixin
from junifer.storage import SQLiteFeatureStorage from junifer.storage import SQLiteFeatureStorage
from junifer.testing.datagrabbers import OasisVBMTestingDatagrabber from junifer.testing.datagrabbers import OasisVBMTestingDatagrabber
@ -90,8 +90,7 @@ def test_marker_collection():
for t_marker in markers: for t_marker in markers:
t_name = t_marker.name t_name = t_marker.name
assert_array_equal( assert_array_equal(
out[t_name]["VBM_GM"]["data"], out[t_name]["VBM_GM"]["data"], out2[t_name]["VBM_GM"]["data"]
out2[t_name]["VBM_GM"]["data"]
) )

View file

@ -0,0 +1,6 @@
"""Provide imports for pipeline sub-package."""
# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from .pipeline_mixin import PipelineStepMixin

View file

@ -6,7 +6,7 @@
import pytest import pytest
from junifer.markers.pipeline_mixin import PipelineStepMixin from junifer.pipeline.pipeline_mixin import PipelineStepMixin
def test_PipelineStepMixin() -> None: def test_PipelineStepMixin() -> None:

View file

@ -13,12 +13,12 @@ from nilearn._utils.niimg_conversions import check_niimg_4d
from nilearn.image import clean_img from nilearn.image import clean_img
from nilearn.masking import compute_brain_mask from nilearn.masking import compute_brain_mask
from ..markers import PipelineStepMixin from ..pipeline import PipelineStepMixin
from ..utils import logger, raise_error from ..utils import logger, raise_error
if TYPE_CHECKING: if TYPE_CHECKING:
from nibabel import Nifti1Image, Nifti2Image, MGHImage from nibabel import MGHImage, Nifti1Image, Nifti2Image
class BaseConfoundRemover(PipelineStepMixin): class BaseConfoundRemover(PipelineStepMixin):

View file

@ -5,7 +5,7 @@
# License: AGPL # License: AGPL
from functools import partial from functools import partial
from typing import Callable, Dict, Any, Optional from typing import Any, Callable, Dict, Optional
import numpy as np import numpy as np
from scipy.stats import trim_mean from scipy.stats import trim_mean
@ -15,8 +15,8 @@ from .utils import logger, raise_error
def get_aggfunc_by_name( def get_aggfunc_by_name(
name: str, name: str, func_params: Optional[Dict[str, Any]]
func_params: Optional[Dict[str, Any]]) -> Callable: ) -> Callable:
"""Get an aggregation function by its name. """Get an aggregation function by its name.
Parameters Parameters

View file

@ -108,8 +108,9 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
) )
prefix = element_to_prefix(element) prefix = element_to_prefix(element)
# Format URI for engine creation # Format URI for engine creation
uri = "sqlite:///" \ uri = (
f"{self.uri.parent}/{prefix}{self.uri.name}" # type: ignore "sqlite:///" f"{self.uri.parent}/{prefix}{self.uri.name}"
) # type: ignore
return create_engine(uri, echo=False) return create_engine(uri, echo=False)
def _save_upsert( def _save_upsert(
@ -231,7 +232,8 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
) )
# Prepare new dataframe # Prepare new dataframe
data_df = pd.DataFrame( data_df = pd.DataFrame(
data, columns=columns, index=idx) # type: ignore data, columns=columns, index=idx
) # type: ignore
# Store dataframe # Store dataframe
self.store_df(df=data_df, meta=meta) self.store_df(df=data_df, meta=meta)
@ -525,8 +527,9 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
""" """
if self.single_output is True: if self.single_output is True:
raise_error(msg="collect() is not implemented for single output.") raise_error(msg="collect() is not implemented for single output.")
logger.info("Collecting data from " logger.info(
f"{self.uri.parent}/*{self.uri.name}") # type: ignore "Collecting data from " f"{self.uri.parent}/*{self.uri.name}"
) # type: ignore
# Create new instance # Create new instance
out_storage = SQLiteFeatureStorage( out_storage = SQLiteFeatureStorage(
uri=self.uri, single_output=True, upsert="ignore" uri=self.uri, single_output=True, upsert="ignore"

View file

@ -4,9 +4,8 @@
# Synchon Mandal <s.mandal@fz-juelich.de> # Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL # License: AGPL
from typing import Union, List
from pathlib import Path from pathlib import Path
from typing import List, Union
import numpy as np import numpy as np
import pandas as pd import pandas as pd
@ -59,8 +58,9 @@ df_ignore = pd.DataFrame(
).set_index(["element", "pk2"]) ).set_index(["element", "pk2"])
def _read_sql(table_name: str, uri: str, def _read_sql(
index_col: Union[str, List[str]]) -> pd.DataFrame: table_name: str, uri: str, index_col: Union[str, List[str]]
) -> pd.DataFrame:
"""Read database table into a pandas DataFrame. """Read database table into a pandas DataFrame.
Parameters Parameters
@ -160,8 +160,7 @@ def test_upsert_replace(tmp_path: Path) -> None:
table_name = storage.store_metadata(meta) table_name = storage.store_metadata(meta)
# Read stored table # Read stored table
c_df1 = _read_sql( c_df1 = _read_sql(
table_name=table_name, uri=uri.as_posix(), table_name=table_name, uri=uri.as_posix(), index_col=["element", "pk2"]
index_col=["element", "pk2"]
) )
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(df1, c_df1) assert_frame_equal(df1, c_df1)
@ -169,8 +168,7 @@ def test_upsert_replace(tmp_path: Path) -> None:
storage._save_upsert(df=df2, name=table_name, if_exists="replace") storage._save_upsert(df=df2, name=table_name, if_exists="replace")
# Read stored table # Read stored table
c_df2 = _read_sql( c_df2 = _read_sql(
table_name=table_name, uri=uri.as_posix(), table_name=table_name, uri=uri.as_posix(), index_col=["element", "pk2"]
index_col=["element", "pk2"]
) )
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(df2, c_df2) assert_frame_equal(df2, c_df2)
@ -198,9 +196,7 @@ def test_upsert_ignore(tmp_path: Path) -> None:
table_name = storage.store_metadata(meta=meta) table_name = storage.store_metadata(meta=meta)
# Read stored table # Read stored table
c_df1 = _read_sql( c_df1 = _read_sql(
table_name=table_name, table_name=table_name, uri=uri.as_posix(), index_col=["element", "pk2"]
uri=uri.as_posix(),
index_col=["element", "pk2"]
) )
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(df1, c_df1) assert_frame_equal(df1, c_df1)
@ -209,7 +205,8 @@ def test_upsert_ignore(tmp_path: Path) -> None:
storage.store_df(df2, meta) storage.store_df(df2, meta)
# Read stored table # Read stored table
c_dfignore = _read_sql( c_dfignore = _read_sql(
table_name, uri=uri.as_posix(), index_col=["element", "pk2"]) table_name, uri=uri.as_posix(), index_col=["element", "pk2"]
)
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(c_dfignore, df_ignore) assert_frame_equal(c_dfignore, df_ignore)
# Check for error # Check for error
@ -236,8 +233,7 @@ def test_upsert_update(tmp_path: Path) -> None:
table_name = storage.store_metadata(meta) table_name = storage.store_metadata(meta)
# Read stored table # Read stored table
c_df1 = _read_sql( c_df1 = _read_sql(
table_name=table_name, uri=uri.as_posix(), table_name=table_name, uri=uri.as_posix(), index_col=["element", "pk2"]
index_col=["element", "pk2"]
) )
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(df1, c_df1) assert_frame_equal(df1, c_df1)
@ -245,8 +241,8 @@ def test_upsert_update(tmp_path: Path) -> None:
storage.store_df(df2, meta) storage.store_df(df2, meta)
# Read stored table # Read stored table
c_dfupdate = _read_sql( c_dfupdate = _read_sql(
table_name, uri=uri.as_posix(), table_name, uri=uri.as_posix(), index_col=["element", "pk2"]
index_col=["element", "pk2"]) )
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(c_dfupdate, df_update) assert_frame_equal(c_dfupdate, df_update)
@ -381,8 +377,9 @@ def test_store_table(tmp_path: Path) -> None:
table_name = storage.store_metadata(meta) table_name = storage.store_metadata(meta)
# Read stored table # Read stored table
c_df = _read_sql( c_df = _read_sql(
table_name=table_name, uri=uri.as_posix(), table_name=table_name,
index_col=["element", "scan"] uri=uri.as_posix(),
index_col=["element", "scan"],
) )
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(df, c_df) assert_frame_equal(df, c_df)
@ -400,8 +397,9 @@ def test_store_table(tmp_path: Path) -> None:
) )
# Read stored table # Read stored table
c_df_new = _read_sql( c_df_new = _read_sql(
table_name=table_name, uri=uri.as_posix(), table_name=table_name,
index_col=["element", "scan"] uri=uri.as_posix(),
index_col=["element", "scan"],
) )
# Check if dataframes are equal # Check if dataframes are equal
assert_frame_equal(df_new, c_df_new) assert_frame_equal(df_new, c_df_new)

View file

@ -4,11 +4,9 @@
# Synchon Mandal <s.mandal@fz-juelich.de> # Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL # License: AGPL
from typing import Any
import hashlib import hashlib
import json import json
from typing import Dict, Optional, Tuple, Union from typing import Any, Dict, Optional, Tuple, Union
import numpy as np import numpy as np
import pandas as pd import pandas as pd

View file

@ -273,8 +273,8 @@ def raise_error(msg: str, klass: Type[Exception] = ValueError) -> NoReturn:
def warn_with_log( def warn_with_log(
msg: str, msg: str, category: Optional[Type[Warning]] = RuntimeWarning
category: Optional[Type[Warning]] = RuntimeWarning) -> None: ) -> None:
"""Warn, but first log it. """Warn, but first log it.
Parameters Parameters