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

View file

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

View file

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

View file

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

View file

@ -10,7 +10,7 @@ import shutil
import tempfile
import zipfile
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 numpy as np
@ -536,7 +536,8 @@ def _retrieve_tian(
if space != "MNI6thgeneration":
raise_error(
f"The parameter `space` ({space}) for 7T needs to be "
f"MNI6thgeneration")
f"MNI6thgeneration"
)
else:
raise_error(
f"The parameter `magneticfield` ({magneticfield}) needs to be "
@ -592,7 +593,7 @@ def _retrieve_tian(
"A simple numbering scheme for distinction was therefore used."
)
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
if not (atlas_fname.exists() and atlas_lname.exists()):
@ -625,9 +626,7 @@ def _retrieve_tian(
def _retrieve_suit(
atlas_dir: Path,
resolution: Optional[float],
space: str = "MNI"
atlas_dir: Path, resolution: Optional[float], space: str = "MNI"
) -> Tuple[Path, List[str]]:
"""Retrieve SUIT atlas.

View file

@ -443,7 +443,8 @@ def test_tian_7T_6thgeneration(
assert fname.name == fname1
assert len(lbl) == n_label
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:

View file

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

View file

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

View file

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

View file

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

View file

@ -5,14 +5,13 @@
# License: AGPL
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 import DefaultDataReader
from ..utils import logger
from .base import BaseMarker
from ..datareader.default import DefaultDataReader
from ..markers.base import BaseMarker
from ..pipeline import PipelineStepMixin
from ..storage.base import BaseFeatureStorage
from ..utils import logger
class MarkerCollection:
@ -32,7 +31,7 @@ class MarkerCollection:
markers: List[BaseMarker],
datareader: Optional[PipelineStepMixin] = None,
preprocessing: Optional[PipelineStepMixin] = None,
storage: Optional[BaseFeatureStorage] = None
storage: Optional[BaseFeatureStorage] = None,
):
"""Initialize the class."""
# Check that the markers have different names

View file

@ -116,7 +116,8 @@ class ParcelAggregation(BaseMarker):
)
logger.debug("Masking")
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
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.markers import MarkerCollection, ParcelAggregation
from junifer.markers.base import PipelineStepMixin
from junifer.pipeline import PipelineStepMixin
from junifer.storage import SQLiteFeatureStorage
from junifer.testing.datagrabbers import OasisVBMTestingDatagrabber
@ -90,8 +90,7 @@ def test_marker_collection():
for t_marker in markers:
t_name = t_marker.name
assert_array_equal(
out[t_name]["VBM_GM"]["data"],
out2[t_name]["VBM_GM"]["data"]
out[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
from junifer.markers.pipeline_mixin import PipelineStepMixin
from junifer.pipeline.pipeline_mixin import PipelineStepMixin
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.masking import compute_brain_mask
from ..markers import PipelineStepMixin
from ..pipeline import PipelineStepMixin
from ..utils import logger, raise_error
if TYPE_CHECKING:
from nibabel import Nifti1Image, Nifti2Image, MGHImage
from nibabel import MGHImage, Nifti1Image, Nifti2Image
class BaseConfoundRemover(PipelineStepMixin):

View file

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

View file

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

View file

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

View file

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

View file

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