docs/changes/newsfragments/185.bugfix

@@ -0,0 +1 @@
+Fix a bug in which fitting a marker (e.g. ``SphereAggregation``) on a specific type (e.g.: ``BOLD``) will fail if another non-supported type (e.g.: ``BOLD_confounds``) is present in the data object by `Fede Raimondo`_

docs/changes/newsfragments/185.enh

@@ -0,0 +1 @@
+Improved logging output for preprocessing, collecting and pipeline building from YAML by `Fede Raimondo`_

docs/changes/newsfragments/201.enh

@@ -1 +1 @@
-Force datalad to be non-interactive on _queued_ jobs by `Fede Raimondo`_
+Force datalad to be non-interactive on *queued* jobs by `Fede Raimondo`_

junifer/api/functions.py

@@ -512,17 +512,21 @@ def _queue_condor(
             )
         if collect == "yes":
             dag_file.write(f"FINAL collect {submit_collect_fname}\n")
-            dag_file.write("SCRIPT PRE collect collect_pre.pl $DAG_STATUS\n")
-            collect_pre_fname = jobdir / "collect_pre.pl"
+            collect_pre_fname = jobdir / "collect_pre.sh"
+            dag_file.write(
+                f"SCRIPT PRE collect {collect_pre_fname.as_posix()} "
+                "$DAG_STATUS\n")
             with open(collect_pre_fname, "w") as pre_file:
-                pre_file.write("#!/usr/bin/env perl\n\n")
-                pre_file.write("if ($ARGV[0] eq 4) {\n")
-                pre_file.write("    exit(1);\n")
-                pre_file.write("}\n")
+                pre_file.write("#!/bin/bash\n\n")
+                pre_file.write("if [ \"${1}\" == \"4\" ]; then\n")
+                pre_file.write("    exit 1\n")
+                pre_file.write("fi\n")
+
+            make_executable(collect_pre_fname)
         elif collect == "on_success_only":
             dag_file.write(f"JOB collect {submit_collect_fname}\n")
             dag_file.write("PARENT ")
-            for i_job, _t_elem in enumerate(elements):
+            for i_job, _ in enumerate(elements):
                 dag_file.write(f"run{i_job} ")
             dag_file.write("CHILD collect\n\n")
 

junifer/api/tests/test_functions.py

@@ -755,9 +755,14 @@ def test_queue_condor_assets_generation(
 
             if has_final_collect_job is True:
                 pre_collect_fname = Path(
-                    tmp_path / "junifer_jobs" / jobname / "collect_pre.pl"
+                    tmp_path / "junifer_jobs" / jobname / "collect_pre.sh"
                 )
                 assert pre_collect_fname.exists()
+                assert (
+                    stat.S_IMODE(pre_collect_fname.stat().st_mode)
+                    & stat.S_IEXEC
+                    != 0
+                )
 
             # Check submit log
             assert (

junifer/datagrabber/datalad_base.py

@@ -72,7 +72,7 @@ class DataladDataGrabber(BaseDataGrabber):
         **kwargs,
     ):
         if datadir is None:
-            logger.warning("`datadir` is None, creating a temporary directory")
+            logger.info("`datadir` is None, creating a temporary directory")
             # Create temporary directory
             tmpdir = Path(tempfile.mkdtemp())
             datadir = tmpdir / "datadir"

junifer/datareader/default.py

@@ -34,7 +34,7 @@ _readers["TSV"] = {"func": pd.read_csv, "params": {"sep": "\t"}}
 class DefaultDataReader(PipelineStepMixin, UpdateMetaMixin):
     """Mixin class for default data reader."""
 
-    def validate_input(self, input: List[str]) -> None:
+    def validate_input(self, input: List[str]) -> List[str]:
         """Validate input.
 
         Parameters
@@ -43,9 +43,14 @@ class DefaultDataReader(PipelineStepMixin, UpdateMetaMixin):
             The input to the pipeline step. The list must contain the
             available Junifer Data dictionary keys.
 
+        Returns
+        -------
+        list of str
+            The actual elements of the input that will be processed by this
+            pipeline step.
         """
         # Nothing to validate, any input is fine
-        pass
+        return input
 
     def get_output_type(self, input: List[str]) -> List[str]:
         """Get output type.

junifer/datareader/tests/test_default_reader.py

@@ -29,7 +29,7 @@ def test_validation(type_) -> None:
 
     """
     reader = DefaultDataReader()
-    assert reader.validate_input(type_) is None
+    assert reader.validate_input(type_) == type_
     assert reader.get_output_type(type_) == type_
     assert reader.validate(type_) == type_
 

junifer/markers/base.py

@@ -58,7 +58,7 @@ class BaseMarker(ABC, PipelineStepMixin, UpdateMetaMixin):
             klass=NotImplementedError,
         )
 
