We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Shim warning when using IPython >= 4.0.0
One of our C-PAC users came across the following deprecation warning during their pipeline run:

> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/kernel/**init**.py:13: ShimWarning: The `IPython.kernel` package has been deprecated. You should import from ipykernel or jupyter_client instead.
> 
>   "You should import from ipykernel or jupyter_client instead.", ShimWarning)
> 
> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/parallel.py:13: ShimWarning: The `IPython.parallel` package has been deprecated. You should import from ipyparallel instead.
> 
>   "You should import from ipyparallel instead.", ShimWarning)

I was able to determine  that this is because, starting in IPython 4.0.0, the IPython developers have chosen to separate IPython.parallel and IPython.kernel into separate packages.  Is there some way that nipype could take this into account by determining if the IPython version number is >= 4.0.0, stopping execution if so (and the packages are not installed), and warning users to download these two packages (or some other appropriate behavior)?  If the packages are installed, the new import statements could be used instead.

Shim warning when using IPython >= 4.0.0
One of our C-PAC users came across the following deprecation warning during their pipeline run:

> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/kernel/**init**.py:13: ShimWarning: The `IPython.kernel` package has been deprecated. You should import from ipykernel or jupyter_client instead.
> 
>   "You should import from ipykernel or jupyter_client instead.", ShimWarning)
> 
> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/parallel.py:13: ShimWarning: The `IPython.parallel` package has been deprecated. You should import from ipyparallel instead.
> 
>   "You should import from ipyparallel instead.", ShimWarning)

I was able to determine  that this is because, starting in IPython 4.0.0, the IPython developers have chosen to separate IPython.parallel and IPython.kernel into separate packages.  Is there some way that nipype could take this into account by determining if the IPython version number is >= 4.0.0, stopping execution if so (and the packages are not installed), and warning users to download these two packages (or some other appropriate behavior)?  If the packages are installed, the new import statements could be used instead.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nipype/pipeline/plugins/ipython.py`
Content:
```
1 # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
2 # vi: set ft=python sts=4 ts=4 sw=4 et:
3 """Parallel workflow execution via IPython controller
4 """
5 
6 from future import standard_library
7 standard_library.install_aliases()
8 
9 from pickle import dumps
10 
11 import sys
12 
13 IPython_not_loaded = False
14 try:
15     from IPython import __version__ as IPyversion
16     from ipyparallel.error import TimeoutError
17 except:
18     IPython_not_loaded = True
19 
20 from .base import (DistributedPluginBase, logger, report_crash)
21 
22 
23 def execute_task(pckld_task, node_config, updatehash):
24     from socket import gethostname
25     from traceback import format_exc
26     from nipype import config, logging
27     traceback = None
28     result = None
29     import os
30     cwd = os.getcwd()
31     try:
32         config.update_config(node_config)
33         logging.update_logging(config)
34         from pickle import loads
35         task = loads(pckld_task)
36         result = task.run(updatehash=updatehash)
37     except:
38         traceback = format_exc()
39         result = task.result
40     os.chdir(cwd)
41     return result, traceback, gethostname()
42 
43 
44 class IPythonPlugin(DistributedPluginBase):
45     """Execute workflow with ipython
46     """
47 
48     def __init__(self, plugin_args=None):
49         if IPython_not_loaded:
50             raise ImportError('ipyparallel could not be imported')
51         super(IPythonPlugin, self).__init__(plugin_args=plugin_args)
52         self.iparallel = None
53         self.taskclient = None
54         self.taskmap = {}
55         self._taskid = 0
56 
57     def run(self, graph, config, updatehash=False):
58         """Executes a pre-defined pipeline is distributed approaches
59         based on IPython's ipyparallel processing interface
60         """
61         # retrieve clients again
62         try:
63             name = 'ipyparallel'
64             __import__(name)
65             self.iparallel = sys.modules[name]
66         except ImportError:
67             raise ImportError("Ipython kernel not found. Parallel execution "
68                               "will be unavailable")
69         try:
70             self.taskclient = self.iparallel.Client()
71         except Exception as e:
72             if isinstance(e, TimeoutError):
73                 raise Exception("No IPython clients found.")
74             if isinstance(e, IOError):
75                 raise Exception("ipcluster/ipcontroller has not been started")
76             if isinstance(e, ValueError):
77                 raise Exception("Ipython kernel not installed")
78             raise e
79         return super(IPythonPlugin, self).run(graph, config, updatehash=updatehash)
80 
81     def _get_result(self, taskid):
82         if taskid not in self.taskmap:
83             raise ValueError('Task %d not in pending list' % taskid)
84         if self.taskmap[taskid].ready():
85             result, traceback, hostname = self.taskmap[taskid].get()
86             result_out = dict(result=None, traceback=None)
87             result_out['result'] = result
88             result_out['traceback'] = traceback
89             result_out['hostname'] = hostname
90             return result_out
91         else:
92             return None
93 
94     def _submit_job(self, node, updatehash=False):
95         pckld_node = dumps(node, 2)
96         result_object = self.taskclient.load_balanced_view().apply(execute_task,
97                                                                    pckld_node,
98                                                                    node.config,
99                                                                    updatehash)
100         self._taskid += 1
101         self.taskmap[self._taskid] = result_object
102         return self._taskid
103 
104     def _report_crash(self, node, result=None):
105         if result and result['traceback']:
106             node._result = result['result']
107             node._traceback = result['traceback']
108             return report_crash(node,
109                                 traceback=result['traceback'])
110         else:
111             return report_crash(node)
112 
113     def _clear_task(self, taskid):
114         if IPyversion >= '0.11':
115             logger.debug("Clearing id: %d" % taskid)
116             self.taskclient.purge_results(self.taskmap[taskid])
117             del self.taskmap[taskid]
118
```

Path: `nipype/pipeline/plugins/ipythonx.py`
Content:
```
1 # emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
2 # vi: set ft=python sts=4 ts=4 sw=4 et:
3 """Parallel workflow execution via IPython controller
4 """
5 
6 import sys
7 
8 IPython_not_loaded = False
9 try:
10     from IPython import __version__ as IPyversion
11     from IPython.kernel.contexts import ConnectionRefusedError
12 except:
13     IPython_not_loaded = True
14 
15 
16 from .base import (DistributedPluginBase, logger, report_crash)
17 
18 
19 class IPythonXPlugin(DistributedPluginBase):
20     """Execute workflow with ipython
21     """
22 
23     def __init__(self, plugin_args=None):
24         if IPython_not_loaded:
25             raise ImportError('ipyparallel could not be imported')
26         super(IPythonXPlugin, self).__init__(plugin_args=plugin_args)
27         self.ipyclient = None
28         self.taskclient = None
29 
30     def run(self, graph, config, updatehash=False):
31         """Executes a pre-defined pipeline is distributed approaches
32         based on IPython's ipyparallel processing interface
33         """
34         # retrieve clients again
35         try:
36             name = 'IPython.kernel.client'
37             __import__(name)
38             self.ipyclient = sys.modules[name]
39         except ImportError:
40             raise ImportError("Ipython kernel not found. Parallel execution "
41                               "will be unavailable")
42         try:
43             self.taskclient = self.ipyclient.TaskClient()
44         except Exception as e:
45             if isinstance(e, ConnectionRefusedError):
46                 raise Exception("No IPython clients found.")
47             if isinstance(e, ValueError):
48                 raise Exception("Ipython kernel not installed")
49         return super(IPythonXPlugin, self).run(graph, config, updatehash=updatehash)
50 
51     def _get_result(self, taskid):
52         return self.taskclient.get_task_result(taskid, block=False)
53 
54     def _submit_job(self, node, updatehash=False):
55         cmdstr = """import sys
56 from traceback import format_exception
57 traceback=None
58 result=None
59 try:
60     result = task.run(updatehash=updatehash)
61 except:
62     etype, eval, etr = sys.exc_info()
63     traceback = format_exception(etype,eval,etr)
64     result = task.result
65 """
66         task = self.ipyclient.StringTask(cmdstr,
67                                          push=dict(task=node,
68                                                    updatehash=updatehash),
69                                          pull=['result', 'traceback'])
70         return self.taskclient.run(task, block=False)
71 
72     def _report_crash(self, node, result=None):
73         if result and result['traceback']:
74             node._result = result['result']
75             node._traceback = result['traceback']
76             return report_crash(node,
77                                 traceback=result['traceback'])
78         else:
79             return report_crash(node)
80 
81     def _clear_task(self, taskid):
82         if IPyversion >= '0.10.1':
83             logger.debug("Clearing id: %d" % taskid)
84             self.taskclient.clear(taskid)
85
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nipype/pipeline/plugins/ipython.py b/nipype/pipeline/plugins/ipython.py
--- a/nipype/pipeline/plugins/ipython.py
+++ b/nipype/pipeline/plugins/ipython.py
@@ -5,7 +5,6 @@
 
 from future import standard_library
 standard_library.install_aliases()
-
 from pickle import dumps
 
 import sys
@@ -47,7 +46,7 @@
 
     def __init__(self, plugin_args=None):
         if IPython_not_loaded:
-            raise ImportError('ipyparallel could not be imported')
+            raise ImportError('Please install ipyparallel to use this plugin.')
         super(IPythonPlugin, self).__init__(plugin_args=plugin_args)
         self.iparallel = None
         self.taskclient = None
@@ -64,7 +63,7 @@
             __import__(name)
             self.iparallel = sys.modules[name]
         except ImportError:
-            raise ImportError("Ipython kernel not found. Parallel execution "
+            raise ImportError("ipyparallel not found. Parallel execution "
                               "will be unavailable")
         try:
             self.taskclient = self.iparallel.Client()
diff --git a/nipype/pipeline/plugins/ipythonx.py b/nipype/pipeline/plugins/ipythonx.py
--- a/nipype/pipeline/plugins/ipythonx.py
+++ b/nipype/pipeline/plugins/ipythonx.py
@@ -5,11 +5,13 @@
 
 import sys
 
+from ...interfaces.base import LooseVersion
 IPython_not_loaded = False
 try:
     from IPython import __version__ as IPyversion
-    from IPython.kernel.contexts import ConnectionRefusedError
-except:
+    if LooseVersion(IPyversion) < LooseVersion('0.11'):
+        from IPython.kernel.contexts import ConnectionRefusedError
+except ImportError:
     IPython_not_loaded = True
 
 
@@ -21,6 +23,12 @@
     """
 
     def __init__(self, plugin_args=None):
+        if LooseVersion(IPyversion) > LooseVersion('0.10.1'):
+            raise EnvironmentError(('The IPythonX plugin can only be used with'
+                                    ' older IPython versions. Please use the '
+                                    'IPython plugin instead.'
+                                    ))
+        DeprecationWarning('This plugin will be deprecated as of version 0.13')
         if IPython_not_loaded:
             raise ImportError('ipyparallel could not be imported')
         super(IPythonXPlugin, self).__init__(plugin_args=plugin_args)

{"golden_diff": "diff --git a/nipype/pipeline/plugins/ipython.py b/nipype/pipeline/plugins/ipython.py\n--- a/nipype/pipeline/plugins/ipython.py\n+++ b/nipype/pipeline/plugins/ipython.py\n@@ -5,7 +5,6 @@\n \n from future import standard_library\n standard_library.install_aliases()\n-\n from pickle import dumps\n \n import sys\n@@ -47,7 +46,7 @@\n \n     def __init__(self, plugin_args=None):\n         if IPython_not_loaded:\n-            raise ImportError('ipyparallel could not be imported')\n+            raise ImportError('Please install ipyparallel to use this plugin.')\n         super(IPythonPlugin, self).__init__(plugin_args=plugin_args)\n         self.iparallel = None\n         self.taskclient = None\n@@ -64,7 +63,7 @@\n             __import__(name)\n             self.iparallel = sys.modules[name]\n         except ImportError:\n-            raise ImportError(\"Ipython kernel not found. Parallel execution \"\n+            raise ImportError(\"ipyparallel not found. Parallel execution \"\n                               \"will be unavailable\")\n         try:\n             self.taskclient = self.iparallel.Client()\ndiff --git a/nipype/pipeline/plugins/ipythonx.py b/nipype/pipeline/plugins/ipythonx.py\n--- a/nipype/pipeline/plugins/ipythonx.py\n+++ b/nipype/pipeline/plugins/ipythonx.py\n@@ -5,11 +5,13 @@\n \n import sys\n \n+from ...interfaces.base import LooseVersion\n IPython_not_loaded = False\n try:\n     from IPython import __version__ as IPyversion\n-    from IPython.kernel.contexts import ConnectionRefusedError\n-except:\n+    if LooseVersion(IPyversion) < LooseVersion('0.11'):\n+        from IPython.kernel.contexts import ConnectionRefusedError\n+except ImportError:\n     IPython_not_loaded = True\n \n \n@@ -21,6 +23,12 @@\n     \"\"\"\n \n     def __init__(self, plugin_args=None):\n+        if LooseVersion(IPyversion) > LooseVersion('0.10.1'):\n+            raise EnvironmentError(('The IPythonX plugin can only be used with'\n+                                    ' older IPython versions. Please use the '\n+                                    'IPython plugin instead.'\n+                                    ))\n+        DeprecationWarning('This plugin will be deprecated as of version 0.13')\n         if IPython_not_loaded:\n             raise ImportError('ipyparallel could not be imported')\n         super(IPythonXPlugin, self).__init__(plugin_args=plugin_args)\n", "issue": "Shim warning when using IPython >= 4.0.0\nOne of our C-PAC users came across the following deprecation warning during their pipeline run:\n\n> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/kernel/**init**.py:13: ShimWarning: The `IPython.kernel` package has been deprecated. You should import from ipykernel or jupyter_client instead.\n> \n>   \"You should import from ipykernel or jupyter_client instead.\", ShimWarning)\n> \n> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/parallel.py:13: ShimWarning: The `IPython.parallel` package has been deprecated. You should import from ipyparallel instead.\n> \n>   \"You should import from ipyparallel instead.\", ShimWarning)\n\nI was able to determine  that this is because, starting in IPython 4.0.0, the IPython developers have chosen to separate IPython.parallel and IPython.kernel into separate packages.  Is there some way that nipype could take this into account by determining if the IPython version number is >= 4.0.0, stopping execution if so (and the packages are not installed), and warning users to download these two packages (or some other appropriate behavior)?  If the packages are installed, the new import statements could be used instead.\n\nShim warning when using IPython >= 4.0.0\nOne of our C-PAC users came across the following deprecation warning during their pipeline run:\n\n> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/kernel/**init**.py:13: ShimWarning: The `IPython.kernel` package has been deprecated. You should import from ipykernel or jupyter_client instead.\n> \n>   \"You should import from ipykernel or jupyter_client instead.\", ShimWarning)\n> \n> /clusterdata/apps/anaconda/lib/python2.7/site-packages/IPython/parallel.py:13: ShimWarning: The `IPython.parallel` package has been deprecated. You should import from ipyparallel instead.\n> \n>   \"You should import from ipyparallel instead.\", ShimWarning)\n\nI was able to determine  that this is because, starting in IPython 4.0.0, the IPython developers have chosen to separate IPython.parallel and IPython.kernel into separate packages.  Is there some way that nipype could take this into account by determining if the IPython version number is >= 4.0.0, stopping execution if so (and the packages are not installed), and warning users to download these two packages (or some other appropriate behavior)?  If the packages are installed, the new import statements could be used instead.\n\n", "before_files": [{"content": "# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-\n# vi: set ft=python sts=4 ts=4 sw=4 et:\n\"\"\"Parallel workflow execution via IPython controller\n\"\"\"\n\nfrom future import standard_library\nstandard_library.install_aliases()\n\nfrom pickle import dumps\n\nimport sys\n\nIPython_not_loaded = False\ntry:\n    from IPython import __version__ as IPyversion\n    from ipyparallel.error import TimeoutError\nexcept:\n    IPython_not_loaded = True\n\nfrom .base import (DistributedPluginBase, logger, report_crash)\n\n\ndef execute_task(pckld_task, node_config, updatehash):\n    from socket import gethostname\n    from traceback import format_exc\n    from nipype import config, logging\n    traceback = None\n    result = None\n    import os\n    cwd = os.getcwd()\n    try:\n        config.update_config(node_config)\n        logging.update_logging(config)\n        from pickle import loads\n        task = loads(pckld_task)\n        result = task.run(updatehash=updatehash)\n    except:\n        traceback = format_exc()\n        result = task.result\n    os.chdir(cwd)\n    return result, traceback, gethostname()\n\n\nclass IPythonPlugin(DistributedPluginBase):\n    \"\"\"Execute workflow with ipython\n    \"\"\"\n\n    def __init__(self, plugin_args=None):\n        if IPython_not_loaded:\n            raise ImportError('ipyparallel could not be imported')\n        super(IPythonPlugin, self).__init__(plugin_args=plugin_args)\n        self.iparallel = None\n        self.taskclient = None\n        self.taskmap = {}\n        self._taskid = 0\n\n    def run(self, graph, config, updatehash=False):\n        \"\"\"Executes a pre-defined pipeline is distributed approaches\n        based on IPython's ipyparallel processing interface\n        \"\"\"\n        # retrieve clients again\n        try:\n            name = 'ipyparallel'\n            __import__(name)\n            self.iparallel = sys.modules[name]\n        except ImportError:\n            raise ImportError(\"Ipython kernel not found. Parallel execution \"\n                              \"will be unavailable\")\n        try:\n            self.taskclient = self.iparallel.Client()\n        except Exception as e:\n            if isinstance(e, TimeoutError):\n                raise Exception(\"No IPython clients found.\")\n            if isinstance(e, IOError):\n                raise Exception(\"ipcluster/ipcontroller has not been started\")\n            if isinstance(e, ValueError):\n                raise Exception(\"Ipython kernel not installed\")\n            raise e\n        return super(IPythonPlugin, self).run(graph, config, updatehash=updatehash)\n\n    def _get_result(self, taskid):\n        if taskid not in self.taskmap:\n            raise ValueError('Task %d not in pending list' % taskid)\n        if self.taskmap[taskid].ready():\n            result, traceback, hostname = self.taskmap[taskid].get()\n            result_out = dict(result=None, traceback=None)\n            result_out['result'] = result\n            result_out['traceback'] = traceback\n            result_out['hostname'] = hostname\n            return result_out\n        else:\n            return None\n\n    def _submit_job(self, node, updatehash=False):\n        pckld_node = dumps(node, 2)\n        result_object = self.taskclient.load_balanced_view().apply(execute_task,\n                                                                   pckld_node,\n                                                                   node.config,\n                                                                   updatehash)\n        self._taskid += 1\n        self.taskmap[self._taskid] = result_object\n        return self._taskid\n\n    def _report_crash(self, node, result=None):\n        if result and result['traceback']:\n            node._result = result['result']\n            node._traceback = result['traceback']\n            return report_crash(node,\n                                traceback=result['traceback'])\n        else:\n            return report_crash(node)\n\n    def _clear_task(self, taskid):\n        if IPyversion >= '0.11':\n            logger.debug(\"Clearing id: %d\" % taskid)\n            self.taskclient.purge_results(self.taskmap[taskid])\n            del self.taskmap[taskid]\n", "path": "nipype/pipeline/plugins/ipython.py"}, {"content": "# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-\n# vi: set ft=python sts=4 ts=4 sw=4 et:\n\"\"\"Parallel workflow execution via IPython controller\n\"\"\"\n\nimport sys\n\nIPython_not_loaded = False\ntry:\n    from IPython import __version__ as IPyversion\n    from IPython.kernel.contexts import ConnectionRefusedError\nexcept:\n    IPython_not_loaded = True\n\n\nfrom .base import (DistributedPluginBase, logger, report_crash)\n\n\nclass IPythonXPlugin(DistributedPluginBase):\n    \"\"\"Execute workflow with ipython\n    \"\"\"\n\n    def __init__(self, plugin_args=None):\n        if IPython_not_loaded:\n            raise ImportError('ipyparallel could not be imported')\n        super(IPythonXPlugin, self).__init__(plugin_args=plugin_args)\n        self.ipyclient = None\n        self.taskclient = None\n\n    def run(self, graph, config, updatehash=False):\n        \"\"\"Executes a pre-defined pipeline is distributed approaches\n        based on IPython's ipyparallel processing interface\n        \"\"\"\n        # retrieve clients again\n        try:\n            name = 'IPython.kernel.client'\n            __import__(name)\n            self.ipyclient = sys.modules[name]\n        except ImportError:\n            raise ImportError(\"Ipython kernel not found. Parallel execution \"\n                              \"will be unavailable\")\n        try:\n            self.taskclient = self.ipyclient.TaskClient()\n        except Exception as e:\n            if isinstance(e, ConnectionRefusedError):\n                raise Exception(\"No IPython clients found.\")\n            if isinstance(e, ValueError):\n                raise Exception(\"Ipython kernel not installed\")\n        return super(IPythonXPlugin, self).run(graph, config, updatehash=updatehash)\n\n    def _get_result(self, taskid):\n        return self.taskclient.get_task_result(taskid, block=False)\n\n    def _submit_job(self, node, updatehash=False):\n        cmdstr = \"\"\"import sys\nfrom traceback import format_exception\ntraceback=None\nresult=None\ntry:\n    result = task.run(updatehash=updatehash)\nexcept:\n    etype, eval, etr = sys.exc_info()\n    traceback = format_exception(etype,eval,etr)\n    result = task.result\n\"\"\"\n        task = self.ipyclient.StringTask(cmdstr,\n                                         push=dict(task=node,\n                                                   updatehash=updatehash),\n                                         pull=['result', 'traceback'])\n        return self.taskclient.run(task, block=False)\n\n    def _report_crash(self, node, result=None):\n        if result and result['traceback']:\n            node._result = result['result']\n            node._traceback = result['traceback']\n            return report_crash(node,\n                                traceback=result['traceback'])\n        else:\n            return report_crash(node)\n\n    def _clear_task(self, taskid):\n        if IPyversion >= '0.10.1':\n            logger.debug(\"Clearing id: %d\" % taskid)\n            self.taskclient.clear(taskid)\n", "path": "nipype/pipeline/plugins/ipythonx.py"}], "after_files": [{"content": "# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-\n# vi: set ft=python sts=4 ts=4 sw=4 et:\n\"\"\"Parallel workflow execution via IPython controller\n\"\"\"\n\nfrom future import standard_library\nstandard_library.install_aliases()\nfrom pickle import dumps\n\nimport sys\n\nIPython_not_loaded = False\ntry:\n    from IPython import __version__ as IPyversion\n    from ipyparallel.error import TimeoutError\nexcept:\n    IPython_not_loaded = True\n\nfrom .base import (DistributedPluginBase, logger, report_crash)\n\n\ndef execute_task(pckld_task, node_config, updatehash):\n    from socket import gethostname\n    from traceback import format_exc\n    from nipype import config, logging\n    traceback = None\n    result = None\n    import os\n    cwd = os.getcwd()\n    try:\n        config.update_config(node_config)\n        logging.update_logging(config)\n        from pickle import loads\n        task = loads(pckld_task)\n        result = task.run(updatehash=updatehash)\n    except:\n        traceback = format_exc()\n        result = task.result\n    os.chdir(cwd)\n    return result, traceback, gethostname()\n\n\nclass IPythonPlugin(DistributedPluginBase):\n    \"\"\"Execute workflow with ipython\n    \"\"\"\n\n    def __init__(self, plugin_args=None):\n        if IPython_not_loaded:\n            raise ImportError('Please install ipyparallel to use this plugin.')\n        super(IPythonPlugin, self).__init__(plugin_args=plugin_args)\n        self.iparallel = None\n        self.taskclient = None\n        self.taskmap = {}\n        self._taskid = 0\n\n    def run(self, graph, config, updatehash=False):\n        \"\"\"Executes a pre-defined pipeline is distributed approaches\n        based on IPython's ipyparallel processing interface\n        \"\"\"\n        # retrieve clients again\n        try:\n            name = 'ipyparallel'\n            __import__(name)\n            self.iparallel = sys.modules[name]\n        except ImportError:\n            raise ImportError(\"ipyparallel not found. Parallel execution \"\n                              \"will be unavailable\")\n        try:\n            self.taskclient = self.iparallel.Client()\n        except Exception as e:\n            if isinstance(e, TimeoutError):\n                raise Exception(\"No IPython clients found.\")\n            if isinstance(e, IOError):\n                raise Exception(\"ipcluster/ipcontroller has not been started\")\n            if isinstance(e, ValueError):\n                raise Exception(\"Ipython kernel not installed\")\n            raise e\n        return super(IPythonPlugin, self).run(graph, config, updatehash=updatehash)\n\n    def _get_result(self, taskid):\n        if taskid not in self.taskmap:\n            raise ValueError('Task %d not in pending list' % taskid)\n        if self.taskmap[taskid].ready():\n            result, traceback, hostname = self.taskmap[taskid].get()\n            result_out = dict(result=None, traceback=None)\n            result_out['result'] = result\n            result_out['traceback'] = traceback\n            result_out['hostname'] = hostname\n            return result_out\n        else:\n            return None\n\n    def _submit_job(self, node, updatehash=False):\n        pckld_node = dumps(node, 2)\n        result_object = self.taskclient.load_balanced_view().apply(execute_task,\n                                                                   pckld_node,\n                                                                   node.config,\n                                                                   updatehash)\n        self._taskid += 1\n        self.taskmap[self._taskid] = result_object\n        return self._taskid\n\n    def _report_crash(self, node, result=None):\n        if result and result['traceback']:\n            node._result = result['result']\n            node._traceback = result['traceback']\n            return report_crash(node,\n                                traceback=result['traceback'])\n        else:\n            return report_crash(node)\n\n    def _clear_task(self, taskid):\n        if IPyversion >= '0.11':\n            logger.debug(\"Clearing id: %d\" % taskid)\n            self.taskclient.purge_results(self.taskmap[taskid])\n            del self.taskmap[taskid]\n", "path": "nipype/pipeline/plugins/ipython.py"}, {"content": "# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-\n# vi: set ft=python sts=4 ts=4 sw=4 et:\n\"\"\"Parallel workflow execution via IPython controller\n\"\"\"\n\nimport sys\n\nfrom ...interfaces.base import LooseVersion\nIPython_not_loaded = False\ntry:\n    from IPython import __version__ as IPyversion\n    if LooseVersion(IPyversion) < LooseVersion('0.11'):\n        from IPython.kernel.contexts import ConnectionRefusedError\nexcept ImportError:\n    IPython_not_loaded = True\n\n\nfrom .base import (DistributedPluginBase, logger, report_crash)\n\n\nclass IPythonXPlugin(DistributedPluginBase):\n    \"\"\"Execute workflow with ipython\n    \"\"\"\n\n    def __init__(self, plugin_args=None):\n        if LooseVersion(IPyversion) > LooseVersion('0.10.1'):\n            raise EnvironmentError(('The IPythonX plugin can only be used with'\n                                    ' older IPython versions. Please use the '\n                                    'IPython plugin instead.'\n                                    ))\n        DeprecationWarning('This plugin will be deprecated as of version 0.13')\n        if IPython_not_loaded:\n            raise ImportError('ipyparallel could not be imported')\n        super(IPythonXPlugin, self).__init__(plugin_args=plugin_args)\n        self.ipyclient = None\n        self.taskclient = None\n\n    def run(self, graph, config, updatehash=False):\n        \"\"\"Executes a pre-defined pipeline is distributed approaches\n        based on IPython's ipyparallel processing interface\n        \"\"\"\n        # retrieve clients again\n        try:\n            name = 'IPython.kernel.client'\n            __import__(name)\n            self.ipyclient = sys.modules[name]\n        except ImportError:\n            raise ImportError(\"Ipython kernel not found. Parallel execution \"\n                              \"will be unavailable\")\n        try:\n            self.taskclient = self.ipyclient.TaskClient()\n        except Exception as e:\n            if isinstance(e, ConnectionRefusedError):\n                raise Exception(\"No IPython clients found.\")\n            if isinstance(e, ValueError):\n                raise Exception(\"Ipython kernel not installed\")\n        return super(IPythonXPlugin, self).run(graph, config, updatehash=updatehash)\n\n    def _get_result(self, taskid):\n        return self.taskclient.get_task_result(taskid, block=False)\n\n    def _submit_job(self, node, updatehash=False):\n        cmdstr = \"\"\"import sys\nfrom traceback import format_exception\ntraceback=None\nresult=None\ntry:\n    result = task.run(updatehash=updatehash)\nexcept:\n    etype, eval, etr = sys.exc_info()\n    traceback = format_exception(etype,eval,etr)\n    result = task.result\n\"\"\"\n        task = self.ipyclient.StringTask(cmdstr,\n                                         push=dict(task=node,\n                                                   updatehash=updatehash),\n                                         pull=['result', 'traceback'])\n        return self.taskclient.run(task, block=False)\n\n    def _report_crash(self, node, result=None):\n        if result and result['traceback']:\n            node._result = result['result']\n            node._traceback = result['traceback']\n            return report_crash(node,\n                                traceback=result['traceback'])\n        else:\n            return report_crash(node)\n\n    def _clear_task(self, taskid):\n        if IPyversion >= '0.10.1':\n            logger.debug(\"Clearing id: %d\" % taskid)\n            self.taskclient.clear(taskid)\n", "path": "nipype/pipeline/plugins/ipythonx.py"}]}