aesthetics pass on AsyncResult.display_outputs

adds a few delimiters when there are rich outputs, and cleans up a few cases of extra trailing whitespace on stdout/err

IPython/parallel/client/asyncresult.py

@@ -15,13 +15,15 b' Authors:'
 # Imports
 #-----------------------------------------------------------------------------
 
+from __future__ import print_function
+
 import sys
 import time
 from datetime import datetime
 
 from zmq import MessageTracker
 
-from IPython.core.display import clear_output, display
+from IPython.core.display import clear_output, display, display_pretty
 from IPython.external.decorator import decorator
 from IPython.parallel import error
 
@@ -38,6 +40,9 b' def _total_seconds(td):'
         # Python 2.6
         return 1e-6 * (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10**6)
 
+def _raw_text(s):
+    display_pretty(s, raw=True)
+
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
@@ -373,10 +378,10 b' class AsyncResult(object):'
         while not self.ready() and (timeout is None or time.time() - tic <= timeout):
             self.wait(interval)
             clear_output()
-            print "%4i/%i tasks finished after %4i s" % (self.progress, N, self.elapsed),
+            print("%4i/%i tasks finished after %4i s" % (self.progress, N, self.elapsed), end="")
             sys.stdout.flush()
-        print
-        print "done"
+        print()
+        print("done")
     
     def _republish_displaypub(self, content, eid):
         """republish individual displaypub content dicts"""
@@ -388,20 +393,31 b' class AsyncResult(object):'
         md = content['metadata'] or {}
         md['engine'] = eid
         ip.display_pub.publish(content['source'], content['data'], md)
+    
+    def _display_stream(self, text, prefix='', file=None):
+        if not text:
+            # nothing to display
+            return
+        if file is None:
+            file = sys.stdout
+        end = '' if text.endswith('\n') else '\n'
         
+        multiline = text.count('\n') > int(text.endswith('\n'))
+        if prefix and multiline and not text.startswith('\n'):
+            prefix = prefix + '\n'
+        print("%s%s" % (prefix, text), file=file, end=end)
         
+    
     def _display_single_result(self):
-        if self.stdout:
-            print self.stdout
-        if self.stderr:
-            print >> sys.stderr, self.stderr
+        self._display_stream(self.stdout)
+        self._display_stream(self.stderr, file=sys.stderr)
         
         try:
             get_ipython()
         except NameError:
             # displaypub is meaningless outside IPython
             return
-            
+        
         for output in self.outputs:
             self._republish_displaypub(output, self.engine_id)
         
@@ -443,9 +459,9 b' class AsyncResult(object):'
             self._display_single_result()
             return
         
-        stdouts = [s.rstrip() for s in self.stdout]
-        stderrs = [s.rstrip() for s in self.stderr]
-        pyouts  = [p for p in self.pyout]
+        stdouts = self.stdout
+        stderrs = self.stderr
+        pyouts  = self.pyout
         output_lists = self.outputs
         results = self.get()
         
@@ -455,10 +471,8 b' class AsyncResult(object):'
             for eid,stdout,stderr,outputs,r,pyout in zip(
                     targets, stdouts, stderrs, output_lists, results, pyouts
                 ):
-                if stdout:
-                    print '[stdout:%i]' % eid, stdout
-                if stderr:
-                    print >> sys.stderr, '[stderr:%i]' % eid, stderr
+                self._display_stream(stdout, '[stdout:%i] ' % eid)
+                self._display_stream(stderr, '[stderr:%i] ' % eid, file=sys.stderr)
                 
                 try:
                     get_ipython()
@@ -466,6 +480,9 b' class AsyncResult(object):'
                     # displaypub is meaningless outside IPython
                     return 
                 
+                if outputs or pyout is not None:
+                    _raw_text('[output:%i]' % eid)
+                
                 for output in outputs:
                     self._republish_displaypub(output, eid)
                 