-    def validate_input(self, input: List[str]) -> None:
+    def validate_input(self, input: List[str]) -> List[str]:
         """Validate input.
 
         Parameters
@@ -67,6 +67,12 @@ class BaseMarker(ABC, PipelineStepMixin, UpdateMetaMixin):
             The input to the pipeline step. The list must contain the
             available Junifer Data dictionary keys.
 
+        Returns
+        -------
+        list of str
+            The actual elements of the input that will be processed by this
+            pipeline step.
+
         Raises
         ------
         ValueError
@@ -79,6 +85,7 @@ class BaseMarker(ABC, PipelineStepMixin, UpdateMetaMixin):
                 f"\t Input: {input}"
                 f"\t Required (any of): {self._on}"
             )
+        return [x for x in self._on if x in input]
 
     @abstractmethod
     def get_output_type(self, input_type: str) -> str:

junifer/markers/tests/test_markers_base.py

@@ -27,7 +27,9 @@ def test_base_marker_subclassing() -> None:
             return ["BOLD", "T1w"]
 
         def get_output_type(self, input):
-            return ["timeseries"]
+            if input == "BOLD":
+                return "timeseries"
+            raise ValueError(f"Cannot compute output type for {input}")
 
         def compute(self, input, extra_input):
             return {
@@ -56,6 +58,8 @@ def test_base_marker_subclassing() -> None:
     with pytest.raises(ValueError, match="not have the required data"):
         marker.validate_input(["T1w"])
 
+    assert marker.validate_input(["BOLD", "Other"]) == ["BOLD"]
+
     output = marker.fit_transform(input=input_)  # process
     # Check output
     assert "BOLD" in output
@@ -75,3 +79,30 @@ def test_base_marker_subclassing() -> None:
 
     # Check attributes
     assert marker.name == "MyBaseMarker"
+
+    # Add one extra input that will not be used to compute
+    input_ = {
+        "BOLD": {
+            "path": ".",
+            "data": "data",
+            "meta": {
+                "datagrabber": "dg",
+                "element": "elem",
+                "datareader": "dr",
+            },
+        },
+        "T2": {
+            "path": ".",
+            "data": "data",
+            "meta": {
+                "datagrabber": "dg",
+                "element": "elem",
+                "datareader": "dr",
+            },
+        }
+    }
+    marker = MyBaseMarker(on=["BOLD"])
+    output = marker.fit_transform(input=input_)  # process
+    # Check output
+    assert "BOLD" in output
+    assert "T2" not in output

junifer/pipeline/pipeline_step_mixin.py

@@ -20,7 +20,7 @@ from .utils import check_ext_dependencies
 class PipelineStepMixin:
     """Mixin class for a pipeline step."""
 
-    def validate_input(self, input: List[str]) -> None:
+    def validate_input(self, input: List[str]) -> List[str]:
         """Validate the input to the pipeline step.
 
         Parameters
@@ -29,6 +29,12 @@ class PipelineStepMixin:
             The input to the pipeline step. The list must contain the
             available Junifer Data dictionary keys.
 
+        Returns
+        -------
+        list of str
+            The actual elements of the input that will be processed by this
+            pipeline step.
+
         Raises
         ------
         ValueError
