Repository: populse/capsul (language: python)

Issue:
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Create a POC engine with Celery
The goal of this work is to test Celery on a local machine to better understand how to use it. 
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Allow loading JSON pipelines as Python package resources
This addition allows to incorporate `.json` files containing custom pipeline definitions into a Python package. This is similar to the `.xml` pipelines that could be defined and imported in Capsul v2.
Create a POC engine with Celery
The goal of this work is to test Celery on a local machine to better understand how to use it. 
Create a POC engine with Celery
The goal of this work is to test Celery on a local machine to better understand how to use it. 
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.
Switch to redis database
This PR requires a change in the Singularity image to work. Therefore it must be discussed. However, it contains a reorganization of database and engine API. There is no much change but many code is now done in the database or in the `Engine` based class. Derived engined (such as `LocalEngine`) are much smaller and easier to understand.

This PR also adds the possibility to have an execution report (dispalying all jobs information including stdout, stderr, etc.). So it may be interesting to consider to merge it. I could go back to populse_db for the database if necessary.


Discussion / hints:

Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 700af2fb4..2a1395803 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -26,3 +26,5 @@ repos:
         name: flake8 under python3
         language_version: python3
         exclude: ^(doc/source|capsul)/sphinxext/
+        args: 
+        - --ignore=F722,F821
diff --git a/appveyor.yml b/appveyor.yml
deleted file mode 100644
index ab9d0f36f..000000000
--- a/appveyor.yml
+++ /dev/null
@@ -1,53 +0,0 @@
-image:
-  - Visual Studio 2019
-
-environment:
-
-  matrix:
-
-    - PYTHON: "C:\\Python39"
-    #- PYTHON: "C:\\Python310"
-
-build: off
-
-install:
-  - "SET PATH=%PYTHON%;%PYTHON%\\Scripts;%PATH%"
-  - "python -m pip install --upgrade pip"
-  - "pip install pydantic"
-  - "pip install numpy"
-  - "pip install six>=1.13"
-  - "pip install traits"
-  - IF "%PYTHON%"=="C:\Python27" pip install subprocess32
-  - "pip install python-dateutil"
-  - "pip install nipype"
-  #- "pip install soma-workflow"
-  - "git clone --depth=50 --branch=pydantic_controller https://github.com/populse/soma-base.git C:\\projects\\soma-base"
-  - "set MAINDIR=%CD%"
-  - "cd C:\\projects\\soma-base"
-  - "python setup.py install"
-  - "git clone --depth=50 --branch=master https://github.com/populse/soma-workflow.git C:\\projects\\soma-workflow"
-  - "cd C:\\projects\\soma-workflow"
-  - "python setup.py install"
-  - "git clone --depth=50 --branch=3.0 https://github.com/populse/populse_db.git C:\\projects\\populse_db"
-  - "cd C:\\projects\\populse_db"
-  - "python setup.py install"
-  ##- "SET PYTHONPATH=C:\\projects\\capsul\\python;%PYTHONPATH%"
-  #- "SET PYTHONPATH=C:\\projects\\soma-base\\python;%PYTHONPATH%"
-  #- "SET PYTHONPATH=C:\\projects\\soma-workflow\\python;%PYTHONPATH%"
-  #- "SET PYTHONPATH=C:\\projects\\populse_db\\python;%PYTHONPATH%"
-  - "cd %MAINDIR%"
-  - "python setup.py install"
-
-test_script:
-  - "python --version"
-  - echo %PYTHONPATH%
-  - "python -c \"import sys; print(sys.path)\""
-  - "python -m capsul.test"
-
-notifications:
-  - provider: Email
-    to:
-      - '{{commitAuthorEmail}}'
-    on_build_failure: true
-    on_build_success: false
-    on_build_status_changed: true
diff --git a/capsul/__main__.py b/capsul/__main__.py
index e69de29bb..80738aeff 100644
--- a/capsul/__main__.py
+++ b/capsul/__main__.py
@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+import sys
+
+from soma.qt_gui.qt_backend import QtGui
+
+from .api import Capsul
+from .config.configuration import ApplicationConfiguration
+from .qt_gui.widgets.settings_editor import SettingsEditor
+
+
+app_config = ApplicationConfiguration('global_config')
+app = QtGui.QApplication(sys.argv)
+capsul = Capsul()
+w = SettingsEditor(capsul.engine())
+w.show()
+app.exec_()
+del w
diff --git a/capsul/application.py b/capsul/application.py
index 210bc70f5..1c5626450 100644
--- a/capsul/application.py
+++ b/capsul/application.py
@@ -96,41 +96,17 @@ def executable(definition, **kwargs):
         '''
         return executable(definition, **kwargs)
 
-    def engine(self, name='local'):
-        ''' Get a :class:`~capsul.engine.CapsulEngine` instance
+    def engine(self, name='builtin'):
+        ''' Get a :class:`~capsul.engine.Engine` instance
         '''
-        # get engine type from config
-        resource_config = getattr(self.config, name, None)
-        if resource_config is None:
-            raise ValueError(f'resource "{name}" is not configured.')
-        engine_type = resource_config.engine_type
-        if engine_type == 'builtin':
-            engine_type = 'local'
-        try:
-            engine_mod = importlib.import_module(
-                f'capsul.engine.{engine_type}')
-        except ImportError:
-            raise ValueError(f'engine type {engine_type} is not known.')
-
         from .engine import Engine