@@ -474,14 +491,12 b' class AsyncResult(object):'
         
         elif groupby in ('type', 'order'):
             # republish stdout:
-            if any(stdouts):
-                for eid,stdout in zip(targets, stdouts):
-                    print '[stdout:%i]' % eid, stdout
+            for eid,stdout in zip(targets, stdouts):
+                self._display_stream(stdout, '[stdout:%i] ' % eid)
         
             # republish stderr:
-            if any(stderrs):
-                for eid,stderr in zip(targets, stderrs):
-                    print >> sys.stderr, '[stderr:%i]' % eid, stderr
+            for eid,stderr in zip(targets, stderrs):
+                self._display_stream(stderr, '[stderr:%i] ' % eid, file=sys.stderr)
         
             try:
                 get_ipython()
@@ -496,10 +511,13 b' class AsyncResult(object):'
                     for eid in targets:
                         outputs = output_dict[eid]
                         if len(outputs) >= N:
+                            _raw_text('[output:%i]' % eid)
                             self._republish_displaypub(outputs[i], eid)
             else:
                 # republish displaypub output
                 for eid,outputs in zip(targets, output_lists):
+                    if outputs:
+                        _raw_text('[output:%i]' % eid)
                     for output in outputs:
                         self._republish_displaypub(output, eid)
         

IPython/parallel/client/client.py

@@ -118,10 +118,14 b' class ExecuteReply(object):'
             out = TermColors.Red
             normal = TermColors.Normal
         
+        if '\n' in text_out and not text_out.startswith('\n'):
+            # add newline for multiline reprs
+            text_out = '\n' + text_out
+        
         p.text(
-            u'[%i] ' % self.metadata['engine_id'] +
-            out + u'Out[%i]: ' % self.execution_count +
-            normal + text_out
+            out + u'Out[%i:%i]: ' % (
+                self.metadata['engine_id'], self.execution_count
+            ) + normal + text_out
         )
     
     def _repr_html_(self):

IPython/parallel/tests/test_asyncresult.py

@@ -212,7 +212,7 b' class AsyncResultTest(ClusterTestCase):'
         with capture_output() as io:
             ar.display_outputs()
         self.assertEquals(io.stderr, '')
-        self.assertTrue('5555' in io.stdout)
+        self.assertEquals('5555\n', io.stdout)
 
         ar = v.execute("a=5")
         ar.get(5)
@@ -232,6 +232,7 b' class AsyncResultTest(ClusterTestCase):'
             ar.display_outputs()
         self.assertEquals(io.stderr, '')
         self.assertEquals(io.stdout.count('5555'), len(v), io.stdout)
+        self.assertFalse('\n\n' in io.stdout, io.stdout)
         self.assertEquals(io.stdout.count('[stdout:'), len(v), io.stdout)
 
         ar = v.execute("a=5")
@@ -252,6 +253,7 b' class AsyncResultTest(ClusterTestCase):'
             ar.display_outputs('engine')
         self.assertEquals(io.stderr, '')
         self.assertEquals(io.stdout.count('5555'), len(v), io.stdout)
+        self.assertFalse('\n\n' in io.stdout, io.stdout)
         self.assertEquals(io.stdout.count('[stdout:'), len(v), io.stdout)
 
         ar = v.execute("a=5")

IPython/parallel/tests/test_magics.py

@@ -16,6 +16,7 b' Authors:'
 # Imports
 #-------------------------------------------------------------------------------
 
+import re
 import sys
 import time
 
@@ -57,9 +58,26 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
             )
         out = io.stdout
         self.assertTrue('[stdout:' in out, out)
+        self.assertFalse('\n\n' in out)
         self.assertTrue(out.rstrip().endswith('10'))
         self.assertRaisesRemote(ZeroDivisionError, ip.magic, 'px 1/0')
     
