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

[ENH]: Introduce WorkDirManager #254

Merged
synchon merged 30 commits from feat/workdirmanager into main 2023-10-13 08:15:05 +00:00
Conversation 8 Commits 30 Files changed 15 +535 -180
15 changed files with 535 additions and 180 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/changes/newsfragments/254.feature Normal file
View file

@ -0,0 +1 @@
Introduce :class:`junifer.pipeline.WorkDirManager` singleton class to manage working and temporary directories across pipeline by `Synchon Mandal`_

3
junifer/api/functions.py
View file

@ -15,6 +15,7 @@ from typing import Dict, List, Optional, Tuple, Union
from ..datagrabber.base import BaseDataGrabber
from ..markers.base import BaseMarker
from ..markers.collection import MarkerCollection
from ..pipeline import WorkDirManager
from ..pipeline.registry import build
from ..preprocess.base import BasePreprocessor
from ..storage.base import BaseFeatureStorage
@ -115,6 +116,8 @@ def run(
# Convert str to Path
if isinstance(workdir, str):
workdir = Path(workdir)
# Initiate working directory manager
WorkDirManager(workdir)
if not isinstance(elements, list) and elements is not None:
elements = [elements]

29
junifer/markers/falff/falff_estimator.py
View file

@ -4,12 +4,9 @@
# Federico Raimondo <f.raimondo@fz-juelich.de>
# License: AGPL
import shutil
import subprocess
import tempfile
import typing
from functools import lru_cache
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
import nibabel as nib
@ -17,8 +14,9 @@ import numpy as np
from nilearn import image as nimg
from scipy.fft import fft, fftfreq
from ...pipeline import WorkDirManager
from ...pipeline.singleton import singleton
from ...utils import logger, raise_error
from ..utils import singleton
if TYPE_CHECKING:
@ -41,19 +39,24 @@ class ALFFEstimator:
use_afni : bool
Whether to use afni for computation. If False, will use python.
Attributes
----------
temp_dir_path : pathlib.Path
Path to the temporary directory for assets storage.
"""
def __init__(self) -> None:
self._file_path = None
# Create temporary directory for intermittent storage of assets during
# computation via afni's 3dReHo
self.temp_dir_path = Path(tempfile.mkdtemp())
# computation via afni's 3dRSFC
self.temp_dir_path = None
def __del__(self) -> None:
"""Cleanup."""
print("Cleaning up temporary directory...")
# Delete temporary directory and ignore errors for read-only files
shutil.rmtree(self.temp_dir_path, ignore_errors=True)
if self.temp_dir_path is not None:
WorkDirManager().delete_tempdir(self.temp_dir_path)
@staticmethod
def _run_afni_cmd(cmd: str) -> None:
@ -123,6 +126,8 @@ class ALFFEstimator:
If the AFNI commands fails due to some issues
"""
# Note: self.temp_dir_path is sure to exist before proceeding, so
# types checks are ignored further on.
# Save niimg to nii.gz
nifti_in_file_path = self.temp_dir_path / "input.nii"
@ -162,7 +167,7 @@ class ALFFEstimator:
self._run_afni_cmd(convert_cmd)
# Cleanup intermediate files
for fname in self.temp_dir_path.glob("temp_*"):
for fname in self.temp_dir_path.glob("temp_*"): # type: ignore
fname.unlink()
# Load niftis
@ -268,6 +273,8 @@ class ALFFEstimator:
fALFF map.
"""
if use_afni:
# Create new temporary directory before using AFNI
self.temp_dir_path = WorkDirManager().get_tempdir(prefix="falff")
output = self._compute_alff_afni(
data=data,
highpass=highpass,
@ -318,8 +325,8 @@ class ALFFEstimator:
# Clear the cache
self._compute.cache_clear()
# Clear temporary directory files
for file_ in self.temp_dir_path.iterdir():
file_.unlink(missing_ok=True)
if self.temp_dir_path is not None:
WorkDirManager().delete_tempdir(self.temp_dir_path)
# Set the new file path
self._file_path = bold_path
else:

139
junifer/markers/falff/tests/test_falff_estimator.py
View file

@ -4,6 +4,7 @@
# License: AGPL
import time
from pathlib import Path
import pytest
from nibabel import Nifti1Image
@ -11,23 +12,36 @@ from scipy.stats import pearsonr
from junifer.datareader import DefaultDataReader
from junifer.markers.falff.falff_estimator import ALFFEstimator
from junifer.pipeline import WorkDirManager
from junifer.pipeline.utils import _check_afni
from junifer.testing.datagrabbers import PartlyCloudyTestingDataGrabber
from junifer.utils import logger
def test_ALFFEstimator_cache_python() -> None:
"""Test that the cache works properly when using python."""
def test_ALFFEstimator_cache_python(tmp_path: Path) -> None:
"""Test that the cache works properly when using python.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Get subject from datagrabber
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
subject = dg["sub-01"]
# Read data for subject
subject_data = DefaultDataReader().fit_transform(subject)
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Setup estimator
estimator = ALFFEstimator()
# Compute without cache
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=False,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -36,14 +50,12 @@ def test_ALFFEstimator_cache_python() -> None:
logger.info(f"ALFF Estimator First time: {first_time}")
assert isinstance(alff, Nifti1Image)
assert isinstance(falff, Nifti1Image)
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now fit again, should be faster
# Compute again with cache, should be faster
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=False,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -51,15 +63,12 @@ def test_ALFFEstimator_cache_python() -> None:
second_time = time.time() - start_time
logger.info(f"ALFF Estimator Second time: {second_time}")
assert second_time < (first_time / 1000)
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now change a parameter, should compute again, without clearing the
# cache
# Change a parameter and compute again without cache
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=False,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.11,
tr=None,
@ -67,14 +76,12 @@ def test_ALFFEstimator_cache_python() -> None:
third_time = time.time() - start_time
logger.info(f"ALFF Estimator Third time: {third_time}")
assert third_time > (first_time / 10)
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now fit again with the previous params, should be fast
# Compute again with cache, should be faster
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=False,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -82,19 +89,17 @@ def test_ALFFEstimator_cache_python() -> None:
fourth = time.time() - start_time
logger.info(f"ALFF Estimator Fourth time: {fourth}")
assert fourth < (first_time / 1000)
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now change the data, it should clear the cache
# Change the data and it should clear the cache
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-02"]
input = DefaultDataReader().fit_transform(input)
subject = dg["sub-02"]
# Read data for new subject
subject_data = DefaultDataReader().fit_transform(subject)
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=False,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -102,25 +107,35 @@ def test_ALFFEstimator_cache_python() -> None:
fifth = time.time() - start_time
logger.info(f"ALFF Estimator Fifth time: {fifth}")
assert fifth > (first_time / 10)
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_ALFFEstimator_cache_afni() -> None:
"""Test that the cache works properly when using afni."""
def test_ALFFEstimator_cache_afni(tmp_path: Path) -> None:
"""Test that the cache works properly when using afni.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Get subject from datagrabber
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
subject = dg["sub-01"]
# Read data for subject
subject_data = DefaultDataReader().fit_transform(subject)
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Setup estimator
estimator = ALFFEstimator()
# Compute with cache
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=True,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -132,11 +147,11 @@ def test_ALFFEstimator_cache_afni() -> None:
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 3 # input + alff + falff
# Now fit again, should be faster
# Compute again with cache, should be faster
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=True,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -147,12 +162,11 @@ def test_ALFFEstimator_cache_afni() -> None:
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 3 # input + alff + falff
# Now change a parameter, should compute again, without clearing the
# cache
# Change a parameter and compute again without cache
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=True,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.11,
tr=None,
@ -161,13 +175,13 @@ def test_ALFFEstimator_cache_afni() -> None:
logger.info(f"ALFF Estimator Third time: {third_time}")
assert third_time > (first_time / 10)
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 5 # input + 2 * alff + 2 * falff
assert n_files == 3 # input + alff + falff
# Now fit again with the previous params, should be fast
# Compute with cache, should be faster
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=True,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -176,18 +190,18 @@ def test_ALFFEstimator_cache_afni() -> None:
logger.info(f"ALFF Estimator Fourth time: {fourth}")
assert fourth < (first_time / 1000)
n_files = len(list(estimator.temp_dir_path.glob("*")))
assert n_files == 5 # input + 2 * alff + 2 * falff
assert n_files == 3 # input + alff + falff
# Now change the data, it should clear the cache
# Change the data and it should clear the cache
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-02"]
input = DefaultDataReader().fit_transform(input)
subject = dg["sub-02"]
# Read data for new subject
subject_data = DefaultDataReader().fit_transform(subject)
start_time = time.time()
alff, falff = estimator.fit_transform(
use_afni=True,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=None,
@ -202,18 +216,29 @@ def test_ALFFEstimator_cache_afni() -> None:
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_ALFFEstimator_afni_vs_python() -> None:
"""Test that the cache works properly when using afni."""
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
def test_ALFFEstimator_afni_vs_python(tmp_path: Path) -> None:
"""Test that the cache works properly when using afni.
input = DefaultDataReader().fit_transform(input)
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Get subject from datagrabber
with PartlyCloudyTestingDataGrabber() as dg:
subject = dg["sub-01"]
# Read data for subject
subject_data = DefaultDataReader().fit_transform(subject)
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Setup estimator
estimator = ALFFEstimator()
# Use an arbitrary TR to test the AFNI vs Python implementation
afni_alff, afni_falff = estimator.fit_transform(
use_afni=True,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=2.5,
@ -221,7 +246,7 @@ def test_ALFFEstimator_afni_vs_python() -> None:
python_alff, python_falff = estimator.fit_transform(
use_afni=False,
input_data=input["BOLD"],
input_data=subject_data["BOLD"],
highpass=0.01,
lowpass=0.1,
tr=2.5,

36
junifer/markers/falff/tests/test_falff_parcels.py
View file

@ -12,6 +12,7 @@ from scipy.stats import pearsonr
from junifer.datareader import DefaultDataReader
from junifer.markers.falff import ALFFParcels
from junifer.pipeline import WorkDirManager
from junifer.pipeline.utils import _check_afni
from junifer.storage import SQLiteFeatureStorage
from junifer.testing.datagrabbers import PartlyCloudyTestingDataGrabber
@ -21,15 +22,20 @@ from junifer.utils import logger
_PARCELLATION = "Schaefer100x7"
def test_ALFFParcels_python() -> None:
"""Test ALFFParcels using python."""
# Get the SPM auditory data:
def test_ALFFParcels_python(tmp_path: Path) -> None:
"""Test ALFFParcels using python.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker = ALFFParcels(
parcellation=_PARCELLATION,
method="mean",
@ -46,14 +52,20 @@ def test_ALFFParcels_python() -> None:
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_ALFFParcels_afni() -> None:
"""Test ALFFParcels using afni."""
# Get the SPM auditory data:
def test_ALFFParcels_afni(tmp_path: Path) -> None:
"""Test ALFFParcels using afni.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker = ALFFParcels(
parcellation=_PARCELLATION,
method="mean",
@ -83,12 +95,15 @@ def test_ALFFParcels_afni() -> None:
"fractional", [True, False], ids=["fractional", "non-fractional"]
)
def test_ALFFParcels_python_vs_afni(
tmp_path: Path,
fractional: bool,
) -> None:
"""Test ALFFParcels using python.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
fractional : bool
Whether to compute fractional ALFF or not.
@ -98,7 +113,7 @@ def test_ALFFParcels_python_vs_afni(
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker_python = ALFFParcels(
parcellation=_PARCELLATION,
method="mean",
@ -137,12 +152,13 @@ def test_ALFFParcels_storage(
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with PartlyCloudyTestingDataGrabber() as dg:
# Use first subject
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker = ALFFParcels(
parcellation=_PARCELLATION,
method="mean",

40
junifer/markers/falff/tests/test_falff_spheres.py
View file

@ -12,6 +12,7 @@ from scipy.stats import pearsonr
from junifer.datareader import DefaultDataReader
from junifer.markers.falff import ALFFSpheres
from junifer.pipeline import WorkDirManager
from junifer.pipeline.utils import _check_afni
from junifer.storage import SQLiteFeatureStorage
from junifer.testing.datagrabbers import PartlyCloudyTestingDataGrabber
@ -21,15 +22,21 @@ from junifer.utils import logger
_COORDINATES = "DMNBuckner"
def test_ALFFSpheres_python() -> None:
"""Test ALFFSpheres using python."""
# Get the SPM auditory data:
def test_ALFFSpheres_python(tmp_path: Path) -> None:
"""Test ALFFSpheres using python.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker = ALFFSpheres(
coords=_COORDINATES,
radius=5,
@ -47,14 +54,21 @@ def test_ALFFSpheres_python() -> None:
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_ALFFSpheres_afni() -> None:
"""Test ALFFSpheres using afni."""
# Get the SPM auditory data:
def test_ALFFSpheres_afni(tmp_path: Path) -> None:
"""Test ALFFSpheres using afni.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker = ALFFSpheres(
coords=_COORDINATES,
radius=5,
@ -88,20 +102,25 @@ def test_ALFFSpheres_afni() -> None:
"fractional", [True, False], ids=["fractional", "non-fractional"]
)
def test_ALFFSpheres_python_vs_afni(
tmp_path: Path,
fractional: bool,
) -> None:
"""Test ALFFSpheres python vs afni results.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
fractional : bool
Whether to compute fractional ALFF or not.
"""
with PartlyCloudyTestingDataGrabber() as dg:
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker_python = ALFFSpheres(
coords=_COORDINATES,
radius=5,
@ -142,12 +161,13 @@ def test_ALFFSpheres_storage(
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with PartlyCloudyTestingDataGrabber() as dg:
# Use first subject
input = dg["sub-01"]
input = DefaultDataReader().fit_transform(input)
# Create ParcelAggregation object
WorkDirManager().workdir = tmp_path
marker = ALFFSpheres(
coords=_COORDINATES,
radius=5,

31
junifer/markers/reho/reho_estimator.py
View file

@ -6,12 +6,9 @@
import hashlib
import shutil
import subprocess
import tempfile
from functools import lru_cache
from itertools import product
from pathlib import Path
from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
import nibabel as nib
@ -20,8 +17,9 @@ from nilearn import image as nimg
from nilearn import masking as nmask
from scipy.stats import rankdata
from ...pipeline import WorkDirManager
from ...pipeline.singleton import singleton
from ...utils import logger, raise_error
from ..utils import singleton
if TYPE_CHECKING:
@ -50,12 +48,13 @@ class ReHoEstimator:
self._file_path = None
# Create temporary directory for intermittent storage of assets during
# computation via afni's 3dReHo
self.temp_dir_path = Path(tempfile.mkdtemp())
self.temp_dir_path = None
def __del__(self) -> None:
"""Cleanup."""
# Delete temporary directory and ignore errors for read-only files
shutil.rmtree(self.temp_dir_path, ignore_errors=True)
if self.temp_dir_path is not None:
WorkDirManager().delete_tempdir(self.temp_dir_path)
def _compute_reho_afni(
self,
@ -137,12 +136,15 @@ class ReHoEstimator:
https://doi.org/10.1089/brain.2013.0154
"""
# Note: self.temp_dir_path is sure to exist before proceeding, so
# types checks are ignored further on.
# Save niimg to nii.gz
nifti_in_file_path = self.temp_dir_path / "input.nii"
nifti_in_file_path = self.temp_dir_path / "input.nii" # type: ignore
nib.save(data, nifti_in_file_path)
# Set 3dReHo command
reho_afni_out_path_prefix = self.temp_dir_path / "reho"
reho_afni_out_path_prefix = self.temp_dir_path / "reho" # type: ignore
reho_cmd: List[str] = [
"3dReHo",
f"-prefix {reho_afni_out_path_prefix.resolve()}",
@ -194,7 +196,8 @@ class ReHoEstimator:
sha256_params = hashlib.sha256(bytes(reho_cmd_str, "utf-8"))
# Convert afni to nifti
reho_afni_to_nifti_out_path = (
self.temp_dir_path / f"output_{sha256_params.hexdigest()}.nii"
self.temp_dir_path
/ f"output_{sha256_params.hexdigest()}.nii" # type: ignore
)
convert_cmd: List[str] = [
"3dAFNItoNIFTI",
@ -225,7 +228,7 @@ class ReHoEstimator:
)
# Cleanup intermediate files
for fname in self.temp_dir_path.glob("reho*"):
for fname in self.temp_dir_path.glob("reho*"): # type: ignore
fname.unlink()
# Load nifti
@ -405,7 +408,7 @@ class ReHoEstimator:
)
output = nimg.new_img_like(data, reho_map, copy_header=False)
return output
return output # type: ignore
@lru_cache(maxsize=None, typed=True)
def _compute(
@ -431,6 +434,8 @@ class ReHoEstimator:
"""
if use_afni:
# Create new temporary directory before using AFNI
self.temp_dir_path = WorkDirManager().get_tempdir(prefix="reho")
output = self._compute_reho_afni(data, **reho_params)
else:
output = self._compute_reho_python(data, **reho_params)
@ -466,8 +471,8 @@ class ReHoEstimator:
# Clear the cache
self._compute.cache_clear()
# Clear temporary directory files
for file_ in self.temp_dir_path.iterdir():
file_.unlink(missing_ok=True)
if self.temp_dir_path is not None:
WorkDirManager().delete_tempdir(self.temp_dir_path)
# Set the new file path
self._file_path = bold_path
else:

80
junifer/markers/reho/tests/test_reho_estimator.py
View file

@ -4,6 +4,7 @@
# License: AGPL
import time
from pathlib import Path
import nibabel as nib
import pytest
@ -11,21 +12,32 @@ from scipy.stats import pearsonr
from junifer.datareader.default import DefaultDataReader
from junifer.markers.reho.reho_estimator import ReHoEstimator
from junifer.pipeline import WorkDirManager
from junifer.pipeline.utils import _check_afni
from junifer.testing.datagrabbers import PartlyCloudyTestingDataGrabber
from junifer.utils.logging import logger
def test_reho_estimator_cache_python() -> None:
"""Test that the cache works properly when using Python implementation."""
def test_reho_estimator_cache_python(tmp_path: Path) -> None:
"""Test that the cache works properly when using Python implementation.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Get subject from datagrabber
with PartlyCloudyTestingDataGrabber() as dg:
subject = dg["sub-01"]
# Read data for subject
subject_data = DefaultDataReader().fit_transform(subject)
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Setup estimator
reho_estimator = ReHoEstimator()
# Compute without cache
first_tic = time.time()
reho_map_without_cache = reho_estimator.fit_transform(
use_afni=False,
@ -37,11 +49,8 @@ def test_reho_estimator_cache_python() -> None:
f"ReHo estimator in Python without cache: {first_toc - first_tic}"
)
assert isinstance(reho_map_without_cache, nib.Nifti1Image)
# Count intermediate files
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now fit again, should be faster
# Compute again with cache, should be faster
second_tic = time.time()
reho_map_with_cache = reho_estimator.fit_transform(
use_afni=False,
@ -55,12 +64,8 @@ def test_reho_estimator_cache_python() -> None:
assert isinstance(reho_map_with_cache, nib.Nifti1Image)
# Check that cache is being used
assert (second_toc - second_tic) < ((first_toc - first_tic) / 1000)
# Count intermediate files
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now change a parameter, should compute again, without clearing the
# cache
# Change a parameter and compute again without cache
third_tic = time.time()
reho_map_with_partial_cache = reho_estimator.fit_transform(
use_afni=False,
@ -74,11 +79,8 @@ def test_reho_estimator_cache_python() -> None:
assert isinstance(reho_map_with_partial_cache, nib.Nifti1Image)
# Should require more time
assert (third_toc - third_tic) > ((first_toc - first_tic) / 10)
# Count intermediate files
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now fit again with the previous params, should be fast
# Compute again with cache, should be faster
fourth_tic = time.time()
reho_map_with_new_cache = reho_estimator.fit_transform(
use_afni=False,
@ -92,11 +94,8 @@ def test_reho_estimator_cache_python() -> None:
assert isinstance(reho_map_with_new_cache, nib.Nifti1Image)
# Should require less time
assert (fourth_toc - fourth_tic) < ((first_toc - first_tic) / 1000)
# Count intermediate files
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
# Now change the data, it should clear the cache
# Change the data and it should clear the cache
with PartlyCloudyTestingDataGrabber() as dg:
subject = dg["sub-02"]
# Read data for new subject
@ -116,24 +115,31 @@ def test_reho_estimator_cache_python() -> None:
assert isinstance(reho_map_with_different_cache, nib.Nifti1Image)
# Should take less time
assert (fifth_toc - fifth_tic) > ((first_toc - first_tic) / 10)
# Count intermediate files
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 0 # no files in python
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_reho_estimator_cache_afni() -> None:
"""Test that the cache works properly when using afni."""
def test_reho_estimator_cache_afni(tmp_path: Path) -> None:
"""Test that the cache works properly when using afni.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Get subject from datagrabber
with PartlyCloudyTestingDataGrabber() as dg:
subject = dg["sub-01"]
# Read data for subject
subject_data = DefaultDataReader().fit_transform(subject)
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Setup estimator
reho_estimator = ReHoEstimator()
# Compute without cache
first_tic = time.time()
reho_map_without_cache = reho_estimator.fit_transform(
use_afni=True,
@ -149,7 +155,7 @@ def test_reho_estimator_cache_afni() -> None:
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 2 # input + reho
# Now fit again, should be faster
# Compute again with cache, should be faster
second_tic = time.time()
reho_map_with_cache = reho_estimator.fit_transform(
use_afni=True,
@ -166,8 +172,7 @@ def test_reho_estimator_cache_afni() -> None:
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 2 # input + reho
# Now change a parameter, should compute again, without clearing the
# cache
# Change a parameter and compute again without cache
third_tic = time.time()
reho_map_with_partial_cache = reho_estimator.fit_transform(
use_afni=True,
@ -183,9 +188,9 @@ def test_reho_estimator_cache_afni() -> None:
assert (third_toc - third_tic) > ((first_toc - first_tic) / 10)
# Count intermediate files
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 3 # input + 2 * reho
assert n_files == 2 # input + reho
# Now fit again with the previous params, should be fast
# Compute with cache, should be faster
fourth_tic = time.time()
reho_map_with_new_cache = reho_estimator.fit_transform(
use_afni=True,
@ -200,9 +205,9 @@ def test_reho_estimator_cache_afni() -> None:
# Should require less time
assert (fourth_toc - fourth_tic) < ((first_toc - first_tic) / 1000)
n_files = len(list(reho_estimator.temp_dir_path.glob("*")))
assert n_files == 3 # input + 2 * reho
assert n_files == 2 # input + reho
# Now change the data, it should clear the cache
# Change the data and it should clear the cache
with PartlyCloudyTestingDataGrabber() as dg:
subject = dg["sub-02"]
# Read data for new subject
@ -229,13 +234,22 @@ def test_reho_estimator_cache_afni() -> None:
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_reho_estimator_afni_vs_python() -> None:
"""Compare afni and Python implementations."""
def test_reho_estimator_afni_vs_python(tmp_path: Path) -> None:
"""Compare afni and Python implementations.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
# Get subject from datagrabber
with PartlyCloudyTestingDataGrabber() as dg:
subject = dg["sub-01"]
# Read data for subject
subject_data = DefaultDataReader().fit_transform(subject)
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Setup estimator
reho_estimator = ReHoEstimator()

29
junifer/markers/reho/tests/test_reho_parcels.py
View file

@ -10,6 +10,7 @@ from nilearn import image as nimg
from scipy.stats import pearsonr
from junifer.markers.reho.reho_parcels import ReHoParcels
from junifer.pipeline import WorkDirManager
from junifer.pipeline.utils import _check_afni
from junifer.storage.sqlite import SQLiteFeatureStorage
from junifer.testing.datagrabbers import SPMAuditoryTestingDataGrabber
@ -18,13 +19,22 @@ from junifer.testing.datagrabbers import SPMAuditoryTestingDataGrabber
PARCELLATION = "Schaefer100x7"
def test_reho_parcels_computation() -> None:
"""Test ReHoParcels fit-transform."""
def test_reho_parcels_computation(tmp_path: Path) -> None:
"""Test ReHoParcels fit-transform.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with SPMAuditoryTestingDataGrabber() as dg:
# Use first subject
subject_data = dg["sub001"]
# Load image to memory
fmri_img = nimg.load_img(subject_data["BOLD"]["path"])
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Initialize marker
reho_parcels_marker = ReHoParcels(parcellation=PARCELLATION)
# Fit transform marker on data
@ -49,14 +59,23 @@ def test_reho_parcels_computation() -> None:
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_reho_parcels_computation_comparison() -> None:
"""Test ReHoParcels fit-transform implementation comparison.."""
def test_reho_parcels_computation_comparison(tmp_path: Path) -> None:
"""Test ReHoParcels fit-transform implementation comparison.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with SPMAuditoryTestingDataGrabber() as dg:
# Use first subject
subject_data = dg["sub001"]
# Load image to memory
fmri_img = nimg.load_img(subject_data["BOLD"]["path"])
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Initialize marker with use_afni=False
reho_parcels_marker_python = ReHoParcels(
parcellation=PARCELLATION, use_afni=False
@ -101,6 +120,8 @@ def test_reho_parcels_storage(tmp_path: Path) -> None:
subject_data = dg["sub001"]
# Load image to memory
fmri_img = nimg.load_img(subject_data["BOLD"]["path"])
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Initialize marker
reho_parcels_marker = ReHoParcels(parcellation=PARCELLATION)
# Initialize storage

29
junifer/markers/reho/tests/test_reho_spheres.py
View file

@ -10,6 +10,7 @@ from nilearn import image as nimg
from scipy.stats import pearsonr
from junifer.markers.reho.reho_spheres import ReHoSpheres
from junifer.pipeline import WorkDirManager
from junifer.pipeline.utils import _check_afni
from junifer.storage.sqlite import SQLiteFeatureStorage
from junifer.testing.datagrabbers import SPMAuditoryTestingDataGrabber
@ -18,13 +19,22 @@ from junifer.testing.datagrabbers import SPMAuditoryTestingDataGrabber
COORDINATES = "DMNBuckner"
def test_reho_spheres_computation() -> None:
"""Test ReHoSpheres fit-transform."""
def test_reho_spheres_computation(tmp_path: Path) -> None:
"""Test ReHoSpheres fit-transform.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with SPMAuditoryTestingDataGrabber() as dg:
# Use first subject
subject_data = dg["sub001"]
# Load image to memory
fmri_img = nimg.load_img(subject_data["BOLD"]["path"])
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Initialize marker
reho_spheres_marker = ReHoSpheres(coords=COORDINATES, radius=10.0)
# Fit transform marker on data
@ -49,14 +59,23 @@ def test_reho_spheres_computation() -> None:
@pytest.mark.skipif(
_check_afni() is False, reason="requires afni to be in PATH"
)
def test_reho_spheres_computation_comparison() -> None:
"""Test ReHoSpheres fit-transform implementation comparison.."""
def test_reho_spheres_computation_comparison(tmp_path: Path) -> None:
"""Test ReHoSpheres fit-transform implementation comparison.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
with SPMAuditoryTestingDataGrabber() as dg:
# Use first subject
subject_data = dg["sub001"]
# Load image to memory
fmri_img = nimg.load_img(subject_data["BOLD"]["path"])
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Initialize marker with use_afni=False
reho_spheres_marker_python = ReHoSpheres(
coords=COORDINATES, radius=10.0, use_afni=False
@ -101,6 +120,8 @@ def test_reho_spheres_storage(tmp_path: Path) -> None:
subject_data = dg["sub001"]
# Load image to memory
fmri_img = nimg.load_img(subject_data["BOLD"]["path"])
# Update workdir to current test's tmp_path
WorkDirManager().workdir = tmp_path
# Initialize marker
reho_spheres_marker = ReHoSpheres(coords=COORDINATES, radius=10.0)
# Initialize storage

41
junifer/markers/utils.py
View file

@ -7,7 +7,7 @@
# Federico Raimondo <f.raimondo@fz-juelich.de>
# License: AGPL
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
from typing import Callable, List, Optional, Tuple, Union
import numpy as np
import pandas as pd
@ -16,45 +16,6 @@ from scipy.stats import zscore
from ..utils import raise_error
def singleton(cls: Type) -> Type:
"""Make a class singleton.
Parameters
----------
cls : class
The class to designate as singleton.
Returns
-------
class
The only instance of the class.
"""
instances: Dict = {}
def get_instance(*args: Any, **kwargs: Any) -> Type:
"""Get the only instance for a class.
Parameters
----------
*args : tuple
The positional arguments to pass to the class.
**kwargs : dict
The keyword arguments to pass to the class.
Returns
-------
class
The only instance of the class.
"""
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
def _ets(
bold_ts: np.ndarray,
roi_names: Union[None, List[str]] = None,

1
junifer/pipeline/__init__.py
View file

@ -6,3 +6,4 @@
from . import registry
from .pipeline_step_mixin import PipelineStepMixin
from .update_meta_mixin import UpdateMetaMixin
from .workdir_manager import WorkDirManager

45
junifer/pipeline/singleton.py Normal file
View file

@ -0,0 +1,45 @@
"""Provide a singleton class to be used by pipeline components."""
# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from typing import Any, Dict, Type
def singleton(cls: Type) -> Type:
"""Make a class singleton.
Parameters
----------
cls : class
The class to designate as singleton.
Returns
-------
class
The only instance of the class.
"""
instances: Dict = {}
def get_instance(*args: Any, **kwargs: Any) -> Type:
"""Get the only instance for a class.
Parameters
----------
*args : tuple
The positional arguments to pass to the class.
**kwargs : dict
The keyword arguments to pass to the class.
Returns
-------
class
The only instance of the class.
"""
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance

53
junifer/pipeline/tests/test_workdir_manager.py Normal file
View file

@ -0,0 +1,53 @@
"""Provide tests for WorkDirManager."""
# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from pathlib import Path
from junifer.pipeline import WorkDirManager
def test_workdir_manager_singleton() -> None:
"""Test that WorkDirManager is a singleton."""
workdir_mgr_1 = WorkDirManager()
workdir_mgr_2 = WorkDirManager()
assert id(workdir_mgr_1) == id(workdir_mgr_2)
def test_workdir_manager_workdir(tmp_path: Path) -> None:
"""Test WorkDirManager correctly sets workdir.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
workdir_mgr = WorkDirManager()
workdir_mgr.workdir = tmp_path
assert workdir_mgr.workdir == tmp_path
def test_workdir_manager_get_and_delete_tempdir(tmp_path: Path) -> None:
"""Test WorkDirManager gets and deletes temporary directories correctly.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
workdir_mgr = WorkDirManager()
workdir_mgr.workdir = tmp_path
# Check no root temporary directory
assert workdir_mgr.root_tempdir is None
tempdir = workdir_mgr.get_tempdir()
# Should create a temporary directory
assert workdir_mgr.root_tempdir is not None
workdir_mgr.delete_tempdir(tempdir)
workdir_mgr._cleanup()
# Should remove temporary directory
assert workdir_mgr.root_tempdir is None

162
junifer/pipeline/workdir_manager.py Normal file
View file

@ -0,0 +1,162 @@
"""Provide a work directory manager class to be used by pipeline components."""
fraimondo commented 2023-10-05 11:30:13 +00:00 (Migrated from github.com)
Copy link

I would not allow to set the workdir like this, for the resasons stated in the constructor

I would not allow to set the workdir like this, for the resasons stated in the constructor
fraimondo commented 2023-10-05 11:30:28 +00:00 (Migrated from github.com)
Copy link

no prefix here, the root tmp dir should be random.

no prefix here, the root tmp dir should be random.
# Authors: Synchon Mandal <s.mandal@fz-juelich.de>
# Federico Raimondo <f.raimondo@fz-juelich.de>
# License: AGPL
import shutil
import tempfile
from pathlib import Path
from typing import Optional, Union
from ..utils import logger
from .singleton import singleton
@singleton
class WorkDirManager:
"""Class for working directory manager.
This class is a singleton and is used for managing temporary and working
directories used across the pipeline by datagrabbers, preprocessors,
markers and so on. It maintains a single super-directory and provides
directories on-demand and cleans after itself thus keeping the user
filesystem clean.
Parameters
----------
workdir : str or pathlib.Path, optional
The path to the super-directory. If None, "TMPDIR/junifer" is used
where TMPDIR is the platform-dependent temporary directory.
Attributes
----------
workdir : pathlib.Path
The path to the working directory.
root_tempdir : pathlib.Path or None
The path to the root temporary directory.
"""
def __init__(self, workdir: Optional[Union[str, Path]] = None) -> None:
fraimondo commented 2023-10-05 11:29:40 +00:00 (Migrated from github.com)
Copy link

Since it's a singleton, any intialization should take care of cleaning up a previous one.

e.g:

WorkdirManager("/tmp/workdir1")
# Do some stuff that creates tempdirs
WorkdirManager("/tmp/workdir2")  # This should cleanup.

However, if a user does use it wrongly, it might cleanup in the middle.

So I would:

  1. add a cleanup=False parameter to the init
  2. Raise an error if needs cleanup but cleanup is False
Since it's a singleton, any intialization should take care of cleaning up a previous one. e.g: ``` WorkdirManager("/tmp/workdir1") # Do some stuff that creates tempdirs WorkdirManager("/tmp/workdir2") # This should cleanup. ``` However, if a user does use it wrongly, it might cleanup in the middle. So I would: 1) add a `cleanup=False` parameter to the init 2) Raise an error if needs cleanup but cleanup is False
synchon commented 2023-10-06 08:53:38 +00:00 (Migrated from github.com)
Copy link

I've made the changes as you proposed. The problem is that the __init__() doesn't fire the second time you try to create an instance as it's a singleton, so the previous instance is returned.

I've made the changes as you proposed. The problem is that the `__init__()` doesn't fire the second time you try to create an instance as it's a singleton, so the previous instance is returned.
fraimondo commented 2023-10-10 09:20:08 +00:00 (Migrated from github.com)
Copy link

Ok, so how can we fix it?

Ok, so how can we fix it?
synchon commented 2023-10-10 09:28:31 +00:00 (Migrated from github.com)
Copy link

The only way I know of interacting with singletons in principle is to use getters and setters.

The only way I know of interacting with singletons in principle is to use getters and setters.
fraimondo commented 2023-10-10 09:36:54 +00:00 (Migrated from github.com)
Copy link

lets do that then

lets do that then
"""Initialize the class."""
self._workdir = workdir
self._root_tempdir = None
self._set_default_workdir()
def _set_default_workdir(self) -> None:
"""Set the default working directory if not set already."""
# Check and set topmost level directory if not provided
if self._workdir is None:
self._workdir = Path(tempfile.gettempdir()) / "junifer"
# Create directory if not found
if not self._workdir.is_dir():
logger.debug(
"Creating working directory at "
f"{self._workdir.resolve()!s}"
)
self._workdir.mkdir(parents=True)
logger.debug(
f"Setting working directory to {self._workdir.resolve()!s}"
)
def __del__(self) -> None:
"""Destructor."""
self._cleanup()
def _cleanup(self) -> None:
"""Clean up the temporary directories."""
# Remove root temporary directory
if self._root_tempdir is not None:
logger.debug(
"Deleting temporary directory at "
f"{self._root_tempdir.resolve()!s}"
)
shutil.rmtree(self._root_tempdir, ignore_errors=True)
self._root_tempdir = None
@property
def workdir(self) -> Path:
"""Get working directory."""
return self._workdir # type: ignore
@workdir.setter
def workdir(self, path: Union[str, Path]) -> None:
"""Set working directory.
The directory path is created if it doesn't exist yet.
Parameters
----------
path : str or pathlib.Path
The path to the working directory.
"""
# Check if existing working directory is same or not;
# if not, then clean up
if self._workdir != Path(path):
self._cleanup()
# Set working directory
self._workdir = Path(path)
logger.debug(
f"Changing working directory to {self._workdir.resolve()!s}"
)
# Create directory if not found
if not self._workdir.is_dir():
logger.debug(
f"Creating working directory at {self._workdir.resolve()!s}"
)
self._workdir.mkdir(parents=True)
@property
def root_tempdir(self) -> Optional[Path]:
"""Get root temporary directory."""
return self._root_tempdir
def get_tempdir(
self, prefix: Optional[str] = None, suffix: Optional[str] = None
) -> Path:
"""Get a temporary directory.
Parameters
----------
prefix : str
The temporary directory prefix. If None, a default prefix is used.
suffix : str
The temporary directory suffix. If None, no suffix is added.
Returns
-------
pathlib.Path
The path to the temporary directory.
"""
# Create root temporary directory if not created already
if self._root_tempdir is None:
logger.debug(
"Setting up temporary directory under "
f"{self._workdir.resolve()!s}" # type: ignore
)
self._root_tempdir = Path(tempfile.mkdtemp(dir=self._workdir))
logger.debug(
f"Creating temporary directory at {self._root_tempdir.resolve()!s}"
)
return Path(
tempfile.mkdtemp(
dir=self._root_tempdir, prefix=prefix, suffix=suffix
)
)
def delete_tempdir(self, tempdir: Path) -> None:
"""Delete a temporary directory.
Parameters
----------
tempdir : pathlib.Path
The temporary directory path to be deleted.
"""
logger.debug(f"Deleting temporary directory at {tempdir}")
shutil.rmtree(tempdir, ignore_errors=True)