Pipeline running configuration (Inria-Empenn#148)

* [BUG] inside unit_tests workflow

* [DOC] runner help

* Creating entry-points for the project

* [DOC] command line tools

* [DOC] command line tools

* Adding a tester command line tool

* [DATALAD] change results url

* Runner configuration

* Remove dir func in

* Runner always stops on first crash

* [TEST][helpers] not failing test if correlation under threshold

* [DOC] narps_open.core.common

.github/workflows/test_changes.yml

@@ -58,5 +58,5 @@ jobs:
     - name: Execute tests with pytest
       run: |
         if [[ "${{ needs.identify-tests.outputs.tests }}" != "" ]]; then
-          pytest -s -q ${{ needs.identify-tests.outputs.tests }}
+          pytest -s -q ${{ needs.identify-tests.outputs.tests }} -m "not pipeline_test"
         fi

docs/core.md

@@ -72,6 +72,15 @@ from narps_open.core.common import remove_file
 remove_file('/path/to/the/image.nii.gz')
 ```
 
+* `remove_directory` remove a directory when it is not needed anymore (to save disk space)
+
+```python
+from narps_open.core.common import remove_directory
+
+# Remove the directory /path/to/
+remove_directory('/path/to/')
+```
+
 * `elements_in_string` : return the first input parameter if it contains one element of second parameter (None otherwise).
 
 ```python

narps_open/core/common.py

@@ -20,6 +20,20 @@ def remove_file(_, file_name: str) -> None:
     except OSError as error:
         print(error)
 
+def remove_directory(_, directory_name: str) -> None:
+    """
+    Fully remove directory generated by a Node, once it is not needed anymore.
+    This function is meant to be used in a Nipype Function Node.
+
+    Parameters:
+    - _: input only used for triggering the Node
+    - directory_name: str, a single absolute path of the directory to remove
+    """
+    # This import must stay inside the function, as required by Nipype
+    from shutil import rmtree
+
+    rmtree(directory_name, ignore_errors = True)
+
 def elements_in_string(input_str: str, elements: list) -> str: #| None:
     """
     Return input_str if it contains one element of the elements list.

narps_open/runner.py

@@ -8,7 +8,7 @@
 from random import choices
 from argparse import ArgumentParser
 
-from nipype import Workflow
+from nipype import Workflow, config
 
 from narps_open.pipelines import Pipeline, implemented_pipelines
 from narps_open.data.participants import (
@@ -96,6 +96,10 @@ def start(self, first_level_only: bool = False, group_level_only: bool = False)
                 (= preprocessing + run level + subject_level)
             - group_level_only: bool (False by default), run the group level workflows only
         """
+        # Set global nipype config for pipeline execution
+        config.update_config(dict(execution = {'stop_on_first_crash': 'True'}))
+
+        # Disclaimer
         print('Starting pipeline for team: '+
             f'{self.team_id}, with {len(self.subjects)} subjects: {self.subjects}')
 
@@ -127,15 +131,15 @@ def start(self, first_level_only: bool = False, group_level_only: bool = False)
                         raise AttributeError('Workflow must be of type nipype.Workflow')
 
                     if nb_procs > 1:
-                        sub_workflow.run('MultiProc', plugin_args={'n_procs': nb_procs})
+                        sub_workflow.run('MultiProc', plugin_args = {'n_procs': nb_procs})
                     else:
                         sub_workflow.run()
             else:
                 if not isinstance(workflow, Workflow):
                     raise AttributeError('Workflow must be of type nipype.Workflow')
 
                 if nb_procs > 1:
-                    workflow.run('MultiProc', plugin_args={'n_procs': nb_procs})
+                    workflow.run('MultiProc', plugin_args = {'n_procs': nb_procs})
                 else:
                     workflow.run()
 

narps_open/utils/configuration/default_config.toml

@@ -10,5 +10,8 @@ narps_results = "data/results/"
 [runner]
 nb_procs = 8 # Maximum number of threads executed by the runner
 
+[pipelines]
+remove_unused_data = true # set to true to activate remove nodes of pipelines
+
 [results]
 neurovault_naming = true # true if results files are saved using the neurovault naming, false if they use naming of narps

narps_open/utils/configuration/testing_config.toml

@@ -13,6 +13,9 @@ test_runs = "run/"
 nb_procs = 8 # Maximum number of threads executed by the runner
 nb_trials = 3 # Maximum number of executions to have the pipeline executed completely
 
+[pipelines]
+remove_unused_data = true # set to true to activate remove nodes of pipelines
+
 [results]
 neurovault_naming = true # true if results files are saved using the neurovault naming, false if they use naming of narps
 

tests/conftest.py

@@ -158,4 +158,5 @@ def test_pipeline_evaluation(team_id: str):
             file.write('success' if passed else 'failure')
             file.write(f' | {[round(i, 2) for i in results]} |\n')
 
-        assert passed
+        if not passed:
+            break

tests/core/test_common.py

@@ -10,7 +10,7 @@
     pytest -q test_common.py
     pytest -q test_common.py -k <selected_test>
 """
-from os import mkdir
+from os import mkdir, makedirs
 from os.path import join, exists, abspath
 from shutil import rmtree
 from pathlib import Path
@@ -59,6 +59,34 @@ def test_remove_file(remove_test_dir):
 
         # Check file is removed
         assert not exists(test_file_path)
+
+    @staticmethod
+    @mark.unit_test
+    def test_remove_directory(remove_test_dir):
+        """ Test the remove_directory function """
+
+        # Create a single inside dir tree
+        dir_path = abspath(join(TEMPORARY_DIR, 'dir_1', 'dir_2'))
+        makedirs(dir_path)
+        file_path = abspath(join(TEMPORARY_DIR, 'dir_1', 'dir_2', 'file1.txt'))
+        Path(file_path).touch()
+        test_dir_path = abspath(join(TEMPORARY_DIR, 'dir_1'))
+
+        # Check file exist
+        assert exists(file_path)
+
+        # Create a Nipype Node using remove_files
+        test_remove_dir_node = Node(Function(
+            function = co.remove_directory,
+            input_names = ['_', 'directory_name'],
+            output_names = []
+            ), name = 'test_remove_dir_node')
+        test_remove_dir_node.inputs._ = ''
+        test_remove_dir_node.inputs.directory_name = test_dir_path
+        test_remove_dir_node.run()
+
+        # Check file is removed
+        assert not exists(test_dir_path)
 
     @staticmethod
     @mark.unit_test