+    def _check_generated_stderr(self, stderr, n):
+        expected = [
+            r'\[stderr:\d+\]',
+            '^stderr$',
+            '^stderr2$',
+        ] * n
+        
+        self.assertFalse('\n\n' in stderr, stderr)
+        lines = stderr.splitlines()
+        self.assertEquals(len(lines), len(expected), stderr)
+        for line,expect in zip(lines, expected):
+            if isinstance(expect, str):
+                expect = [expect]
+            for ex in expect:
+                self.assertTrue(re.search(ex, line) is not None, "Expected %r in %r" % (ex, line))
+        
     def test_cellpx_block_args(self):
         """%%px --[no]block flags work"""
         ip = get_ipython()
@@ -97,13 +115,16 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
         with capture_output() as io:
             ip.run_cell_magic('px', '--group-outputs=engine', 'generate_output()')
         
-        lines = io.stdout.strip().splitlines()[1:]
+        self.assertFalse('\n\n' in io.stdout)
+        lines = io.stdout.splitlines()[1:]
         expected = [
-            ('[stdout:', '] stdout'),
+            r'\[stdout:\d+\]',
+            'stdout',
             'stdout2',
-            'IPython.core.display.HTML',
-            'IPython.core.display.Math',
-            ('] Out[', 'IPython.core.display.Math')
+            r'\[output:\d+\]',
+            r'IPython\.core\.display\.HTML',
+            r'IPython\.core\.display\.Math',
+            r'Out\[\d+:\d+\]:.*IPython\.core\.display\.Math',
             ] * len(v)
         
         self.assertEquals(len(lines), len(expected), io.stdout)
@@ -111,20 +132,9 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
             if isinstance(expect, str):
                 expect = [expect]
             for ex in expect:
-                self.assertTrue(ex in line, "Expected %r in %r" % (ex, line))
-        
-        expected = [
-            ('[stderr:', '] stderr'),
-            'stderr2',
-        ] * len(v)
+                self.assertTrue(re.search(ex, line) is not None, "Expected %r in %r" % (ex, line))
         
-        lines = io.stderr.strip().splitlines()
-        self.assertEquals(len(lines), len(expected), io.stderr)
-        for line,expect in zip(lines, expected):
-            if isinstance(expect, str):
-                expect = [expect]
-            for ex in expect:
-                self.assertTrue(ex in line, "Expected %r in %r" % (ex, line))
+        self._check_generated_stderr(io.stderr, len(v))
 
 
     def test_cellpx_groupby_order(self):
@@ -139,20 +149,24 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
         with capture_output() as io:
             ip.run_cell_magic('px', '--group-outputs=order', 'generate_output()')
         
-        lines = io.stdout.strip().splitlines()[1:]
+        self.assertFalse('\n\n' in io.stdout)
+        lines = io.stdout.splitlines()[1:]
         expected = []
         expected.extend([
-            ('[stdout:', '] stdout'),
+            r'\[stdout:\d+\]',
+            'stdout',
             'stdout2',
         ] * len(v))
         expected.extend([
+            r'\[output:\d+\]',
             'IPython.core.display.HTML',
         ] * len(v))
         expected.extend([
+            r'\[output:\d+\]',
             'IPython.core.display.Math',
         ] * len(v))
         expected.extend([
-            ('] Out[', 'IPython.core.display.Math')
+            r'Out\[\d+:\d+\]:.*IPython\.core\.display\.Math'
         ] * len(v))
         
         self.assertEquals(len(lines), len(expected), io.stdout)
@@ -160,22 +174,11 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
             if isinstance(expect, str):
                 expect = [expect]
             for ex in expect:
-                self.assertTrue(ex in line, "Expected %r in %r" % (ex, line))
+                self.assertTrue(re.search(ex, line) is not None, "Expected %r in %r" % (ex, line))
         