-
-        # find en Engine subclass in the module
-        engines = []
-        for k, v in engine_mod.__dict__.items():
-            if isinstance(v, type) and v is not Engine \
-                    and issubclass(v, Engine):
-                engines.append((k, v))
-        if len(engines) == 0:
-            raise ValueError(
-                f'No Engine defined in module {engine_mod.__name__}')
-        if len(engines) > 1:
-            raise ValueError(
-                f'Several Engines are defined in module {engine_mod.__name__}'
-                f': {[e[1].__name__ for e in engines]}')
-
-        engine_class = engines[0][1]
-        return engine_class(name, resource_config)
-
+        
+        # get engine type from config
+        engine_config = getattr(self.config, name, None)
+        if engine_config is None:
+            raise ValueError(f'engine "{name}" is not configured.')
+        return Engine(name, engine_config)
+    
     def dataset(self, path):
         ''' Get a :class:`~.dataset.DataSet` instance associated with the given path
 
diff --git a/capsul/config/configuration.py b/capsul/config/configuration.py
index d41ee62aa..cee1a7cbe 100644
--- a/capsul/config/configuration.py
+++ b/capsul/config/configuration.py
@@ -4,6 +4,7 @@
 import os
 import sys
 import importlib
+import tempfile
 
 from soma.controller import (Controller, field,
                              OpenKeyDictController, File)
@@ -11,6 +12,12 @@
 
 from ..dataset import Dataset
 
+
+default_workers_type = 'builtin'
+default_database_url = f'sqlite://{tempfile.gettempdir()}{os.sep}capsul_engine_database.sqlite'
+# default_workers_type = 'celery'
+# default_database_url = f'redis+socket://{tempfile.gettempdir()}{os.sep}capsul_engine_database.redis'
+
 def full_module_name(module_name):
     '''
     Return a complete module name (which must be a valid Python module
@@ -141,7 +148,8 @@ class EngineConfiguration(Controller):
     dataset: OpenKeyDictController[Dataset]
     config_modules: list[str]
     python_modules: list[str]
-    engine_type: str = 'builtin'
+    workers_type: str = default_workers_type
+    database_url: str = default_database_url
 
     def add_module(self, module_name, allow_existing=False):
         ''' Loads a modle and adds it in the engine configuration.
@@ -207,14 +215,14 @@ class ConfigurationLayer(OpenKeyDictController[EngineConfiguration]):
 
     A ConfigurationLayer contains sub-configs corresponding to computing
     resources, which are keys in this :class:`Controller`. A "default" config
-    resource could be named "local".
+    resource could be named "builtin".
     '''
 
     def __init__(self):
         super().__init__()
         self.add_field(
-            'local', EngineConfiguration, default_factory=EngineConfiguration,
-            doc='Default local computing resource config. Elements are config '
+            'builtin', EngineConfiguration, default_factory=EngineConfiguration,
+            doc='Default builtin computing resource config. Elements are config '
             'modules which should be registered in the application (spm, fsl, '
             '...)')
 
@@ -234,8 +242,8 @@ def load(self, filename):
     def add_field(self, name, *args, **kwargs):
         if 'doc' not in kwargs:
             kwargs = dict(kwargs)
-            if name == 'local':
-                kwargs['doc'] = 'Default local computing resource config. ' \
+            if name == 'builtin':
+                kwargs['doc'] = 'Default builtin computing resource config. ' \
                     'Elements are config modules which should be registered ' \
                     'in the application (spm, fsl, ...)'
             else:
@@ -291,7 +299,7 @@ class ApplicationConfiguration(Controller):
         app_config = ApplicationConfiguration(
             'my_app_name', site_file='/usr/local/etc/my_app_name.json')
         user_conf_dict = {
-            'local': {
+            'builtin': {
                 'spm': {
                     'spm12_standalone': {
                         'directory': '/usr/local/spm12_standalone',
@@ -313,9 +321,9 @@ class ApplicationConfiguration(Controller):
     In each configuration (``user``, ``site``,, ``merged_config``):
 
     * The first level, "environment" corresponds to a "computing resource"
-      name. The default (and always existing) is "local" and means the local
-      computer configuration. Additional configs may be added to store settings
-      for remote computing resources.
+      name. The default (and always existing) is "builtin" and is using the local
+      computer configuration without parallelism. Additional configs may be added
+      to store settings for remote computing resources.
 
     * the second level corresponds to configuration modules. Each module has to
       be known in the Capsul config system, and is accessed as a module. The
diff --git a/capsul/database/__init__.py b/capsul/database/__init__.py
new file mode 100644
index 000000000..f836abc44
--- /dev/null
+++ b/capsul/database/__init__.py
@@ -0,0 +1,311 @@
+# -*- coding: utf-8 -*-
+from datetime import datetime
+import dateutil.parser
+import importlib
+from pprint import pprint
+from urllib.parse import urlsplit, urlunsplit
+import sys
+import time
+
+from populse_db.database import json_encode, json_decode
+
+from soma.api import DictWithProxy, undefined
+
+from ..application import Capsul
+from ..execution_context import ExecutionContext
+from ..pipeline.pipeline import Process, Pipeline
+
+database_classes = {
+    'sqlite': 'capsul.database.populse_db:Populse_dbExecutionDatabase',
+    'redis+socket': 'capsul.database.redis:RedisExecutionDatabase',
+}
+
+def execution_database(database_url):
+    url = urlsplit(database_url)
+    class_string = database_classes.get(url.scheme)
+    if class_string is None:
+        raise ValueError(f'Invalid database URL {database_url}: scheme {url.scheme} is not supported')
+    module_name, class_name = class_string.rsplit(':', 1)
+    module = importlib.import_module(module_name)
+    database_class = getattr(module, class_name)
+    return database_class(url)
+
+
+class ExecutionDatabase:
+    def __init__(self, url):
+        self._url = url
+
+    @property
+    def url(self):
+        url = urlunsplit(self._url)
+        if not self._url.netloc:
+            url = url.replace(':', '://', 1)
+        return url
+    
+    def new_execution(self, executable, execution_context, workflow, start_time):
+        executable_json = json_encode(executable.json(include_parameters=False))
+        execution_context_json = execution_context.json()
+        workflow_parameters_json = json_encode(workflow.parameters.json())
+        ready = []
+        waiting = []
+        jobs = [self._job_to_json(job.copy()) for job in workflow.jobs.values()]
+        for job in jobs:
+            if job['wait_for']:
+                waiting.append(job['uuid'])
+            else:
+                ready.append(job['uuid'])
+        execution_id = self.store_execution(
+            start_time=self._time_to_json(start_time),
+            executable_json=executable_json,
+            execution_context_json=execution_context_json,
+            workflow_parameters_json=workflow_parameters_json,
+            jobs=jobs,
+            ready=ready,
+            waiting=waiting,
+        )
+        return execution_id
+
+    def _execution_context_from_json(self, execution_context_json):
+        return ExecutionContext(config=execution_context_json)
+
+    def execution_context(self, execution_id):
+        j = self.execution_context_json(execution_id)
+        if j is not None:
+            return self._execution_context_from_json(j)
+
+    def _executable_from_json(self, executable_json):
+        return Capsul.executable(json_decode(executable_json))
+    
+    def executable(self, execution_id):
+        j = self.executable_json(execution_id)
+        if j is not None:
+            return self._executable_from_json(j)
+
+    def _workflow_parameters_from_json(self, workflow_parameters_json):
+        return DictWithProxy.from_json(json_decode(workflow_parameters_json))
+    
+    def workflow_parameters(self, execution_id):
+        j = self.workflow_parameters_json(execution_id)
+        if j:
+            return self._workflow_parameters_from_json(j)
+
+    def set_workflow_parameters(self, execution_id, workflow_parameters):
+        self.set_workflow_parameters_json(execution_id, json_encode(workflow_parameters.json()))
+
+    def update_workflow_parameters(self, execution_id, parameters_location, output_values):
+        self.update_workflow_parameters_json(execution_id, parameters_location, json_encode(output_values))
+
+    @staticmethod
+    def _time_from_json(time_json):
+        return dateutil.parser.parse(time_json)
+
+    @staticmethod
+    def _time_to_json(time):
+        return time.isoformat()
+    
+    def start_time(self, execution_id):
+        j = self.start_time_json(execution_id)
+        if j:
+            return self._time_from_json(j)
+        return None
+
+    def end_time(self, execution_id):
+        j = self.end_time_json(execution_id)
+        if j:
+            return self._time_from_json(j)
+        return None
+
+    def _job_from_json(self, job):
+        for k in ('start_time', 'end_time'):
+            t = job.get(k)
+            if t:
+                job[k] = self._time_from_json(t)
+        return job
+
+    def _job_to_json(self, job):
+        for k in ('start_time', 'end_time'):
+            t = job.get(k)
+            if t:
+                job[k] = self._time_to_json(t)
+        return job
+
+    def jobs(self, execution_id):
+        return [self._job_from_json(i) for i in self.jobs_json(execution_id)]
+
+    def job(self, execution_id, job_uuid):
+        j = self.job_json(execution_id, job_uuid)
+        if j:
+            return self._job_from_json(j)
+        return None
+
+    def execution_report(self, execution_id):
+        report = self.execution_report_json(execution_id)
+        report['execution_id'] = execution_id
+        for n in ('executable', 'execution_context', 'workflow_parameters'):
+            convert = getattr(self, f'_{n}_from_json')
+            report[n] = convert(report[n])
+        for n in ('start_time', 'end_time'):
+            j = report.get(n)
+            if j:
+                report[n] = self._time_from_json(j)
+        
+        for job in report['jobs']:
+            self._job_from_json(job)
+            if job['uuid'] in report['done']:
+                job['status'] = 'done'
+            elif job['uuid'] in report['failed']:
+                job['status'] = 'failed'
+            elif job['uuid'] in report['ongoing']:
+                job['status'] = 'ongoing'
+            elif job['uuid'] in report['ready']:
+                job['status'] = 'ready'
+            elif job['uuid'] in report['waiting']:
+                job['status'] = 'waiting'
+            else:
+                job['status'] = 'unknown'
+        return report
+    
+    def print_execution_report(self, report, file=sys.stdout):
+        print('====================\n'
+              '| Execution report |\n'
+              '====================\n', file=file)
+        print('executable:', report['executable'].definition, file=file)
+        print('status:', report['status'], file=file)
+        print('start time:', report['start_time'], file=file)
+        print('end time:', report['end_time'], file=file)
+        print('execution_id:', report['execution_id'], file=file)
+        print('execution context:', file=file)
+        pprint(report['execution_context'].asdict(), stream=file)
+        if report['error']:
+            print('error:', report['error'], file=file)
+        if report['error_detail']:
+            print('-' * 50, file=file)
+            print(report['error_detail'], file=file)
+            print('-' * 50, file=file)
+        print('\n---------------\n'
+              '| Jobs status |\n'
+              '---------------\n', file=file)
+        print('waiting:', report['waiting'])
+        print('ready:', report['ready'])
+        print('ongoing:', report['ongoing'])
+        print('done:', report['done'])
+        print('failed:', report['failed'])
+
+        now = datetime.now()
+        for job in sorted(report['jobs'], key=lambda j: (j.get('start_time') if j.get('start_time') else now)):
+            job_uuid = job['uuid']
+            process_definition = job.get('process', {}).get('definition')
+            start_time = job.get('start_time')
+            end_time = job.get('end_time')
+            pipeline_node = '.'.join(i for i in job.get('parameters_location', '') if i != 'nodes')
+            returncode = job.get('returncode')
+            status = job['status']
+            command = ' '.join(f"'{i}'" for i in job.get('command', []))
+            stdout = job.get('stdout')
+            stderr = job.get('stderr')
+            parameters = report['workflow_parameters']
+            if parameters:
+                for index in job.get('parameters_location', []):
+                    if index.isnumeric():
+                        index = int(index)
+                    parameters = parameters[index]
+            wait_for = job.get('wait_for', [])
+            waited_by = job.get('waited_by', [])
+
+            print('=' * 50, file=file)
+            print('job uuid:', job_uuid, file=file)
+            print('process:', process_definition, file=file)
+            print('pipeline node:', pipeline_node, file=file)
+            print('status:', status, file=file)
+            print('returncode:', returncode, file=file)
+            print('start time:', start_time, file=file)
+            print('end time:', end_time, file=file)
+            print('command:', command, file=file)
+            print('wait for:', wait_for, file=file)
+            print('waited_by:', waited_by, file=file)
+            if parameters:
+                print('parameters:', file=file)
+                pprint(parameters.no_proxy(), stream=file)
+            else:
+                print('parameters: none', file=file)
+            if stdout:
+                print('---------- standard output ----------', file=file)
+                print(stdout, file=file)
+            if stderr:
+                print('---------- error output ----------', file=file)
+                print(stderr, file=file)
+
+        if report['engine_debug']:
+            print('\n----------------\n'
+                '| Engine debug |\n'
+                '----------------\n', file=file)
+            for k, v in report['engine_debug'].items():
+                print(k, file=file)
+                print('-' * len(k), file=file)
+                print(v, file=file)
+                print(file=file)
+
+    def wait(self, execution_id, timeout=None):
+        start = time.time()
+        status = self.status(execution_id)
+        if status == 'ready':
+            for i in range(50):
+                time.sleep(0.1)
+                status = self.status(execution_id)
+                if status != 'ready':
+                    break
+            else:
+                self.print_execution_report(self.execution_report(execution_id), file=sys.stderr)
+                raise SystemError(f'workers are too slow to start execution ({datetime.now()})')
+        status = self.status(execution_id)
+        while status == 'running':
+            if timeout is not None and (time.time() - start) > timeout:
+                raise TimeoutError('Process execution timeout')
+            time.sleep(0.1)
+            status = self.status(execution_id)
+
+    def raise_for_status(self, execution_id):
+        error = self.error(execution_id)
+        if error:
+            self.print_execution_report(self.execution_report(execution_id), file=sys.stderr)
+            raise RuntimeError(error)
+
+    def update_executable(self, executable, execution_id):
+        parameters = self.workflow_parameters(execution_id)
+        # print('!update_executable!')
+        # from pprint import pprint
+        # pprint(parameters.proxy_values)
+        # pprint(parameters.content)
+        # pprint(parameters.no_proxy())
+        if isinstance(executable, Pipeline):
+            enable_parameter_links = executable.enable_parameter_links
+            executable.enable_parameter_links = False
+        else:
+            enable_parameter_links = None
+        try:
+            stack = [(executable, parameters)]
+            # print('!update_executable! stack', executable.full_name)
+            # pprint(parameters.content)
+            while stack:
+                node, parameters = stack.pop(0)
+                for field in node.user_fields():
+                    value = parameters.get(field.name, undefined)
+                    if value is not undefined:
+                        value = parameters.no_proxy(value)
+                        if value is None:
+                            value = undefined
+                        # print('!update_executable!', node.full_name, field.name, '<-', repr(value))
+                        setattr(node, field.name, value)
+                    # else:
+                    #     print('!update_executable! ignore', node.full_name, field.name, repr(value))
+                if isinstance(node, Pipeline):
+                    stack.extend((n, parameters['nodes'][n.name]) for n in node.nodes.values() if n is not node and isinstance(n, Process) and n.activated)
+        finally:
+            if enable_parameter_links is not None:
+                executable.enable_parameter_links = enable_parameter_links
+
+    def start_one_job(self, execution_id, start_time):
+        return self.start_one_job_json(execution_id, self._time_to_json(start_time))
+    
+    def job_finished(self, execution_id, job_uuid, end_time, returncode, stdout, stderr):
+        return self.job_finished_json(execution_id, job_uuid, self._time_to_json(end_time), returncode, stdout, stderr)
diff --git a/capsul/database/populse_db.py b/capsul/database/populse_db.py
new file mode 100644
index 000000000..a5e43a38a
--- /dev/null
+++ b/capsul/database/populse_db.py
@@ -0,0 +1,283 @@
+# -*- coding: utf-8 -*-
+import os
+from uuid import uuid4
+
+from populse_db import Database
+
+from . import ExecutionDatabase
+
+class Populse_dbExecutionDatabase(ExecutionDatabase):
+    def __init__(self, url):
+        super().__init__(url)
+        if url.query or url.fragment:
+            raise ValueError(f'Invalid URL: {self.url}')
+        self.sqlite_file = f'{url.netloc}{url.path}'
+        self.database = Database(f'sqlite://{self.sqlite_file}')
+        with self.database as session:
+            if not session.has_collection('execution'):
+                session.add_collection('global')
+                session['global'].add_field('capsul_connection_count', int)
+                session['global'][''] = {
+                    'capsul_connection_count': 1,
+                }
+                session.add_collection('execution')
+                execution = session['execution']
+                execution.add_field('status', str)
+                execution.add_field('error', str)
+                execution.add_field('error_detail', str)
+                execution.add_field('execution_context', dict)
+                execution.add_field('executable', dict)
+                execution.add_field('start_time', str)
+                execution.add_field('end_time', str)
+                execution.add_field('waiting', list[str])
+                execution.add_field('ready', list[str])
+                execution.add_field('ongoing', list[str])
+                execution.add_field('done', list[str])
+                execution.add_field('failed', list[str])
+                execution.add_field('workflow_parameters', dict)
+                execution.add_field('dispose', bool)
+                execution.add_field('tmp', str)
+
+                session.add_collection('job', ('execution_id', 'uuid'))
+            else:
+                connection_count = session['global']['']['capsul_connection_count'] + 1
+                session['global'][''] = {'capsul_connection_count': connection_count}
+    
+    def close(self):
+        with self.database as session:
+            connection_count = session['global']['']['capsul_connection_count']
+            if connection_count > 0:
+               connection_count -= 1
+               session['global'][''] = {'capsul_connection_count': connection_count}
+               delete_database = (connection_count == 0)
+            else:
+                delete_database = False
+        
+        del self.database
+        if delete_database:
+            os.remove(self.sqlite_file)
+    
+    def _get(self, execution_id, name):
+        with self.database as session:
+            row = session['execution'].document(execution_id, fields=[name], as_list=True)
+            if row is not None:
+                return row[0]
+            return None
+    
+    def _set(self, execution_id, name, value):
+        with self.database as session:
+            session['execution'].update_document(execution_id, {name: value})
+
+    def store_execution(self,
+            start_time, 
+            executable_json,
+            execution_context_json,
+            workflow_parameters_json,
+            jobs,
+            ready,
+            waiting
+        ):
+        execution_id = str(uuid4())
+        execution = {
+            'status': 'ready',
+            'start_time': start_time,
+            'executable': executable_json,
+            'execution_context': execution_context_json,
+            'workflow_parameters': workflow_parameters_json,
+            'ready': ready,
+            'waiting': waiting,
+            'ongoing': [],
+            'done': [],
+            'failed': [],
+            'dispose': False,
+        }
+        if not ready:
+            execution['status']  = 'ended'
+            execution['end_time'] = start_time
+        with self.database as session:
+            session['execution'][execution_id] = execution
+            for job in jobs:
+                session['job'][(execution_id, job['uuid'])] = job
+        return execution_id
+    
+    def status(self, execution_id):
+        return self._get(execution_id, 'status')
+
+    def error(self, execution_id):
+        return self._get(execution_id, 'error')
+
+    def error_detail(self, execution_id):
+        return self._get(execution_id, 'error_detail')
+
+    def set_error(self, execution_id, error, error_detail=None):
+        with self.database as session:
+            session['execution'].update_document(execution_id, {
+                'error': error,
+                'error_detail': error_detail
+            })
+    
+    def execution_context_json(self, execution_id):
+        return self._get(execution_id, 'execution_context')
+
+    def executable_json(self, execution_id):
+        return self._get(execution_id, 'executable')
+    
+    def workflow_parameters_json(self, execution_id):
+        return self._get(execution_id, 'workflow_parameters')
+
+    def set_workflow_parameters_json(self, execution_id, workflow_parameters_json):
+        with self.database as session:
+            session['execution'].update_document(
+                execution_id,
+                {
+                    'workflow_parameters': workflow_parameters_json
+                }
+            )
+
+    def update_workflow_parameters_json(self, execution_id, parameters_location, output_values):
+        workflow_parameters = self.workflow_parameters(execution_id)
+        parameters = workflow_parameters
+        for index in parameters_location:
+            if index.isnumeric():
+                index = int(index)
+            parameters = parameters[index]
+        for k, v in output_values.items():
+            parameters[k] = v
+        self.set_workflow_parameters(execution_id, workflow_parameters)
+
+    def start_time_json(self, execution_id):
+        return self._get(execution_id, 'start_time')
+
+    def end_time_json(self, execution_id):
+        return self._get(execution_id, 'end_time')
+
+    def jobs_json(self, execution_id):
+        with self.database as session:
+            yield from session['job'].filter(f'{{execution_id}}=="{execution_id}"')
+
+    def job_json(self, execution_id, job_uuid):
+        with self.database as session:
+            result = session['job'][(execution_id, job_uuid)]
+            return result
+
+    def ready(self ,execution_id):
+        return self._get(execution_id, 'ready')
+
+    def waiting(self ,execution_id):
+        return self._get(execution_id, 'waiting')
+
+    def ongoing(self ,execution_id):
+        return self._get(execution_id, 'ongoing')
+
+    def done(self ,execution_id):
+        return self._get(execution_id, 'done')
+
+    def failed(self ,execution_id):
+        return self._get(execution_id, 'failed')
+   
+    def start_one_job_json(self, execution_id, start_time):
+        with self.database as session:
+            execution= {}
+            executions = session['execution']
+            ready, ongoing = executions.document(
+                execution_id,
+                fields=('ready', 'ongoing'),
+                as_list=True)
+            if ready:
+                execution['status'] = 'running'
+                job_uuid = ready.pop(0)
+                execution['ready'] = ready
+                job = session['job'][(execution_id, job_uuid)]
+                job['start_time'] = start_time
+                session['job'][(execution_id, job_uuid)] = job
+                ongoing.append(job_uuid)
+                execution['ongoing'] = ongoing
+                result = job
+            else:
+                result = None
+            executions.update_document(execution_id, execution)
+            return result
+    
+    def job_finished_json(self, execution_id, job_uuid, end_time, returncode, stdout, stderr):
+        with self.database as session:
+            session['job'].update_document(
+                (execution_id, job_uuid),
+                {
+                    'end_time': end_time,
+                    'returncode': returncode,
+                    'stdout': stdout,
+                    'stderr': stderr,
+                })
+            execution = session['execution'].document(
+                execution_id, fields=('ready', 'ongoing', 'waiting', 'failed', 'done')
+            )
+            execution['ongoing'].remove(job_uuid)
+
+            job = session['job'][(execution_id, job_uuid)]
+            if returncode:
+                execution['failed'].append(job_uuid)
+                stack = list(job.get('waited_by', []))
+                while stack:
+                    uuid = stack.pop(0)
+                    job = session['job'][(execution_id, uuid)]
+                    job['returncode'] = 'Not started because de dependent job failed'
+                    session['job'][(execution_id, uuid)] = job
+                    execution['waiting'].remove(uuid)
+                    execution['failed'].append(uuid)
+                    stack.extend(job.get('waited_by', []))
+            else:
+                execution['done'].append(job_uuid)
+                done = set(execution['done'])
+                for waiting_uuid in job.get('waited_by', []):
+                    waiting_job = self.job(execution_id, waiting_uuid)
+                    for waited in waiting_job.get('wait_for', []):
+                        if waited not in done:
+                            break
+                    else:
+                        execution['waiting'].remove(waiting_uuid)
+                        execution['ready'].append(waiting_uuid)
+            if execution['ongoing'] or execution['ready']:
+                result = False
+            else:
+                if execution['failed']:
+                    execution['error'] = 'Some jobs failed'
+                execution['status'] = 'ended'
+                execution['end_time'] = end_time
+                result = True
+            session['execution'].update_document(execution_id, execution)
+            return result
+
+    def execution_report_json(self, execution_id):
+        with self.database as session:
+            result = dict(
+                executable = self.executable_json(execution_id),
+                execution_context = self.execution_context_json(execution_id),
+                status = self.status(execution_id),
+                error = self.error(execution_id),
+                error_detail = self.error_detail(execution_id),
+                start_time = self.start_time_json(execution_id),
+                end_time = self.end_time_json(execution_id),
+                waiting = self.waiting(execution_id),
+                ready = self.ready(execution_id),
+                ongoing = self.ongoing(execution_id),
+                done = self.done(execution_id),
+                failed = self.failed(execution_id),
+                jobs = self.jobs_json(execution_id),
+                workflow_parameters = self.workflow_parameters_json(execution_id),
+                engine_debug = {}
+            )
+        return result
+        
+    def dispose(self, execution_id):
+        with self.database as session:
+            if self.status(execution_id) == 'ended':
+                session['job'].delete(f'{{execution_id}}=="{execution_id}"')
+                del session['execution'][execution_id]
+            else:
+                session['execution'].update_document(execution_id, {'dispose': True})
+        
+    def set_tmp(self, execution_id, tmp):
+        self._set(execution_id, 'tmp', tmp)
+
+    def tmp(self, execution_id):
+        return self._get(execution_id, 'tmp')
diff --git a/capsul/database/redis.py b/capsul/database/redis.py
new file mode 100644
index 000000000..88b41c606
--- /dev/null
+++ b/capsul/database/redis.py
@@ -0,0 +1,484 @@
+# -*- coding: utf-8 -*-
+from datetime import datetime
+import shutil
+import tempfile
+import json
+import os
+import subprocess
+import time
+from uuid import uuid4
+
+import redis
+
+from . import ExecutionDatabase
+
+
+class RedisExecutionPipeline:
+    def __init__(self, pipeline, execution_id):
+        self.pipeline = pipeline
+        self.execution_id = execution_id
+
+    def key(self, name):
+        return f'capsul:{self.execution_id}:{name}'
+    
+    def get(self, name):
+        return self.pipeline.get(self.key(name))
+
+    def set(self, name, value):
+        self.pipeline.set(self.key(name), value)
+
+    def set_list(self, name, values):
+        key = self.key(name)
+        self.pipeline.delete(key)
+        if values:
+            self.pipeline.rpush(key, *values)
+
+    def hset(self, name, item, value):
+        self.pipeline.hset(self.key(name), item, value)
+
+    def rpush(self, name, value):
+        self.pipeline.rpush(self.key(name), value)
+
+    def delete(self, name):
+        self.pipeline.delete(self.key(name))
+
+    def execute(self):
+        return self.pipeline.execute()
+
+class RedisExecutionDatabase(ExecutionDatabase):
+    def __init__(self, url):
+        super().__init__(url)
+        self.uuid = str(uuid4())
+        if url.scheme != 'redis+socket':
+            raise NotImplementedError(f'Redis connection with {self.url} is not implemented')
+
+        self.redis_socket = f'{url.netloc}{url.path}'
+        if not os.path.exists(self.redis_socket):
+            # Start the redis server
+            tmp = tempfile.mkdtemp(prefix='capsul_redis_')
+            try:
+                pid_file = f'{tmp}/redis.pid'
+                log_file = f'{tmp}/redis.log'
+                cmd = [
+                    'redis-server',
+                    '--unixsocket', self.redis_socket,
+                    '--port', '0', # port 0 means no TCP connection
+                    '--daemonize', 'yes',
+                    '--pidfile', pid_file,
+                    '--dir', tmp,
+                    '--logfile', log_file,
+                    '--dbfilename', 'redis.rdb',
+                ]
+                subprocess.run(cmd)
+                for i in range(20):
+                    if os.path.exists(self.redis_socket):
+                        break
+                    time.sleep(0.1)
+                self.redis  = redis.Redis(unix_socket_path=self.redis_socket,
+                                            decode_responses=True)
+                self.redis.set('capsul_redis_tmp', tmp)
+                self.redis.set('capsul_redis_pid_file', pid_file)
+            except Exception:
+                shutil.rmtree(tmp)
+        else:
+            self.redis  = redis.Redis(unix_socket_path=self.redis_socket,
+                                      decode_responses=True)
+        
+        self.redis.hset('capsul_connections', self.uuid, datetime.now().isoformat())
+
+        # Some functions are implemented as a Lua script in redis
+        # in order to be atomic. In redis these scripts must always
+        # be registered before using them.
+        self._start_one_job = self.redis.register_script('''
+            local function key(name)
+                return 'capsul:' .. ARGV[1] .. ':' .. name
+            end
+
+            local job_uuid = redis.call('lpop', key('ready'))
+            if job_uuid then
+                redis.call('set', key('status'), 'running')
+                local job = cjson.decode(redis.call('hget', key('jobs'), job_uuid))
+                job['start_time'] = ARGV[2]
+                job = cjson.encode(job)
+                redis.call('hset', key('jobs'), job_uuid, job)
+                redis.call('rpush', key('ongoing'), job_uuid)
+                return job
+            end
+            return nil
+        ''')
+
+        self._job_finished = self.redis.register_script('''
+            local function key(name)
+                return 'capsul:' .. ARGV[1] .. ':' .. name
+            end
+
+            local execution_id = ARGV[1]
+            local job_uuid = ARGV[2]
+            local end_time = ARGV[3]
+            local returncode = ARGV[4]
+            local stdout = ARGV[5]
+            local stderr = ARGV[6]
+
+            local job = cjson.decode(redis.call('hget', key('jobs'), job_uuid))
+            job['end_time'] = end_time
+            job['returncode'] = returncode
+            job['stdout'] = stdout
+            job['stderr'] = stderr
+            redis.call('hset', key('jobs'), job_uuid, cjson.encode(job))
+
+            redis.call('lrem', key('ongoing'), 1, job_uuid)
+
+            if returncode ~= '0' then
+                redis.call('rpush', key('failed'), job_uuid)
+                local stack = {}
+                if job['waited_by'] then
+                    for key, value in pairs(job['waited_by']) do
+                        stack[key] = value
+                    end
+                end
+
+                local uuid = table.remove(stack)
+                while uuid do
+                    job = cjson.decode(redis.call('hget', key('jobs'), uuid))
+                    job['returncode'] = 'Not started because de dependent job failed'
+                    redis.call('hset', key('jobs'), uuid, cjson.encode(job))
+                    redis.call('lrem', key('waiting'), 1, uuid)
+                    redis.call('rpush', key('failed'), uuid)
+                    if job['waited_by'] then
+                        for key, value in ipairs(job['waited_by']) do
+                            table.insert(stack, value)
+                        end
+                    end
+                    uuid = table.remove(stack)
+                end
+            else
+                redis.call('rpush', key('done'), job_uuid)
+                if job['waited_by'] then
+                    for _, waiting_uuid in ipairs(job['waited_by']) do
+                        local waiting_job = cjson.decode(redis.call('hget', key('jobs'), waiting_uuid))
+                        local ready_to_go = true
+                        if waiting_job['wait_for'] then
+                            for key2, waited in ipairs(waiting_job['wait_for']) do
+                                if not redis.call('lpos', key('done'), waited) then
+                                    ready_to_go = false
+                                    break
+                                end
+                            end
+                        end
+                        if ready_to_go then
+                        redis.call('lrem', key('waiting'), 1, waiting_uuid)
+                            redis.call('rpush', key('ready'), waiting_uuid)
+                        end
+                    end
+                end
+            end
+
+            if (redis.call('llen', key('ongoing')) ~= 0) or (redis.call('llen', key('ready')) ~= 0) then
+                return false
+            else
+                redis.call('hdel', 'capsul_ongoing_executions', execution_id)
+                if redis.call('hexists', 'capsul_undisposed_executions', execution_id) then
+                    if redis.call('llen', key('failed')) ~= 0 then
+                        redis.call('set', key('error'), 'Some jobs failed')
+                    end
+                    redis.call('set', key('status'), 'ended')
+                    redis.call('set', key('end_time'), end_time)
+                    return true
+                else
+                    -- TODO: reuse _dispose script
+                    redis.call('del', key('executable'))
+                    redis.call('del', key('execution_context'))
+                    redis.call('del', key('status'))
+                    redis.call('del', key('error'))
+                    redis.call('del', key('error_detail'))
+                    redis.call('del', key('start_time'))
+                    redis.call('del', key('end_time'))
+                    redis.call('del', key('waiting'))
+                    redis.call('del', key('ready'))
+                    redis.call('del', key('ongoing'))
+                    redis.call('del', key('done'))
+                    redis.call('del', key('failed'))
+                    redis.call('del', key('jobs'))
+                    redis.call('del', key('workflow_parameters'))
+                    local tmp = redis.call('get', key('tmp'))
+                    redis.call('del', key('tmp'))
+                    return tmp
+                end
+            end
+        ''')
+
+        self._full_report = self.redis.register_script('''
+            local function key(name)
+                return 'capsul:' .. ARGV[1] .. ':' .. name
+            end
+
+            local executable = redis.call('get', key('executable'))
+            local execution_context = redis.call('get', key('execution_context'))
+            local workflow_parameters = redis.call('get', key('workflow_parameters'))
+            local status = redis.call('get', key('status'))
+            local error = redis.call('get', key('error'))
+            local error_detail = redis.call('get', key('error_detail'))
+            local start_time = redis.call('get', key('start_time'))
+            local end_time = redis.call('get', key('end_time'))
+            local waiting = redis.call('lrange', key('waiting'), 0, -1)
+            local ready = redis.call('lrange', key('ready'), 0, -1)
+            local ongoing = redis.call('lrange', key('ongoing'), 0, -1)
+            local done = redis.call('lrange', key('done'), 0, -1)
+            local failed = redis.call('lrange', key('failed'), 0, -1)
+            local jobs = {}
+            for i, v in ipairs(redis.call('hgetall', key('jobs'))) do
+                if i % 2 == 0 then
+                    table.insert(jobs, v)
+                end
+            end
+            return {executable, execution_context, workflow_parameters, 
+                status, error, error_detail, start_time, end_time, 
+                waiting, ready, ongoing, done, failed, jobs}                
+            ''')
+
+        self._dispose = self.redis.register_script('''
+            local function key(name)
+                return 'capsul:' .. ARGV[1] .. ':' .. name
+            end
+
+            local execution_id = ARGV[1]
+
+            redis.call('hdel', 'capsul_undisposed_executions', execution_id)
+            if redis.call('get', key('status')) == 'ended' then
+                redis.call('del', key('executable'))
+                redis.call('del', key('execution_context'))
+                redis.call('del', key('status'))
+                redis.call('del', key('error'))
+                redis.call('del', key('error_detail'))
+                redis.call('del', key('start_time'))
+                redis.call('del', key('end_time'))
+                redis.call('del', key('waiting'))
+                redis.call('del', key('ready'))
+                redis.call('del', key('ongoing'))
+                redis.call('del', key('done'))
+                redis.call('del', key('failed'))
+                redis.call('del', key('jobs'))
+                redis.call('del', key('workflow_parameters'))
+                local tmp = redis.call('get', key('tmp'))
+                redis.call('del', key('tmp'))
+                return tmp
+            else
+                return false
+            end
+            ''')
+
+        self._check_shutdown = self.redis.register_script('''
+            if redis.call('hlen', 'capsul_connections') == 0 and
+                redis.call('hlen', 'capsul_ongoing_executions') == 0 and
+                redis.call('llen', 'capsul_undisposed_executions') == 0
+            then
+                -- setting capsul_connection to a string value will prevent
+                -- the creation of new connections because they will use
+                -- hset command that raises an error value type is not a hash
+                redis.call('set', 'capsul_connections', 'shutting down')
+                return true
+            else
+                return false
+            end
+            ''')
+
+        self._update_workflow_parameters = self.redis.register_script('''
+            local function key(name)
+                return 'capsul:' .. ARGV[1] .. ':' .. name
+            end
+
+            local execution_id = ARGV[1]
+            local parameters_location = cjson.decode(ARGV[2])
+            local output_parameters = cjson.decode(ARGV[3])
+            local workflow_parameters = cjson.decode(redis.call('get', key('workflow_parameters')))
+
+            local parameters = workflow_parameters['content']
+            for index, value in ipairs(parameters_location) do
+                local i = tonumber(value)
+                if i then
+                    parameters = parameters[i+1]
+                else
+                    parameters = parameters[value]
+                end
+            end
+
+            for k, v in pairs(output_parameters) do
+                workflow_parameters['proxy_values'][parameters[k][2]+1] = v
+            end
+            redis.call('set', key('workflow_parameters'), cjson.encode(workflow_parameters))
+            ''')
+      
+    def close(self):
+        self.redis.hdel('capsul_connections', self.uuid)
+        self.check_shutdown()
+   
+    def key(self, execution_id, name):
+        return f'capsul:{execution_id}:{name}'
+
+    def get(self, execution_id, name):
+        return self.redis.get(self.key(execution_id, name))
+    
+    def set(self, execution_id, name, value):
+        self.redis.set(self.key(execution_id, name), value)
+
+    def pipeline(self, execution_id):
+        return RedisExecutionPipeline(self.redis.pipeline(), execution_id)
+
+    def store_execution(self,
+            start_time, 
+            executable_json,
+            execution_context_json,
+            workflow_parameters_json,
+            jobs,
+            ready,
+            waiting
+        ):
+        execution_id = str(uuid4())
+        pipe = self.pipeline(execution_id)
+        pipe.pipeline.hset('capsul_ongoing_executions', execution_id, datetime.now().isoformat())
+        pipe.pipeline.hset('capsul_undisposed_executions', execution_id, datetime.now().isoformat())
+        pipe.set( 'status', ('ready' if ready else 'ended'))
+        pipe.set('start_time', start_time)
+        pipe.set('executable', json.dumps(executable_json))
+        pipe.set('execution_context', json.dumps(execution_context_json))
+        pipe.set('workflow_parameters', json.dumps(workflow_parameters_json))
+        pipe.set_list('ready', ready)
+        pipe.set_list('waiting', waiting)
+        pipe.delete('ongoing')
+        pipe.delete('done')
+        pipe.delete('failed')
+        if not ready:
+            pipe.set('end_time', start_time)
+        for job in jobs:
+            pipe.hset('jobs', job['uuid'], json.dumps(job))
+        pipe.execute()
+        return execution_id
+    
+    def status(self, execution_id):
+        return self.get(execution_id, 'status')
+
+    def error(self, execution_id):
+        return self.get(execution_id, 'error')
+
+    def error_detail(self, execution_id):
+        return self.get(execution_id, 'error_detail')
+
+    def set_error(self, execution_id, error, error_detail=None):
+        pipe = self.pipeline(execution_id)
+        pipe.set( 'error', error)
+        pipe.set( 'error_detail', error_detail)
+        pipe.execute()
+    
+    def execution_context_json(self, execution_id):
+        return json.loads(self.get(execution_id, 'execution_context'))
+
+    def executable_json(self, execution_id):
+        return json.loads(self.get(execution_id, 'executable'))
+    
+    def workflow_parameters_json(self, execution_id):
+        return json.loads(self.get(execution_id, 'workflow_parameters'))
+
+    def set_workflow_parameters_json(self, execution_id, workflow_parameters_json):
+        self.set(execution_id, 'workflow_parameters', json.dumps(workflow_parameters_json))
+
+    def update_workflow_parameters_json(self, execution_id, parameters_location, output_values):
+        self._update_workflow_parameters(args=[execution_id, json.dumps(parameters_location), json.dumps(output_values)])
+    
+    def start_time_json(self, execution_id):
+        return self.get(execution_id, 'start_time')
+
+    def end_time_json(self, execution_id):
+        return self.get(execution_id, 'end_time')
+
+    def jobs_json(self, execution_id):
+        jobs_key = self.key(execution_id, 'jobs')
+        for job_uuid in self.redis.hkeys(jobs_key):
+            yield json.loads(self.redis.hget(jobs_key, job_uuid))
+
+
+    def job_json(self, execution_id, job_uuid):
+        job = json.loads(self.redis.hget(self.key(execution_id, 'jobs'), job_uuid))
+        return job
+
+    def ready(self ,execution_id):
+        return self.get_list(execution_id, 'ready')
+
+    def waiting(self ,execution_id):
+        return self.get_list(execution_id, 'waiting')
+
+    def ongoing(self ,execution_id):
+        return self.get_list(execution_id, 'ongoing')
+
+    def done(self ,execution_id):
+        return self.get_list(execution_id, 'done')
+
+    def failed(self ,execution_id):
+        return self.get_list(execution_id, 'failed')
+   
+    def start_one_job_json(self, execution_id, start_time):
+        job = self._start_one_job(args=[execution_id, start_time])
+        if job is not None:
+            job = json.loads(job)
+        return job
+    
+    def job_finished_json(self, execution_id, job_uuid, end_time, returncode, stdout, stderr):
+        r = self._job_finished(args=[execution_id, job_uuid, end_time, returncode, stdout, stderr])
+        if isinstance(r, str) and os.path.exists(r):
+            shutil.rmtree(r)
+            return True
+        else:
+            return bool(r)
+
+    def execution_report_json(self, execution_id):
+        (executable, execution_context, workflow_parameters, status, error,
+         error_detail, start_time, end_time, waiting, ready,
+         ongoing, done, failed, jobs) = self._full_report(args=[execution_id])
+        result = dict(
+            executable=json.loads(executable),
+            execution_context=json.loads(execution_context),
+            status=status,
+            error=error,
+            error_detail=error_detail,
+            start_time=start_time,
+            end_time=end_time,
+            waiting=waiting,
+            ready=ready,
+            ongoing=ongoing,
+            done=done,
+            failed=failed,
+            jobs=[json.loads(i) for i in jobs],
+            workflow_parameters =json.loads(workflow_parameters),
+            engine_debug = {},
+        )
+        return result
+        
+    def dispose(self, execution_id):
+        tmp = self._dispose(args=[execution_id])
+        if tmp and os.path.exists(tmp):
+            shutil.rmtree(tmp)
+    
+    def check_shutdown(self):
+        pipe = self.redis.pipeline()
+        pipe.hlen('capsul_connections')
+        if self._check_shutdown():
+            tmp = self.redis.get('capsul_redis_tmp')
+            pid_file = self.redis.get('capsul_redis_pid_file')
+            
+            # Kill the redis server
+            self.redis.shutdown()
+
+            # Ensure the pid file is deleted
+            for i in range(20):
+                if not os.path.exists(pid_file):
+                    break
+                time.sleep(0.1)
+
+            shutil.rmtree(tmp)
+    
+    def set_tmp(self, execution_id, tmp):
+        self.set(execution_id, 'tmp', tmp)
+
+    def tmp(self, execution_id):
+        return self.get(execution_id, 'tmp')
+
+    
diff --git a/capsul/dataset.py b/capsul/dataset.py
index 588f608f6..a639d9e98 100644
--- a/capsul/dataset.py
+++ b/capsul/dataset.py
@@ -133,10 +133,10 @@ class BIDSSchema(MetadataSchema):
     schema_name = 'bids'
 
     folder: Literal['sourcedata', 'rawdata', 'derivative']
-    pipeline: str = None
+    process: str = None
     sub: str
     ses: str
-    data_type: str
+    data_type: str = None
     task: str = None
     acq: str = None
     ce: str = None
@@ -144,7 +144,7 @@ class BIDSSchema(MetadataSchema):
     run: str = None
     echo: str = None
     part: str = None
-    suffix: str
+    suffix: str = None
     extension: str
 
     path_pattern = re.compile(
@@ -160,7 +160,7 @@ class BIDSSchema(MetadataSchema):
         r'(?:_run-(?P<run>[^-_/]*))?'
         r'(?:_echo-(?P<echo>[^-_/]*))?'
         r'(?:_part-(?P<part>[^-_/]*))?'
-        r'_(?P<suffix>[^-_/]*)\.(?P<extension>.*)$'
+        r'(?:_(?P<suffix>[^-_/]*))?\.(?P<extension>.*)$'
     )
 
     def __init__(self, base_path='', **kwargs):
@@ -176,11 +176,14 @@ def _path_list(self):
           sub-{sub}/ses-{ses}/{data_type}/sub-{sub}_ses-{ses}[_task-{task}][_acq-{acq}][_ce-{ce}][_rec-{rec}][_run-{run}][_echo-{echo}][_part-{part}]{modality}.{extension}
         '''
         path_list = [self.folder]
-        if self.pipeline:
-            path_list += [self.pipeline]
+        if self.process:
+            path_list += [self.process]
         path_list += [f'sub-{self.sub}',
-                      f'ses-{self.ses}',
-                      self.data_type]
+                      f'ses-{self.ses}']
+        if self.data_type:
+            path_list.append(self.data_type)
+        elif not self.process:
+            raise ValueError('BIDS schema requires a value for either data_type or process')
 
         filename = [f'sub-{self.sub}',
                     f'ses-{self.ses}']
@@ -188,7 +191,10 @@ def _path_list(self):
             value = getattr(self, key, undefined)
             if value:
                 filename.append(f'{key}-{value}')
-        filename.append(f'{self.suffix}.{self.extension}')
+        if self.suffix:
+            filename.append(f'{self.suffix}.{self.extension}')
+        else:
+            filename[-1] += f'.{self.extension}'
         path_list.append('_'.join(filename))
         return path_list
     
@@ -266,7 +272,10 @@ def find(self, **kwargs):
         -------
         Yields a path for every match.
         '''
-        layout = self.__class__(self.base_path, **kwargs)
+        if 'data_type' not in kwargs and 'process' not in kwargs:
+            layout = self.__class__(self.base_path, data_type='*', **kwargs)
+        else:
+            layout = self.__class__(self.base_path, **kwargs)
         for field in layout.fields():
             value = getattr(layout, field.name, undefined) 
             if value is undefined:
@@ -622,7 +631,11 @@ def _generate_paths(self, executable, metadatas, iteration_index, debug):
                     # merge attributes
                     level = len(path_schemas) - 1
                     for pipeline_name, process, process_parameter in reversed(path):
-                        dprint(debug, 'pname:', pipeline_name, ', def:', process.definition, ', par:', process_parameter)
+                        dprint(debug, 
+                               'pname:', pipeline_name, 
+                               ', def:', process.definition, 
+                               ', par:', process_parameter,
+                               ', lev:', level)
                         if isinstance(process, ProcessIteration):
                             metadata_schema = path_schemas[level]
                             if metadata_schema:
diff --git a/capsul/engine/__init__.py b/capsul/engine/__init__.py
index f68f86f7a..dae9e2a81 100644
--- a/capsul/engine/__init__.py
+++ b/capsul/engine/__init__.py
@@ -1,178 +1,160 @@
 # -*- coding: utf-8 -*-
-
-import time
+from datetime import datetime
 import importlib
-import os
+import sys
 
 from soma.controller import Controller
 
-from ..execution_context import ExecutionContext
-from ..pipeline.process_iteration import ProcessIteration
-from ..api import Pipeline, Process
+from ..execution_context import CapsulWorkflow, ExecutionContext
 from ..config.configuration import ModuleConfiguration
 from ..dataset import Dataset
+from ..database import execution_database
+
+
+def execution_context(engine_label, engine_config, executable):
+    execution_context = ExecutionContext(executable=executable)
+    python_modules = getattr(engine_config, 'python_modules', ())
+    if python_modules:
+        execution_context.python_modules = python_modules
+    for name, cfg in getattr(engine_config, 'dataset', {}).items():
+        setattr(execution_context.dataset, name, Dataset(path=cfg.path, metadata_schema=cfg.metadata_schema))
+
+    req_to_check = execution_context.executable_requirements(executable)
+    done_req = []  # record requirements to avoid loops
+    valid_configs = {}
+    needed_modules = set()
+
+    while req_to_check:
+        module_name, requirements = req_to_check.popitem()
+        if (module_name, requirements) in done_req:
+            continue
+        done_req.append((module_name, requirements))
+        needed_modules.add(module_name)
+
+        module_configs = getattr(engine_config, module_name, {})
+        if not isinstance(module_configs, Controller):
+            raise ValueError(f'Unknown requirement: "{module_name}"')
+        for module_field in module_configs.fields():
+            module_config = getattr(module_configs, module_field.name)
+            added_req = module_config.is_valid_config(requirements)
+            if added_req not in (False, None):
+                valid_configs.setdefault(
+                    module_name, {})[module_field] = module_config
+                if isinstance(added_req, dict):
+                    req_to_check.update(added_req)
+
+    # now check we have only one module for each
+    for module_name in needed_modules:
+        valid_module_configs = valid_configs.get(module_name)
+        if not valid_module_configs:
+            raise RuntimeError(
+                f'Execution environment "{engine_label}" has no '
+                f'valid configuration for module {module_name}')
+        if len(valid_module_configs) > 1:
+            raise RuntimeError(
+                f'Execution environment "{engine_label}" has '
+                f'{len(valid_configs)} possible configurations for '
+                f'module {module_name}')
+        # get the single remaining config
+        valid_config = next(iter(valid_module_configs.values()))
+        execution_context.add_field(module_name, type_=ModuleConfiguration)
+        setattr(execution_context, module_name,  valid_config)
+    return execution_context
+
+class Engine(Controller):
 
-
-class Engine:
     def __init__(self, label, config):
+        super().__init__()
         self.label = label
         self.config = config
+        workers_type = self.config.workers_type
+        try:
+            workers_module = importlib.import_module(
+                f'capsul.engine.{workers_type}')
+        except ImportError:
+            raise ValueError(f'engine type {workers_type} is not known.')
+        self.workers_class = getattr(workers_module, f'{workers_type.capitalize()}Workers')
 
     def __enter__(self):
-        return self
+        self.database = execution_database(self.config.database_url)
+        self.workers = self.workers_class(self.label, self.config, self.database)
+        return self.workers
 
     def __exit__(self, exception_type, exception_value, exception_traceback):
-        pass
-
-    @staticmethod
-    def module(module_name):
-        return importlib.import_module(f'capsul.config.{module_name}')
-
-    def executable_requirements(self, executable):
-        result = {}
-        if isinstance(executable, ProcessIteration):
-            for process in executable.iterate_over_process_parmeters():
-                if process.activated:
-                    result.update(self.executable_requirements(process))
-        elif isinstance(executable, Pipeline):
-            for node in executable.all_nodes():
-                if node is not executable and isinstance(node, Process) and node.activated:
-                    result.update(self.executable_requirements(node))
-        result.update(getattr(executable, 'requirements', {}))
-        return result
-
+        del self.workers
+        self.database.close()
+        del self.database
+    
     def execution_context(self, executable):
-        execution_context = ExecutionContext(executable=executable)
-        python_modules = getattr(self.config, 'python_modules', ())
-        if python_modules:
-            execution_context.python_modules = python_modules
-        for name, cfg in getattr(self.config, 'dataset', {}).items():
-            setattr(execution_context.dataset, name, Dataset(path=cfg.path, metadata_schema=cfg.metadata_schema))
-
-        req_to_check = self.executable_requirements(executable)
-        done_req = []  # record requirements to avoid loops
-        valid_configs = {}
-        needed_modules = set()
-
-        while req_to_check:
-            module_name, requirements = req_to_check.popitem()
-            if (module_name, requirements) in done_req:
-                continue
-            done_req.append((module_name, requirements))
-            needed_modules.add(module_name)
-
-            module_configs = getattr(self.config, module_name, {})
-            if not isinstance(module_configs, Controller):
-                raise ValueError(f'Unknown requirement: "{module_name}"')
-            for module_field in module_configs.fields():
-                module_config = getattr(module_configs, module_field.name)
-                added_req = module_config.is_valid_config(requirements)
-                if added_req not in (False, None):
-                    valid_configs.setdefault(
-                        module_name, {})[module_field] = module_config
-                    if isinstance(added_req, dict):
-                        req_to_check.update(added_req)
-
-        # now check we have only one module for each
-        for module_name in needed_modules:
-            valid_module_configs = valid_configs.get(module_name)
-            if not valid_module_configs:
-                raise RuntimeError(
-                    f'Execution environment "{self.label}" has no '
-                    f'valid configuration for module {module_name}')
-            if len(valid_module_configs) > 1:
-                raise RuntimeError(
-                    f'Execution environment "{self.label}" has '
-                    f'{len(valid_configs)} possible configurations for '
-                    f'module {module_name}')
-            # get the single remaining config
-            valid_config = next(iter(valid_module_configs.values()))
-            execution_context.add_field(module_name, type_=ModuleConfiguration)
-            setattr(execution_context, module_name,  valid_config)
-        return execution_context
-
-    @staticmethod
-    def filename_from_url(url):
-        return url.split('://', 1)[-1]
+        return execution_context(self.label, self.config, executable)
+
+
+class Workers(Controller):
+    def __init__(self, engine_label, engine_config, database):
+        self.engine_label = engine_label
+        self.engine_config = engine_config
+        self.database = database
 
     def start(self, executable, **kwargs):
+        for name, value in kwargs.items():
+            setattr(executable, name, value)
+        econtext = execution_context(self.engine_label, self.engine_config, executable)
+        workflow = CapsulWorkflow(executable)
+        # from pprint import pprint
+        # print('!start!', flush=True)
+        # pprint(workflow.jobs)
+        # pprint(workflow.parameters.proxy_values)
+        # pprint(workflow.parameters.content)
+        # pprint(workflow.parameters.no_proxy())
+        # print('----')
+        # pprint(workflow.jobs)
+        execution_id = self.database.new_execution(executable, econtext, workflow, start_time=datetime.now())
+        self._start(execution_id)
+        return execution_id
+
+    def _start(self, execution_id):
         raise NotImplementedError(
-            'start must be implemented in Engine subclasses.')
+            '_start must be implemented in Workers subclasses.')
 
-    def status(self, execution_id, keys=['status', 'start_time']):
-        raise NotImplementedError(
-            'status must be implemented in Engine subclasses.')
-
-    def wait(self, execution_id, timeout=None):
-        start = time.time()
-        status = self.status(execution_id, keys='status')
-        if status['status'] == 'submited':
-            for i in range(50):
-                time.sleep(0.1)
-                status = self.status(execution_id, keys='status')
-                if status['status'] != 'submited':
-                    break
-            else:
-                raise SystemError('executable too slow to start')
-        status = self.status(execution_id, keys='status')
-        while status['status'] == 'running':
-            if timeout is not None and (time.time() - start) > timeout:
-                raise TimeoutError('Process execution timeout')
-            time.sleep(0.1)
-            status = self.status(execution_id, keys='status')
-
-    def raise_for_status(self, status):
-        output = status.get('engine_output')
-        if output is not None:
-            output = output.strip()
-        if output:
-            print('----- local engine output -----')
-            print(output)
-            print('-------------------------------')
-        error = status.get('error')
-        if error:
-            detail = status.get('error_detail')
-            if detail:
-                raise RuntimeError(f'{error}\n\n{detail}')
-            else:
-                raise RuntimeError(error)
-
-    def update_executable(self, executable, execution_id):
+    def debug_info(self, execution_id):
         raise NotImplementedError(
-            'update_executable must be implemented in Engine subclasses.')
+            'debug_info must be implemented in Workers subclasses.')
 
-    def run(self, executable, timeout=None, **kwargs):
+    def status(self, execution_id):
+        return self.database.status(execution_id)
+    
+    def wait(self, *args, **kwargs):
+        self.database.wait(*args, **kwargs)
+
+    def raise_for_status(self, *args, **kwargs):
+        self.database.raise_for_status(*args, **kwargs)
+
+    def execution_report(self, *args, **kwargs):
+        return self.database.execution_report(*args, **kwargs)
+
+    def print_execution_report(self, *args, **kwargs):
+        self.database.print_execution_report(*args, **kwargs)
+
+    def update_executable(self, *args, **kwargs):
+        self.database.update_executable(*args, **kwargs)
+
+    def dispose(self, *args, **kwargs):
+        self.database.dispose(*args, **kwargs)
+
+    def run(self, executable, timeout=None, print_report=False, **kwargs):
         execution_id = self.start(executable, **kwargs)
         try:
-            self.wait(execution_id, timeout=timeout)
-            status = self.status(execution_id, keys=['status', 'error', 'error_detail', 'debug_messages', 'output_parameters'])
-            self.print_debug_messages(status)
-            self.raise_for_status(status)
+            try:
+                self.wait(execution_id, timeout=timeout)
+            except TimeoutError:
+                self.print_execution_report(self.execution_report(execution_id), sys.stderr)
+                raise
+            status = self.status(execution_id)
+            self.raise_for_status(execution_id)
+            if print_report:
+                self.print_execution_report(self.execution_report(execution_id), file=sys.stdout)
             self.update_executable(executable, execution_id)
         finally:
-            self.dispose(execution_id, retry=0.5)
+            self.dispose(execution_id)
         return status
-
-    def dispose(self, execution_id, retry=0):
-        db_file = self.filename_from_url(execution_id)
-        std = f'{db_file}.stdouterr'
-        if os.path.exists(std):
-            os.remove(std)
-        t0 = time.time()
-        ok = False
-        err = None
-        while not ok and time.time() - t0 < retry:
-            if os.path.exists(db_file):
-                try:
-                    os.remove(db_file)
-                except PermissionError as e:
-                    err = str(e)
-                    time.sleep(0.02)
-                    continue
-            ok = True
-        if not ok:
-            raise PermissionError(err)
-
-    def print_debug_messages(self, status):
-        for debug in status.get('debug_messages', []):
-            print('!', *debug)
+    
diff --git a/capsul/engine/builtin.py b/capsul/engine/builtin.py
new file mode 100644
index 000000000..87152f0aa
--- /dev/null
+++ b/capsul/engine/builtin.py
@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+from datetime import datetime
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import traceback
+
+from capsul.database import execution_database
+
+
+from . import Workers
+
+      
+class BuiltinWorkers(Workers): 
+    def _start(self, execution_id):
+        subprocess.run(
+            [sys.executable, '-m', 'capsul.engine.builtin', self.database.url, execution_id],
+            capture_output=False,
+            check=True
+        )
+        
+    def _debug_info(self, execution_id):
+        return {}
+    
+if __name__ == '__main__':
+    if len(sys.argv) != 3:
+        raise ValueError('This command must be called with two '
+            'parameters: a database URL and an execution_id')
+    database_url = sys.argv[1]
+    execution_id = sys.argv[2]
+
+    # Really detach the process from the parent.
+    # Whthout this fork, performing Capsul tests shows warning
+    # about child processes not properly waited for.
+    if sys.platform.startswith('win'):
+        pid = 0
+    else:
+        pid = os.fork()
+    if pid == 0:
+        if not sys.platform.startswith('win'):
+            os.setsid()
+        tmp = tempfile.mkdtemp(prefix='caspul_builtin_')
+        try:
+            database = execution_database(database_url)
+            try:
+                env = os.environ.copy()
+                env['CAPSUL_DATABASE'] = database_url
+                env['CAPSUL_EXECUTION_ID'] = execution_id
+                database.set_tmp(execution_id, tmp)
+                job = database.start_one_job(execution_id, start_time=datetime.now())
+                while job is not None:
+                    command = job['command']
+                    if command is not None:
+                        result = subprocess.run(command, env=env, capture_output=True)
+                        returncode = result.returncode
+                        stdout = result.stdout.decode()
+                        stderr = result.stderr.decode()
+                    else:
+                        returncode = stdout = stderr = None
+                    all_done = database.job_finished(execution_id, job['uuid'], 
+                        end_time=datetime.now(),
+                        returncode=returncode,
+                        stdout=stdout,
+                        stderr=stderr)
+                    if all_done:
+                        break
+                    job = database.start_one_job(execution_id, start_time=datetime.now())
+            except Exception as e:
+                database.set_error(execution_id,
+                    error=f'Builtin engine loop failure: {e}',
+                    error_detail=f'{traceback.format_exc()}'
+                )
+                raise
+            finally:
+                database.close()
+        finally:
+            shutil.rmtree(tmp)
diff --git a/capsul/engine/celery.py b/capsul/engine/celery.py
new file mode 100644
index 000000000..4bf9da4e2
--- /dev/null
+++ b/capsul/engine/celery.py
@@ -0,0 +1,126 @@
+# -*- coding: utf-8 -*-
+from datetime import datetime
+from glob import glob
+import os
+import shutil
+import subprocess
+import tempfile
+import sys
+
+# Remove a warning message: see https://github.com/celery/kombu/issues/1339
+import warnings
+warnings.filterwarnings('ignore', 'SelectableGroups dict interface')
+from celery import Celery
+
+from ..database import execution_database
+from ..database.redis import RedisExecutionDatabase
+from . import Workers
+
+shutdown_countdown = 10
+
+database_url = os.environ.get('CAPSUL_DATABASE')
+if database_url:
+    celery_app = Celery('capsul.engine.celery', broker=database_url)
+
+    @celery_app.task(bind=True, ignore_result=True)
+    def check_shutdown(self):
+        print('!check_shutdown!')
+        global database_url
+        celery_app.control.revoke(self.request.id) # prevent this task from being executed again
+        try:
+            database = execution_database(database_url)
+            try:
+                executions_count = database.redis.llen('capsul_ongoing_executions')
+                print(f'!executions_count! {executions_count}')
+                #TODO: possible race condition here if a connection to the celery workers
+                # is done right now
+                if not executions_count:
+                    # Release special database connection
+                    print('!shutting down celery!')
+                    database.redis.hdel('capsul_connections', 'celery_workers_connection')
+                    celery_app.control.shutdown() # send shutdown signal to all workers
+                    return
+            finally:
+                database.close()
+        except Exception:
+            pass
+        celery_app.send_task('capsul.engine.celery.check_shutdown', countdown=shutdown_countdown)
+
+
+
+    @celery_app.task(ignore_result=True)
+    def execute_process(execution_id, job_uuid):
+        global database_url
+        
+        database = execution_database(database_url)
+        try:
+            job = database.job(execution_id, job_uuid)
+            print(f"execute_process {job.get('process', {}).get('definition')} (execution_id={execution_id}, job {job_uuid})")
+            command = job['command']
+            if command is not None:
+                env = os.environ.copy()
+                env['CAPSUL_DATABASE'] = database_url
+                env['CAPSUL_EXECUTION_ID'] = execution_id
+                result = subprocess.run(command, env=env, capture_output=True)
+                returncode = result.returncode
+                stdout = result.stdout.decode()
+                stderr = result.stderr.decode()
+            else:
+                returncode = stdout = stderr = None
+            database.job_finished(execution_id, job['uuid'], 
+                end_time=datetime.now(),
+                returncode=returncode,
+                stdout=stdout,
+                stderr=stderr)
+            job = database.start_one_job(execution_id, start_time=datetime.now())
+            while job is not None:
+                # print(f"  push job {job.get('process', {}).get('definition')} (job {job_uuid})")
+                celery_app.send_task('capsul.engine.celery.execute_process', args=(execution_id, job['uuid'],))
+                job = database.start_one_job(execution_id, start_time=datetime.now())
+        finally:
+            database.close()
+
+
+class CeleryWorkers(Workers):
+    def _start(self, execution_id):
+        if not isinstance(self.database, RedisExecutionDatabase):
+            raise TypeError('Celery workers can only work with a Redis execution database')
+        tmp = self.database.redis.get('capsul_redis_tmp')
+        workers_pid_file = f'{tmp}/capsul_celery_workers.pid'
+        if not os.path.exists(workers_pid_file):
+            env = os.environ.copy()
+            env.update({
+                'CAPSUL_DATABASE': self.database.url,
+            })
+            cmd = [
+                sys.executable, '-m', 'celery',
+                '-A', 'capsul.engine.celery',
+                'multi', 'start',
+                'capsul_celery_workers',
+                f'--pidfile={tmp}/%n.pid',
+                f'--logfile={tmp}/%n%I.log'
+            ]
+            subprocess.Popen(cmd, env=env, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
+            celery_app = Celery('capsul.engine.celery', broker=self.database.url)
+            celery_app.send_task('capsul.engine.celery.check_shutdown', countdown=shutdown_countdown)
+            # keep a special database connection to prevent the database to shutdown before Celery workers
+            self.database.redis.hset('capsul_connections', 'celery_workers_connection', datetime.now().isoformat())
+        else:
+            celery_app = Celery('capsul.engine.celery', broker=self.database.url)
+        tmp = tempfile.mkdtemp(prefix='caspul_celery_')
+        try:
+            self.database.set_tmp(execution_id, tmp)
+            job = self.database.start_one_job(execution_id, start_time=datetime.now())
+            while job is not None:
+                celery_app.send_task('capsul.engine.celery.execute_process', args=(execution_id, job['uuid'],))
+                job = self.database.start_one_job(execution_id, start_time=datetime.now())
+        except Exception:
+            shutil.rmtree(tmp)
+
+ 
+    def _cleanup(self, execution_directory):
+        pass
+
+    def _debug_info(self, execution_id):
+        #TODO: return the workers log
+        return {}
diff --git a/capsul/engine/local.py b/capsul/engine/local.py
deleted file mode 100644
index 0473efdf5..000000000
--- a/capsul/engine/local.py
+++ /dev/null
@@ -1,172 +0,0 @@
-# -*- coding: utf-8 -*-
-from datetime import datetime
-import os
-import shutil
-import subprocess
-import sys
-import tempfile
-import traceback
-
-from soma.undefined import undefined
-
-from ..api import Pipeline, Process
-from ..execution_context import ExecutionDatabase, CapsulWorkflow
-from . import Engine
-
-      
-class LocalEngine(Engine):
-    
-    def start(self, executable, **kwargs):
-        db_file = tempfile.NamedTemporaryFile(prefix='capsul_local_engine_',
-                                              suffix='.sqlite', delete=False)
-        try:
-            for name, value in kwargs.items():
-                setattr(executable, name, value)
-            execution_context = self.execution_context(executable)
-            workflow = CapsulWorkflow(executable)
-            # from pprint import pprint
-            # print('!start!')
-            # pprint(workflow.parameters.proxy_values)
-            # pprint(workflow.parameters.content)
-            # pprint(workflow.parameters.no_proxy())
-            # print('----')
-            # pprint(workflow.jobs)
-            db_url = f'sqlite://{db_file.name}'
-            with ExecutionDatabase(db_url) as db:
-                db.execution_context = execution_context
-                db.executable = executable
-                db.save_workflow(workflow)
-                db.start_time =  datetime.now()
-                db.status = 'ready'
-            p = subprocess.Popen(
-                [sys.executable, '-m', 'capsul.engine.local', db_url],
-            )
-            p.wait()
-            return db_url
-        except Exception:
-            db_file.close()
-            os.remove(db_file.name)
-            raise
-
-    def status(self, execution_id, keys=['status', 'start_time']):
-        if isinstance(keys, str):
-            keys = [keys]
-        with ExecutionDatabase(execution_id) as db:
-            status = db.session['status'].document('', fields=keys)
-        filename = self.filename_from_url(execution_id) + '.stdouterr'
-        if os.path.exists(filename):
-            with open(filename) as f:
-                output = f.read()
-            status['engine_output'] = output
-        return status
-
-    def update_executable(self, executable, execution_id):
-        with ExecutionDatabase(execution_id) as db:
-            parameters = db.workflow_parameters
-        # print('!update_executable!')
-        # from pprint import pprint
-        # pprint(parameters.proxy_values)
-        # pprint(parameters.content)
-        # pprint(parameters.no_proxy())
-        if isinstance(executable, Pipeline):
-            enable_parameter_links = executable.enable_parameter_links
-            executable.enable_parameter_links = False
-        else:
-            enable_parameter_links = None
-        try:
-            stack = [(executable, parameters)]
-            # print('!update_executable! stack', executable.full_name)
-            # pprint(parameters.content)
-            while stack:
-                node, parameters = stack.pop(0)
-                for field in node.user_fields():
-                    value = parameters.get(field.name, undefined)
-                    if value is not undefined:
-                        value = parameters.no_proxy(value)
-                        if value is None:
-                            value = undefined
-                        # print('!update_executable!', node.full_name, field.name, '<-', value)
-                        setattr(node, field.name, value)
-                    # else:
-                    #     print('!update_executable! ignore', node.full_name, field.name, value)
-                if isinstance(node, Pipeline):
-                    stack.extend((n, parameters['nodes'][n.name]) for n in node.nodes.values() if n is not node and isinstance(n, Process) and n.activated)
-        finally:
-            if enable_parameter_links is not None:
-                executable.enable_parameter_links = enable_parameter_links
-    
-    
-if __name__ == '__main__':
-    import contextlib
-
-    if len(sys.argv) != 2:
-        raise ValueError('This command must be called with a single '
-            'parameter containing a capsul execution database file name')
-    db_url = sys.argv[1]
-    db_file = LocalEngine.filename_from_url(db_url)
-    output = open(db_file + '.stdouterr', 'w')
-    contextlib.redirect_stdout(output)
-    contextlib.redirect_stderr(output)
-    database = ExecutionDatabase(db_url)
-    with database as db:
-        db.status = 'submited'
-
-    # Really detach the process from the parent.
-    # Whthout this fork, performing Capsul tests shows warning
-    # about child processes not properly wait for.
-    if sys.platform.startswith('win'):
-        pid = 0
-    else:
-        pid = os.fork()
-    if pid == 0:
-        if not sys.platform.startswith('win'):
-            os.setsid()
-        # Create temporary directory
-        tmp = tempfile.mkdtemp(prefix='capsul_local_engine_')
-        db_update = {}
-        try:
-            # create environment variables for jobs
-            env = os.environ.copy()
-            env.update({
-                'CAPSUL_DATABASE': db_url,
-                'CAPSUL_TMP': tmp,
-            })
-            # Read jobs workflow
-            with database as db:
-                db.status = 'running'
-                db.start_time = datetime.now()
-                jobs = {}
-                ready = set()
-                waiting = set()
-                done = set()
-                for job in db.jobs():
-                    jobs[job['uuid']] = job
-                    if job['wait_for']:
-                        waiting.add(job['uuid'])
-                    else:
-                        ready.add(job['uuid'])
-            
-            # Execute jobs sequentially
-            while ready:
-                job_uuid = ready.pop()
-                job = jobs[job_uuid]
-                command = job['command']
-                if command is not None:
-                    subprocess.check_call(command, env=env, stdout=sys.stdout, 
-                        stderr=subprocess.STDOUT,)
-                done.add(job_uuid)
-                for waiting_uuid in list(waiting):
-                    waiting_job = jobs[waiting_uuid]
-                    if not any(i for i in waiting_job['wait_for'] if i not in done):
-                        waiting.remove(waiting_uuid)
-                        ready.add(waiting_uuid)
-        except Exception as e:
-            db_update['error'] = f'{e}'
-            db_update['error_detail'] = f'{traceback.format_exc()}'
-        finally:
-            shutil.rmtree(tmp)
-            db_update['status'] = 'ended'
-            db_update['end_time'] = datetime.now()
-            with database as db:
-                db.session['status'].update_document('', db_update)
-            del database
diff --git a/capsul/execution_context.py b/capsul/execution_context.py
index 038be88fc..4e1ff36c3 100644
--- a/capsul/execution_context.py
+++ b/capsul/execution_context.py
@@ -1,21 +1,18 @@
 # -*- coding: utf-8 -*-
 
-from datetime import datetime
-from itertools import chain
 from uuid import uuid4
 import importlib
 
-from populse_db import Database
 from soma.controller import Controller, OpenKeyDictController
 from soma.api import DictWithProxy
 from soma.undefined import undefined
 
-from .application import Capsul, executable
 from .dataset import Dataset
 from .pipeline.pipeline import Process, Pipeline
 from capsul.process.process import NipypeProcess
 from .pipeline.process_iteration import ProcessIteration
-from capsul.config.configuration import get_config_class, ModuleConfiguration
+from capsul.config.configuration import get_config_class
+from .config.configuration import ModuleConfiguration
 
 
 class ExecutionContext(Controller):
@@ -56,115 +53,32 @@ def __init__(self, config=None, executable=None):
             if hasattr(cls, 'init_execution_context'):
                 cls.init_execution_context(self)
 
-
-class ExecutionDatabase:
-    def __init__(self, path):
-        self.db = Database(path)
-        self.collection = None
-    
-    def __enter__(self):
-        session = self.session = self.db.__enter__()
-        if not session.has_collection('status'):
-            session.add_collection('status')
-            status = session['status']
-            status.add_field('status', str)
-            status.add_field('execution_context', dict)
-            status.add_field('start_time', datetime)
-            status.add_field('end_time', datetime)
-            # status.add_field('debug_messages', list[str])
-            status.add_field('error', str)
-            status.add_field('error_detail', str)
-            status.add_field('executable', dict)
-            status[''] = {}
-
-            session.add_collection('jobs', 'uuid')
-            jobs = session['jobs']
-            jobs.add_field('command', list[str])
-            jobs.add_field('wait_for', list[str])
-            jobs.add_field('process', dict)
-            jobs.add_field('parameters_location', list[str])
-
-            session.add_collection('workflow', 'uuid')
-            workflow = session['workflow']
-            workflow.add_field('temporaries', dict)
-            workflow.add_field('parameters', dict[str, list])
-        return self
-    
-    def __exit__(self, *args):
-        self.db.__exit__(*args)
-        self.session = None
-        
-    @property
-    def status(self):
-        row = self.session['status'].document('', fields=['status'], as_list=True)
-        if row is not None:
-            return row[0]
-
-    @status.setter
-    def status(self, status):
-        self.session['status'].update_document('', {'status': status})
-
-    @property
-    def execution_context(self):
-        row = self.session['status'].document('', fields=['execution_context'], as_list=True)
-        if row is not None:
-            return ExecutionContext(config=row[0])
-
-    @execution_context.setter
-    def execution_context(self, execution_context):
-        json_context = execution_context.json()
+    def json(self):
+        json_context = super().json()
         for k in list(json_context):
             # replace module classes with long name, if they are not in the
             # standard location (capsul.config)
-            m = getattr(execution_context, k)
+            m = getattr(self, k)
             if isinstance(m, ModuleConfiguration) \
                     and m.__class__.__module__.split('.') \
                         != ['capsul', 'config', m.name]:
                 cls_name = f'{m.__class__.__module__}.{m.__class__.__name__}'
                 json_context[cls_name] = json_context[k]
                 del json_context[k]
+        return json_context
 
-        self.session['status'].update_document('', {'execution_context': json_context})
-    
-    @property
-    def executable(self):
-        row = self.session['status'].document('', fields=['executable'], as_list=True)
-        if row is not None:
-            return Capsul.executable(row[0])
-
-    @executable.setter
-    def executable(self, executable):
-        self.session['status'].update_document('', {'executable': executable.json()})
-    
-    @property
-    def start_time(self):
-        row = self.session['status'].document('', fields=['start_time'], as_list=True)
-        if row is not None:
-            return row[0]
-
-    @start_time.setter
-    def start_time(self, value):
-        self.session['status'].update_document('', {'start_time': value})
-
-    def save_workflow(self, workflow):
-        for job_uuid, job in workflow.jobs.items():
-            self.session['jobs'][job_uuid] = job
-        self.session['workflow'][''] = {}
-        self.workflow_parameters = workflow.parameters
-
-    @property
-    def workflow_parameters(self):
-        row = self.session['workflow'].document('', fields=['parameters'], as_list=True)
-        if row is not None:
-            return DictWithProxy.from_json(row[0])
-        return None
-    
-    @workflow_parameters.setter
-    def workflow_parameters(self, parameters):
-        self.session['workflow'].update_document('', {'parameters': parameters.json()})
-
-    def jobs(self):
-        return self.session['jobs'].documents()
+    def executable_requirements(self, executable):
+        result = {}
+        if isinstance(executable, ProcessIteration):
+            for process in executable.iterate_over_process_parmeters():
+                if process.activated:
+                    result.update(self.executable_requirements(process))
+        elif isinstance(executable, Pipeline):
+            for node in executable.all_nodes():
+                if node is not executable and isinstance(node, Process) and node.activated:
+                    result.update(self.executable_requirements(node))
+        result.update(getattr(executable, 'requirements', {}))
+        return result
 
 
 class CapsulWorkflow(Controller):
@@ -225,9 +139,13 @@ def __init__(self, executable):
             del self.jobs[disabled_job[0]]
     
         # Transform wait_for sets to lists for json storage
-        for job in self.jobs.values():
-            job['wait_for'] = list(job['wait_for'])
-    
+        # and add waited_by
+        for waiting, job in self.jobs.items():
+            wait_for = list(job['wait_for'])
+            job['wait_for'] = wait_for
+            for waited in wait_for:
+                self.jobs[waited].setdefault('waited_by',[]).append(waiting)
+        
     def _create_jobs(self,
                      executable,
                      jobs_per_process,
@@ -336,12 +254,14 @@ def _create_jobs(self,
             job_uuid = str(uuid4())
             if disabled:
                 self.jobs[job_uuid] = {
+                    'uuid': job_uuid,
                     'command': None,
                     'wait_for': set(),
                     'waited_by': set(),
                 }
             else:
                 self.jobs[job_uuid] = {
+                    'uuid': job_uuid,
                     'command': ['python', '-m', 'capsul.run', 'process', job_uuid],
                     'wait_for': set(),
                     'process': process.json(include_parameters=False),
diff --git a/capsul/pipeline/pipeline.py b/capsul/pipeline/pipeline.py
index 202c50136..36ab6537d 100644
--- a/capsul/pipeline/pipeline.py
+++ b/capsul/pipeline/pipeline.py
@@ -345,7 +345,7 @@ def add_process(self, name, process, do_not_export=None,
         A last note about invalid nodes:
 
         When saving a pipeline (through the :class:`graphical editor
-        <capsul.qt_gui.widgets.pipeline_developper_view.PipelineDeveloperView>`
+        <capsul.qt_gui.widgets.pipeline_developer_view.PipelineDeveloperView>`
         typically), missing nodes *will not be saved* because they are not
         actually in the pipeline. So be careful to save only pipelines with full
         features.
diff --git a/capsul/pipeline/pipeline_workflow.py b/capsul/pipeline/pipeline_workflow.py
deleted file mode 100644
index 01578c1fb..000000000
--- a/capsul/pipeline/pipeline_workflow.py
+++ /dev/null
@@ -1,1670 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Capsul Pipeline conversion into soma-workflow workflow.
-
-Standard use case::
-
-    workflow = workflow_from_pipeline(pipeline)
-    controller, wf_id = workflow_run(workflow_name, workflow, engine)
-
-Functions
-=========
-:func:`workflow_from_pipeline`
-------------------------------
-:func:`workflow_run`
---------------------
-"""
-
-from __future__ import print_function
-from __future__ import absolute_import
-import os
-import socket
-import sys
-import weakref
-
-import soma_workflow.client as swclient
-import soma_workflow.info as swinfo
-
-from capsul.pipeline.pipeline import Pipeline, Switch
-from capsul.pipeline import pipeline_tools
-from capsul.process.process import Process
-from soma.controller import undefined, Any
-from soma.sorted_dictionary import OrderedDict, SortedDictionary
-from .process_iteration import ProcessIteration
-from soma_workflow.custom_jobs import MapJob, ReduceJob
-
-
-def workflow_from_pipeline(pipeline, engine=None, disabled_nodes=None,
-                           jobs_priority=0, create_directories=True,
-                           environment='global', check_requirements=True,
-                           complete_parameters=False):
-    """ Create a soma-workflow workflow from a Capsul Pipeline
-
-    Parameters
-    ----------
-    pipeline: Pipeline (mandatory)
-        a CAPSUL pipeline
-    engine: CapsulEngine (optional), or dict
-        holds information about file transfers and shared resource paths.
-        If not specified, it will be accessed through the pipeline.
-    disabled_nodes: sequence of pipeline nodes (Node instances) (optional)
-        such nodes will be disabled on-the-fly in the pipeline, file transfers
-        will be adapted accordingly (outputs may become inputs in the resulting
-        workflow), and temporary files will be checked. If a disabled node was
-        to produce a temporary files which is still used in an enabled node,
-        then a ValueError exception will be raised.
-        If disabled_nodes is not passed, they will possibly be taken from the
-        pipeline (if available) using disabled steps:
-        see Pipeline.define_steps()
-    jobs_priority: int (optional, default: 0)
-        set this priority on soma-workflow jobs.
-    create_directories: bool (optional, default: True)
-        if set, needed output directories (which will contain output files)
-        will be created in a first job, which all other ones depend on.
-    environment: str (default: "global")
-        configuration environment name (default: "global"). See
-        :class:`capsul.engine.CapsulEngine` and
-        :class:`capsul.engine.settings.Settings`.
-    check_requirements: bool (default: True)
-        if True, check the pipeline nodes requirements in the capsul engine
-        settings, and issue an exception if they are not met instead of
-        proceeding with the workflow.
-    complete_parameters: bool (default: False)
-        Perform parameters completion on the input pipeline while building
-        the workflow. The default is False because we should avoid to do things
-        several times when it's already done, but in iteration nodes,
-        completion needs to be done anyway for each iteration, so this option
-        offers to do the rest of the "parent" pipeline completion.
-
-    Returns
-    -------
-    workflow: Workflow
-        a soma-workflow workflow
-    """
-
-    class TempFile(str):
-        # class needed temporary to identify temporary paths in the pipeline.
-        # must inherit a string type since it is used as a field value
-        def __init__(self, string=''):
-            # in python3 super(..).__init__() cannot take an argument
-            # moreover the str value is assigned anyway.
-            super(TempFile, self).__init__()
-            if isinstance(string, TempFile):
-                self.pattern = string.pattern
-                self.value = string.value
-                self.ref = string.ref if string.ref else string
-            else:
-                self.pattern = '%s'
-                self.value = string
-                self.ref = None
-
-        def referent(self):
-            return self.ref if self.ref else self
-
-        def get_value(self):
-            return self.referent().value
-
-        def __add__(self, other):
-            res = TempFile(str(self) + str(other))
-            res.pattern = self.pattern + str(other)
-            res.value = self.value
-            res.ref = self.referent()
-            return res
-
-        def __radd__(self, other):
-            res = TempFile(str(other) + str(self))
-            res.pattern = str(other) + self.pattern
-            res.value = self.value
-            res.ref = self.referent()
-            return res
-
-        def __iadd__(self, other):
-            self.pattern += str(other)
-            str(TempFile, self).__iadd__(str(other))
-
-        def __str__(self):
-            return self.pattern % self.get_value()
-
-        def __hash__(self):
-            if self.ref:
-                return self.referent().__hash__()
-            return super(TempFile, self).__hash__()
-
-        def __eq__(self, other):
-            if not isinstance(other, TempFile):
-                return False
-            return self.referent() is other.referent()
-
-
-    def _files_group(path, merged_formats):
-        bname = os.path.basename(path)
-        l0 = len(path) - len(bname)
-        p0 = 0
-        paths = [path]
-        while True:
-            p = bname.find('.', p0)
-            if p < 0:
-                break
-            ext = bname[p:]
-            p0 = p + 1
-            format_def = merged_formats.get(ext)
-            if format_def:
-                path0 = path[:l0 + p]
-                paths += [path0 + e[0] for e in format_def]
-                break
-        paths.append(path + '.minf')
-        return paths
-
-    def _translated_path(path, shared_map, shared_paths, field=None):
-        if path is None or path is undefined \
-                or not shared_paths \
-                or (field is not None
-                    and not field.is_path()):
-            return None # not a path
-        item = shared_map.get(path)
-        if item is not None:
-            # already in map
-            return item
-
-        for base_dir, (namespace, uuid) in shared_paths.items():
-            if path.startswith(base_dir + os.sep):
-                rel_path = path[len(base_dir)+1:]
-                #uuid = path
-                item = swclient.SharedResourcePath(
-                    rel_path, namespace, uuid=uuid)
-                shared_map[path] = item
-                return item
-        return None
-
-    def _replace_in_list(rlist, temp_map):
-        for i, item in enumerate(rlist):
-            if isinstance(item, (list, tuple)):
-                deeperlist = list(item)
-                _replace_in_list(deeperlist, temp_map)
-                if isinstance(item, tuple):
-                    deeperlist = tuple(deeperlist)
-                elif isinstance(item, set):
-                    deeperlist = set(deeperlist)
-                rlist[i] = deeperlist
-            #elif item is undefined:
-                #rlist[i] = ''
-            elif isinstance(item, (dict, OrderedDict, SortedDictionary)):
-                _replace_in_dict(item, temp_map)
-            elif item in temp_map:
-                value = temp_map[item]
-                value = value.__class__(value)
-                if hasattr(item, 'pattern'):
-                    # temp case (differs from shared case)
-                    value.pattern = item.pattern
-                rlist[i] = value
-
-    def _replace_in_dict(rdict, temp_map):
-        for name, item in rdict.items():
-            if isinstance(item, (list, tuple, set)):
-                deeperlist = list(item)
-                _replace_in_list(deeperlist, temp_map)
-                if isinstance(item, tuple):
-                    deeperlist = tuple(deeperlist)
-                elif isinstance(item, set):
-                    deeperlist = set(deeperlist)
-                rdict[name] = deeperlist
-            #elif item is undefined:
-                #rdict[name] = ''
-            elif isinstance(item, (dict, OrderedDict, SortedDictionary)):
-                _replace_in_dict(item, temp_map)
-            elif item in temp_map:
-                value = temp_map[item]
-                value = value.__class__(value)
-                if hasattr(item, 'pattern'):
-                    # temp case (differs from shared case)
-                    value.pattern = item.pattern
-                rdict[name] = value
-
-    def _get_replaced(rlist, temp_map):
-        if isinstance(rlist, (dict, OrderedDict, SortedDictionary)):
-            return _get_replaced(list(rlist.values()), temp_map)
-        if isinstance(rlist, (list, tuple)):
-            l = [x for x in [_get_replaced(y, temp_map) for y in rlist]
-                 if x is not None]
-            out = []
-            while l:
-                item = l.pop(0)
-                if isinstance(item, list):
-                    l = item + l
-                else:
-                    out.append(item)
-            return out
-        if rlist not in temp_map:
-            if isinstance(rlist, swclient.SpecialPath):
-                return rlist
-            return None
-        value = temp_map[rlist]
-        if hasattr(rlist, 'pattern'):
-            # temp case (differs from shared case)
-            value.pattern = rlist.pattern
-        return [value]
-
-    def _replace_transfers(rlist, process, itransfers, otransfers):
-        param_name = None
-        i = 3
-        for item in rlist[3:]:
-            if param_name is None:
-                param_name = item
-            else:
-                transfer = itransfers.get(param_name)
-                if transfer is None:
-                    transfer = otransfers.get(param_name)
-                if transfer is not None:
-                    value = transfer[0]
-                    if isinstance(item, (list, tuple)):
-                        # TODO: handle lists of files [transfers]
-                        #deeperlist = list(item)
-                        #_replace_in_list(deeperlist, transfers)
-                        #rlist[i] = deeperlist
-                        print('*** LIST! ***')
-                    else:
-                        rlist[i] = value
-                param_name = None
-            i += 1
-
-    def _replace_dict_transfers(rdict, process, itransfers, otransfers):
-        for param_name, item in rdict.items():
-            transfer = itransfers.get(param_name)
-            if transfer is None:
-                transfer = otransfers.get(param_name)
-            if transfer is not None:
-                value = transfer[0]
-                if isinstance(item, (list, tuple)):
-                    # TODO: handle lists of files [transfers]
-                    #deeperlist = list(item)
-                    #_replace_in_list(deeperlist, transfers)
-                    #rdict[param_name] = deeperlist
-                    print('*** LIST! ***')
-                else:
-                    rdict[param_name] = value
-
-    def build_job(process, temp_map={}, shared_map={}, transfers=[{}, {}],
-                  shared_paths={}, forbidden_temp=set(), name='', priority=0,
-                  step_name='', engine=None, environment='global'):
-        """ Create a soma-workflow Job from a Capsul Process
-
-        Parameters
-        ----------
-        process: Process (mandatory) or custom Node
-            a CAPSUL process instance or custom Node instance
-        temp_map: dict (optional)
-            temporary paths map.
-        shared_map: dict (optional)
-            file shared translated paths, global pipeline dict.
-            This dict is updated when needed during the process.
-        shared_paths: dict (optional)
-            holds information about shared resource paths base dirs for
-            soma-workflow.
-            If not specified, no translation will be used.
-        forbidden_temp: dict
-            values forbidden for temporary files (?)
-        name: string (optional)
-            job name. If empty, use the process name.
-        priority: int (optional)
-            priority assigned to the job
-        step_name: str (optional)
-            the step name will be stored in the job user_storage variable
-        engine: CapsulEngine (optional)
-            used to check configuration requirements for non-process nodes
-        environment: str (default: "global")
-            configuration environment name (default: "global"). See
-            :class:`capsul.engine.CapsulEngine` and
-            :class:`capsul.engine.settings.Settings`.
-
-        Returns
-        -------
-        job: Job
-            a soma-workflow Job instance that will execute the CAPSUL process
-        """
-        job_name = name
-        if not job_name:
-            job_name = process.name
-
-        # check for special modified paths in parameters
-        input_replaced_paths = []
-        output_replaced_paths = []
-        param_dict = {}
-        has_outputs = False
-        for parameter in process.user_fields():
-            param_name = parameter.name
-            if process.metadata(parameter, 'write', False) \
-                    and (process.metadata(parameter, 'output', False)
-                          or not process.metadata(
-                            parameter, 'format', '').startswith('path/')):
-                has_outputs = True
-            value = getattr(process, param_name, undefined)
-            if isinstance(value, list):
-                values = value
-            else:
-                values = [value]
-            for value in values:
-                if isinstance(value, TempFile):
-                    # duplicate swf temp and copy pattern into it
-                    tval = temp_map[value]
-                    tval = tval.__class__(tval)
-                    tval.pattern = value.pattern
-                    if parameter.output:
-                        output_replaced_paths.append(tval)
-                    else:
-                        if value in forbidden_temp:
-                            raise ValueError(
-                                'Temporary value used cannot be generated '
-                                'in the workflkow: %s.%s'
-                                % (job_name, param_name))
-                        input_replaced_paths.append(tval)
-                else:
-                    _translated_path(value, shared_map, shared_paths,
-                                    parameter)
-
-        # Get the process command line
-        #process_cmdline = process.get_commandline()
-        process_cmdline = process.params_to_command()
-        # and replace in commandline
-        iproc_transfers = transfers[0].get(process, {})
-        oproc_transfers = transfers[1].get(process, {})
-        #proc_transfers = dict(iproc_transfers)
-        #proc_transfers.update(oproc_transfers)
-        _replace_in_list(process_cmdline, temp_map)
-        _replace_in_list(process_cmdline, shared_map)
-        _replace_transfers(
-            process_cmdline, process, iproc_transfers, oproc_transfers)
-
-        config = {}
-        config = process.check_requirements(environment)
-        if config is None:
-            # here we bypass unmet requirements, it's not our job here.
-            config = {}
-
-        use_input_params_file = False
-        if process_cmdline[0] == 'capsul_job':
-            # use python executable from config, if any
-            pconf = config.get('capsul.engine.module.python')
-            if not pconf:
-                pconf = process.get_engine().settings. \
-                    select_configurations(environment, {'python': 'any'})
-                if pconf:
-                    if not config:
-                        config = pconf
-                    else:
-                        if 'capsul.engine.module.python' in pconf:
-                            config['capsul.engine.module.python'] \
-                                = pconf['capsul.engine.module.python']
-                        uses = pconf.get('capsul_engine', {}).get('uses', {})
-                        if uses:
-                            config.setdefault('capsul_engine', {}).setdefault(
-                                'uses', {}).update(uses)
-            python_command = pconf.get(
-                'capsul.engine.module.python', {}).get('executable')
-            if not python_command:
-                python_command = os.path.basename(sys.executable)
-            path_trick = ''
-            # python path cannot be passed in a library since the access to
-            # this library (capsul module typically) may be conditioned by
-            # this path. We cannot use PYTHONPATH env either because it would
-            # completely erase any user settings (.bashrc). So we add it here.
-            ppath = pconf.get(
-                'capsul.engine.module.python', {}).get('path')
-            if ppath:
-                path_trick = 'import sys; sys.path = %s + sys.path; ' \
-                    % repr(ppath)
-            process_cmdline = [
-                'capsul_job', python_command, '-c',
-                '%sfrom capsul.api import Process; '
-                'Process.run_from_commandline("%s")'
-                % (path_trick, process_cmdline[1])]
-            use_input_params_file = True
-            param_dict = process.asdict(exclude_undefined=True)
-        elif process_cmdline[0] in ('json_job', 'custom_job'):
-            use_input_params_file = True
-            param_dict = process.asdict(exclude_undefined=True)
-        for name in Process.nonplug_names:
-            if name in param_dict:
-                del param_dict[name]
-
-        _replace_in_dict(param_dict, temp_map)
-        _replace_in_dict(param_dict, shared_map)
-        _replace_dict_transfers(
-            param_dict, process, iproc_transfers, oproc_transfers)
-
-        # handle native specification (cluster-specific specs as in
-        # soma-workflow)
-        native_spec = getattr(process, 'native_specification', None)
-
-        # Return the soma-workflow job
-        if process_cmdline[0] == 'custom_job':
-            job = build_custom_job(
-                process, process_cmdline, name=job_name,
-                referenced_input_files
-                    =input_replaced_paths \
-                        + [x[0] for x in iproc_transfers.values()],
-                referenced_output_files
-                    =output_replaced_paths \
-                        + [x[0] for x in oproc_transfers.values()],
-                param_dict=param_dict)
-        else:
-            job = swclient.Job(
-                name=job_name,
-                command=process_cmdline[1:],
-                referenced_input_files
-                    =input_replaced_paths \
-                        + [x[0] for x in iproc_transfers.values()],
-                referenced_output_files
-                    =output_replaced_paths \
-                        + [x[0] for x in oproc_transfers.values()],
-                priority=priority,
-                native_specification=native_spec,
-                param_dict=param_dict,
-                use_input_params_file=use_input_params_file,
-                has_outputs=has_outputs)
-            # print('job command:', job.command)
-            # handle parallel job info (as in soma-workflow)
-            parallel_job_info = getattr(process, 'parallel_job_info', None)
-            if parallel_job_info:
-                job.parallel_job_info = parallel_job_info
-        if step_name:
-            job.user_storage = step_name
-        if hasattr(process, 'uuid'):
-            # propagate the process uuid, if any, to maintain link with it
-            job.uuid = process.uuid
-        job.configuration = config
-        # associate job with process
-        job.process = weakref.ref(process)
-        job._do_not_pickle = ['process']
-        job.process_hash = id(process)
-        return job
-
-    def build_custom_job(node, process_cmdline, name,
-                         referenced_input_files, referenced_output_files,
-                         param_dict):
-        ''' Build a custom job (generally running on engine side) from a
-        custom pipeline node.
-        '''
-        if hasattr(node, 'build_job'):
-            job = node.build_job(
-                name=name, referenced_input_files=referenced_input_files,
-                referenced_output_files=referenced_output_files,
-                param_dict=param_dict)
-            return job
-        return None
-
-    def build_group(name, jobs):
-        """ Create a group of jobs
-
-        Parameters
-        ----------
-        name: str (mandatory)
-            the group name
-        jobs: list of Job (mandatory)
-            the jobs we want to insert in the group
-
-        Returns
-        -------
-        group: Group
-            the soma-workflow Group instance
-        """
-        return swclient.Group(jobs, name=name)
-
-    def get_jobs(group, groups):
-        gqueue = list(group.elements)
-        jobs = []
-        while gqueue:
-            group_or_job = gqueue.pop(0)
-            if group_or_job in groups:
-                gqueue += group_or_job.elements
-            else:
-                jobs.append(group_or_job)
-        return jobs
-
-    def assign_temporary_filenames(pipeline, count_start=0):
-        ''' Find and temporarily assign necessary temporary file names'''
-        temp_filenames = pipeline.find_empty_parameters()
-        temp_map = {}
-        count = count_start
-        for node, plug_name, optional in temp_filenames:
-            if hasattr(node, 'process'):
-                process = node.process
-            else:
-                process = node
-            field = process.field(plug_name)
-            if process.metadata(field, 'output', False):
-               # filename is explicitly an output: not a temporary.
-                continue
-            is_list = field.is_list()
-            values = []
-            if is_list:
-                todo = getattr(process, plug_name, undefined)
-                # FIXME for now this test is redundant and useless
-                #if process.metadata(field).get('output', False):
-                    ## filename is explicitly an output: not a temporary.
-                    #continue
-            else:
-                todo = [undefined]
-            for item in todo:
-                if item not in (undefined, '', None):
-                    # non-empty list element
-                    values.append(item)
-                    continue
-                is_directory = field.is_directory()
-                if process.metadata(field, 'allowed_extensions', None):
-                    suffix = process.metadata(field, 'allowed_extensions')[0]
-                else:
-                    suffix = ''
-                swf_tmp = swclient.TemporaryPath(is_directory=is_directory,
-                    suffix=suffix, name='temporary_%d' % count)
-                tmp_file = TempFile('%d' % count)
-                count += 1
-                temp_map[tmp_file] = (swf_tmp, node, plug_name, optional)
-                values.append(tmp_file)
-            # set a TempFile value to identify the params / value
-            if is_list:
-                setattr(process, plug_name, values)
-            else:
-                setattr(process, plug_name, values[0])
-        return temp_map
-
-    def restore_empty_filenames(temporary_map):
-      ''' Set back undefined values to temporarily assigned file names (using
-      assign_temporary_filenames()
-      '''
-      for tmp_file, item in temporary_map.items():
-          node, plug_name = item[1:3]
-          if hasattr(node, 'process'):
-              process = node.process
-          else:
-              process = node
-          value = getattr(process, plug_name)
-          if isinstance(value, list):
-              # FIXME TODO: only restore values in list which correspond to
-              # a temporary.
-              # Problem: they are sometimes transformed into strings
-              # FIXME: several temp items can be part of the same list,
-              # so this assignment is likely to be done several times.
-              # It could probably be optimized.
-              # WARNING: we set "" values instead of undefined because they may
-              # be mandatory
-              setattr(process, plug_name, [''] * len(value))
-          else:
-              setattr(process, plug_name, undefined)
-
-    def _get_swf_paths(engine):
-        study_config = engine.study_config
-        computing_resource = getattr(
-            study_config, 'somaworkflow_computing_resource', None)
-        if computing_resource in (None, undefined):
-            return [], {}
-        resources_conf = getattr(
-            study_config, 'somaworkflow_computing_resources_config', None)
-        if resources_conf in (None, undefined):
-            return [], {}
-        resource_conf = getattr(resources_conf, computing_resource, None)
-        if resource_conf in (None, undefined):
-            return [], {}
-        return (resource_conf.transfer_paths,
-                resource_conf.path_translations.export_to_dict())
-
-    def _propagate_transfer(node, param, path, output, transfers,
-                            transfer_item):
-        todo_plugs = [(node, param, output)]
-        done_plugs = set()
-        while todo_plugs:
-            node, param, output = todo_plugs.pop()
-            plug = node.plugs[param]
-            is_pipeline = False
-            if plug is None or not plug.enabled or not plug.activated \
-                    or plug in done_plugs:
-                continue
-            done_plugs.add(plug)
-            if isinstance(node, Switch):
-                if output:
-                    # propagate to active input
-                    other_param = node.switch + '_switch_' + param
-                    #plug = node.plugs[input_param]
-                else:
-                    other_param = param[len(node.switch + '_switch_'):]
-                    #other_plug = node.plugs[other_param]
-                todo_plugs.append((node, other_param, not output))
-            else:
-                process = node.process
-                if hasattr(process, 'nodes'):
-                    is_pipeline = True
-                    #plug = process.nodes[''].plugs.get(param)
-                else:
-                    # process: replace its param
-                    # check field type (must be path, not Any)
-                    field = process.field(param)
-                    #plug = node.plugs[param]
-                    if field.is_path():
-                        transfers[
-                            bool(process.metadata(field, 'write'))
-                        ].setdefault(process, {})[param] = (transfer_item,
-                                                            path)
-                    #output = not output # invert IO status
-
-            #if plug is None or not plug.enabled or not plug.activated \
-                    #or plug in done_plugs:
-                #continue
-            if output:
-                links = plug.links_to
-            else:
-                links = plug.links_from
-            for proc_name, param_name, node, other_plug, act in links:
-                if not node.activated or not node.enabled \
-                        or not other_plug.activated or not other_plug.enabled \
-                        or other_plug in done_plugs:
-                    continue
-                todo_plugs.append((node, param_name, not output))
-            if is_pipeline:
-                # in a pipeline node, go both directions
-                if output:
-                    links = plug.links_from
-                else:
-                    links = plug.links_to
-                for proc_name, param_name, node, plug, act in links:
-                    if not node.activated or not node.enabled \
-                            or not plug.activated or not plug.enabled \
-                            or plug in done_plugs:
-                        continue
-                    todo_plugs.append((node, param_name, output))
-
-    def _get_transfers(pipeline, transfer_paths, merged_formats):
-        """ Create and list FileTransfer objects needed in the pipeline.
-
-        Parameters
-        ----------
-        pipeline: Pipeline
-            pipeline to build workflow for
-        transfer_paths: list
-            paths basedirs for translations from soma-worflow config
-
-        Returns
-        -------
-        [in_transfers, out_transfers]
-        each of which is a dict: { Process: proc_dict }
-            proc_dict is a dict: { file_path : FileTransfer object }
-        FileTransfer objects are reused when referring to the same path
-        used from different processes within the pipeline.
-        """
-        in_transfers = {}
-        out_transfers = {}
-        transfers = [in_transfers, out_transfers]
-        todo_nodes = [pipeline.pipeline_node]
-        while todo_nodes:
-            node = todo_nodes.pop(0)
-            if hasattr(node, 'process'):
-                process = node.process
-            else:
-                process = node
-            for field in process.fields():
-                param = field.name
-                if field.is_file() or field.type is Any:
-                    # is value in paths
-                    path = getattr(process, param, undefined)
-                    if path is None or path is undefined:
-                        continue
-                    output = bool(process.metadata(field, 'write'))
-                    existing_transfers = transfers[output].get(process, {})
-                    existing_transfer = existing_transfers.get(param)
-                    if existing_transfer:
-                        continue
-                    for tpath in transfer_paths:
-                        if path.startswith(os.path.join(tpath, '')):
-                            transfer_item = swclient.FileTransfer(
-                                is_input=not output,
-                                client_path=path,
-                                client_paths=_files_group(path,
-                                                          merged_formats))
-                            _propagate_transfer(node, param,
-                                                path, not output, transfers,
-                                                transfer_item)
-                            break
-            if hasattr(process, 'nodes'):
-                todo_nodes += [sub_node
-                               for name, sub_node in process.nodes.items()
-                               if name != ''
-                                  and not isinstance(sub_node, Switch)]
-        return transfers
-
-    def _expand_nodes(nodes):
-        '''Expands the nodes list or set to leaf nodes by replacing pipeline
-        nodes by their children list.
-
-        Returns
-        -------
-        set of leaf nodes.
-        '''
-        nodes_list = list(nodes)
-        expanded_nodes = set()
-        while nodes_list:
-            node = nodes_list.pop(0)
-            if not hasattr(node, 'process'):
-                continue # switch or something
-            if isinstance(node.process, Pipeline):
-                nodes_list.extend([p for p
-                                   in node.process.nodes.values()
-                                   if p is not node])
-            else:
-                expanded_nodes.add(node)
-        return expanded_nodes
-
-    def _handle_disable_nodes(pipeline, temp_map, transfers, disabled_nodes):
-        '''Take into account disabled nodes by changing FileTransfer outputs
-        for such nodes to inputs, and recording output temporary files, so as
-        to ensure that missing temporary outputs will not be used later in
-        the workflow.
-
-        disabled_nodes should be a list, or preferably a set, of leaf process
-        nodes. Use _expand_nodes() if needed before calling
-        _handle_disable_nodes().
-        '''
-        move_to_input = {}
-        remove_temp = set()
-        for node in disabled_nodes:
-            if not hasattr(node, 'process'):
-                continue # switch or something else
-            process = node.process
-            otrans = transfers[1].get(process, None)
-            for field in process.fields():
-                param = field.name
-                if field.metadata(field, 'write') and (
-                    field.is_path() or field.type is Any):
-                    path = getattr(process, param, undefined)
-                    if otrans is not None:
-                        transfer, path2 = otrans.get(param, (None, None))
-                    else:
-                        transfer = None
-                    if transfer is not None:
-                        print('transfered output path:', path,
-                              'from: %s.%s changes to input.'
-                              % (node.name, param))
-                        move_to_input[path] = transfer
-                        transfer.initial_status \
-                            = swclient.constants.FILES_ON_CLIENT
-                    elif path in temp_map:
-                        print('temp path in: %s.%s will not be produced.'
-                              % (node.name, param))
-                        remove_temp.add(path)
-        return move_to_input, remove_temp
-
-    def iter_to_workflow(process, node_name, step_name, temp_map,
-                         shared_map, transfers, shared_paths,
-                         disabled_nodes, remove_temp,
-                         steps, engine, iteration, map_job=None,
-                         reduce_job=None, environment='global'):
-        '''
-        Build a workflow for a single iteration step of a process/sub-pipeline
-
-        is called for each iteration by build_iteration()
-
-        Jobs inserted are tuples (process, iteration) because each job will be
-        converted into as many jobs as the number of iterations.
-
-        Returns
-        -------
-        (jobs, dependencies, groups, root_jobs, links)
-        '''
-        if isinstance(process, Pipeline):
-            temp_map2 = assign_temporary_filenames(process, len(temp_map))
-            temp_subst_list = [(x1, x2[0]) for x1, x2 in temp_map2.items()]
-            temp_subst_map = dict(temp_subst_list)
-            temp_subst_map.update(temp_map)
-            try:
-                (jobs1, dependencies, groups, sub_root_jobs, plinks, nodes) = \
-                    workflow_from_pipeline_structure(
-                        process, temp_subst_map, shared_map, transfers,
-                        shared_paths, disabled_nodes=disabled_nodes,
-                        forbidden_temp=remove_temp, steps=steps,
-                        engine=engine, environment=environment)
-                jobs = {}
-                for proc, job in jobs1.items():
-                    jobs[(proc, iteration)] = job
-                groot = []
-                for j in list(sub_root_jobs.values()):
-                    if isinstance(j, list):
-                        groot += j
-                    else:
-                        groot.append(j)
-                links = {}
-                for dproc, dplink in plinks.items():
-                    dlink = {}
-                    links[(dproc, iteration)] = dlink
-                    for param, linkl in dplink.items():
-                        dlink[param] = [((link[0], iteration), link[1])
-                                        for link in linkl]
-                group = build_group(node_name, groot)
-                groups[(process, iteration)] = group
-                root_jobs = {(process, iteration): group}
-            finally:
-                restore_empty_filenames(temp_map2)
-        elif isinstance(process, ProcessIteration):
-            # sub-iteration
-            pipeline = Pipeline()
-            pipeline.set_engine(engine)  # FIXME doesn't exist
-            pipeline.add_process('iter', process.process)
-            return build_iteration(pipeline, step_name, temp_map=temp_map,
-                                   shared_map=shared_map,
-                                   transfers=transfers,
-                                   shared_paths=shared_paths,
-                                   disabled_nodes=disabled_nodes,
-                                   remove_temp=remove_temp,
-                                   steps=steps,
-                                   engine=engine,
-                                   environment=environment)
-        else:
-            # single process
-            job = build_job(process, temp_map, shared_map,
-                            transfers, shared_paths,
-                            forbidden_temp=remove_temp,
-                            name=node_name,
-                            priority=jobs_priority,
-                            step_name=step_name,
-                            engine=engine,
-                            environment=environment)
-            jobs = {(process, iteration): job}
-            groups = {}
-            dependencies = {}
-            links = {}
-            root_jobs = {(process, iteration): job}
-            nodes = []
-
-        return (jobs, dependencies, groups, root_jobs, links, []) # nodes)
-
-    def build_iteration(it_node, step_name, temp_map,
-                        shared_map, transfers, shared_paths, disabled_nodes,
-                        remove_temp, steps, engine={},
-                        environment='global'):
-        '''
-        Build workflow for an iterative process: the process / sub-pipeline is
-        filled with appropriate parameters for each iteration, and its
-        workflow is generated.
-
-        Returns
-        -------
-        (jobs, dependencies, groups, root_jobs, links, nodes)
-        '''
-        it_process = it_node.process
-        completion_engine = ProcessCompletionEngine.get_completion_engine(
-            it_process)
-        # # check if it is an iterative completion engine
-        # if hasattr(completion_engine, 'complete_iteration_step'):
-        size = completion_engine.iteration_size()
-
-        #no_output_value = None
-        #size = None
-        #size_error = False
-
-        ## calculate the number of iterations
-        #for parameter in it_process.iterative_parameters:
-            #field = it_process.field(parameter)
-            #psize = len(getattr(it_process, parameter))
-            #if psize:
-                #if size is None:
-                    #size = psize
-                #elif size != psize:
-                    #size_error = True
-                    #break
-                #if it_process.is_output(field):
-                    #if no_output_value is None:
-                        #no_output_value = False
-                    #elif no_output_value:
-                        #size_error = True
-                        #break
-            #else:
-                #if it_process.is_output(field):
-                    #if no_output_value is None:
-                        #no_output_value = True
-                    #elif not no_output_value:
-                        #size_error = True
-                        #break
-
-        #if size_error:
-            #raise ValueError('Iterative parameter values must be lists of the '
-                #'same size: %s' % ','.join('%s=%d'
-                    #% (n, len(getattr(it_process, n)))
-                    #for n in it_process.iterative_parameters))
-
-        jobs = {}
-        workflows = []
-        for parameter in it_process.regular_parameters:
-            setattr(it_process.process, parameter,
-                    getattr(it_process, parameter))
-
-        jobs = {}
-        dependencies = set()
-        groups = {}
-        root_jobs = {}
-        links = {}
-        nodes = []
-
-        if size == 0:
-            return (jobs, dependencies, groups, root_jobs, links, nodes)
-
-        #if no_output_value:
-            ## this case is a "really" dynamic iteration, the number of
-            ## iterations and parameters are determined in runtime, so we
-            ## cannot handle it at the moment.
-            #raise ValueError('Dynamic iteration is not handled in this '
-                #'version of CAPSUL / Soma-Workflow')
-
-            #for parameter in it_process.iterative_parameters:
-                #trait = it_process.trait(parameter)
-                #if trait.output:
-                    #setattr(it_process, parameter, [])
-            #outputs = {}
-            #for iteration in range(size):
-                #for parameter in it_process.iterative_parameters:
-                    #if not no_output_value \
-                            #or not it_process.trait(parameter).output:
-                        #values = getattr(it_process, parameter)
-                        #if len(values) != 0:
-                            #if len(values) > iteration:
-                                #setattr(it_process.process, parameter,
-                                        #values[iteration])
-                            #else:
-                                #setattr(it_process.process, parameter,
-                                        #values[-1])
-
-                ## operate completion
-                #complete_iteration(it_process, iteration)
-
-                ##workflow = workflow_from_pipeline(it_process.process,
-                                                  ##engine=engine)
-                ##workflows.append(workflow)
-
-                #for parameter in it_process.iterative_parameters:
-                    #trait = it_process.trait(parameter)
-                    #if trait.output:
-                        #outputs.setdefault(
-                            #parameter,[]).append(getattr(it_process.process,
-                                                         #parameter))
-            #for parameter, value in outputs.items():
-                #setattr(it_process, parameter, value)
-        #else:
-
-        if True:
-
-            # iterations are built using
-            # * a map job to dispatch input lists
-            # * the iterated process or pipeline jobs, duplicated for each
-            #   iteration
-            # * a reduce job to gather outputs into lists
-            # dependencies and parameters links have to be built.
-            # links to processes outside the iteration are made, in order to
-            # connect the iteration node to its neighbors.
-
-            # collect iterated inputs / outputs
-            map_param_dict = {}
-            # copy non-iterative inputs
-            for field in it_process.user_fields():
-                param = field.name
-                value = getattr(it_process, param, undefined)
-                if not process.is_output(field):
-                    map_param_dict[param] = value
-
-            in_params = [p for p in it_process.iterative_parameters
-                         if not it_process.is_output(p)]
-            out_params = [p for p in it_process.iterative_parameters
-                          if it_process.is_output(p)]
-            in_values = [getattr(it_process, param) for param in in_params]
-            out_values = [getattr(it_process, param) for param in out_params]
-
-            # build map and reduce nodes
-            map_param_dict.update({
-                'input_names': in_params,
-                'output_names': ['%s' % p + '_%d' for p in in_params],
-            })
-            reduce_param_dict = {}
-            for field in it_process.user_fields():
-                param = field.name
-                if it_process.is_output(field):
-                    value = getattr(it_process, param, undefined)
-                    reduce_param_dict[param] = value
-                    if param in out_params:
-                        # set input params of reduce node for a non-dynamic
-                        # reduce case (outputs are set from outside)
-                        for i, p in enumerate(value):
-                            reduce_param_dict['%s_%d' % (param, i)] = p
-
-            reduce_param_dict.update({
-                'input_names': ['%s' % p + '_%d' for p in out_params],
-                'output_names': out_params,
-                'lengths': [size] * len(out_params),
-            })
-
-            # replace special paths in map/reduce parameters
-
-            _replace_in_list(in_values, temp_map)
-            in_values = _get_replaced(in_values, shared_map)
-            _replace_in_list(out_values, temp_map)
-            out_values = _get_replaced(out_values, shared_map)
-            _replace_in_dict(map_param_dict, temp_map)
-            _replace_in_dict(map_param_dict, shared_map)
-            _replace_dict_transfers(
-                map_param_dict, it_process, transfers[0], transfers[1])
-            _replace_in_dict(reduce_param_dict, temp_map)
-            _replace_in_dict(reduce_param_dict, shared_map)
-            _replace_dict_transfers(
-                reduce_param_dict, it_process, transfers[0], transfers[1])
-
-            map_job = MapJob(
-                referenced_input_files=in_values,
-                referenced_output_files=in_values,
-                name=it_process.process.name + '_map',
-                param_dict=map_param_dict)
-            reduce_job = ReduceJob(
-                referenced_input_files=out_values,
-                referenced_output_files=out_values,
-                name=it_process.process.name + '_reduce',
-                param_dict=reduce_param_dict)
-            map_job.process_hash = id(it_process)
-            reduce_job.process_hash = id(it_process)
-
-            # connect inputs of the map node, outputs to reduce node,
-            # and record connections to iterated jobs
-            map_links = {}
-            map_iter_links = {}
-            reduce_iter_links = {}
-            red_iter_links = {}
-            for param, plug in it_node.plugs.items():
-                if param in Process.nonplug_names:
-                    continue
-                if not plug.output:
-                    # connect inputs of the map node
-                    sources = pipeline_tools.find_plug_connection_sources(
-                        it_node.plugs[param], it_node)
-                    for pnode, pparam, pparent in sources:
-                        pproc = pnode
-                        if hasattr(pnode, 'process'):
-                            pproc = pnode.process
-                        map_links.setdefault(param, []).append((pproc, pparam))
-                    # record dest of links in iterated nodes
-                    if isinstance(it_process.process, Pipeline):
-                        dest = \
-                            pipeline_tools.find_plug_connection_destinations(
-                                it_process.process.pipeline_node.plugs[param],
-                                it_process.process.pipeline_node)
-                        for pnode, pparam, pparent in dest:
-                            pproc = pnode
-                            if hasattr(pnode, 'process'):
-                                pproc = pnode.process
-                            map_iter_links.setdefault(pproc, {}) \
-                                .setdefault(pparam, []).append(
-                                    (map_job, param))
-                    else:
-                        map_iter_links.setdefault(it_process.process, {}) \
-                            .setdefault(param, []).append(
-                                (map_job, param))
-                else:
-                    # connect outputs of the reduce node
-                    dest = pipeline_tools.find_plug_connection_destinations(
-                        it_node.plugs[param], it_node)
-                    for pnode, pparam, pparent in dest:
-                        pproc = pnode
-                        if hasattr(pnode, 'process'):
-                            pproc = pnode.process
-                        links.setdefault(pproc, {}).setdefault(pparam, []) \
-                            .append((reduce_job, param))
-                    # record source of links in iterated nodes
-                    if isinstance(it_process.process, Pipeline):
-                        #print('reduce from pipeline', param)
-                        sources = \
-                            pipeline_tools.find_plug_connection_sources(
-                                it_process.process.pipeline_node.plugs[param],
-                                it_process.process.pipeline_node)
-                        #print('sources:', sources)
-                        for pnode, pparam, pparent in sources:
-                            pproc = pnode
-                            if hasattr(pnode, 'process'):
-                                pproc = pnode.process
-                            red_iter_links.setdefault(param, []).append(
-                                (pproc, pparam))
-                    else:
-                        red_iter_links.setdefault(param, []).append(
-                            (it_process.process, param))
-            links[map_job] = map_links
-            reduce_iter_links[reduce_job] = red_iter_links
-            jobs[map_job] = map_job
-            jobs[reduce_job] = reduce_job
-            root_jobs[map_job] = map_job
-            root_jobs[reduce_job] = reduce_job
-
-            # iterate the iterates process / pipeline
-
-            iter_values = {}
-            for iteration in range(size):
-                for parameter in it_process.iterative_parameters:
-                    if it_process.process.metadata(parameter, 'output'):
-                        # dynamic output has no forced value
-                        continue
-                    values = getattr(it_process, parameter, [])
-                    if len(values) != 0:
-                        if len(values) > iteration:
-                            setattr(it_process.process, parameter,
-                                    values[iteration])
-                        else:
-                            setattr(it_process.process, parameter, values[-1])
-
-                # operate completion
-                complete_iteration(it_process, iteration)
-
-                # get iteration values to set on the parent iter node
-                for parameter in it_process.iterative_parameters:
-                    if it_process.process.metadata(parameter, 'output'):
-                        # dynamic output has no forced value
-                        continue
-                    value = getattr(it_process.process, parameter, [])
-                    iter_values.setdefault(parameter, []).append(value)
-
-                # build a workflow for the job / pipeline iteration
-                process_name = it_process.process.name + '_%d' % iteration
-                (sub_jobs, sub_dependencies, sub_groups, sub_root_jobs,
-                 sub_links, sub_nodes) = \
-                    iter_to_workflow(it_process.process, process_name,
-                        step_name,
-                        temp_map, shared_map, transfers,
-                        shared_paths, disabled_nodes, remove_temp, steps,
-                        engine, iteration, map_job=map_job,
-                        reduce_job=reduce_job, environment=environment)
-                nodes += sub_nodes
-                jobs.update(sub_jobs)
-                dependencies.update(sub_dependencies)
-                groups.update(sub_groups)
-                root_jobs.update(sub_root_jobs)
-                links.update(sub_links)
-
-                # connect map / reduce nodes to iterated jobs
-                for proc, dlink in map_iter_links.items():
-                    slink = links.setdefault((proc, iteration), {})
-                    for dparam, linkl in dlink.items():
-                        l = slink.setdefault(dparam, [])
-                        for link in linkl:
-                            if link[1] in in_params:
-                                # iterative param
-                                l.append((link[0],
-                                          '%s_%d' % (link[1], iteration)))
-                            else:
-                                l.append(link)
-                for proc, dlink in reduce_iter_links.items():
-                    for dparam, linkl in dlink.items():
-                        if dparam in out_params:
-                            # iterative param
-                            dparam = '%s_%d' % (dparam, iteration)
-                        for link in linkl:
-                            links.setdefault(proc, {}) \
-                                .setdefault(dparam, []) \
-                                .append(((link[0], iteration), link[1]))
-
-        # set values on the iter node (what a full completion does, but we're
-        # doing it only partially here)
-        for param, value in iter_values.items():
-            setattr(it_process, param, value)
-
-        # the iteration process is not a single job, but can be reached
-        # (for links) through the map and reduce nodes. So we record a tuple
-        # (map, reduce) here for this special node.
-        jobs[it_process] = (map_job, reduce_job)  # special job(s)
-        return (jobs, dependencies, groups, root_jobs, links, nodes)
-
-
-    def complete_iteration(it_process, iteration):
-        completion_engine = ProcessCompletionEngine.get_completion_engine(
-            it_process)
-        # check if it is an iterative completion engine
-        if hasattr(completion_engine, 'complete_iteration_step'):
-            completion_engine.complete_iteration_step(iteration)
-
-
-    def workflow_from_pipeline_structure(
-            pipeline, temp_map={}, shared_map={},
-            transfers=[{}, {}], shared_paths={},
-            disabled_nodes=set(), forbidden_temp=set(),
-            jobs_priority=0, steps={}, current_step='',
-            engine=None, with_links=True, environment='global'):
-        """ Convert a CAPSUL pipeline into a soma-workflow workflow
-
-        Parameters
-        ----------
-        pipeline: Pipeline (mandatory)
-            a CAPSUL pipeline
-        temp_map: dict (optional)
-            temporary files to replace by soma_workflow TemporaryPath objects
-        shared_map: dict (optional)
-            shared translated paths maps (global to pipeline).
-            This dict is updated when needed during the process.
-        transfers: list of 2 dicts (optional)
-            File transfers dicts (input / output), indexed by process, then by
-            file path.
-        shared_paths: dict (optional)
-            holds information about shared resource paths from soma-worflow
-            section in study config.
-            If not specified, no translation will be used.
-        jobs_priority: int (optional, default: 0)
-            set this priority on soma-workflow jobs.
-        steps: dict (optional)
-            node name -> step name dict
-        current_step: str (optional)
-            the parent node step name
-        engine: CapsulEngine instance (optional)
-            used only for iterative nodes, to be passed to create sub-workflows
-        environment: str (default: "global")
-            configuration environment name (default: "global"). See
-            :class:`capsul.engine.CapsulEngine` and
-            :class:`capsul.engine.settings.Settings`.
-
-        Returns
-        -------
-        workflow: tuple (jobs, dependencies, groups, root_jobs, links, nodes)
-            the corresponding soma-workflow workflow definition (to be passed
-            to Workflow constructor)
-        """
-        def _update_links(links1, links2):
-            for dest_node, slink in links2.items():
-                nlink = links1.setdefault(dest_node, {})
-                nlink.update(slink)
-
-        jobs = {}
-        groups = {}
-        root_jobs = {}
-        dependencies = set()
-        group_nodes = {}
-        links = {}
-
-        nodes = [(pipeline, node_name, node)
-                 for node_name, node in pipeline.nodes.items()
-                 if node.activated and node.enabled
-                    and node is not pipeline.pipeline_node]
-        all_nodes = []
-
-        # Go through all graph nodes
-        for node_desc in nodes:
-
-            n_pipeline, node_name, node = node_desc
-
-            if node in disabled_nodes \
-                    or not pipeline_tools.is_node_enabled(n_pipeline,
-                                                          node_name=node_name):
-                continue
-
-            if not node.is_job():
-                continue
-
-            all_nodes.append(node_desc)
-
-            if isinstance(node, Pipeline):
-                # sub-pipeline
-                group_nodes[node_name] = node
-
-                step_name = current_step or steps.get(node_name, '')
-                (sub_jobs, sub_deps, sub_groups, sub_root_jobs, sub_links,
-                 sub_nodes) \
-                    = workflow_from_pipeline_structure(
-                        node.process, temp_map, shared_map, transfers,
-                        shared_paths, disabled_nodes,
-                        jobs_priority=jobs_priority,
-                        steps=steps, current_step=step_name, with_links=False,
-                        environment=environment)
-                group = build_group(node_name,
-                                    sum(list(sub_root_jobs.values()), []))
-                groups[node] = group
-                root_jobs[node] = [group]
-                jobs.update(sub_jobs)
-                groups.update(sub_groups)
-                dependencies.update(sub_deps)
-                all_nodes += sub_nodes
-                #_update_links(links, sub_links)
-            # Otherwise convert all the processes to jobs
-            else:
-                sub_jobs = {}
-                process = None
-                if isinstance(node, Process):
-                    process = node
-                elif hasattr(node, 'process'):
-                    # process node
-                    process = node.process
-                else:
-                    process = node  # custom node
-                step_name = current_step or steps.get(node.name)
-                if isinstance(process, ProcessIteration):
-                    # iterative node
-                    group_nodes.setdefault(
-                        node_name, []).append(node)
-                    sub_workflows = build_iteration(
-                        node, step_name, temp_map,
-                        shared_map, transfers, shared_paths,
-                        disabled_nodes,
-                        {}, steps, engine=engine)
-                    (sub_jobs, sub_deps, sub_groups, sub_root_jobs,
-                      sub_links, sub_nodes) = sub_workflows
-                    group = build_group(node_name, list(sub_root_jobs.values()))
-                    groups.setdefault(process, []).append(group)
-                    root_jobs.setdefault(process, []).append(group)
-                    groups.update(sub_groups)
-                    jobs.update(sub_jobs)
-                    dependencies.update(sub_deps)
-                    all_nodes += sub_nodes
-                    _update_links(links, sub_links)
-                else:
-                    job = build_job(process, temp_map, shared_map,
-                                    transfers, shared_paths,
-                                    forbidden_temp=forbidden_temp,
-                                    name=node_name,
-                                    priority=jobs_priority,
-                                    step_name=step_name,
-                                    engine=engine, environment=environment)
-                    if job:
-                        sub_jobs[process] = job
-                        root_jobs[process] = [job]
-                        jobs.update(sub_jobs)
-
-        # links / dependencies
-        if with_links:
-            for node_desc in all_nodes:
-                sub_pipeline, node_name, node = node_desc
-                dproc = getattr(node, 'process', node)
-                if isinstance(dproc, Pipeline):
-                    continue  # pipeline nodes are virtual
-                for param, plug in node.plugs.items():
-                    sources = pipeline_tools.find_plug_connection_sources(
-                        plug, True)
-                    for source in sources:
-                        snode, param_name, parent = source
-                        if node in disabled_nodes \
-                                or not pipeline_tools.is_node_enabled(
-                                    pipeline, node=snode):
-                            continue
-                        process = getattr(snode, 'process', snode)
-                        if not isinstance(snode, Process) \
-                                and snode not in jobs:
-                            # the node is a phantom node (switch). Add deps to
-                            # all upstream nodes
-                            new_nodes = [snode]
-                            while new_nodes:
-                                mnode = new_nodes.pop(0)
-                                moredep = [
-                                    pipeline_tools.find_plug_connection_sources
-                                    (mlink[3])
-                                    for mplug in mnode.plugs.values()
-                                    for mlink in mplug.links_from
-                                    if not mplug.output
-                                ]
-                                dependencies.update(
-                                    [(x[0].process, node.process)
-                                     for x in moredep
-                                     if isinstance(x[0], Process)])
-                                new_nodes += [x[0] for x in moredep
-                                              if x[0] is not None
-                                                  and not isinstance(
-                                                      x[0], Process)]
-                        elif process is pipeline:
-                            continue  # from main input: no dependency link
-                        else:  # Process
-                            sjob = jobs[process]
-                            if isinstance(sjob, tuple):  # iteration
-                                sjob = sjob[1]  # source
-                            djob = jobs[dproc]
-                            if isinstance(djob, tuple):  # iteration
-                                djob = djob[0]  # destination
-                            dependencies.add((sjob, djob))
-                            field = process.field(param_name)
-                            if process.metadata(field, 'output', False) \
-                                    or not process.metadfata(
-                                        field, 'format',
-                                        '').startswith('path/'):
-                                links.setdefault(dproc, {}).setdefault(
-                                    param, []).append((process, param_name))
-
-        return jobs, dependencies, groups, root_jobs, links, all_nodes
-
-    def _create_directories_job(pipeline, shared_map={}, shared_paths={},
-                                priority=0, transfer_paths=[]):
-        def _is_transfer(d, transfer_paths):
-            for path in transfer_paths:
-                if d.startswith(os.path.join(path, '')):
-                    return True
-            return False
-
-        directories = [d
-                       for d in pipeline_tools.get_output_directories(
-                          pipeline)[1]
-                       if not _is_transfer(d, transfer_paths)]
-        if len(directories) == 0:
-            return None # no dirs to create.
-        paths = []
-        # check for path translations
-        for path in directories:
-            new_path = _translated_path(path, shared_map, shared_paths)
-            paths.append(new_path or path)
-        # use a python command to avoid the shell command mkdir
-        cmdline = ['python', '-c',
-                   'import sys, os; [os.makedirs(p) if not os.path.exists(p) '
-                   'else None '
-                   'for p in sys.argv[1:]]'] \
-                  + paths
-
-        job = swclient.Job(
-            name='output directories creation',
-            command=cmdline,
-            priority=priority)
-        return job
-
-
-    # TODO: handle formats in a separate, centralized place
-    # formats: {name: ext_props}
-    #     ext_props: {ext: [dependent_exts]}
-    #     dependent_exts: (ext, mandatory)
-    formats = {
-        'NIFTI-1': {'.nii': [], '.img': [('.hdr', True)], '.nii.gz': []},
-        'GIS': {'.ima': [('.dim', True)]},
-        'GIFTI': {'.gii': []},
-        'MESH': {'.mesh': []},
-        'ARG': {'.arg': [('.data', False)]},
-    }
-    # transform it to an ext-based dict
-    # merged_formats: {ext: [dependent_exts]}
-    # (formats names are lost here)
-    merged_formats = {}
-    for format, values in formats.items():
-        merged_formats.update(values)
-
-    if engine is None:
-        engine = pipeline.get_engine()  # FIXME does not exist
-
-    if check_requirements:
-        if pipeline.check_requirements(environment) is None:
-            raise ValueError(
-                'The pipeline requirements are not met in the current '
-                'configuration settings.')
-
-    if not isinstance(pipeline, Pipeline):
-        # "pipeline" is actually a single process (or should, if it is not a
-        # pipeline). Get it into a pipeline (with a single node) to make the
-        # workflow.
-        new_pipeline = Pipeline()
-        new_pipeline.set_engine(engine)  # FIXME does not exist
-        new_pipeline.add_process('main', pipeline)
-        new_pipeline.autoexport_nodes_parameters(include_optional=True)
-        pipeline = new_pipeline
-
-    if complete_parameters:
-        completion = ProcessCompletionEngine.get_completion_engine(pipeline)
-        if completion:
-            attributes = completion.get_attribute_values()
-            completion.complete_parameters(complete_iterations=False)
-
-    temp_map = assign_temporary_filenames(pipeline)
-    temp_subst_list = [(x1, x2[0]) for x1, x2 in temp_map.items()]
-    temp_subst_map = dict(temp_subst_list)
-    shared_map = {}
-
-    swf_paths = _get_swf_paths(engine)
-    transfers = _get_transfers(pipeline, swf_paths[0], merged_formats)
-    # get complete list of disabled leaf nodes
-    if disabled_nodes is None:
-        disabled_nodes = pipeline.disabled_pipeline_steps_nodes()
-    disabled_nodes.update(name for name, node in pipeline.nodes.items()
-           if node.node_type != 'processing_node')
-    disabled_nodes = _expand_nodes(disabled_nodes)
-    move_to_input, remove_temp = _handle_disable_nodes(
-        pipeline, temp_subst_map, transfers, disabled_nodes)
-    #print('changed transfers:', move_to_input)
-    #print('removed temp:', remove_temp)
-    #print('temp_map:', temp_map, '\n')
-    #print('SWF transfers:', swf_paths[0])
-    #print('shared paths:', swf_paths[1])
-
-    if create_directories:
-        # create job
-        dirs_job = _create_directories_job(
-            pipeline, shared_map=shared_map, shared_paths=swf_paths[1],
-            transfer_paths=swf_paths[0])
-
-    # build steps map
-    steps = {}
-    if hasattr(pipeline, 'pipeline_steps'):
-        for step in pipeline.pipeline_steps.user_fields():
-            step_name = step.name
-            nodes = step.nodes
-            steps.update(dict([(node, step_name) for node in nodes]))
-
-    # actually build the workflow, recursively if needed.
-    try:
-        (jobs, dependencies, groups, root_jobs, links, nodes) \
-            = workflow_from_pipeline_structure(
-                pipeline, temp_subst_map, shared_map, transfers,
-                swf_paths[1],
-                disabled_nodes=disabled_nodes, forbidden_temp=remove_temp,
-                steps=steps, engine=engine,
-                environment=environment)
-    finally:
-        restore_empty_filenames(temp_map)
-
-    # post-process links to replace nodes with jobs
-    param_links = {}
-    # expand jobs map because jobs keys may be tuples (for iterations)
-    jobs_map = {}
-    for process, job in jobs.items():
-        while isinstance(process, tuple):
-            process = process[0]
-        jobs_map.setdefault(process, []).append(job)
-    for dnode, dlinks in links.items():
-        if isinstance(dnode, (Pipeline, ProcessIteration)) \
-                or (isinstance(dnode, tuple)
-                    and isinstance(dnode[0], (Pipeline, ProcessIteration))):
-            continue  # FIXME handle this
-        djlinks = {}
-        for param, linkl in dlinks.items():
-            for link in linkl:
-                if link[0] is not pipeline \
-                        and not isinstance(link[0],
-                                          (Pipeline, ProcessIteration)):
-                    # FIXME handle ProcessIteration cases
-                    if isinstance(link[0], tuple):
-                        if not isinstance(link[0][0],
-                                          (Pipeline, ProcessIteration)):
-                            djlinks.setdefault(param, []) \
-                                .append((jobs[link[0]], link[1]))
-                    else:
-                        for job in jobs_map[link[0]]:
-                            djlinks.setdefault(param, []) \
-                                .append((job, link[1]))
-        if isinstance(dnode, tuple):
-            param_links[jobs[dnode]] = djlinks
-        else:
-            for job in jobs_map[dnode]:
-                param_links[job] = djlinks
-
-    all_jobs = [job for job in list(jobs.values()) if not isinstance(job, tuple)]
-    root_jobs = sum(list(root_jobs.values()), [])
-
-    # if directories have to be created, all other primary jobs will depend
-    # on this first one
-    if create_directories and dirs_job is not None:
-        dependend_jobs = set()
-        for dependency in dependencies:
-            dependend_jobs.add(dependency[1])
-        new_deps = [(dirs_job, job) for job in all_jobs
-                    if job not in dependend_jobs]
-        dependencies.update(new_deps)
-        all_jobs.insert(0, dirs_job)
-        root_jobs.insert(0, dirs_job)
-
-    workflow = swclient.Workflow(jobs=all_jobs,
-        dependencies=dependencies,
-        root_group=root_jobs,
-        name=pipeline.name,
-        param_links=param_links)
-
-    # mark workflow with pipeline
-    workflow.pipeline = weakref.ref(pipeline)
-    workflow._do_not_pickle = ['pipeline']
-    if hasattr(pipeline, 'uuid'):
-        workflow.uuid = pipeline.uuid
-
-    return workflow
-
-
-def workflow_run(workflow_name, workflow, engine):
-    """ Create a soma-workflow controller and submit a workflow
-
-    Parameters
-    ----------
-    workflow_name: str (mandatory)
-        the name of the workflow
-    workflow: Workflow (mandatory)
-        the soma-workflow workflow
-    engine: CapsulEngine (mandatory)
-        contains needed configuration through the SomaWorkflowConfig module
-    """
-    swm = engine.modules['soma-workflow']
-    swm.connect_resource()
-    controller = swm.get_workflow_controller()
-    resource_id = swm.get_resource_id()
-    queue = None
-    if hasattr(engine.somaworkflow_computing_resources_config,
-               resource_id):
-        res_conf = getattr(
-            engine.somaworkflow_computing_resources_config, resource_id)
-        queue = getattr(res_conf, 'queue', undefined)
-        if queue is undefined:
-            queue = None
-    wf_id = controller.submit_workflow(workflow=workflow, name=workflow_name,
-                                       queue=queue)
-    swclient.Helper.transfer_input_files(wf_id, controller)
-    swclient.Helper.wait_workflow(wf_id, controller)
-
-    # get output values
-    pipeline = getattr(workflow, 'pipeline', None)
-    if pipeline:
-        pipeline = pipeline()  # dereference the weakref
-        proc_map = {}
-        todo = [pipeline]
-        while todo:
-            process = todo.pop(0)
-            if isinstance(process, Pipeline):
-                todo += [n.process for n in process.nodes.values()
-                         if n is not process.pipeline_node
-                             and isinstance(n, Process)]
-            else:
-                proc_map[id(process)] = process
-
-        eng_wf = controller.workflow(wf_id)
-        for job in eng_wf.jobs:
-            if job.has_outputs:
-                out_params = controller.get_job_output_params(
-                    eng_wf.job_mapping[job].job_id)
-                if out_params:
-                    process = proc_map.get(job.process_hash)
-                    if process is None:
-                        # iteration or non-process job
-                        continue
-                    for param in list(out_params.keys()):
-                        if process.field(param) is None:
-                            del out_params[param]
-                    process.import_dict(out_params)
-
-    # TODO: should we transfer if the WF fails ?
-    swclient.Helper.transfer_output_files(wf_id, controller)
-    return controller, wf_id
diff --git a/capsul/qt_gui/widgets/__init__.py b/capsul/qt_gui/widgets/__init__.py
index 78ca92432..fa5498402 100644
--- a/capsul/qt_gui/widgets/__init__.py
+++ b/capsul/qt_gui/widgets/__init__.py
@@ -1,7 +1,4 @@
 # -*- coding: utf-8 -*-
-from .pipeline_developper_view import PipelineDeveloperView
+from .pipeline_developer_view import PipelineDeveloperView
 from .pipeline_user_view import PipelineUserView
 from .links_debugger import CapsulLinkDebuggerView
-
-# deprecated. Imported for compatibility
-from .pipeline_developper_view import PipelineDevelopperView
diff --git a/capsul/qt_gui/widgets/activation_inspector.py b/capsul/qt_gui/widgets/activation_inspector.py
index 83232ce96..a43b32f31 100644
--- a/capsul/qt_gui/widgets/activation_inspector.py
+++ b/capsul/qt_gui/widgets/activation_inspector.py
@@ -80,7 +80,7 @@ def init_window(self):
         self.controller_window = ControllerWidget(self.pipeline)
         self.activation_window = ActivationInspector(
             self.pipeline, ui_file, self.record_file,
-            developper_view=self.pipeline_window)
+            developer_view=self.pipeline_window)
         self.pipeline_window.show()
         self.activation_window.show()
         self.controller_window.show()
@@ -91,7 +91,7 @@ class ActivationInspector(QtGui.QWidget):
     """ A Widget to display the pipeline activation process step by step.
     """
     def __init__(self, pipeline, ui_file=None, record_file=None,
-                 developper_view=None, parent=None):
+                 developer_view=None, parent=None):
         """ Initialize the ActivationInspector class.
 
         Parameters
@@ -104,7 +104,7 @@ def __init__(self, pipeline, ui_file=None, record_file=None,
         record_file: str (optional)
             a file path where the activation steps are recorded.
             If not specified (None), it will create a temporary file.
-        developper_view: PipelineDeveloperView (optional)
+        developer_view: PipelineDeveloperView (optional)
             if specified it is possible to click on a plug to set a filter
             pattern and to update the pipeline activation accordingly.
         """
@@ -147,7 +147,7 @@ def __del__(self):
         # Store class parameters
         self.pipeline = pipeline
         self.record_file = record_file
-        self.developper_view = developper_view
+        self.developer_view = developer_view
 
         # Set the pipeline record file if folder exists
         if os.path.isdir(os.path.dirname(self.record_file)):
@@ -173,8 +173,8 @@ def __del__(self):
 
         # Dynamically select a filter rule by clicking on the pipeline view
         # plugs
-        if developper_view is not None:
-          developper_view.plug_clicked.connect(self.set_filter_pattern)
+        if developer_view is not None:
+          developer_view.plug_clicked.connect(self.set_filter_pattern)
 
     def show(self):
         """ Shows the widget and its child widgets.
diff --git a/capsul/qt_gui/widgets/pipeline_developper_view.py b/capsul/qt_gui/widgets/pipeline_developper_view.py
deleted file mode 100644
index e0c37ef60..000000000
--- a/capsul/qt_gui/widgets/pipeline_developper_view.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-This is a compatibility module for the former, misspelled,
-pipeline_developper_view (with two "p"s) which should be spelled
-pipeline_developer_view.
-It contains a compatibility class, PipelineDevelopperView (with two "p"s) which
-should be spelled PipelineDeveloperView now. This class issues a deprecation
-warning and otherwise behaves the same.
-"""
-
-from .pipeline_developer_view import *
-import warnings
-
-
-warnings.warn(
-    'capsul.pipeline_developper_view is deprecated. Please use '
-    'capsul.pipeline_developer_view (with one "p") instead.',
-    DeprecationWarning)
-
-
-class PipelineDevelopperView(PipelineDeveloperView):
-    '''
-    This is a compatibility class for the former, misspelled,
-    PipelineDevelopperView (with two "p"s) which should be spelled
-    PipelineDeveloperView now. This class issues a deprecation warning and
-    otherwise behaves the same.
-    '''
-    def __init__(self, pipeline=None, parent=None, show_sub_pipelines=False,
-                 allow_open_controller=False, logical_view=False,
-                 enable_edition=False, userlevel=0):
-
-        warnings.warn(
-            'PipelineDevelopperView is deprecated. Please use '
-            'PipelineDeveloperView (with one "p") instead.',
-            DeprecationWarning)
-        super(PipelineDevelopperView, self).__init__(
-            pipeline=pipeline, parent=parent,
-            show_sub_pipelines=show_sub_pipelines,
-            allow_open_controller=allow_open_controller,
-            logical_view=logical_view,
-            enable_edition=enable_edition,
-            userlevel=userlevel)
diff --git a/capsul/qt_gui/widgets/settings_editor.py b/capsul/qt_gui/widgets/settings_editor.py
index fd7524e0a..4226d6933 100644
--- a/capsul/qt_gui/widgets/settings_editor.py
+++ b/capsul/qt_gui/widgets/settings_editor.py
@@ -3,9 +3,8 @@
 from __future__ import print_function
 
 import sys
-from capsul.engine import CapsulEngine
 from soma.qt_gui.qt_backend import Qt
-from soma.qt_gui.qvtabbar import QVTabBar, QVTabWidget
+from soma.qt_gui.qvtabbar import QVTabWidget
 
 
 class SettingsEditor(Qt.QDialog):
diff --git a/capsul/run.py b/capsul/run.py
index 5a8dce83b..4f9994473 100644
--- a/capsul/run.py
+++ b/capsul/run.py
@@ -6,119 +6,97 @@
 from soma.undefined import undefined
 
 from .application import Capsul
-from . import execution_context
+from .database import execution_database
 
 debug = False
 
-def filename_from_url(url):
-    return url.split('://', 1)[-1]
-
 if __name__ == '__main__':
-    tmp = os.environ.get('CAPSUL_TMP')
-    if not tmp:
-        print('Capsul cannot run command because CAPSUL_TMP is not defined: '
+    if len(sys.argv) < 2:
+        print('Capsul cannot run command because parameters are missing: '
               f'command={sys.argv}',
               file=sys.stderr)
         sys.exit(1)
-    if not os.path.exists(tmp):
-        print('Capsul cannot run command because temporary directory does not exist: '
-              f'tmp="{tmp}", command={sys.argv}',
-              file=sys.stderr)
-        sys.exit(1)
-    db_url = os.environ.get('CAPSUL_DATABASE')
-    if not db_url:
-        print('Capsul cannot run command because CAPSUL_DATABASE is not defined: '
-              f'tmp="{tmp}", command={sys.argv}',
-              file=sys.stderr)
-        sys.exit(1)
-    if not os.path.exists(filename_from_url(db_url)):
-        print('Capsul cannot run command because database file does not exist: '
-              f'tmp="{tmp}", database="{db_url}", command={sys.argv}',
+    database_url = os.environ.get('CAPSUL_DATABASE')
+    if not database_url:
+        print('Capsul cannot run job because CAPSUL_DATABASE is not defined: '
+              f'command={sys.argv}',
               file=sys.stderr)
-        sys.exit(1)
-    if len(sys.argv) < 2:
-        print('Capsul cannot run command because parameters are missing: '
-              f'tmp="{tmp}", database="{db_url}", command={sys.argv}',
+    execution_id = os.environ.get('CAPSUL_EXECUTION_ID')
+    if not execution_id:
+        print('Capsul cannot run job because CAPSUL_EXECUTION_ID is not defined: '
+              f'command={sys.argv}',
               file=sys.stderr)
         sys.exit(1)
-
-    database = execution_context.ExecutionDatabase(db_url)
-    with database as db:
-        execution_context = db.execution_context
+    database = execution_database(database_url)
+    try:
+        execution_context = database.execution_context(execution_id)
         execution_context.dataset.tmp = {
-            'path': tmp
+            'path': database.tmp(execution_id),
         }
-    command = sys.argv[1]
-    args = sys.argv[2:]
-    invalid_parameters = False
-    if command == 'process':
-        if len(args) == 1:
-            job_uuid = args[0]
-            with database as db:
-                row = db.session['jobs'].document(job_uuid, fields=['process', 'parameters_location'], as_list=True)
-            if row is None:
-                raise ValueError(f'No such job: {job_uuid}')
-            process_json, parameters_location = row
-            process = Capsul.executable(process_json)
-            if debug:
-                print(f'---- init {process.definition} {parameters_location} ----')
-            with database as db:
-                workflow_parameters = db.workflow_parameters
-            parameters = workflow_parameters
-            for index in parameters_location:
-                if index.isnumeric():
-                    index = int(index)
-                parameters = parameters[index]
-            if debug:
-                from pprint import pprint
-                pprint(parameters.no_proxy())
-                print(f'----')
-            for field in process.user_fields():
-                if field.name in parameters and parameters[field.name] is not None:
-                    value = parameters.no_proxy(parameters[field.name])
-                    setattr(process, field.name, value)
-            execution_context.executable = process
-            process.resolve_paths(execution_context)
-            if debug:
-                print(f'---- start {process.definition} ----')
-                for field in process.user_fields():
-                    if not field.output:
-                        value = getattr(process, field.name, undefined)
-                        if value is not undefined:
-                            print ('   ', field.name, '=', value)
-            process.before_execute(execution_context)
-            result = process.execute(execution_context)
-            process.after_execute(result, execution_context)
-            if debug:
-                print(f'---- stop {process.definition} ----')
+        command = sys.argv[1]
+        args = sys.argv[2:]
+        invalid_parameters = False
+        if command == 'process':
+            if len(args) == 1:
+                job_uuid = args[0]
+                job = database.job(execution_id, job_uuid)
+                if job is None:
+                    raise ValueError(f'No such job: {job_uuid}')
+                process_json = job['process']
+                parameters_location = job['parameters_location']
+                process = Capsul.executable(process_json)
+                if debug:
+                    print(f'---- init {process.definition} {parameters_location} ----')
+                workflow_parameters = database.workflow_parameters(execution_id)
+                parameters = workflow_parameters
+                for index in parameters_location:
+                    if index.isnumeric():
+                        index = int(index)
+                    parameters = parameters[index]
+                if debug:
+                    from pprint import pprint
+                    pprint(parameters.no_proxy())
+                    print(f'----')
                 for field in process.user_fields():
-                    if field.output:
-                        value = getattr(process, field.name, undefined)
-                        if value is not undefined:
-                            print ('   ', field.name, '=', value)
-            workflow_parameters = None
-            with database as db:
+                    if field.name in parameters and parameters[field.name] is not None:
+                        value = parameters.no_proxy(parameters[field.name])
+                        setattr(process, field.name, value)
+                execution_context.executable = process
+                process.resolve_paths(execution_context)
+                if debug:
+                    print(f'---- start {process.definition} ----')
+                    for field in process.user_fields():
+                        if not field.output:
+                            value = getattr(process, field.name, undefined)
+                            if value is not undefined:
+                                print ('   ', field.name, '=', value)
+                process.before_execute(execution_context)
+                result = process.execute(execution_context)
+                process.after_execute(result, execution_context)
+                if debug:
+                    print(f'---- stop {process.definition} ----')
+                    for field in process.user_fields():
+                        if field.output:
+                            value = getattr(process, field.name, undefined)
+                            if value is not undefined:
+                                print ('   ', field.name, '=', value)
+                output_values = {}
                 for field in process.user_fields():
                     if field.output:
                         value = getattr(process, field.name, undefined)
                         if value is not undefined:
-                            if workflow_parameters is None:
-                                workflow_parameters = db.workflow_parameters
-                                parameters = workflow_parameters
-                                for index in parameters_location:
-                                    if index.isnumeric():
-                                        index = int(index)
-                                    parameters = parameters[index]
-                            parameters[field.name] = value
-                if workflow_parameters is not None:
-                    db.workflow_parameters = workflow_parameters
+                            output_values[field.name] = value
+                if output_values is not None:
+                    database.update_workflow_parameters(execution_id, parameters_location, output_values)
+            else:
+                invalid_parameters = True  
         else:
-            invalid_parameters = True  
-    else:
-        invalid_parameters = True
-
+            invalid_parameters = True
+    finally:
+        database.close()
+    
     if invalid_parameters:
         print('Capsul cannot run command because parameters are invalid: '
-            f'tmp="{tmp}", database="{db_url}", command={sys.argv}',
+            f'command={sys.argv}',
             file=sys.stderr)
         sys.exit(1)
diff --git a/completion.md b/completion.md
new file mode 100644
index 000000000..a46c46be0
--- /dev/null
+++ b/completion.md
@@ -0,0 +1,153 @@
+# Path generation API
+
+We call path generation the automatic creation of paths values for a process instance. This document describes the path generation system of Capsul.
+
+
+## General principles
+
+The path generation system is designed to be fully automatic. Path generation must not need any user input except the selection of a few input parameters of the process. This is important to be able to use path generation on iteration of hundreds of executions. 
+
+Path generation is done using metadata and metadata schema. The metadata contains the values that are used to build various part of the path. For instance, the subject code is often used in path names as well as an extension defining the file type. These two values are included in the metadata. There is no single way to create a path given metadata. There are many possible layouts for path names using various metadata. For instance, BrainVISA has defined a path organisation layout. BIDS is another path organisation layout that is the actual standard for neuroimaging. Capsul can support many different systems; each one beign defined in a `MetadataSchema` class (see below).
+
+Metadata given to create a path name can have various origins. For instance, the extension of the file is most often dependent on the process. An image parameter uses an image extension (such as `.nii`) whereas a mesh parameter uses a mesh format (such as `.gii`). This kind of metatada, called process metadata, is defined globally for a process, usually by the process developer. On the other hand, metadata such a subject identifier depends on the usage context of the process. This kind of metadata is called user metadata becaus it is given at runtime as the result of a user action (manual, entry, database selection, etc.).
+
+The path generation system must be able to deal with several metadata schemas for a single process. If not it woul mean that all process parameters must be in the same schema. This is the case if all input and output data are following the BIDS standard. However, in many cases a process will have to deal with several metadata schemas. For instance, there could be one schema for input data (i.e. BIDS), another schema for output data (for use cases not covered by BIDS) and other schemas for third party data (for instance template images in SPM software directory). To support several metadata schema, Capsul make a link between a dataset (i.e. a directory) and the metadata schema used throughout that directory using he `Dataset` class.
+
+### Using path generation
+
+Path generation is done for all path parameters of a process. The following diagram illustrate the case of a process with one `input` parameter supposed to follow BIDS schema and one `output` parameter following another schema called BrainVISA. In oder to use path generation, the user must create a ̀ProcessMetadata` instance that allows to set user metadata for all schemas used by the process. These user metadata are combined with process metadata in order to generate values for all path parameters.
+
+```mermaid
+graph LR
+    P["Process<br/><code>input</code> → 'bids' schema<br/><code>output</code> → 'brainvisa' schema"]
+    -- 1 - create metadata for process -->
+    M["ProcessMetadata<br/><code>bids</code> → metadata for bids schema<br/><code>brainvisa</code> → metadata for brainvisa schema"]
+    -- 2 - Generate paths --> P
+```
+
+## Datasets and relative directory reference
+
+In Capsul, during a process creation, the authors can define how many different datasets the process will use. For instance, a normalization process could be defined with three datasets : one for input data, one for output data and one containing normalization templates. Except the list of datasets, no other information is known about them at process creation time (neither the base path of the dataset nor the schema it uses). Therefore, a process simply defines one name for each dataset it uses. By default, two datasets are defined : ̀`input` for input files and `output`for output files. But a processes can define there own dataset symbolic names. For instance `SPM templates` could be used to designate the directory containing the template images bundled with the SPM software. Of course, it is necessary that all processes uses the same symbolic name for a given directory.
+
+To ba able to do something with a dataset name, one must know two things:
+* What is the path of the base directory for this dataset ?
+* How the file names are organized in this directory ? In other words, what is the metadata schema associated with this dataset.
+
+This information depends on the execution environment. Indeed, the path of a directory that will be used by a process may only be valid on the execution environment. If the user is on a remote computer this path may be invalid on this machine. Therefore, the link between a dataset global name and its actual location and schema is done in Capsul's execution environment configuration. This configuration contains a series of `Dataset` objects making the link between the global name and the base path location as well as the metadata schema. A metadata schema represents the orgranization of the paths within the dataset. For instance their is a metadata schema for BIDS organization. Each metadata schema is a subclass of `MetadataSchema` that derives from `Controller`. A schema contains controller fields describing the metadata that can be used to create a single path. For instance, a BIDS dataset class could be defined as follows:
+
+```
+class BIDSSchema(MetadataSchema):
+    ''' Metadata schema for BIDS datasets
+    '''
+    schema_name = 'bids'
+
+    folder: Literal['sourcedata', 'rawdata', 'derivative']
+    process: str = None
+    sub: str
+    ses: str
+    data_type: str = None
+    task: str = None
+    acq: str = None
+    ce: str = None
+    rec: str = None
+    run: str = None
+    echo: str = None
+    part: str = None
+    suffix: str = None
+    extension: str
+```
+
+The fields are the various elements that one can find in a complete BIDS path. Given valid values for these fields, it is possible to build any valid BIDS paths (relative to a base path).
+
+## Simple example
+
+```python
+from capsul.api import Capsul, Process
+from capsul.dataset import ProcessMetadata, ProcessSchema
+
+from soma.controller import field, File
+
+
+class SegmentHemispheres(Process):
+    input: File
+    voronoi: field(type_=File, write=True)
+    left_output: field(type_=File, write=True)
+    right_output: field(type_=File, write=True)
+
+
+
+class SegmentHemispheresBids(ProcessSchema, 
+                            schema='bids',
+                            process=SegmentHemispheres):
+    _ = {
+        '*': {
+            'process': 'segment_hemispheres',
+            'extension': 'nii'
+        }
+    }
+    voronoi = {'suffix': 'voronoi'}
+    left_output = {'suffix': 'lhemi'}
+    right_output = {'suffix': 'rhemi'}
+
+
+if __name__ == '__main__':
+    config = {
+        'builtin': {
+            'dataset': {
+                'input': {
+                    'path': '/input',
+                    'metadata_schema': 'bids',
+                },
+                'output': {
+                    'path': '/output',
+                    'metadata_schema': 'bids',
+                },
+            }
+        }
+    }
+    capsul = Capsul()
+    capsul.config.import_dict(config)
+    process = Capsul.executable('__main__.SegmentHemispheres')
+    engine = capsul.engine()
+    execution_context = engine.execution_context(process)
+    metadata = ProcessMetadata(process, execution_context)
+    metadata.bids.folder = 'derivative'
+    metadata.bids.sub = 'thesubject'
+    metadata.bids.ses = 'thesession'
+    metadata.generate_paths(process)
+
+    print('Paths before execution:')
+    for field in process.user_fields():
+        value =  getattr(process, field.name)
+        print(field.name, f"= '{value}'",)
+
+    print('Paths during execution:')
+    process.resolve_paths(execution_context)
+    for field in process.user_fields():
+        value =  getattr(process, field.name)
+        print(field.name, f"= '{value}'",)
+```
+
+In this example, a fake process with one input parameter and three outputs is created. Then the `SegmentHemispheresBids` definition gives the BIDS specific metadata for this process. This metadata indicate that when a `BIDSSchema` is created for this process, two attributes will be set for all parameters: `process='segment_hemisphere'` and `extension='nii'`; and each output parameter have its own `suffix` value.
+
+A configuration is created containing two datasets for the symbolic names `input` and `output`. These dataset names are used by default for input and output parameters.
+
+After the creation of a process instance and an execution context, a `ProcessMetadata` class is created, some user metadata attributes are set (BIDS values for `folder`, `sub`, `ses`) and `generate_paths()` is called to create path values for each parameters given process and user and metadatas. 
+
+Finally, this script prints the process parameters twice. The output is:
+```
+Paths before execution:
+input = '!{dataset.input.path}/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession.nii'
+voronoi = '!{dataset.output.path}/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession_voronoi.nii'
+left_output = '!{dataset.output.path}/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession_lhemi.nii'
+right_output = '!{dataset.output.path}/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession_rhemi.nii'
+Paths during execution:
+input = '/input/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession.nii'
+voronoi = '/output/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession_voronoi.nii'
+left_output = '/output/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession_lhemi.nii'
+right_output = '/output/derivative/segment_hemispheres/sub-thesubject/ses-thesession/sub-thesubject_ses-thesession_rhemi.nii'
+```
+
+First, the parameters are printed as they are created by `generate_paths()`. These paths (starting with `!` character) are using expressions that will be resolved at runtime using the execution context. This allows to make paths that uses symbolic dataset names (`input` and `output` in this case) and can be used on client side independently of the final configuration. 
+
+Then, the final real path value after the resolution of special expressions is printed.
diff --git a/doc/source/tutorial/capsul_tutorial.ipynb.in b/doc/source/tutorial/capsul_tutorial.ipynb.in
index 92644783d..d732009cb 100644
--- a/doc/source/tutorial/capsul_tutorial.ipynb.in
+++ b/doc/source/tutorial/capsul_tutorial.ipynb.in
@@ -345,7 +345,7 @@
     "    import sys\n",
     "    # note that the qt backend may be either PyQt4 or PySide.\n",
     "    from soma.qt_gui.qt_backend import QtGui\n",
-    "    from capsul.qt_gui.widgets import PipelineDevelopperView\n",
+    "    from capsul.qt_gui.widgets import PipelineDeveloperView\n",
     "\n",
     "    # here we determine whether the Qt GUI is already running or not.\n",
     "    run_qt_loop = False\n",
@@ -357,7 +357,7 @@
     "    # in the following we will reuse this run_qt_loop variable for simplicity\n",
     "\n",
     "    # now the real thing for pipeline viewing    \n",
-    "    view1 = PipelineDevelopperView(pipeline1)\n",
+    "    view1 = PipelineDeveloperView(pipeline1)\n",
     "    view1.show()\n",
     "\n",
     "    if run_qt_loop:\n",
@@ -445,7 +445,7 @@
    "source": [
     "pipeline2 = Pipeline2()\n",
     "if globals().get('use_gui', True):\n",
-    "    view2 = PipelineDevelopperView(pipeline2)\n",
+    "    view2 = PipelineDeveloperView(pipeline2)\n",
     "    view2.show()\n",
     "    if run_qt_loop:\n",
     "        app.exec_()"
@@ -556,10 +556,10 @@
     "if globals().get('use_gui', True):\n",
     "    import sys\n",
     "    from soma.qt_gui.qt_backend import QtGui\n",
-    "    from capsul.qt_gui.widgets import PipelineDevelopperView\n",
+    "    from capsul.qt_gui.widgets import PipelineDeveloperView\n",
     "    from soma.qt_gui.controller_widget import ControllerWidget\n",
     "\n",
-    "    view = PipelineDevelopperView(xmlpipe)\n",
+    "    view = PipelineDeveloperView(xmlpipe)\n",
     "    controller = ControllerWidget(xmlpipe, live=True)\n",
     "    view.show()\n",
     "    controller.show()\n",
@@ -618,7 +618,7 @@
    "source": [
     "pipeline3 = Pipeline3()\n",
     "if globals().get('use_gui', True):\n",
-    "    view3 = PipelineDevelopperView(pipeline3, allow_open_controller=True, show_sub_pipelines=True)\n",
+    "    view3 = PipelineDeveloperView(pipeline3, allow_open_controller=True, show_sub_pipelines=True)\n",
     "    view3.show()\n",
     "    if run_qt_loop:\n",
     "        app.exec_()"
@@ -741,10 +741,10 @@
    "source": [
     "if globals().get('use_gui', True):\n",
     "    from soma.qt_gui.qt_backend import QtGui\n",
-    "    from capsul.qt_gui.widgets import PipelineDevelopperView\n",
+    "    from capsul.qt_gui.widgets import PipelineDeveloperView\n",
     "    from soma.qt_gui.controller_widget import ControllerWidget\n",
     "\n",
-    "    view = PipelineDevelopperView(xmlpipe)\n",
+    "    view = PipelineDeveloperView(xmlpipe)\n",
     "    controller = ControllerWidget(xmlpipe, live=True)\n",
     "    view.show()\n",
     "    controller.show()\n",
@@ -812,7 +812,7 @@
    "source": [
     "pipeline4 = Pipeline4()\n",
     "if globals().get('use_gui', True):\n",
-    "    view4 = PipelineDevelopperView(pipeline4, allow_open_controller=True, show_sub_pipelines=True)\n",
+    "    view4 = PipelineDeveloperView(pipeline4, allow_open_controller=True, show_sub_pipelines=True)\n",
     "    view4.show()\n",
     "    if run_qt_loop:\n",
     "        app.exec_()"
@@ -923,7 +923,7 @@
    "source": [
     "pipeline5 = Pipeline5()\n",
     "if globals().get('use_gui', True):\n",
-    "    view5 = PipelineDevelopperView(pipeline5, allow_open_controller=True, show_sub_pipelines=True)\n",
+    "    view5 = PipelineDeveloperView(pipeline5, allow_open_controller=True, show_sub_pipelines=True)\n",
     "    view5.show()\n",
     "    if run_qt_loop:\n",
     "        app.exec_()"
@@ -1020,7 +1020,7 @@
    "source": [
     "pipeline6 = Pipeline6()\n",
     "if globals().get('use_gui', True):\n",
-    "    view6 = PipelineDevelopperView(pipeline6, allow_open_controller=True, show_sub_pipelines=True)\n",
+    "    view6 = PipelineDeveloperView(pipeline6, allow_open_controller=True, show_sub_pipelines=True)\n",
     "    view6.show()\n",
     "    if run_qt_loop:\n",
     "        app.exec_()"
@@ -1752,9 +1752,9 @@
    "outputs": [],
    "source": [
     "if globals().get('use_gui', True):\n",
-    "    from capsul.qt_gui.widgets import PipelineDevelopperView\n",
+    "    from capsul.qt_gui.widgets import PipelineDeveloperView\n",
     "\n",
-    "    pv = PipelineDevelopperView(pipeline, allow_open_controller=True, show_sub_pipelines=True)\n",
+    "    pv = PipelineDeveloperView(pipeline, allow_open_controller=True, show_sub_pipelines=True)\n",
     "    pv.show()\n",
     "\n",
     "    if run_qt_loop:\n",
@@ -1895,7 +1895,7 @@
     "    ap_view = AttributedProcessWidget(mp)\n",
     "    ap_view.show()\n",
     "\n",
-    "    pv = PipelineDevelopperView(mp, allow_open_controller=True, show_sub_pipelines=True)\n",
+    "    pv = PipelineDeveloperView(mp, allow_open_controller=True, show_sub_pipelines=True)\n",
     "    pv.show()\n",
     "\n",
     "    if run_qt_loop:\n",
diff --git a/doc/source/user_guide_tree/index.rst b/doc/source/user_guide_tree/index.rst
index a4b46c082..40855e3a9 100644
--- a/doc/source/user_guide_tree/index.rst
+++ b/doc/source/user_guide_tree/index.rst
@@ -144,7 +144,7 @@ Documenting CAPSUL processes and pipelines with Sphinx
 
 Sphinx documentation can be built automatically for all Capsul processes. See :mod:`capsul.sphinxext`.
 
-Hints may be stored in process classes to point to the documentation. This doc may be accessed through the :class:`~capsul.qt_gui.widgets.pipeline_developper_view.PipelineDevelopperView` pipeline viewer (at least). Several mechanisms can be used to find the HTML documentation of a process or node:
+Hints may be stored in process classes to point to the documentation. This doc may be accessed through the :class:`~capsul.qt_gui.widgets.pipeline_developer_view.PipelineDeveloperView` pipeline viewer (at least). Several mechanisms can be used to find the HTML documentation of a process or node:
 
 * a :class:`~capsul.process.process.Process` or :class:`~capsul.pipeline.pipeline_nodes.Node` class (or instance) may contain a ``_doc_path`` attribute. It points to the HTML document corresponding to the process documentation. The path may be absolute (``/path/to/file.html``), or an URL (``https://populse.github.io/capsul/api/pipeline.html#leaveoneoutnode``), or a relative URL. In the latter case the link is relative to to root of the project documentation, found via the process or node modules hierarchy. See below.
 
diff --git a/README.rst b/readme.md
similarity index 62%
rename from README.rst
rename to readme.md
index 38dd6f440..d3ebc8fdb 100644
--- a/README.rst
+++ b/readme.md
@@ -1,33 +1,32 @@
-======
-CAPSUL 
-======
+# Capsul v3
+
+[![test_v3](https://github.com/populse/capsul/actions/workflows/test-v3.yml/badge.svg?branch=celery_redis)](https://github.com/populse/capsul/actions/workflows/test-v3.yml)
 
-Future Capsul version 3
-=======================
 
 This branch contains the code of the future 3.0 release. This is still work in progress and using it is not possible through standards installation means such as Pypi.
 
-Using development version od Capsul v3
-======================================
+## Using development version of Capsul v3
 
-To use Capsul v3 it is mandatory to have at least Python 3.9 and the @pydantic_controller@ branch of projects `soma-base <https://github.com/populse/soma-base>`_ and `capsul <https://github.com/populse/capsul>`_ .
+To use Capsul v3 it is mandatory to have at least Python 3.9 and the @pydantic_controller@ branch of projects [soma-base](https://github.com/populse/soma-base>) and [capsul](https://github.com/populse/capsul>).
 
-The simplest is to use a `casa-distro <https://github.com/brainvisa/casa-distro>`_ container for developers, and setup a minimalist dev environment, based on an Ubuntu 22.04 container with singularity:
+The simplest is to use a [casa-distro](https://github.com/brainvisa/casa-distro>) container for developers, and setup a minimalist dev environment, based on an Ubuntu 22.04 container with singularity:
 
-* `install singularity <https://brainvisa.info/web/download.html#prerequisites-for-singularity-on-linux>`_
+* [install singularity](https://brainvisa.info/web/download.html#prerequisites-for-singularity-on-linux>)
 
 
-* get a recent `developer image <https://brainvisa.info/web/download.html#installing-a-singularity-developer-environment>`_::
-  
+* get a recent [developer image](https://brainvisa.info/web/download.html#installing-a-singularity-developer-environment>):
+```  
         wget https://brainvisa.info/download/casa-dev-5.3-6.sif
+```
 
-* setup a developer environment::
-
+* setup a developer environment:
+```
       mkdir capsul3
       singularity run -B capsul3:/casa/setup casa-dev-5.3-6.sif distro=opensource
+```
 
-* change the `bv_maker.cfg` file for a ligher one, which switches to the expected branches::
-
+* change the `bv_maker.cfg` file for a ligher one, which switches to the expected branches:
+```
       cat > capsul3/conf/bv_maker.cfg << EOF
       [ source \$CASA_SRC ]
         brainvisa brainvisa-cmake \$CASA_BRANCH
@@ -53,15 +52,17 @@ The simplest is to use a `casa-distro <https://github.com/brainvisa/casa-distro>
         + \$CASA_SRC/populse_db
         + \$CASA_SRC/capsul
       EOF
+```
 
-* get the code and build::
-
+* get the code and build:
+```
       capsul3/bin/bv_maker
+```
 
 * It's ready. You can use it using either::
-
+```
       capsul3/bin/bv bash
       capsul3/bin/bv ipython3
       capsul3/bin/bv python <script>
-
-  You can also set the python3/bin directory into your `PATH` configuration.
+```
+  You can also set the `python3/bin` directory into your `PATH` configuration.


Test patch:
diff --git a/.github/workflows/test-v3.yml b/.github/workflows/test-v3.yml
new file mode 100644
index 000000000..fe1404870
--- /dev/null
+++ b/.github/workflows/test-v3.yml
@@ -0,0 +1,41 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
+
+name: test_v3
+
+on:
+  push:
+    branches:
+    - master
+    - celery_redis
+    
+  pull_request:
+    branches:
+    - celery_redis
+    - pydantic_controller
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python 3.9
+      uses: actions/setup-python@v3
+      with:
+        python-version: "3.9"
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install git+https://github.com/populse/soma-base.git@pydantic_controller
+        pip install git+https://github.com/populse/soma-workflow.git
+        pip install git+https://github.com/populse/populse_db.git@3.0
+        pip install .
+    - name: Test
+      run: |
+        python -m capsul.test
+        
diff --git a/capsul/config/test/test_config.py b/capsul/config/test/test_config.py
index e97d07270..0ebd60429 100644
--- a/capsul/config/test/test_config.py
+++ b/capsul/config/test/test_config.py
@@ -1,14 +1,15 @@
 # -*- coding: utf-8 -*-
-import unittest
-from capsul.config import (ApplicationConfiguration, ConfigurationLayer,
-                           EngineConfiguration)
-from soma.controller import undefined
-import sys
+import json
 import os
 import os.path as osp
 import shutil
 import tempfile
-import json
+import unittest
+
+from soma.controller import undefined
+
+from capsul.config import ApplicationConfiguration
+from capsul.config.configuration import default_workers_type, default_database_url
 
 
 class TestConfiguration(unittest.TestCase):
@@ -26,8 +27,9 @@ def test_single_configuration(self):
 
         user_file = osp.join(self.tmp_dir, 'user_conf.json')
         conf_dict = {
-            'local': {
-                'engine_type': 'builtin',
+            'builtin': {
+                'workers_type': default_workers_type,
+                'database_url': default_database_url,
                 'matlab': {},
                 'spm': {
                     'spm12_standalone': {
@@ -44,18 +46,26 @@ def test_single_configuration(self):
                                               user_file=user_file)
         self.maxDiff = 2000
         self.assertEqual(
-            app_config.asdict(),
-            {'site': {'local': {'engine_type': 'builtin'}},
-             'app_name': 'single_conf',
-             'user': conf_dict,
-             'merged_config': conf_dict,
-             'user_file': user_file})
+            app_config.asdict(), {
+                'site': {
+                    'builtin': {
+                        'workers_type': default_workers_type,
+                        'database_url': default_database_url,
+                    }
+                },
+                'app_name': 'single_conf',
+                'user': conf_dict,
+                'merged_config': conf_dict,
+                'user_file': user_file
+            }
+        )
 
     def test_config_assignment(self):
 
         conf_dict = {
-            'local': {
-                'engine_type': 'builtin',
+            'builtin': {
+                'workers_type': default_workers_type,
+                'database_url': default_database_url,
                 'matlab': {},
                 'spm': {
                     'spm12_standalone': {
@@ -72,25 +82,30 @@ def test_config_assignment(self):
                                               user_file=None)
         app_config.user = conf_dict
 
-        # print(app_config.asdict())
         self.maxDiff = None
         self.assertEqual(
             app_config.asdict(),
-            {'site': {'local': {'engine_type': 'builtin'}},
+            {'site': {'builtin': {
+                'workers_type': default_workers_type,
+                'database_url': default_database_url,
+              }},
              'app_name': 'single_conf2',
              'user': conf_dict,
-             'merged_config': {'local': {'engine_type': 'builtin'}}})
+             'merged_config': {'builtin': {
+                'workers_type': default_workers_type,
+                'database_url': default_database_url,
+             }}})
 
     def test_config_merge(self):
         user_conf_dict = {
-            'local': {
+            'builtin': {
                 'spm': {
                     'spm12_standalone': {
                         'directory': '/usr/local/spm12_standalone',
                         'standalone': True},
                     }}}
         site_conf_dict = {
-            'local': {
+            'builtin': {
                 'spm': {
                     'spm12_standalone': {
                         'directory': '/i2bm/local/spm12_standalone',
@@ -107,8 +122,9 @@ def test_config_merge(self):
                         'setup_script': '/i2bm/local/fsl/etc/fslconf/fsl.sh'
                     }}}}
         merged_conf_dict = {
-            'local': {
-                'engine_type': 'builtin',
+            'builtin': {
+                'workers_type': default_workers_type,
+                'database_url': default_database_url,
                 'matlab': {},
                 'spm': {
                     'spm12_standalone': {
diff --git a/capsul/pipeline/test/test_custom_nodes.py b/capsul/pipeline/test/test_custom_nodes.py
index 5706a9f3e..bc777f763 100644
--- a/capsul/pipeline/test/test_custom_nodes.py
+++ b/capsul/pipeline/test/test_custom_nodes.py
@@ -2,7 +2,6 @@
 
 import unittest
 from capsul.api import Process, Pipeline, executable
-from capsul.pipeline import pipeline_workflow
 from capsul.pipeline import python_export
 from soma.controller import File
 import os
diff --git a/capsul/pipeline/test/test_iterative_process.py b/capsul/pipeline/test/test_iterative_process.py
index c2fb65285..b68085c17 100644
--- a/capsul/pipeline/test/test_iterative_process.py
+++ b/capsul/pipeline/test/test_iterative_process.py
@@ -17,26 +17,6 @@
 
 debug = False
 
-def setUpModule():
-    global old_home
-    global temp_home_dir
-    # Run tests with a temporary HOME directory so that they are isolated from
-    # the user's environment
-    temp_home_dir = None
-    old_home = os.environ.get('HOME')
-    try:
-        temp_home_dir = tempfile.mkdtemp('', prefix='soma_workflow')
-        os.environ['HOME'] = temp_home_dir
-        swconfig.change_soma_workflow_directory(temp_home_dir)
-    except BaseException:  # clean up in case of interruption
-        if old_home is None:
-            del os.environ['HOME']
-        else:
-            os.environ['HOME'] = old_home
-        if temp_home_dir:
-            shutil.rmtree(temp_home_dir)
-        raise
-
 def setUpModule():
     global old_home
     global temp_home_dir
diff --git a/capsul/pipeline/test/test_pipeline_with_temp.py b/capsul/pipeline/test/test_pipeline_with_temp.py
index 095948aef..cf9e354b5 100644
--- a/capsul/pipeline/test/test_pipeline_with_temp.py
+++ b/capsul/pipeline/test/test_pipeline_with_temp.py
@@ -150,7 +150,8 @@ def test_iterative_pipeline_with_temp(self):
             self.assertTrue(os.path.exists(output_name))
             with open(input_name) as f:
                 with open(output_name) as g:
-                    self.assertEqual(f.read() * 3, g.read())
+                    r = g.read()
+                    self.assertEqual(f.read() * 3, r)
             # check temporary filenames
             o = self.iter_pipeline.nodes['node1'].output_image
             self.assertEqual(len(o), 3)
@@ -175,7 +176,6 @@ def test():
 
 
 if __name__ == "__main__":
-    import sys
     from soma.qt_gui.qt_backend import QtGui
     from capsul.qt_gui.widgets import PipelineDeveloperView
 
diff --git a/capsul/pipeline/test/test_pipeline_workflow.py b/capsul/pipeline/test/test_pipeline_workflow.py
index 2cbc27b04..d37cf3f51 100644
--- a/capsul/pipeline/test/test_pipeline_workflow.py
+++ b/capsul/pipeline/test/test_pipeline_workflow.py
@@ -170,9 +170,10 @@ def test_requirements(self):
         with engine.settings as session:
             session.remove_config('spm', 'global', 'spm12-standalone')
         self.pipeline.enable_all_pipeline_steps()
-        with self.assertRaises(ValueError):
-            wf = pipeline_workflow.workflow_from_pipeline(
-                self.pipeline, study_config=self.study_config)
+        # Commented out to avoid error with flake8
+        # with self.assertRaises(ValueError):
+        #     wf = pipeline_workflow.workflow_from_pipeline(
+        #         self.pipeline, study_config=self.study_config)
 
 
     def test_full_wf(self):
@@ -243,7 +244,7 @@ def test_iter_workflow_without_temp(self):
         with Capsul().engine() as engine:
             status = engine.run(pipeline)
 
-        self.assertEqual(status['status'], 'ended')
+        self.assertEqual(status, 'ended')
         self.assertTrue(osp.exists(pipeline.output))
         olen = 12
         with open(pipeline.output) as f:
@@ -271,7 +272,7 @@ def test_iter_workflow(self):
         with Capsul().engine() as engine:
             status = engine.run(pipeline)
 
-        self.assertEqual(status['status'], 'ended')
+        self.assertEqual(status, 'ended')
         self.assertTrue(osp.exists(pipeline.output1))
         self.assertTrue(osp.exists(pipeline.output2))
         self.assertTrue(osp.exists(pipeline.output3))
diff --git a/capsul/pipeline/test/test_temporary.py b/capsul/pipeline/test/test_temporary.py
index 6fbc5db7c..bbfae6dd8 100644
--- a/capsul/pipeline/test/test_temporary.py
+++ b/capsul/pipeline/test/test_temporary.py
@@ -155,13 +155,13 @@ def setUp(self):
         self.output = tmpout[1]
         self.pipeline.input = '/tmp/file_in.nii'
         self.pipeline.output = self.output
-        study_config = StudyConfig(modules=['SomaWorkflowConfig'])
-        study_config.input_directory = '/tmp'
-        study_config.somaworkflow_computing_resource = 'localhost'
-        study_config.somaworkflow_computing_resources_config.localhost = {
-            'transfer_paths': [],
-        }
-        self.study_config = study_config
+        # study_config = StudyConfig(modules=['SomaWorkflowConfig'])
+        # study_config.input_directory = '/tmp'
+        # study_config.somaworkflow_computing_resource = 'localhost'
+        # study_config.somaworkflow_computing_resources_config.localhost = {
+        #     'transfer_paths': [],
+        # }
+        # self.study_config = study_config
 
     def tearDown(self):
         swm = self.study_config.modules['SomaWorkflowConfig']
diff --git a/capsul/process/test/test_metadata_schema.py b/capsul/process/test/test_metadata_schema.py
index 419c6ab33..42c79b780 100644
--- a/capsul/process/test/test_metadata_schema.py
+++ b/capsul/process/test/test_metadata_schema.py
@@ -92,7 +92,7 @@ def setUp(self):
             output.mkdir()
             with (config / f'{app_name}.json').open('w') as f:
                 json.dump({
-                    'local': {
+                    'builtin': {
                         'python_modules': [
                             'capsul.process.test.test_metadata_schema'
                         ],
diff --git a/capsul/process/test/test_nipype_wrap.py b/capsul/process/test/test_nipype_wrap.py
index 5e43c0dc2..c5678aa7c 100644
--- a/capsul/process/test/test_nipype_wrap.py
+++ b/capsul/process/test/test_nipype_wrap.py
@@ -56,7 +56,7 @@ def init_spm_config():
 
     config = ApplicationConfiguration('capsul_test_nipype_spm')
     user_conf_dict = {
-        'local': {
+        'builtin': {
             'spm': {
                 'spm12_standalone': {
                     'directory': spm_dir,
@@ -135,7 +135,7 @@ def test_nipype_spm_exec(self):
                          'spm12']
         for template_dir_s in template_dirs:
             template_dir = osp.join(
-                c.config.local.spm.spm12_standalone.directory, template_dir_s)
+                c.config.builtin.spm.spm12_standalone.directory, template_dir_s)
             if osp.isdir(template_dir):
                 break
         self.assertTrue(
diff --git a/capsul/process/test/test_runprocess.py b/capsul/process/test/test_runprocess.py
index 3a2bd9ffd..7227dc9ce 100644
--- a/capsul/process/test/test_runprocess.py
+++ b/capsul/process/test/test_runprocess.py
@@ -20,6 +20,7 @@ def execute(self, context=None):
 
 class TestRunProcess(unittest.TestCase):
     """Test case for CAPSUL command-line usage."""
+    @unittest.skip('reimplementation expected for capsul v3')
     def test_python_m_capsul(self):
         with open(os.devnull, 'wb') as f:
             ret = subprocess.call([sys.executable, "-m", "capsul", "--help"],
diff --git a/capsul/test/test_completion.py b/capsul/test/test_completion.py
index afbd7b27d..bc93e50e6 100644
--- a/capsul/test/test_completion.py
+++ b/capsul/test/test_completion.py
@@ -80,7 +80,7 @@ class FakeSPMNormalization12BrainVISA(
 class TestPipelineBIDS(ProcessSchema, schema='bids',
                        process=TestPipeline):
     _ = {
-        '*': {'process': 'test_pipeline'}
+        '*': {'pipeline': 'test_pipeline'}
     }
 
 class TestPipelineBrainVISA(ProcessSchema, schema='brainvisa',
@@ -115,7 +115,7 @@ def setUp(self):
 
         # Configuration base dictionary
         config = {
-            'local': {
+            'builtin': {
                 'config_modules': [
                     'capsul.test.test_fake_morphologist',
                 ],
diff --git a/capsul/test/test_fake_morphologist.py b/capsul/test/test_fake_morphologist.py
index 6867d95ef..a8effd185 100644
--- a/capsul/test/test_fake_morphologist.py
+++ b/capsul/test/test_fake_morphologist.py
@@ -12,7 +12,7 @@
 from soma.controller import Directory, undefined
 
 from capsul.api import Capsul, Process, Pipeline
-from capsul.config.configuration import ModuleConfiguration
+from capsul.config.configuration import ModuleConfiguration, default_workers_type, default_database_url
 from capsul.dataset import ProcessMetadata, ProcessSchema, MetadataSchema
 
 class FakeSPMConfiguration(ModuleConfiguration):
@@ -666,7 +666,7 @@ def setUp(self):
 
         # Configuration base dictionary
         config = {
-            'local': {
+            'builtin': {
                 'config_modules': [
                     'capsul.test.test_fake_morphologist',
                 ],
@@ -699,7 +699,7 @@ def setUp(self):
                 'directory': str(fakespm),
                 'version': version,
             }
-            config['local'].setdefault('fakespm', {})[f'fakespm_{version}'] = fakespm_config
+            config['builtin'].setdefault('fakespm', {})[f'fakespm_{version}'] = fakespm_config
             
 
         # Create a configuration file
@@ -721,8 +721,9 @@ def tearDown(self):
     def test_fake_morphologist_config(self):
         self.maxDiff = 2000
         expected_config = {
-            'local': {
-                'engine_type': 'builtin',
+            'builtin': {
+                'workers_type': default_workers_type,
+                'database_url': default_database_url,
                 'dataset': {
                     'input': {
                         'path': str(self.tmp / 'bids'),
@@ -1052,7 +1053,7 @@ def test_path_generation(self):
                 metadata.bids.asdict(),
                 {
                     'folder': 'rawdata',
-                    'pipeline': None,
+                    'process': None,
                     'sub': 'aleksander',
                     'ses': 'm0',
                     'data_type': 'anat',
@@ -1322,7 +1323,7 @@ def test_pipeline_iteration(self):
         spm_normalization_version = []
 
         for path in sorted(
-                self.capsul.config.local.dataset.input.find(suffix='T1w',
+                self.capsul.config.builtin.dataset.input.find(suffix='T1w',
                                                             extension='nii')):
             input_metadata \
                 = execution_context.dataset['input'].schema.metadata(path)
diff --git a/capsul/test/test_tiny_morphologist.py b/capsul/test/test_tiny_morphologist.py
index 6fe04e401..b151ef7e8 100644
--- a/capsul/test/test_tiny_morphologist.py
+++ b/capsul/test/test_tiny_morphologist.py
@@ -10,9 +10,10 @@
 from soma.controller import Directory, undefined
 
 from capsul.api import Capsul, Process, Pipeline
-from capsul.config.configuration import ModuleConfiguration
+from capsul.config.configuration import ModuleConfiguration, default_workers_type, default_database_url
 from capsul.dataset import ProcessMetadata, ProcessSchema
 
+
 class FakeSPMConfiguration(ModuleConfiguration):
     ''' SPM configuration module
     '''
@@ -190,6 +191,7 @@ class TinyMorphologistBIDS(ProcessSchema, schema='bids', process=TinyMorphologis
     _ = {
         '*': {'process': 'tinymorphologist'}
     }
+    input = {'process': None}
     left_hemisphere = {'part': 'left_hemi'}
     right_hemisphere = {'part': 'right_hemi'}
 
@@ -197,6 +199,7 @@ class TinyMorphologistBrainVISA(ProcessSchema, schema='brainvisa', process=TinyM
     _ = {
         '*': {'process': 'tinymorphologist'}
     }
+    input = {'process': None}
     left_hemisphere = {'prefix': 'left_hemi'}
     right_hemisphere = {'prefix': 'right_hemi'}
 
@@ -292,7 +295,7 @@ def setUp(self):
 
         # Configuration base dictionary
         config = {
-            'local': {
+            'builtin': {
                 'config_modules': [
                     'capsul.test.test_tiny_morphologist',
                 ],
@@ -321,7 +324,7 @@ def setUp(self):
                 'directory': str(fakespm),
                 'version': version,
             }
-            config['local'].setdefault('fakespm', {})[f'fakespm_{version}'] = fakespm_config
+            config['builtin'].setdefault('fakespm', {})[f'fakespm_{version}'] = fakespm_config
             
 
         # Create a configuration file
@@ -341,8 +344,9 @@ def tearDown(self):
     def test_tiny_morphologist_config(self):
         self.maxDiff = 2000
         expected_config = {
-            'local': {
-                'engine_type': 'builtin',
+            'builtin': {
+                'workers_type': default_workers_type,
+                'database_url': default_database_url,
                 'dataset': {
                     'input': {
                         'path': str(self.tmp / 'bids'),
@@ -413,7 +417,7 @@ def test_tiny_morphologist_config(self):
         dict_context = context.asdict()
         self.assertEqual(dict_context, expected_context)
 
-    def test_path_generation(self):
+    def test_tiny_path_generation(self):
         expected = {
             'none': {
                 'template': '!{fakespm.directory}/template',
@@ -474,7 +478,7 @@ def test_path_generation(self):
                 metadata.bids.asdict(),
                 {
                     'folder': 'rawdata',
-                    'pipeline': None, 
+                    'process': None, 
                     'sub': 'aleksander', 
                     'ses': 'm0', 
                     'data_type': 'anat', 
@@ -498,15 +502,9 @@ def test_path_generation(self):
             #     value = getattr(tiny_morphologist, field.name, undefined)
             #     print(f'!{normalization}!', field.name, value)
 
-            # run it
-            # Note: to run via soma-workflow, just set this:
-            # self.capsul.config.local.engine_type = 'soma_workflow'
             with self.capsul.engine() as engine:
-                status = engine.run(tiny_morphologist)
-            self.assertEqual(
-                status,
-                {'status': 'ended', 'error': None, 'error_detail': None,
-                 'engine_output': ''})
+                status = engine.run(tiny_morphologist, timeout=5)
+                self.assertEqual(status, 'ended')
 
 
     def test_pipeline_iteration(self):
@@ -732,7 +730,7 @@ class TinyMorphologistIterationBrainVISA(ProcessSchema, schema='brainvisa', proc
         # BIDS fields (sub, ses, acq, etc.)
         inputs = []
         normalizations = []
-        for path in sorted(self.capsul.config.local.dataset.input.find(suffix='T1w', extension='nii')):
+        for path in sorted(self.capsul.config.builtin.dataset.input.find(suffix='T1w', extension='nii')):
             input_metadata = execution_context.dataset['input'].schema.metadata(path)
             inputs.extend([input_metadata]*3)
             normalizations += ['none', 'aims', 'fakespm8']
@@ -755,7 +753,7 @@ class TinyMorphologistIterationBrainVISA(ProcessSchema, schema='brainvisa', proc
 
         # run it
         # Note: to run via soma-workflow, just set this:
-        # self.capsul.config.local.engine_type = 'soma_workflow'
+        # self.capsul.config.builtin.engine_type = 'soma_workflow'
 
         #status = None
         #try:
@@ -770,21 +768,21 @@ class TinyMorphologistIterationBrainVISA(ProcessSchema, schema='brainvisa', proc
             #{'status': 'ended', 'error': None, 'error_detail': None,
               #'engine_output': ''})
 
-def test():
-    suite = unittest.TestLoader().loadTestsFromTestCase(TestTinyMorphologist)
-    runtime = unittest.TextTestRunner(verbosity=2).run(suite)
-    return runtime.wasSuccessful()
 
 if __name__ == '__main__':
-    import sys
-    sys.stdout.flush()
-    from soma.qt_gui.qt_backend import QtGui
-    from capsul.qt_gui.widgets import PipelineDeveloperView
-    tiny_morphologist = Capsul.executable('capsul.test.test_tiny_morphologist.TinyMorphologist')
-    app = QtGui.QApplication.instance()
-    if not app:
-        app = QtGui.QApplication(sys.argv)
-    view1 = PipelineDeveloperView(tiny_morphologist, show_sub_pipelines=True, allow_open_controller=True, enable_edition=True)
-    view1.show()
-    app.exec_()
-    del view1
+    # import sys
+    # sys.stdout.flush()
+    # from soma.qt_gui.qt_backend import QtGui
+    # from capsul.qt_gui.widgets import PipelineDeveloperView
+    # tiny_morphologist = Capsul.executable('capsul.test.test_tiny_morphologist.TinyMorphologist')
+    # app = QtGui.QApplication.instance()
+    # if not app:
+    #     app = QtGui.QApplication(sys.argv)
+    # view1 = PipelineDeveloperView(tiny_morphologist, show_sub_pipelines=True, allow_open_controller=True, enable_edition=True)
+    # view1.show()
+    # app.exec_()
+    # del view1
+    t = TestTinyMorphologist()
+    t.setUp()
+    t.test_tiny_path_generation()
+    t.tearDown()