[WIP,ENH] Conditional Nodes and Workflows

CHANGES

@@ -1,6 +1,7 @@
 Next release
 ============
 
+* ENH: Introduced runtime decisions (https://github.com/nipy/nipype/pull/1299)
 * ENH: Provides a Nipype wrapper for ANTs DenoiseImage (https://github.com/nipy/nipype/pull/1291)
 * FIX: Minor bugfix logging hash differences (https://github.com/nipy/nipype/pull/1298)
 * FIX: Use released Prov python library (https://github.com/nipy/nipype/pull/1279)

doc/users/index.rst

@@ -33,6 +33,7 @@
    function_interface
    mapnode_and_iterables
    joinnode_and_itersource
+   runtime_decisions
    model_specification
    saving_workflows
    spmmcr

doc/users/runtime_decisions.rst

@@ -0,0 +1,120 @@
+.. runtime_decisions:
+
+===========================
+Runtime decisions in nipype
+===========================
+
+Adding conditional execution (https://github.com/nipy/nipype/issues/878)
+other runtime decisions (https://github.com/nipy/nipype/issues/819) in
+nipype is an old request. Here we introduce some logic and signalling into
+the workflows.
+
+Disable signal in nodes
+=======================
+
+The :class:`nipype.pipeline.engine.Node` now provides a `signals` attribute
+with a `disable` signal by default.
+When the `run()` member of a node is called, the interface will run
+normally *iff* `disable` is `False` (default case).
+
+Example
+-------
+
+For instance, the following code will run the BET interface from fsl:
+
+    >>> from nipype.pipeline.engine import Node
+    >>> from nipype.interfaces import fsl
+    >>> bet = Node(fsl.BET(), 'BET')
+    >>> bet.inputs.in_file = 'T1.nii'
+    >>> bet.run() # doctest: +SKIP
+
+However, if we set the disable signal, then the interface is not run.
+
+    >>> bet.signals.disable = True
+    >>> bet.run() is None
+    True
+
+Disable signal in Workflow
+==========================
+
+:class:`nipype.pipeline.engine.Workflow` also provides signals, including
+`disable` by default.
+It is also allowed to connect the output of a node to a signal in a workflow,
+using the `signalnode.<name-of-signal>` port.
+
+
+Example
+-------
+
+    >>> from nipype.pipeline import engine as pe
+    >>> from nipype.interfaces import utility as niu
+    >>> def _myfunc(val):
+    ...     return val + 1
+    >>> wf = pe.Workflow('TestDisableWorkflow')
+    >>> inputnode = pe.Node(niu.IdentityInterface(
+    ...                     fields=['in_value']), 'inputnode')
+    >>> outputnode = pe.Node(niu.IdentityInterface(
+    ...                      fields=['out_value']), 'outputnode')
+    >>> func = pe.Node(niu.Function(
+    ...     input_names=['val'], output_names=['out'],
+    ...     function=_myfunc), 'functionnode')
+    >>> wf.connect([
+    ...     (inputnode, func, [('in_value', 'val')]),
+    ...     (ifset, outputnode, [('out', 'out_value')])
+    ... ])
+    >>> wf.inputs.inputnode.in_value = 0
+    >>> wf.run() # Will produce 1 in outputnode.out_value
+
+The workflow can be disabled:
+
+    >>> wf.signals.disabled = True
+    >>> wf.run() # The outputnode.out_value remains <undefined>
+
+
+CachedWorkflow
+==============
+
+The :class:`nipype.pipeline.engine.CachedWorkflow` is a type of workflow
+that implements a conditional workflow that is executed *iff* the set of
+cached inputs is not set.
+More precisely, this workflow is able to decide whether its nodes should
+be executed or not if all the inputs of the input node called `cachenode`
+are set.
+For instance, in https://github.com/nipy/nipype/pull/1081 this feature
+is requested.
+The implementation makes use of :class:`nipype.interfaces.utility.CheckInterface`
+which produces an output `out` set to `True` if any/all the inputs are defined
+and `False` otherwise.
+The input `operation` allows to switch between the any and all conditions.
+
+
+Example
+-------
+
+    >>> from nipype.pipeline import engine as pe
+    >>> from nipype.interfaces import utility as niu
+    >>> def _myfunc(a, b):
+    ...     return a + b
+    >>> wf = pe.CachedWorkflow('InnerWorkflow',
+    ...                        cache_map=('c', 'out'))
+    >>> inputnode = pe.Node(niu.IdentityInterface(
+    ...                     fields=['a', 'b']), 'inputnode')
+    >>> func = pe.Node(niu.Function(
+    ...     input_names=['a', 'b'], output_names=['out'],
+    ...     function=_myfunc), 'functionnode')
+    >>> wf.connect([
+    ...     (inputnode, func, [('a', 'a'), ('b', 'b')]),
+    ...     (func, 'output', [('out', 'out')])
+    ... ])
+    >>> wf.inputs.inputnode.a = 2
+    >>> wf.inputs.inputnode.b = 3
+    >>> wf.run() # Will generate 5 in outputnode.out
+
+Please note that the output node should be referred to as 'output' in
+the *connect()* call.
+
+If we set all the inputs of the cache, then the workflow is skipped and
+the output is mapped from the cache:
+
+    >>> wf.inputs.cachenode.c = 7
+    >>> wf.run() # Will produce 7 in outputnode.out

nipype/__init__.py

@@ -83,6 +83,6 @@ def _test_local_install():
     pass
 
 
-from .pipeline import Node, MapNode, JoinNode, Workflow
+from .pipeline import engine
 from .interfaces import (DataGrabber, DataSink, SelectFiles,
                          IdentityInterface, Rename, Function, Select, Merge)

nipype/caching/memory.py

@@ -21,7 +21,7 @@
 
 from ..interfaces.base import BaseInterface
 from ..pipeline.engine import Node
-from ..pipeline.utils import modify_paths
+from ..pipeline.engine.utils import modify_paths
 
 ################################################################################
 # PipeFunc object: callable interface to nipype.interface objects

nipype/caching/tests/test_memory.py

@@ -7,7 +7,7 @@
 from nose.tools import assert_equal
 
 from .. import Memory
-from ...pipeline.tests.test_engine import TestInterface
+from ...pipeline.engine.tests.test_engine import TestInterface
 
 from ... import config
 config.set_default_config()

nipype/interfaces/base.py

@@ -40,9 +40,8 @@
                                TraitListObject, TraitError,
                                isdefined, File, Directory,
                                has_metadata)
-from ..utils.filemanip import (md5, hash_infile, FileNotFoundError,
-                               hash_timestamp, save_json,
-                               split_filename)
+from ..utils.filemanip import md5, \
+    FileNotFoundError, save_json, split_filename
 from ..utils.misc import is_container, trim, str2bool
 from ..utils.provenance import write_provenance
 from .. import config, logging, LooseVersion
@@ -463,6 +462,7 @@ def _deprecated_warn(self, obj, name, old, new):
 
     def _hash_infile(self, adict, key):
         """ Inject file hashes into adict[key]"""
+        from nipype.utils.filemanip import hash_infile, hash_timestamp
         stuff = adict[key]
         if not is_container(stuff):
             stuff = [stuff]
@@ -578,6 +578,7 @@ def get_hashval(self, hash_method=None):
 
     def _get_sorteddict(self, object, dictwithhash=False, hash_method=None,
                         hash_files=True):
+        from nipype.utils.filemanip import hash_infile, hash_timestamp
         if isinstance(object, dict):
             out = []
             for key, val in sorted(object.items()):

nipype/interfaces/io.py

@@ -920,7 +920,8 @@ class SelectFiles(IOBase):
     --------
 
     >>> import pprint
-    >>> from nipype import SelectFiles, Node
+    >>> from nipype.pipeline.engine import Node
+    >>> from nipype.interfaces.io import SelectFiles
     >>> templates={"T1": "{subject_id}/struct/T1.nii",
     ...            "epi": "{subject_id}/func/f[0, 1].nii"}
     >>> dg = Node(SelectFiles(templates), "selectfiles")

nipype/interfaces/utility.py

@@ -566,3 +566,84 @@ def _list_outputs(self):
                 entry = self._parse_line(line)
                 outputs = self._append_entry(outputs, entry)
         return outputs
+
+
+class CheckInterfaceOutputSpec(TraitedSpec):
+    out = traits.Bool(False, desc='Inputs meet condition')
+
+
+class CheckInterface(IOBase):
+    """
+    Interface that performs checks on inputs
+
+    Examples
+    --------
+
+    >>> from nipype.interfaces.utility import CheckInterface
+    >>> checkif = CheckInterface(fields=['a', 'b'], operation='any')
+    >>> checkif._list_outputs()['out']
+    False
+
+    >>> checkif.inputs.a = 'foo'
+    >>> out = checkif.run()
+    >>> checkif._list_outputs()['out']
+    True
+
+    >>> checkif.inputs.operation = 'all'
+    >>> out = checkif.run()
+    >>> checkif._list_outputs()['out']
+    False
+
+    >>> checkif.inputs.b = 'bar'
+    >>> out = checkif.run()
+    >>> checkif._list_outputs()['out']
+    True
+    """
+    input_spec = DynamicTraitedSpec
+    output_spec = CheckInterfaceOutputSpec
+    _always_run = True
+
+    def __init__(self, fields=None, operation='all', **inputs):
+        super(CheckInterface, self).__init__(**inputs)
+
+        if fields is None or not fields:
+            raise ValueError('CheckInterface fields must be a non-empty '
+                             'list')
+
+        if 'operation' in fields:
+            raise ValueError('CheckInterface does not allow fields using'
+                             ' special name \'operation\'')
+        # Each input must be in the fields.
+        for in_field in inputs:
+            if in_field not in fields:
+                raise ValueError('CheckInterface input is not in the '
+                                 'fields: %s' % in_field)
+        self._fields = fields
+        add_traits(self.inputs, fields + ['operation'])
+
+        # Adding any traits wipes out all input values set in superclass initialization,
+        # even it the trait is not in the add_traits argument. The work-around is to reset
+        # the values after adding the traits.
+        self.inputs.set(**inputs)
+
+        if operation not in ['all', 'any']:
+            raise ValueError('CheckInterface does not accept key word '
+                             '\'%s\' as operation input' % operation)
+        self.inputs.operation = operation
+
+    def _check_result(self):
+        if self.inputs.operation not in ['all', 'any']:
+            raise ValueError('CheckInterface does not accept keyword '
+                             '\'%s\' as operation input' % operation)
+
+        results = [isdefined(getattr(self.inputs, key))
+                   for key in self._fields if key != 'operation']
+
+        if self.inputs.operation == 'any':
+            return any(results)
+        return all(results)
+
+    def _list_outputs(self):
+        outputs = self._outputs().get()
+        outputs['out'] = self._check_result()
+        return outputs

nipype/pipeline/__init__.py

@@ -7,4 +7,5 @@
 
 from __future__ import absolute_import
 __docformat__ = 'restructuredtext'
-from .engine import Node, MapNode, JoinNode, Workflow
+
+from .engine import *

nipype/pipeline/engine/__init__.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
+# vi: set ft=python sts=4 ts=4 sw=4 et:
+
+from .workflows import *
+from .nodes import *