-        expected = [
-            ('[stderr:', '] stderr'),
-            'stderr2',
-        ] * len(v)
-        
-        lines = io.stderr.strip().splitlines()
-        self.assertEquals(len(lines), len(expected), io.stderr)
-        for line,expect in zip(lines, expected):
-            if isinstance(expect, str):
-                expect = [expect]
-            for ex in expect:
-                self.assertTrue(ex in line, "Expected %r in %r" % (ex, line))
+        self._check_generated_stderr(io.stderr, len(v))
 
-    def test_cellpx_groupby_atype(self):
+    def test_cellpx_groupby_type(self):
         """%%px --group-outputs=type"""
         ip = get_ipython()
         v = self.client[:]
@@ -187,19 +190,22 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
         with capture_output() as io:
             ip.run_cell_magic('px', '--group-outputs=type', 'generate_output()')
         
-        lines = io.stdout.strip().splitlines()[1:]
+        self.assertFalse('\n\n' in io.stdout)
+        lines = io.stdout.splitlines()[1:]
         
         expected = []
         expected.extend([
-            ('[stdout:', '] stdout'),
+            r'\[stdout:\d+\]',
+            'stdout',
             'stdout2',
         ] * len(v))
         expected.extend([
-            'IPython.core.display.HTML',
-            'IPython.core.display.Math',
+            r'\[output:\d+\]',
+            r'IPython\.core\.display\.HTML',
+            r'IPython\.core\.display\.Math',
         ] * len(v))
         expected.extend([
-            ('] Out[', 'IPython.core.display.Math')
+            (r'Out\[\d+:\d+\]', r'IPython\.core\.display\.Math')
         ] * len(v))
         
         self.assertEquals(len(lines), len(expected), io.stdout)
@@ -207,20 +213,9 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
             if isinstance(expect, str):
                 expect = [expect]
             for ex in expect:
-                self.assertTrue(ex in line, "Expected %r in %r" % (ex, line))
+                self.assertTrue(re.search(ex, line) is not None, "Expected %r in %r" % (ex, line))
         
-        expected = [
-            ('[stderr:', '] stderr'),
-            'stderr2',
-        ] * len(v)
-        
-        lines = io.stderr.strip().splitlines()
-        self.assertEquals(len(lines), len(expected), io.stderr)
-        for line,expect in zip(lines, expected):
-            if isinstance(expect, str):
-                expect = [expect]
-            for ex in expect:
-                self.assertTrue(ex in line, "Expected %r in %r" % (ex, line))
+        self._check_generated_stderr(io.stderr, len(v))
 
 
     def test_px_nonblocking(self):
@@ -238,6 +233,8 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
         self.assertTrue(isinstance(ar, AsyncResult))
         self.assertTrue('Async' in io.stdout)
         self.assertFalse('[stdout:' in io.stdout)
+        self.assertFalse('\n\n' in io.stdout)
+        
         ar = ip.magic('px 1/0')
         self.assertRaisesRemote(ZeroDivisionError, ar.get)
     
@@ -256,12 +253,12 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
             ip.run_cell("b/c")
             ip.magic('autopx')
         
-        output = io.stdout.strip()
+        output = io.stdout
         
         self.assertTrue(output.startswith('%autopx enabled'), output)
-        self.assertTrue(output.endswith('%autopx disabled'), output)
+        self.assertTrue(output.rstrip().endswith('%autopx disabled'), output)
         self.assertTrue('RemoteError: ZeroDivisionError' in output, output)
-        self.assertTrue('] Out[' in output, output)
+        self.assertTrue('\nOut[' in output, output)
         self.assertTrue(': 24690' in output, output)
         ar = v.get_result(-1)
         self.assertEquals(v['a'], 5)
@@ -283,7 +280,7 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
             ip.run_cell('b*=2')
             ip.magic('autopx')
         
-        output = io.stdout.strip()
+        output = io.stdout.rstrip()
         
         self.assertTrue(output.startswith('%autopx enabled'))
         self.assertTrue(output.endswith('%autopx disabled'))
