Merge pull request #388 from djarecka/fix/call

fixing __call__ function

Author: satra

pydra/engine/core.py

@@ -392,26 +392,28 @@ def output_dir(self):
             return [self._cache_dir / checksum for checksum in self.checksum_states()]
         return self._cache_dir / self.checksum
 
-    def __call__(self, submitter=None, plugin=None, rerun=False, **kwargs):
+    def __call__(
+        self, submitter=None, plugin=None, plugin_kwargs=None, rerun=False, **kwargs
+    ):
         """Make tasks callable themselves."""
         from .submitter import Submitter
 
         if submitter and plugin:
             raise Exception("Specify submitter OR plugin, not both")
-        plugin = plugin or self.plugin
-        if plugin:
-            submitter = Submitter(plugin=plugin)
-        elif self.state:
-            submitter = Submitter()
+        elif submitter:
+            pass
+        # if there is plugin provided or the task is a Workflow or has a state,
+        # the submitter will be created using provided plugin, self.plugin or "cf"
+        elif plugin or self.state or is_workflow(self):
+            plugin = plugin or self.plugin or "cf"
+            if plugin_kwargs is None:
+                plugin_kwargs = {}
+            submitter = Submitter(plugin=plugin, **plugin_kwargs)
 
         if submitter:
             with submitter as sub:
                 res = sub(self)
-        else:
-            if is_workflow(self):
-                raise NotImplementedError(
-                    "TODO: linear workflow execution - assign submitter or plugin for now"
-                )
+        else:  # tasks without state could be run without a submitter
             res = self._run(rerun=rerun, **kwargs)
         return res
 

pydra/engine/tests/test_numpy_examples.py

@@ -17,15 +17,15 @@ def arrayout(val):
     return np.array([val, val])
 
 
-def test_multiout(plugin, tmpdir):
+def test_multiout(tmpdir):
     """ testing a simple function that returns a numpy array"""
     wf = Workflow("wf", input_spec=["val"], val=2)
     wf.add(arrayout(name="mo", val=wf.lzin.val))
 
     wf.set_output([("array", wf.mo.lzout.b)])
     wf.cache_dir = tmpdir
 
-    with Submitter(plugin=plugin, n_procs=2) as sub:
+    with Submitter(plugin="cf", n_procs=2) as sub:
         sub(runnable=wf)
 
     results = wf.result(return_inputs=True)
@@ -34,7 +34,7 @@ def test_multiout(plugin, tmpdir):
     assert np.array_equal(results[1].output.array, np.array([2, 2]))
 
 
-def test_multiout_st(plugin, tmpdir):
+def test_multiout_st(tmpdir):
     """ testing a simple function that returns a numpy array, adding splitter"""
     wf = Workflow("wf", input_spec=["val"], val=[0, 1, 2])
     wf.add(arrayout(name="mo", val=wf.lzin.val))
@@ -43,7 +43,7 @@ def test_multiout_st(plugin, tmpdir):
     wf.set_output([("array", wf.mo.lzout.b)])
     wf.cache_dir = tmpdir
 
-    with Submitter(plugin=plugin, n_procs=2) as sub:
+    with Submitter(plugin="cf", n_procs=2) as sub:
         sub(runnable=wf)
 
     results = wf.result(return_inputs=True)

pydra/engine/tests/test_submitter.py

@@ -23,17 +23,30 @@ def sleep_add_one(x):
 
 def test_callable_wf(plugin, tmpdir):
     wf = gen_basic_wf()
+    res = wf()
+    assert res.output.out == 9
+    del wf, res
 
-    with pytest.raises(NotImplementedError):
-        wf()
-
+    # providing plugin
+    wf = gen_basic_wf()
     res = wf(plugin="cf")
     assert res.output.out == 9
     del wf, res
 
+    # providing plugin_kwargs
     wf = gen_basic_wf()
-    wf.cache_dir = tmpdir
+    res = wf(plugin="cf", plugin_kwargs={"n_procs": 2})
+    assert res.output.out == 9
+    del wf, res
+
+    # providing wrong plugin_kwargs
+    wf = gen_basic_wf()
+    with pytest.raises(TypeError, match="an unexpected keyword argument"):
+        wf(plugin="cf", plugin_kwargs={"sbatch_args": "-N2"})
 
+    # providing submitter
+    wf = gen_basic_wf()
+    wf.cache_dir = tmpdir
     sub = Submitter(plugin)
     res = wf(submitter=sub)
     assert res.output.out == 9