@@ -132,8 +138,8 @@ class PipelineStepMixin:
                     # Set attribute for using external tools
                     setattr(self, f"use_{dependency['name']}", out)
 
-        self.validate_input(input=input)
-        outputs = [self.get_output_type(t_input) for t_input in input]
+        fit_input = self.validate_input(input=input)
+        outputs = [self.get_output_type(t_input) for t_input in fit_input]
         return outputs
 
     def fit_transform(

junifer/pipeline/registry.py

@@ -132,9 +132,22 @@ def build(
     if init_params is None:
         init_params = {}
     # Get class of the registered function
+    logger.debug(f"Building {step}/{name}")
     klass = get_class(step=step, name=name)
+    logger.debug(f"\tClass: {klass.__name__}")
+    logger.debug(f"\tInit params: {init_params}")
+    try:
         # Create instance of the class
         object_ = klass(**init_params)
+    except Exception as e:
+        raise_error(
+            msg=(
+                f"Failed to create {step} ({name}). "
+                f"Error: {e}"
+            ),
+            klass=RuntimeError,
+            exception=e,
+        )
     # Verify created instance belongs to the base class
     if not isinstance(object_, baseclass):
         raise_error(

junifer/pipeline/tests/test_pipeline_step_mixin.py

@@ -32,8 +32,8 @@ def test_pipeline_step_mixin_validate_correct_dependencies() -> None:
 
         _DEPENDENCIES = {"setuptools"}
 
-        def validate_input(self, input: List[str]) -> None:
-            print(input)
+        def validate_input(self, input: List[str]) -> List[str]:
+            return input
 
         def get_output_type(self, input_type: str) -> str:
             return input_type
@@ -53,8 +53,8 @@ def test_pipeline_step_mixin_validate_incorrect_dependencies() -> None:
 
         _DEPENDENCIES = {"foobar"}
 
-        def validate_input(self, input: List[str]) -> None:
-            print(input)
+        def validate_input(self, input: List[str]) -> List[str]:
+            return input
 
         def get_output_type(self, input_type: str) -> str:
             return input_type
@@ -78,8 +78,8 @@ def test_pipeline_step_mixin_validate_correct_ext_dependencies() -> None:
 
         _EXT_DEPENDENCIES = [{"name": "afni", "optional": False}]
 
-        def validate_input(self, input: List[str]) -> None:
-            print(input)
+        def validate_input(self, input: List[str]) -> List[str]:
+            return input
 
         def get_output_type(self, input_type: str) -> str:
             return input_type
@@ -104,8 +104,8 @@ def test_pipeline_step_mixin_validate_ext_deps_correct_commands() -> None:
             {"name": "afni", "optional": False, "commands": ["3dReHo"]}
         ]
 
-        def validate_input(self, input: List[str]) -> None:
-            print(input)
+        def validate_input(self, input: List[str]) -> List[str]:
+            return input
 
         def get_output_type(self, input_type: str) -> str:
             return input_type
@@ -132,8 +132,8 @@ def test_pipeline_step_mixin_validate_ext_deps_incorrect_commands() -> None:
             {"name": "afni", "optional": False, "commands": ["3d"]}
         ]
 
-        def validate_input(self, input: List[str]) -> None:
-            print(input)
+        def validate_input(self, input: List[str]) -> List[str]:
+            return input
 
         def get_output_type(self, input_type: str) -> str:
             return input_type
@@ -154,8 +154,8 @@ def test_pipeline_step_mixin_validate_incorrect_ext_dependencies() -> None:
 
         _EXT_DEPENDENCIES = [{"name": "foobar", "optional": True}]
 
-        def validate_input(self, input: List[str]) -> None:
-            print(input)
+        def validate_input(self, input: List[str]) -> List[str]:
+            return input
 
         def get_output_type(self, input_type: str) -> str:
             return input_type

junifer/pipeline/tests/test_registry.py

@@ -139,3 +139,12 @@ def test_build():
     # Check error
     with pytest.raises(ValueError, match="Must inherit"):
         build(step="datagrabber", name="concrete", baseclass=np.ndarray)
+
+    # Check error
+    with pytest.raises(RuntimeError, match="Failed to create"):
+        build(
+            step="datagrabber",
+            name="concrete",
+            baseclass=SuperClass,
+            init_params={"wrong": 2},
+        )

junifer/preprocess/base.py

@@ -36,7 +36,7 @@ class BasePreprocessor(ABC, PipelineStepMixin, UpdateMetaMixin):
             raise ValueError(f"{name} cannot be computed on {wrong_on}")
         self._on = on
 
-    def validate_input(self, input: List[str]) -> None:
+    def validate_input(self, input: List[str]) -> List[str]:
         """Validate input.
 
         Parameters
@@ -45,6 +45,12 @@ class BasePreprocessor(ABC, PipelineStepMixin, UpdateMetaMixin):
             The input to the pipeline step. The list must contain the
             available Junifer Data dictionary keys.
 
+        Returns
+        -------
+        list of str
+            The actual elements of the input that will be processed by this
+            pipeline step.
+
         Raises
         ------
         ValueError
@@ -56,6 +62,7 @@ class BasePreprocessor(ABC, PipelineStepMixin, UpdateMetaMixin):
                 f"\t Input: {input}"
                 f"\t Required (any of): {self._on}"
             )
+        return [x for x in self._on if x in input]
 
     @abstractmethod
     def get_output_type(self, input: List[str]) -> List[str]:
@@ -113,22 +120,27 @@ class BasePreprocessor(ABC, PipelineStepMixin, UpdateMetaMixin):
         out = input
         for type_ in self._on:
             if type_ in input.keys():
-                logger.info(f"Computing {type_}")
+                logger.info(f"Preprocessing {type_}")
                 t_input = input[type_]
 
                 # Pass the other data types as extra input, removing
                 # the current type
                 extra_input = input
                 extra_input.pop(type_)
+                logger.debug(
+                    f"Extra input for preprocess: {extra_input.keys()}"
+                )
                 key, t_out = self.preprocess(
                     input=t_input, extra_input=extra_input
                 )
 
                 # Add the output to the Junifer Data object
+                logger.debug(f"Adding {key} to output")
                 out[key] = t_out
 
                 # In case we are creating a new type, re-add the original input
                 if key != type_:
+                    logger.debug("Adding original input back to output")
                     out[type_] = t_input
 
                 self.update_meta(out[key], "preprocess")

junifer/preprocess/confounds/fmriprep_confound_remover.py

@@ -200,7 +200,7 @@ class fMRIPrepConfoundRemover(BasePreprocessor):
             )
         super().__init__()
 
-    def validate_input(self, input: List[str]) -> None:
+    def validate_input(self, input: List[str]) -> List[str]:
         """Validate the input to the pipeline step.
 
         Parameters
@@ -208,6 +208,11 @@ class fMRIPrepConfoundRemover(BasePreprocessor):
         input : list of str
             The input to the pipeline step. The list must contain the
             available Junifer Data object keys.
+        Returns
+        -------
+        list of str
+            The actual elements of the input that will be processed by this
+            pipeline step.
 
         Raises
         ------
@@ -224,6 +229,8 @@ class fMRIPrepConfoundRemover(BasePreprocessor):
                 klass=ValueError,
             )
 
+        return [x for x in self._on if x in input]
+
     def get_output_type(self, input: List[str]) -> List[str]:
         """Get the kind of the pipeline step.
 

junifer/storage/hdf5.py

@@ -940,6 +940,14 @@ class HDF5FeatureStorage(BaseFeatureStorage):
             f"{self.uri.parent}/*_{self.uri.name}"  # type: ignore
         )
         logger.info(f"Will collect {len(elements_per_feature_md5)} features.")
+
+        # Print info before to avoid tqdm progress bar interference
+        for feature_md5, element_files in elements_per_feature_md5.items():
+            logger.info(
+                f"Collecting {len(element_files)} files for feature MD5: "
+                f"{feature_md5}."
+            )
+
         for feature_md5, element_files in tqdm(
             elements_per_feature_md5.items(), desc="feature"
         ):