@@ -314,7 +311,7 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
                 ]:
             with capture_output() as io:
                 ip.magic('result ' + idx)
-            output = io.stdout.strip()
+            output = io.stdout
             msg = "expected %s output to include %s, but got: %s" % \
                 ('%result '+idx, str(data[name]), output)
             self.assertTrue(str(data[name]) in output, msg)
@@ -336,7 +333,7 b' class TestParallelMagics(ClusterTestCase, ParametricTestCase):'
         with capture_output() as io:
             ip.magic("px plot(rand(100))")
         
-        self.assertTrue('] Out[' in io.stdout, io.stdout)
+        self.assertTrue('Out[' in io.stdout, io.stdout)
         self.assertTrue('matplotlib.lines' in io.stdout, io.stdout)
         
 

docs/examples/parallel/Parallel Magics.ipynb

@@ -86,6 +86,15 b''
      "outputs": []
     },
     {
+     "cell_type": "code",
+     "collapsed": false,
+     "input": [
+      "%px print >> sys.stderr, \"ERROR\""
+     ],
+     "language": "python",
+     "outputs": []
+    },
+    {
      "cell_type": "markdown",
      "source": [
       "You don't have to wait for results:"
@@ -207,6 +216,7 b''
       "x = linspace(0,pi,1000)",
       "for n in range(id,12, stride):",
       "    print n",
+      "    plt.figure()",
       "    plt.plot(x,sin(n*x))",
       "plt.title(\"Plot %i\" % id)"
      ],
@@ -214,10 +224,85 b''
      "outputs": []
     },
     {
+     "cell_type": "markdown",
+     "source": [
+      "When you specify 'order', then individual display outputs (e.g. plots) will be interleaved:"
+     ]
+    },
+    {
+     "cell_type": "code",
+     "collapsed": false,
+     "input": [
+      "%%px --group-outputs=order",
+      "x = linspace(0,pi,1000)",
+      "for n in range(id,12, stride):",
+      "    print n",
+      "    plt.figure()",
+      "    plt.plot(x,sin(n*x))",
+      "plt.title(\"Plot %i\" % id)"
+     ],
+     "language": "python",
+     "outputs": []
+    },
+    {
+     "cell_type": "heading",
+     "level": 2,
+     "source": [
+      "Single-engine views"
+     ]
+    },
+    {
+     "cell_type": "markdown",
+     "source": [
+      "When a DirectView has a single target, the output is a bit simpler (no prefixes on stdout/err, etc.):"
+     ]
+    },
+    {
+     "cell_type": "code",
+     "collapsed": true,
+     "input": [
+      "def generate_output():",
+      "    \"\"\"function for testing output",
+      "    ",
+      "    publishes two outputs of each type, and returns something",
+      "    \"\"\"",
+      "    ",
+      "    import sys,os",
+      "    from IPython.core.display import display, HTML, Math",
+      "    ",
+      "    print \"stdout\"",
+      "    print >> sys.stderr, \"stderr\"",
+      "    ",
+      "    display(HTML(\"<b>HTML</b>\"))",
+      "    ",
+      "    print \"stdout2\"",
+      "    print >> sys.stderr, \"stderr2\"",
+      "    ",
+      "    display(Math(r\"\\alpha=\\beta\"))",
+      "    ",
+      "    return os.getpid()",
+      "",
+      "dv['generate_output'] = generate_output"
+     ],
+     "language": "python",
+     "outputs": []
+    },
+    {
      "cell_type": "code",
      "collapsed": true,
      "input": [
-      ""
+      "e0 = rc[-1]",
+      "e0.block = True",
+      "e0.activate()"
+     ],
+     "language": "python",
+     "outputs": []
+    },
+    {
+     "cell_type": "code",
+     "collapsed": false,
+     "input": [
+      "%px generate_output()"
      ],
      "language": "python",
      "outputs": []