Remove unused imports in IPython.kernel

IPython/kernel/inprocess/tests/test_kernelmanager.py

@@ -1,112 +1,111 b''
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2012  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Standard library imports
 import unittest
 
 # Local imports
 from IPython.kernel.inprocess.blocking import BlockingInProcessKernelClient
-from IPython.kernel.inprocess.ipkernel import InProcessKernel
 from IPython.kernel.inprocess.manager import InProcessKernelManager
 
 #-----------------------------------------------------------------------------
 # Test case
 #-----------------------------------------------------------------------------
 
 class InProcessKernelManagerTestCase(unittest.TestCase):
 
     def test_interface(self):
         """ Does the in-process kernel manager implement the basic KM interface?
         """
         km = InProcessKernelManager()
         self.assert_(not km.has_kernel)
 
         km.start_kernel()
         self.assert_(km.has_kernel)
         self.assert_(km.kernel is not None)
 
         kc = BlockingInProcessKernelClient(kernel=km.kernel)
         self.assert_(not kc.channels_running)
 
         kc.start_channels()
         self.assert_(kc.channels_running)
 
         old_kernel = km.kernel
         km.restart_kernel()
         self.assert_(km.kernel is not None)
         self.assertNotEquals(km.kernel, old_kernel)
 
         km.shutdown_kernel()
         self.assert_(not km.has_kernel)
 
         self.assertRaises(NotImplementedError, km.interrupt_kernel)
         self.assertRaises(NotImplementedError, km.signal_kernel, 9)
 
         kc.stop_channels()
         self.assert_(not kc.channels_running)
 
     def test_execute(self):
         """ Does executing code in an in-process kernel work?
         """
         km = InProcessKernelManager()
         km.start_kernel()
         kc = BlockingInProcessKernelClient(kernel=km.kernel)
         kc.start_channels()
         kc.execute('foo = 1')
         self.assertEquals(km.kernel.shell.user_ns['foo'], 1)
 
     def test_complete(self):
         """ Does requesting completion from an in-process kernel work?
         """
         km = InProcessKernelManager()
         km.start_kernel()
         kc = BlockingInProcessKernelClient(kernel=km.kernel)
         kc.start_channels()
         km.kernel.shell.push({'my_bar': 0, 'my_baz': 1})
         kc.complete('my_ba', 'my_ba', 5)
         msg = kc.get_shell_msg()
         self.assertEqual(msg['header']['msg_type'], 'complete_reply')
         self.assertEqual(sorted(msg['content']['matches']),
                           ['my_bar', 'my_baz'])
 
     def test_object_info(self):
         """ Does requesting object information from an in-process kernel work?
         """
         km = InProcessKernelManager()
         km.start_kernel()
         kc = BlockingInProcessKernelClient(kernel=km.kernel)
         kc.start_channels()
         km.kernel.shell.user_ns['foo'] = 1
         kc.object_info('foo')
         msg = kc.get_shell_msg()
         self.assertEquals(msg['header']['msg_type'], 'object_info_reply')
         self.assertEquals(msg['content']['name'], 'foo')
         self.assertEquals(msg['content']['type_name'], 'int')
 
     def test_history(self):
         """ Does requesting history from an in-process kernel work?
         """
         km = InProcessKernelManager()
         km.start_kernel()
         kc = BlockingInProcessKernelClient(kernel=km.kernel)
         kc.start_channels()
         kc.execute('%who')
         kc.history(hist_access_type='tail', n=1)
         msg = kc.shell_channel.get_msgs()[-1]
         self.assertEquals(msg['header']['msg_type'], 'history_reply')
         history = msg['content']['history']
         self.assertEquals(len(history), 1)
         self.assertEquals(history[0][2], '%who')
 
 
 if __name__ == '__main__':
     unittest.main()

IPython/kernel/ioloop/manager.py

@@ -1,63 +1,62 b''
 """A kernel manager with a tornado IOLoop"""
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2013  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import absolute_import
 
-import zmq
 from zmq.eventloop import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
 
 from IPython.utils.traitlets import (
     Instance
 )
 
 from IPython.kernel.manager import KernelManager
 from .restarter import IOLoopKernelRestarter
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 
 def as_zmqstream(f):
     def wrapped(self, *args, **kwargs):
         socket = f(self, *args, **kwargs)
         return ZMQStream(socket, self.loop)
     return wrapped
 
 class IOLoopKernelManager(KernelManager):
 
     loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
     def _loop_default(self):
         return ioloop.IOLoop.instance()
 
     _restarter = Instance('IPython.kernel.ioloop.IOLoopKernelRestarter')
 
     def start_restarter(self):
         if self.autorestart and self.has_kernel:
             if self._restarter is None:
                 self._restarter = IOLoopKernelRestarter(
                     kernel_manager=self, loop=self.loop,
                     parent=self, log=self.log
                 )
             self._restarter.start()
 
     def stop_restarter(self):
         if self.autorestart:
             if self._restarter is not None:
                 self._restarter.stop()
 
     connect_shell = as_zmqstream(KernelManager.connect_shell)
     connect_iopub = as_zmqstream(KernelManager.connect_iopub)
     connect_stdin = as_zmqstream(KernelManager.connect_stdin)
     connect_hb = as_zmqstream(KernelManager.connect_hb)

IPython/kernel/ioloop/restarter.py

@@ -1,55 +1,54 b''
 """A basic in process kernel monitor with autorestarting.
 
 This watches a kernel's state using KernelManager.is_alive and auto
 restarts the kernel if it dies.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2013  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import absolute_import
 
-import zmq
 from zmq.eventloop import ioloop
 
 
 from IPython.kernel.restarter import KernelRestarter
 from IPython.utils.traitlets import (
-    Instance, Float, List,
+    Instance,
 )
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 class IOLoopKernelRestarter(KernelRestarter):
     """Monitor and autorestart a kernel."""
 
     loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
     def _loop_default(self):
         return ioloop.IOLoop.instance()
 
     _pcallback = None
 
     def start(self):
         """Start the polling of the kernel."""
         if self._pcallback is None:
             self._pcallback = ioloop.PeriodicCallback(
                 self.poll, 1000*self.time_to_dead, self.loop
             )
             self._pcallback.start()
 
     def stop(self):
         """Stop the kernel polling."""
         if self._pcallback is not None:
             self._pcallback.stop()
             self._pcallback = None
 

IPython/kernel/multikernelmanager.py

@@ -1,301 +1,301 b''
 """A kernel manager for multiple kernels
 
 Authors:
 
 * Brian Granger
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2013  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import absolute_import
 
 import os
 import uuid
 
 import zmq
 
 from IPython.config.configurable import LoggingConfigurable
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import (
-    Instance, Dict, Unicode, Any, DottedObjectName, Bool
+    Instance, Dict, Unicode, Any, DottedObjectName
 )
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class DuplicateKernelError(Exception):
     pass
 
 
 
 def kernel_method(f):
     """decorator for proxying MKM.method(kernel_id) to individual KMs by ID"""
     def wrapped(self, kernel_id, *args, **kwargs):
         # get the kernel
         km = self.get_kernel(kernel_id)
         method = getattr(km, f.__name__)
         # call the kernel's method
         r = method(*args, **kwargs)
         # last thing, call anything defined in the actual class method
         # such as logging messages
         f(self, kernel_id, *args, **kwargs)
         # return the method result
         return r
     return wrapped
 
 
 class MultiKernelManager(LoggingConfigurable):
     """A class for managing multiple kernels."""
     
     kernel_manager_class = DottedObjectName(
         "IPython.kernel.ioloop.IOLoopKernelManager", config=True,
         help="""The kernel manager class.  This is configurable to allow
         subclassing of the KernelManager for customized behavior.
         """
     )
     def _kernel_manager_class_changed(self, name, old, new):
         self.kernel_manager_factory = import_item(new)
     
     kernel_manager_factory = Any(help="this is kernel_manager_class after import")
     def _kernel_manager_factory_default(self):
         return import_item(self.kernel_manager_class)
     
     context = Instance('zmq.Context')
     def _context_default(self):
         return zmq.Context.instance()
 
     connection_dir = Unicode('')
 
     _kernels = Dict()
 
     def list_kernel_ids(self):
         """Return a list of the kernel ids of the active kernels."""
         # Create a copy so we can iterate over kernels in operations
         # that delete keys.
         return list(self._kernels.keys())
 
     def __len__(self):
         """Return the number of running kernels."""
         return len(self.list_kernel_ids())
 
     def __contains__(self, kernel_id):
         return kernel_id in self._kernels
 
     def start_kernel(self, **kwargs):
         """Start a new kernel.
 
         The caller can pick a kernel_id by passing one in as a keyword arg,
         otherwise one will be picked using a uuid.
 
         To silence the kernel's stdout/stderr, call this using::
 
             km.start_kernel(stdout=PIPE, stderr=PIPE)
 
         """
         kernel_id = kwargs.pop('kernel_id', unicode(uuid.uuid4()))
         if kernel_id in self:
             raise DuplicateKernelError('Kernel already exists: %s' % kernel_id)
         # kernel_manager_factory is the constructor for the KernelManager
         # subclass we are using. It can be configured as any Configurable,
         # including things like its transport and ip.
         km = self.kernel_manager_factory(connection_file=os.path.join(
                     self.connection_dir, "kernel-%s.json" % kernel_id),
                     parent=self, autorestart=True, log=self.log
         )
         km.start_kernel(**kwargs)
         self._kernels[kernel_id] = km
         return kernel_id
 
     @kernel_method
     def shutdown_kernel(self, kernel_id, now=False):
         """Shutdown a kernel by its kernel uuid.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to shutdown.
         now : bool
             Should the kernel be shutdown forcibly using a signal.
         """
         self.log.info("Kernel shutdown: %s" % kernel_id)
         self.remove_kernel(kernel_id)
 
     def remove_kernel(self, kernel_id):
         """remove a kernel from our mapping.
 
         Mainly so that a kernel can be removed if it is already dead,
         without having to call shutdown_kernel.
 
         The kernel object is returned.
         """
         return self._kernels.pop(kernel_id)
 
     def shutdown_all(self, now=False):
         """Shutdown all kernels."""
         for kid in self.list_kernel_ids():
             self.shutdown_kernel(kid, now=now)
 
     @kernel_method
     def interrupt_kernel(self, kernel_id):
         """Interrupt (SIGINT) the kernel by its uuid.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to interrupt.
         """
         self.log.info("Kernel interrupted: %s" % kernel_id)
 
     @kernel_method
     def signal_kernel(self, kernel_id, signum):
         """Sends a signal to the kernel by its uuid.
 
         Note that since only SIGTERM is supported on Windows, this function
         is only useful on Unix systems.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to signal.
         """
         self.log.info("Signaled Kernel %s with %s" % (kernel_id, signum))
 
     @kernel_method
     def restart_kernel(self, kernel_id):
         """Restart a kernel by its uuid, keeping the same ports.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel to interrupt.
         """
         self.log.info("Kernel restarted: %s" % kernel_id)
 
     @kernel_method
     def is_alive(self, kernel_id):
         """Is the kernel alive.
 
         This calls KernelManager.is_alive() which calls Popen.poll on the
         actual kernel subprocess.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel.
         """
 
     def _check_kernel_id(self, kernel_id):
         """check that a kernel id is valid"""
         if kernel_id not in self:
             raise KeyError("Kernel with id not found: %s" % kernel_id)
 
     def get_kernel(self, kernel_id):
         """Get the single KernelManager object for a kernel by its uuid.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel.
         """
         self._check_kernel_id(kernel_id)
         return self._kernels[kernel_id]
 
     @kernel_method
     def add_restart_callback(self, kernel_id, callback, event='restart'):
         """add a callback for the KernelRestarter"""
 
     @kernel_method
     def remove_restart_callback(self, kernel_id, callback, event='restart'):
         """remove a callback for the KernelRestarter"""
 
     @kernel_method
     def get_connection_info(self, kernel_id):
         """Return a dictionary of connection data for a kernel.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel.
 
         Returns
         =======
         connection_dict : dict
             A dict of the information needed to connect to a kernel.
             This includes the ip address and the integer port
             numbers of the different channels (stdin_port, iopub_port,
             shell_port, hb_port).
         """
 
     @kernel_method
     def connect_iopub(self, kernel_id, identity=None):
         """Return a zmq Socket connected to the iopub channel.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel
         identity : bytes (optional)
             The zmq identity of the socket
 
         Returns
         =======
         stream : zmq Socket or ZMQStream
         """
 
     @kernel_method
     def connect_shell(self, kernel_id, identity=None):
         """Return a zmq Socket connected to the shell channel.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel
         identity : bytes (optional)
             The zmq identity of the socket
 
         Returns
         =======
         stream : zmq Socket or ZMQStream
         """
 
     @kernel_method
     def connect_stdin(self, kernel_id, identity=None):
         """Return a zmq Socket connected to the stdin channel.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel
         identity : bytes (optional)
             The zmq identity of the socket
 
         Returns
         =======
         stream : zmq Socket or ZMQStream
         """
 
     @kernel_method
     def connect_hb(self, kernel_id, identity=None):
         """Return a zmq Socket connected to the hb channel.
 
         Parameters
         ==========
         kernel_id : uuid
             The id of the kernel
         identity : bytes (optional)
             The zmq identity of the socket
 
         Returns
         =======
         stream : zmq Socket or ZMQStream
         """

IPython/kernel/tests/test_launcher.py

@@ -1,63 +1,60 b''
 """Tests for kernel utility functions
 
 Authors
 -------
 * MinRK
 """
 #-----------------------------------------------------------------------------
 # Copyright (c) 2011, the IPython Development Team.
 #
 # Distributed under the terms of the Modified BSD License.
 #
 # The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
-# Stdlib imports
-from unittest import TestCase
-
 # Third-party imports
 import nose.tools as nt
 
 # Our own imports
 from IPython.testing import decorators as dec
 from IPython.kernel.launcher import swallow_argv
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 @dec.parametric
 def test_swallow_argv():
     tests = [
         # expected  , argv       , aliases, flags
         (['-a', '5'], ['-a', '5'], None, None),
         (['5'], ['-a', '5'], None, ['a']),
         ([], ['-a', '5'], ['a'], None),
         ([], ['-a', '5'], ['a'], ['a']),
         ([], ['--foo'], None, ['foo']),
         ([], ['--foo'], ['foobar'], []),
         ([], ['--foo', '5'], ['foo'], []),
         ([], ['--foo=5'], ['foo'], []),
         (['--foo=5'], ['--foo=5'], [], ['foo']),
         (['5'], ['--foo', '5'], [], ['foo']),
         (['bar'], ['--foo', '5', 'bar'], ['foo'], ['foo']),
         (['bar'], ['--foo=5', 'bar'], ['foo'], ['foo']),
         (['5','bar'], ['--foo', '5', 'bar'], None, ['foo']),
         (['bar'], ['--foo', '5', 'bar'], ['foo'], None),
         (['bar'], ['--foo=5', 'bar'], ['foo'], None),
     ]
     for expected, argv, aliases, flags in tests:
         stripped = swallow_argv(argv, aliases=aliases, flags=flags)
         message = '\n'.join(['',
             "argv: %r" % argv,
             "aliases: %r" % aliases,
             "flags : %r" % flags,
             "expected : %r" % expected,
             "returned : %r" % stripped,
         ])
         yield nt.assert_equal(expected, stripped, message)
 

IPython/kernel/tests/test_message_spec.py

@@ -1,511 +1,507 b''
 """Test suite for our zeromq-based messaging specification.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING.txt, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 import re
-import sys
-import time
 from subprocess import PIPE
 from Queue import Empty
 
 import nose.tools as nt
 
 from IPython.kernel import KernelManager
 
-
 from IPython.testing import decorators as dec
-from IPython.utils import io
 from IPython.utils.traitlets import (
     HasTraits, TraitError, Bool, Unicode, Dict, Integer, List, Enum, Any,
 )
 
 #-----------------------------------------------------------------------------
 # Global setup and utilities
 #-----------------------------------------------------------------------------
 
 def setup():
     global KM, KC
     KM = KernelManager()
     KM.start_kernel(stdout=PIPE, stderr=PIPE)
     KC = KM.client()
     KC.start_channels()
     
     # wait for kernel to be ready
     KC.execute("pass")
     KC.get_shell_msg(block=True, timeout=5)
     flush_channels()
 
 
 def teardown():
     KC.stop_channels()
     KM.shutdown_kernel()
 
 
 def flush_channels(kc=None):
     """flush any messages waiting on the queue"""
     if kc is None:
         kc = KC
     for channel in (kc.shell_channel, kc.iopub_channel):
         while True:
             try:
                 msg = channel.get_msg(block=True, timeout=0.1)
             except Empty:
                 break
             else:
                 list(validate_message(msg))
 
 
 def execute(code='', kc=None, **kwargs):
     """wrapper for doing common steps for validating an execution request"""
     msg_id = KC.execute(code=code, **kwargs)
     reply = KC.get_shell_msg(timeout=2)
     list(validate_message(reply, 'execute_reply', msg_id))
     busy = KC.get_iopub_msg(timeout=2)
     list(validate_message(busy, 'status', msg_id))
     nt.assert_equal(busy['content']['execution_state'], 'busy')
     
     if not kwargs.get('silent'):
         pyin = KC.get_iopub_msg(timeout=2)
         list(validate_message(pyin, 'pyin', msg_id))
         nt.assert_equal(pyin['content']['code'], code)
     
     return msg_id, reply['content']
 
 #-----------------------------------------------------------------------------
 # MSG Spec References
 #-----------------------------------------------------------------------------
 
 
 class Reference(HasTraits):
 
     """
     Base class for message spec specification testing.
 
     This class is the core of the message specification test.  The
     idea is that child classes implement trait attributes for each
     message keys, so that message keys can be tested against these
     traits using :meth:`check` method.
 
     """
 
     def check(self, d):
         """validate a dict against our traits"""
         for key in self.trait_names():
             yield nt.assert_true(key in d, "Missing key: %r, should be found in %s" % (key, d))
             # FIXME: always allow None, probably not a good idea
             if d[key] is None:
                 continue
             try:
                 setattr(self, key, d[key])
             except TraitError as e:
                 yield nt.assert_true(False, str(e))
 
 
 class RMessage(Reference):
     msg_id = Unicode()
     msg_type = Unicode()
     header = Dict()
     parent_header = Dict()
     content = Dict()
 
 class RHeader(Reference):
     msg_id = Unicode()
     msg_type = Unicode()
     session = Unicode()
     username = Unicode()
 
 class RContent(Reference):
     status = Enum((u'ok', u'error'))
 
 
 class ExecuteReply(Reference):
     execution_count = Integer()
     status = Enum((u'ok', u'error'))
     
     def check(self, d):
         for tst in Reference.check(self, d):
             yield tst
         if d['status'] == 'ok':
             for tst in ExecuteReplyOkay().check(d):
                 yield tst
         elif d['status'] == 'error':
             for tst in ExecuteReplyError().check(d):
                 yield tst
 
 
 class ExecuteReplyOkay(Reference):
     payload = List(Dict)
     user_variables = Dict()
     user_expressions = Dict()
 
 
 class ExecuteReplyError(Reference):
     ename = Unicode()
     evalue = Unicode()
     traceback = List(Unicode)
 
 
 class OInfoReply(Reference):
     name = Unicode()
     found = Bool()
     ismagic = Bool()
     isalias = Bool()
     namespace = Enum((u'builtin', u'magics', u'alias', u'Interactive'))
     type_name = Unicode()
     string_form = Unicode()
     base_class = Unicode()
     length = Integer()
     file = Unicode()
     definition = Unicode()
     argspec = Dict()
     init_definition = Unicode()
     docstring = Unicode()
     init_docstring = Unicode()
     class_docstring = Unicode()
     call_def = Unicode()
     call_docstring = Unicode()
     source = Unicode()
     
     def check(self, d):
         for tst in Reference.check(self, d):
             yield tst
         if d['argspec'] is not None:
             for tst in ArgSpec().check(d['argspec']):
                 yield tst
 
 
 class ArgSpec(Reference):
     args = List(Unicode)
     varargs = Unicode()
     varkw = Unicode()
     defaults = List()
 
 
 class Status(Reference):
     execution_state = Enum((u'busy', u'idle', u'starting'))
 
 
 class CompleteReply(Reference):
     matches = List(Unicode)
 
 
 def Version(num, trait=Integer):
     return List(trait, default_value=[0] * num, minlen=num, maxlen=num)
 
 
 class KernelInfoReply(Reference):
 
     protocol_version = Version(2)
     ipython_version = Version(4, Any)
     language_version = Version(3)
     language = Unicode()
 
     def _ipython_version_changed(self, name, old, new):
         for v in new:
             nt.assert_true(
                 isinstance(v, int) or isinstance(v, basestring),
                 'expected int or string as version component, got {0!r}'
                 .format(v))
 
 
 # IOPub messages
 
 class PyIn(Reference):
     code = Unicode()
     execution_count = Integer()
 
 
 PyErr = ExecuteReplyError
 
 
 class Stream(Reference):
     name = Enum((u'stdout', u'stderr'))
     data = Unicode()
 
 
 mime_pat = re.compile(r'\w+/\w+')
 
 class DisplayData(Reference):
     source = Unicode()
     metadata = Dict()
     data = Dict()
     def _data_changed(self, name, old, new):
         for k,v in new.iteritems():
             nt.assert_true(mime_pat.match(k))
             nt.assert_true(isinstance(v, basestring), "expected string data, got %r" % v)
 
 
 class PyOut(Reference):
     execution_count = Integer()
     data = Dict()
     def _data_changed(self, name, old, new):
         for k,v in new.iteritems():
             nt.assert_true(mime_pat.match(k))
             nt.assert_true(isinstance(v, basestring), "expected string data, got %r" % v)
 
 
 references = {
     'execute_reply' : ExecuteReply(),
     'object_info_reply' : OInfoReply(),
     'status' : Status(),
     'complete_reply' : CompleteReply(),
     'kernel_info_reply': KernelInfoReply(),
     'pyin' : PyIn(),
     'pyout' : PyOut(),
     'pyerr' : PyErr(),
     'stream' : Stream(),
     'display_data' : DisplayData(),
 }
 """
 Specifications of `content` part of the reply messages.
 """
 
 
 def validate_message(msg, msg_type=None, parent=None):
     """validate a message
     
     This is a generator, and must be iterated through to actually
     trigger each test.
     
     If msg_type and/or parent are given, the msg_type and/or parent msg_id
     are compared with the given values.
     """
     RMessage().check(msg)
     if msg_type:
         yield nt.assert_equal(msg['msg_type'], msg_type)
     if parent:
         yield nt.assert_equal(msg['parent_header']['msg_id'], parent)
     content = msg['content']
     ref = references[msg['msg_type']]
     for tst in ref.check(content):
         yield tst
 
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 # Shell channel
 
 @dec.parametric
 def test_execute():
     flush_channels()
     
     msg_id = KC.execute(code='x=1')
     reply = KC.get_shell_msg(timeout=2)
     for tst in validate_message(reply, 'execute_reply', msg_id):
         yield tst
 
 
 @dec.parametric
 def test_execute_silent():
     flush_channels()
     msg_id, reply = execute(code='x=1', silent=True)
     
     # flush status=idle
     status = KC.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(status, 'status', msg_id):
         yield tst
     nt.assert_equal(status['content']['execution_state'], 'idle')
 
     yield nt.assert_raises(Empty, KC.iopub_channel.get_msg, timeout=0.1)
     count = reply['execution_count']
     
     msg_id, reply = execute(code='x=2', silent=True)
     
     # flush status=idle
     status = KC.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(status, 'status', msg_id):
         yield tst
     yield nt.assert_equal(status['content']['execution_state'], 'idle')
     
     yield nt.assert_raises(Empty, KC.iopub_channel.get_msg, timeout=0.1)
     count_2 = reply['execution_count']
     yield nt.assert_equal(count_2, count)
 
 
 @dec.parametric
 def test_execute_error():
     flush_channels()
     
     msg_id, reply = execute(code='1/0')
     yield nt.assert_equal(reply['status'], 'error')
     yield nt.assert_equal(reply['ename'], 'ZeroDivisionError')
     
     pyerr = KC.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(pyerr, 'pyerr', msg_id):
         yield tst
 
 
 def test_execute_inc():
     """execute request should increment execution_count"""
     flush_channels()
 
     msg_id, reply = execute(code='x=1')
     count = reply['execution_count']
     
     flush_channels()
     
     msg_id, reply = execute(code='x=2')
     count_2 = reply['execution_count']
     nt.assert_equal(count_2, count+1)
 
 
 def test_user_variables():
     flush_channels()
 
     msg_id, reply = execute(code='x=1', user_variables=['x'])
     user_variables = reply['user_variables']
     nt.assert_equal(user_variables, {u'x': {
         u'status': u'ok',
         u'data': {u'text/plain': u'1'},
         u'metadata': {},
     }})
 
 
 def test_user_variables_fail():
     flush_channels()
 
     msg_id, reply = execute(code='x=1', user_variables=['nosuchname'])
     user_variables = reply['user_variables']
     foo = user_variables['nosuchname']
     nt.assert_equal(foo['status'], 'error')
     nt.assert_equal(foo['ename'], 'KeyError')
 
 
 def test_user_expressions():
     flush_channels()
 
     msg_id, reply = execute(code='x=1', user_expressions=dict(foo='x+1'))
     user_expressions = reply['user_expressions']
     nt.assert_equal(user_expressions, {u'foo': {
         u'status': u'ok',
         u'data': {u'text/plain': u'2'},
         u'metadata': {},
     }})
 
 
 def test_user_expressions_fail():
     flush_channels()
 
     msg_id, reply = execute(code='x=0', user_expressions=dict(foo='nosuchname'))
     user_expressions = reply['user_expressions']
     foo = user_expressions['foo']
     nt.assert_equal(foo['status'], 'error')
     nt.assert_equal(foo['ename'], 'NameError')
 
 
 @dec.parametric
 def test_oinfo():
     flush_channels()
 
     msg_id = KC.object_info('a')
     reply = KC.get_shell_msg(timeout=2)
     for tst in validate_message(reply, 'object_info_reply', msg_id):
         yield tst
 
 
 @dec.parametric
 def test_oinfo_found():
     flush_channels()
 
     msg_id, reply = execute(code='a=5')
     
     msg_id = KC.object_info('a')
     reply = KC.get_shell_msg(timeout=2)
     for tst in validate_message(reply, 'object_info_reply', msg_id):
         yield tst
     content = reply['content']
     yield nt.assert_true(content['found'])
     argspec = content['argspec']
     yield nt.assert_true(argspec is None, "didn't expect argspec dict, got %r" % argspec)
 
 
 @dec.parametric
 def test_oinfo_detail():
     flush_channels()
 
     msg_id, reply = execute(code='ip=get_ipython()')
     
     msg_id = KC.object_info('ip.object_inspect', detail_level=2)
     reply = KC.get_shell_msg(timeout=2)
     for tst in validate_message(reply, 'object_info_reply', msg_id):
         yield tst
     content = reply['content']
     yield nt.assert_true(content['found'])
     argspec = content['argspec']
     yield nt.assert_true(isinstance(argspec, dict), "expected non-empty argspec dict, got %r" % argspec)
     yield nt.assert_equal(argspec['defaults'], [0])
 
 
 @dec.parametric
 def test_oinfo_not_found():
     flush_channels()
 
     msg_id = KC.object_info('dne')
     reply = KC.get_shell_msg(timeout=2)
     for tst in validate_message(reply, 'object_info_reply', msg_id):
         yield tst
     content = reply['content']
     yield nt.assert_false(content['found'])
 
 
 @dec.parametric
 def test_complete():
     flush_channels()
 
     msg_id, reply = execute(code="alpha = albert = 5")
     
     msg_id = KC.complete('al', 'al', 2)
     reply = KC.get_shell_msg(timeout=2)
     for tst in validate_message(reply, 'complete_reply', msg_id):
         yield tst
     matches = reply['content']['matches']
     for name in ('alpha', 'albert'):
         yield nt.assert_true(name in matches, "Missing match: %r" % name)
 
 
 @dec.parametric
 def test_kernel_info_request():
     flush_channels()
 
     msg_id = KC.kernel_info()
     reply = KC.get_shell_msg(timeout=2)
     for tst in validate_message(reply, 'kernel_info_reply', msg_id):
         yield tst
 
 
 # IOPub channel
 
 
 @dec.parametric
 def test_stream():
     flush_channels()
 
     msg_id, reply = execute("print('hi')")
 
     stdout = KC.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(stdout, 'stream', msg_id):
         yield tst
     content = stdout['content']
     yield nt.assert_equal(content['name'], u'stdout')
     yield nt.assert_equal(content['data'], u'hi\n')
 
 
 @dec.parametric
 def test_display_data():
     flush_channels()
 
     msg_id, reply = execute("from IPython.core.display import display; display(1)")
     
     display = KC.iopub_channel.get_msg(timeout=2)
     for tst in validate_message(display, 'display_data', parent=msg_id):
         yield tst
     data = display['content']['data']
     yield nt.assert_equal(data['text/plain'], u'1')
 

IPython/kernel/zmq/heartbeat.py

@@ -1,57 +1,56 b''
 """The client and server for a basic ping-pong style heartbeat.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 import socket
-import sys
 from threading import Thread
 
 import zmq
 
 from IPython.utils.localinterfaces import LOCALHOST
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 
 class Heartbeat(Thread):
     "A simple ping-pong style heartbeat that runs in a thread."
 
     def __init__(self, context, addr=('tcp', LOCALHOST, 0)):
         Thread.__init__(self)
         self.context = context
         self.transport, self.ip, self.port = addr
         if self.port == 0:
             if addr[0] == 'tcp':
                 s = socket.socket()
                 # '*' means all interfaces to 0MQ, which is '' to socket.socket
                 s.bind(('' if self.ip == '*' else self.ip, 0))
                 self.port = s.getsockname()[1]
                 s.close()
             elif addr[0] == 'ipc':
                 self.port = 1
                 while os.path.exists("%s-%s" % (self.ip, self.port)):
                     self.port = self.port + 1
             else:
                 raise ValueError("Unrecognized zmq transport: %s" % addr[0])
         self.addr = (self.ip, self.port)
         self.daemon = True
 
     def run(self):
         self.socket = self.context.socket(zmq.REP)
         c = ':' if self.transport == 'tcp' else '-'
         self.socket.bind('%s://%s' % (self.transport, self.ip) + c + str(self.port))
         zmq.device(zmq.FORWARDER, self.socket, self.socket)
 

IPython/kernel/zmq/ipkernel.py

@@ -1,804 +1,803 b''
 #!/usr/bin/env python
 """A simple interactive kernel that talks to a frontend over 0MQ.
 
 Things to do:
 
 * Implement `set_parent` logic. Right before doing exec, the Kernel should
   call set_parent on all the PUB objects with the message about to be executed.
 * Implement random port and security key logic.
 * Implement control messages.
 * Implement event loop and poll version.
 """
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Standard library imports
 import __builtin__
 import sys
 import time
 import traceback
 import logging
 import uuid
 
 from datetime import datetime
 from signal import (
-        signal, getsignal, default_int_handler, SIGINT, SIG_IGN
+        signal, default_int_handler, SIGINT
 )
 
 # System library imports
 import zmq
 from zmq.eventloop import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
 
 # Local imports
 from IPython.config.configurable import Configurable
 from IPython.core.error import StdinNotImplementedError
 from IPython.core import release
-from IPython.utils import io
 from IPython.utils import py3compat
 from IPython.utils.jsonutil import json_clean
 from IPython.utils.traitlets import (
-    Any, Instance, Float, Dict, CaselessStrEnum, List, Set, Integer, Unicode,
+    Any, Instance, Float, Dict, List, Set, Integer, Unicode,
     Type
 )
 
 from serialize import serialize_object, unpack_apply_message
 from session import Session
 from zmqshell import ZMQInteractiveShell
 
 
 #-----------------------------------------------------------------------------
 # Main kernel class
 #-----------------------------------------------------------------------------
 
 protocol_version = list(release.kernel_protocol_version_info)
 ipython_version = list(release.version_info)
 language_version = list(sys.version_info[:3])
 
 
 class Kernel(Configurable):
 
     #---------------------------------------------------------------------------
     # Kernel interface
     #---------------------------------------------------------------------------
 
     # attribute to override with a GUI
     eventloop = Any(None)
     def _eventloop_changed(self, name, old, new):
         """schedule call to eventloop from IOLoop"""
         loop = ioloop.IOLoop.instance()
         loop.add_timeout(time.time()+0.1, self.enter_eventloop)
 
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
     shell_class = Type(ZMQInteractiveShell)
 
     session = Instance(Session)
     profile_dir = Instance('IPython.core.profiledir.ProfileDir')
     shell_streams = List()
     control_stream = Instance(ZMQStream)
     iopub_socket = Instance(zmq.Socket)
     stdin_socket = Instance(zmq.Socket)
     log = Instance(logging.Logger)
     
     user_module = Any()
     def _user_module_changed(self, name, old, new):
         if self.shell is not None:
             self.shell.user_module = new
     
     user_ns = Dict(default_value=None)
     def _user_ns_changed(self, name, old, new):
         if self.shell is not None:
             self.shell.user_ns = new
             self.shell.init_user_ns()
 
     # identities:
     int_id = Integer(-1)
     ident = Unicode()
 
     def _ident_default(self):
         return unicode(uuid.uuid4())
 
 
     # Private interface
     
     # Time to sleep after flushing the stdout/err buffers in each execute
     # cycle.  While this introduces a hard limit on the minimal latency of the
     # execute cycle, it helps prevent output synchronization problems for
     # clients.
     # Units are in seconds.  The minimum zmq latency on local host is probably
     # ~150 microseconds, set this to 500us for now.  We may need to increase it
     # a little if it's not enough after more interactive testing.
     _execute_sleep = Float(0.0005, config=True)
 
     # Frequency of the kernel's event loop.
     # Units are in seconds, kernel subclasses for GUI toolkits may need to
     # adapt to milliseconds.
     _poll_interval = Float(0.05, config=True)
 
     # If the shutdown was requested over the network, we leave here the
     # necessary reply message so it can be sent by our registered atexit
     # handler.  This ensures that the reply is only sent to clients truly at
     # the end of our shutdown process (which happens after the underlying
     # IPython shell's own shutdown).
     _shutdown_message = None
 
     # This is a dict of port number that the kernel is listening on. It is set
     # by record_ports and used by connect_request.
     _recorded_ports = Dict()
 
     # A reference to the Python builtin 'raw_input' function.
     # (i.e., __builtin__.raw_input for Python 2.7, builtins.input for Python 3)
     _sys_raw_input = Any()
 
     # set of aborted msg_ids
     aborted = Set()
 
 
     def __init__(self, **kwargs):
         super(Kernel, self).__init__(**kwargs)
 
         # Initialize the InteractiveShell subclass
         self.shell = self.shell_class.instance(parent=self,
             profile_dir = self.profile_dir,
             user_module = self.user_module,
             user_ns     = self.user_ns,
         )
         self.shell.displayhook.session = self.session
         self.shell.displayhook.pub_socket = self.iopub_socket
         self.shell.displayhook.topic = self._topic('pyout')
         self.shell.display_pub.session = self.session
         self.shell.display_pub.pub_socket = self.iopub_socket
         self.shell.data_pub.session = self.session
         self.shell.data_pub.pub_socket = self.iopub_socket
 
         # TMP - hack while developing
         self.shell._reply_content = None
 
         # Build dict of handlers for message types
         msg_types = [ 'execute_request', 'complete_request',
                       'object_info_request', 'history_request',
                       'kernel_info_request',
                       'connect_request', 'shutdown_request',
                       'apply_request',
                     ]
         self.shell_handlers = {}
         for msg_type in msg_types:
             self.shell_handlers[msg_type] = getattr(self, msg_type)
         
         control_msg_types = msg_types + [ 'clear_request', 'abort_request' ]
         self.control_handlers = {}
         for msg_type in control_msg_types:
             self.control_handlers[msg_type] = getattr(self, msg_type)
 
     def dispatch_control(self, msg):
         """dispatch control requests"""
         idents,msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.unserialize(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Control Message", exc_info=True)
             return
 
         self.log.debug("Control received: %s", msg)
 
         header = msg['header']
         msg_id = header['msg_id']
         msg_type = header['msg_type']
 
         handler = self.control_handlers.get(msg_type, None)
         if handler is None:
             self.log.error("UNKNOWN CONTROL MESSAGE TYPE: %r", msg_type)
         else:
             try:
                 handler(self.control_stream, idents, msg)
             except Exception:
                 self.log.error("Exception in control handler:", exc_info=True)
     
     def dispatch_shell(self, stream, msg):
         """dispatch shell requests"""
         # flush control requests first
         if self.control_stream:
             self.control_stream.flush()
         
         idents,msg = self.session.feed_identities(msg, copy=False)
         try:
             msg = self.session.unserialize(msg, content=True, copy=False)
         except:
             self.log.error("Invalid Message", exc_info=True)
             return
 
         header = msg['header']
         msg_id = header['msg_id']
         msg_type = msg['header']['msg_type']
         
         # Print some info about this message and leave a '--->' marker, so it's
         # easier to trace visually the message chain when debugging.  Each
         # handler prints its message at the end.
         self.log.debug('\n*** MESSAGE TYPE:%s***', msg_type)
         self.log.debug('   Content: %s\n   --->\n   ', msg['content'])
 
         if msg_id in self.aborted:
             self.aborted.remove(msg_id)
             # is it safe to assume a msg_id will not be resubmitted?
             reply_type = msg_type.split('_')[0] + '_reply'
             status = {'status' : 'aborted'}
             md = {'engine' : self.ident}
             md.update(status)
             reply_msg = self.session.send(stream, reply_type, metadata=md,
                         content=status, parent=msg, ident=idents)
             return
         
         handler = self.shell_handlers.get(msg_type, None)
         if handler is None:
             self.log.error("UNKNOWN MESSAGE TYPE: %r", msg_type)
         else:
             # ensure default_int_handler during handler call
             sig = signal(SIGINT, default_int_handler)
             try:
                 handler(stream, idents, msg)
             except Exception:
                 self.log.error("Exception in message handler:", exc_info=True)
             finally:
                 signal(SIGINT, sig)
     
     def enter_eventloop(self):
         """enter eventloop"""
         self.log.info("entering eventloop")
         # restore default_int_handler
         signal(SIGINT, default_int_handler)
         while self.eventloop is not None:
             try:
                 self.eventloop(self)
             except KeyboardInterrupt:
                 # Ctrl-C shouldn't crash the kernel
                 self.log.error("KeyboardInterrupt caught in kernel")
                 continue
             else:
                 # eventloop exited cleanly, this means we should stop (right?)
                 self.eventloop = None
                 break
         self.log.info("exiting eventloop")
 
     def start(self):
         """register dispatchers for streams"""
         self.shell.exit_now = False
         if self.control_stream:
             self.control_stream.on_recv(self.dispatch_control, copy=False)
 
         def make_dispatcher(stream):
             def dispatcher(msg):
                 return self.dispatch_shell(stream, msg)
             return dispatcher
 
         for s in self.shell_streams:
             s.on_recv(make_dispatcher(s), copy=False)
 
         # publish idle status
         self._publish_status('starting')
     
     def do_one_iteration(self):
         """step eventloop just once"""
         if self.control_stream:
             self.control_stream.flush()
         for stream in self.shell_streams:
             # handle at most one request per iteration
             stream.flush(zmq.POLLIN, 1)
             stream.flush(zmq.POLLOUT)
 
 
     def record_ports(self, ports):
         """Record the ports that this kernel is using.
 
         The creator of the Kernel instance must call this methods if they
         want the :meth:`connect_request` method to return the port numbers.
         """
         self._recorded_ports = ports
 
     #---------------------------------------------------------------------------
     # Kernel request handlers
     #---------------------------------------------------------------------------
     
     def _make_metadata(self, other=None):
         """init metadata dict, for execute/apply_reply"""
         new_md = {
             'dependencies_met' : True,
             'engine' : self.ident,
             'started': datetime.now(),
         }
         if other:
             new_md.update(other)
         return new_md
     
     def _publish_pyin(self, code, parent, execution_count):
         """Publish the code request on the pyin stream."""
 
         self.session.send(self.iopub_socket, u'pyin',
                             {u'code':code, u'execution_count': execution_count},
                             parent=parent, ident=self._topic('pyin')
         )
     
     def _publish_status(self, status, parent=None):
         """send status (busy/idle) on IOPub"""
         self.session.send(self.iopub_socket,
                           u'status',
                           {u'execution_state': status},
                           parent=parent,
                           ident=self._topic('status'),
                           )
         
 
     def execute_request(self, stream, ident, parent):
         """handle an execute_request"""
         
         self._publish_status(u'busy', parent)
         
         try:
             content = parent[u'content']
             code = content[u'code']
             silent = content[u'silent']
             store_history = content.get(u'store_history', not silent)
         except:
             self.log.error("Got bad msg: ")
             self.log.error("%s", parent)
             return
         
         md = self._make_metadata(parent['metadata'])
 
         shell = self.shell # we'll need this a lot here
 
         # Replace raw_input. Note that is not sufficient to replace
         # raw_input in the user namespace.
         if content.get('allow_stdin', False):
             raw_input = lambda prompt='': self._raw_input(prompt, ident, parent)
         else:
             raw_input = lambda prompt='' : self._no_raw_input()
 
         if py3compat.PY3:
             self._sys_raw_input = __builtin__.input
             __builtin__.input = raw_input
         else:
             self._sys_raw_input = __builtin__.raw_input
             __builtin__.raw_input = raw_input
 
         # Set the parent message of the display hook and out streams.
         shell.displayhook.set_parent(parent)
         shell.display_pub.set_parent(parent)
         shell.data_pub.set_parent(parent)
         try:
             sys.stdout.set_parent(parent)
         except AttributeError:
             pass
         try:
             sys.stderr.set_parent(parent)
         except AttributeError:
             pass
 
         # Re-broadcast our input for the benefit of listening clients, and
         # start computing output
         if not silent:
             self._publish_pyin(code, parent, shell.execution_count)
 
         reply_content = {}
         try:
             # FIXME: the shell calls the exception handler itself.
             shell.run_cell(code, store_history=store_history, silent=silent)
         except:
             status = u'error'
             # FIXME: this code right now isn't being used yet by default,
             # because the run_cell() call above directly fires off exception
             # reporting.  This code, therefore, is only active in the scenario
             # where runlines itself has an unhandled exception.  We need to
             # uniformize this, for all exception construction to come from a
             # single location in the codbase.
             etype, evalue, tb = sys.exc_info()
             tb_list = traceback.format_exception(etype, evalue, tb)
             reply_content.update(shell._showtraceback(etype, evalue, tb_list))
         else:
             status = u'ok'
         finally:
             # Restore raw_input.
              if py3compat.PY3:
                  __builtin__.input = self._sys_raw_input
              else:
                  __builtin__.raw_input = self._sys_raw_input
 
         reply_content[u'status'] = status
 
         # Return the execution counter so clients can display prompts
         reply_content['execution_count'] = shell.execution_count - 1
 
         # FIXME - fish exception info out of shell, possibly left there by
         # runlines.  We'll need to clean up this logic later.
         if shell._reply_content is not None:
             reply_content.update(shell._reply_content)
             e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method='execute')
             reply_content['engine_info'] = e_info
             # reset after use
             shell._reply_content = None
         
         if 'traceback' in reply_content:
             self.log.info("Exception in execute request:\n%s", '\n'.join(reply_content['traceback']))
         
 
         # At this point, we can tell whether the main code execution succeeded
         # or not.  If it did, we proceed to evaluate user_variables/expressions
         if reply_content['status'] == 'ok':
             reply_content[u'user_variables'] = \
                          shell.user_variables(content.get(u'user_variables', []))
             reply_content[u'user_expressions'] = \
                          shell.user_expressions(content.get(u'user_expressions', {}))
         else:
             # If there was an error, don't even try to compute variables or
             # expressions
             reply_content[u'user_variables'] = {}
             reply_content[u'user_expressions'] = {}
 
         # Payloads should be retrieved regardless of outcome, so we can both
         # recover partial output (that could have been generated early in a
         # block, before an error) and clear the payload system always.
         reply_content[u'payload'] = shell.payload_manager.read_payload()
         # Be agressive about clearing the payload because we don't want
         # it to sit in memory until the next execute_request comes in.
         shell.payload_manager.clear_payload()
 
         # Flush output before sending the reply.
         sys.stdout.flush()
         sys.stderr.flush()
         # FIXME: on rare occasions, the flush doesn't seem to make it to the
         # clients... This seems to mitigate the problem, but we definitely need
         # to better understand what's going on.
         if self._execute_sleep:
             time.sleep(self._execute_sleep)
 
         # Send the reply.
         reply_content = json_clean(reply_content)
         
         md['status'] = reply_content['status']
         if reply_content['status'] == 'error' and \
                         reply_content['ename'] == 'UnmetDependency':
                 md['dependencies_met'] = False
 
         reply_msg = self.session.send(stream, u'execute_reply',
                                       reply_content, parent, metadata=md,
                                       ident=ident)
         
         self.log.debug("%s", reply_msg)
 
         if not silent and reply_msg['content']['status'] == u'error':
             self._abort_queues()
 
         self._publish_status(u'idle', parent)
 
     def complete_request(self, stream, ident, parent):
         txt, matches = self._complete(parent)
         matches = {'matches' : matches,
                    'matched_text' : txt,
                    'status' : 'ok'}
         matches = json_clean(matches)
         completion_msg = self.session.send(stream, 'complete_reply',
                                            matches, parent, ident)
         self.log.debug("%s", completion_msg)
 
     def object_info_request(self, stream, ident, parent):
         content = parent['content']
         object_info = self.shell.object_inspect(content['oname'],
                         detail_level = content.get('detail_level', 0)
         )
         # Before we send this object over, we scrub it for JSON usage
         oinfo = json_clean(object_info)
         msg = self.session.send(stream, 'object_info_reply',
                                 oinfo, parent, ident)
         self.log.debug("%s", msg)
 
     def history_request(self, stream, ident, parent):
         # We need to pull these out, as passing **kwargs doesn't work with
         # unicode keys before Python 2.6.5.
         hist_access_type = parent['content']['hist_access_type']
         raw = parent['content']['raw']
         output = parent['content']['output']
         if hist_access_type == 'tail':
             n = parent['content']['n']
             hist = self.shell.history_manager.get_tail(n, raw=raw, output=output,
                                                             include_latest=True)
 
         elif hist_access_type == 'range':
             session = parent['content']['session']
             start = parent['content']['start']
             stop = parent['content']['stop']
             hist = self.shell.history_manager.get_range(session, start, stop,
                                                         raw=raw, output=output)
 
         elif hist_access_type == 'search':
             n = parent['content'].get('n')
             unique = parent['content'].get('unique', False)
             pattern = parent['content']['pattern']
             hist = self.shell.history_manager.search(
                 pattern, raw=raw, output=output, n=n, unique=unique)
 
         else:
             hist = []
         hist = list(hist)
         content = {'history' : hist}
         content = json_clean(content)
         msg = self.session.send(stream, 'history_reply',
                                 content, parent, ident)
         self.log.debug("Sending history reply with %i entries", len(hist))
 
     def connect_request(self, stream, ident, parent):
         if self._recorded_ports is not None:
             content = self._recorded_ports.copy()
         else:
             content = {}
         msg = self.session.send(stream, 'connect_reply',
                                 content, parent, ident)
         self.log.debug("%s", msg)
 
     def kernel_info_request(self, stream, ident, parent):
         vinfo = {
             'protocol_version': protocol_version,
             'ipython_version': ipython_version,
             'language_version': language_version,
             'language': 'python',
         }
         msg = self.session.send(stream, 'kernel_info_reply',
                                 vinfo, parent, ident)
         self.log.debug("%s", msg)
 
     def shutdown_request(self, stream, ident, parent):
         self.shell.exit_now = True
         content = dict(status='ok')
         content.update(parent['content'])
         self.session.send(stream, u'shutdown_reply', content, parent, ident=ident)
         # same content, but different msg_id for broadcasting on IOPub
         self._shutdown_message = self.session.msg(u'shutdown_reply',
                                                   content, parent
         )
 
         self._at_shutdown()
         # call sys.exit after a short delay
         loop = ioloop.IOLoop.instance()
         loop.add_timeout(time.time()+0.1, loop.stop)
 
     #---------------------------------------------------------------------------
     # Engine methods
     #---------------------------------------------------------------------------
 
     def apply_request(self, stream, ident, parent):
         try:
             content = parent[u'content']
             bufs = parent[u'buffers']
             msg_id = parent['header']['msg_id']
         except:
             self.log.error("Got bad msg: %s", parent, exc_info=True)
             return
 
         self._publish_status(u'busy', parent)
 
         # Set the parent message of the display hook and out streams.
         shell = self.shell
         shell.displayhook.set_parent(parent)
         shell.display_pub.set_parent(parent)
         shell.data_pub.set_parent(parent)
         try:
             sys.stdout.set_parent(parent)
         except AttributeError:
             pass
         try:
             sys.stderr.set_parent(parent)
         except AttributeError:
             pass
 
         # pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
         # self.iopub_socket.send(pyin_msg)
         # self.session.send(self.iopub_socket, u'pyin', {u'code':code},parent=parent)
         md = self._make_metadata(parent['metadata'])
         try:
             working = shell.user_ns
 
             prefix = "_"+str(msg_id).replace("-","")+"_"
 
             f,args,kwargs = unpack_apply_message(bufs, working, copy=False)
 
             fname = getattr(f, '__name__', 'f')
 
             fname = prefix+"f"
             argname = prefix+"args"
             kwargname = prefix+"kwargs"
             resultname = prefix+"result"
 
             ns = { fname : f, argname : args, kwargname : kwargs , resultname : None }
             # print ns
             working.update(ns)
             code = "%s = %s(*%s,**%s)" % (resultname, fname, argname, kwargname)
             try:
                 exec code in shell.user_global_ns, shell.user_ns
                 result = working.get(resultname)
             finally:
                 for key in ns.iterkeys():
                     working.pop(key)
 
             result_buf = serialize_object(result,
                 buffer_threshold=self.session.buffer_threshold,
                 item_threshold=self.session.item_threshold,
             )
         
         except:
             # invoke IPython traceback formatting
             shell.showtraceback()
             # FIXME - fish exception info out of shell, possibly left there by
             # run_code.  We'll need to clean up this logic later.
             reply_content = {}
             if shell._reply_content is not None:
                 reply_content.update(shell._reply_content)
                 e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method='apply')
                 reply_content['engine_info'] = e_info
                 # reset after use
                 shell._reply_content = None
             
             self.session.send(self.iopub_socket, u'pyerr', reply_content, parent=parent,
                                 ident=self._topic('pyerr'))
             self.log.info("Exception in apply request:\n%s", '\n'.join(reply_content['traceback']))
             result_buf = []
 
             if reply_content['ename'] == 'UnmetDependency':
                 md['dependencies_met'] = False
         else:
             reply_content = {'status' : 'ok'}
 
         # put 'ok'/'error' status in header, for scheduler introspection:
         md['status'] = reply_content['status']
 
         # flush i/o
         sys.stdout.flush()
         sys.stderr.flush()
         
         reply_msg = self.session.send(stream, u'apply_reply', reply_content,
                     parent=parent, ident=ident,buffers=result_buf, metadata=md)
 
         self._publish_status(u'idle', parent)
 
     #---------------------------------------------------------------------------
     # Control messages
     #---------------------------------------------------------------------------
 
     def abort_request(self, stream, ident, parent):
         """abort a specifig msg by id"""
         msg_ids = parent['content'].get('msg_ids', None)
         if isinstance(msg_ids, basestring):
             msg_ids = [msg_ids]
         if not msg_ids:
             self.abort_queues()
         for mid in msg_ids:
             self.aborted.add(str(mid))
 
         content = dict(status='ok')
         reply_msg = self.session.send(stream, 'abort_reply', content=content,
                 parent=parent, ident=ident)
         self.log.debug("%s", reply_msg)
 
     def clear_request(self, stream, idents, parent):
         """Clear our namespace."""
         self.shell.reset(False)
         msg = self.session.send(stream, 'clear_reply', ident=idents, parent=parent,
                 content = dict(status='ok'))
 
 
     #---------------------------------------------------------------------------
     # Protected interface
     #---------------------------------------------------------------------------
 
     def _wrap_exception(self, method=None):
         # import here, because _wrap_exception is only used in parallel,
         # and parallel has higher min pyzmq version
         from IPython.parallel.error import wrap_exception
         e_info = dict(engine_uuid=self.ident, engine_id=self.int_id, method=method)
         content = wrap_exception(e_info)
         return content
 
     def _topic(self, topic):
         """prefixed topic for IOPub messages"""
         if self.int_id >= 0:
             base = "engine.%i" % self.int_id
         else:
             base = "kernel.%s" % self.ident
         
         return py3compat.cast_bytes("%s.%s" % (base, topic))
     
     def _abort_queues(self):
         for stream in self.shell_streams:
             if stream:
                 self._abort_queue(stream)
 
     def _abort_queue(self, stream):
         poller = zmq.Poller()
         poller.register(stream.socket, zmq.POLLIN)
         while True:
             idents,msg = self.session.recv(stream, zmq.NOBLOCK, content=True)
             if msg is None:
                 return
 
             self.log.info("Aborting:")
             self.log.info("%s", msg)
             msg_type = msg['header']['msg_type']
             reply_type = msg_type.split('_')[0] + '_reply'
 
             status = {'status' : 'aborted'}
             md = {'engine' : self.ident}
             md.update(status)
             reply_msg = self.session.send(stream, reply_type, metadata=md,
                         content=status, parent=msg, ident=idents)
             self.log.debug("%s", reply_msg)
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
             poller.poll(50)
 
 
     def _no_raw_input(self):
         """Raise StdinNotImplentedError if active frontend doesn't support
         stdin."""
         raise StdinNotImplementedError("raw_input was called, but this "
                                        "frontend does not support stdin.") 
         
     def _raw_input(self, prompt, ident, parent):
         # Flush output before making the request.
         sys.stderr.flush()
         sys.stdout.flush()
         # flush the stdin socket, to purge stale replies
         while True:
             try:
                 self.stdin_socket.recv_multipart(zmq.NOBLOCK)
             except zmq.ZMQError as e:
                 if e.errno == zmq.EAGAIN:
                     break
                 else:
                     raise
         
         # Send the input request.
         content = json_clean(dict(prompt=prompt))
         self.session.send(self.stdin_socket, u'input_request', content, parent,
                           ident=ident)
 
         # Await a response.
         while True:
             try:
                 ident, reply = self.session.recv(self.stdin_socket, 0)
             except Exception:
                 self.log.warn("Invalid Message:", exc_info=True)
             else:
                 break
         try:
             value = reply['content']['value']
         except:
             self.log.error("Got bad raw_input reply: ")
             self.log.error("%s", parent)
             value = ''
         if value == '\x04':
             # EOF
             raise EOFError
         return value
 
     def _complete(self, msg):
         c = msg['content']
         try:
             cpos = int(c['cursor_pos'])
         except:
             # If we don't get something that we can convert to an integer, at
             # least attempt the completion guessing the cursor is at the end of
             # the text, if there's any, and otherwise of the line
             cpos = len(c['text'])
             if cpos==0:
                 cpos = len(c['line'])
         return self.shell.complete(c['text'], c['line'], cpos)
 
     def _at_shutdown(self):
         """Actions taken at shutdown by the kernel, called by python's atexit.
         """
         # io.rprint("Kernel at_shutdown") # dbg
         if self._shutdown_message is not None:
             self.session.send(self.iopub_socket, self._shutdown_message, ident=self._topic('shutdown'))
             self.log.debug("%s", self._shutdown_message)
         [ s.flush(zmq.POLLOUT) for s in self.shell_streams ]
 

IPython/kernel/zmq/log.py

@@ -1,23 +1,21 b''
-import logging
 from logging import INFO, DEBUG, WARN, ERROR, FATAL
 
-import zmq
 from zmq.log.handlers import PUBHandler
 
 class EnginePUBHandler(PUBHandler):
     """A simple PUBHandler subclass that sets root_topic"""
     engine=None
     
     def __init__(self, engine, *args, **kwargs):
         PUBHandler.__init__(self,*args, **kwargs)
         self.engine = engine
         
     @property
     def root_topic(self):
         """this is a property, in case the handler is created
         before the engine gets registered with an id"""
         if isinstance(getattr(self.engine, 'id', None), int):
             return "engine.%i"%self.engine.id
         else:
             return "engine"
     

IPython/kernel/zmq/pylab/backend_inline.py

@@ -1,215 +1,212 b''
 """Produce SVG versions of active plots for display by the rich Qt frontend.
 """
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
-# Standard library imports
-import sys
-
 # Third-party imports
 import matplotlib
 from matplotlib.backends.backend_agg import new_figure_manager, FigureCanvasAgg
 from matplotlib._pylab_helpers import Gcf
 
 # Local imports.
 from IPython.config.configurable import SingletonConfigurable
 from IPython.core.display import display
 from IPython.core.displaypub import publish_display_data
 from IPython.core.pylabtools import print_figure, select_figure_format
 from IPython.utils.traitlets import Dict, Instance, CaselessStrEnum, Bool
 from IPython.utils.warn import warn
 
 #-----------------------------------------------------------------------------
 # Configurable for inline backend options
 #-----------------------------------------------------------------------------
 # inherit from InlineBackendConfig for deprecation purposes
 class InlineBackendConfig(SingletonConfigurable):
     pass
 
 class InlineBackend(InlineBackendConfig):
     """An object to store configuration of the inline backend."""
 
     def _config_changed(self, name, old, new):
         # warn on change of renamed config section
         if new.InlineBackendConfig != old.InlineBackendConfig:
             warn("InlineBackendConfig has been renamed to InlineBackend")
         super(InlineBackend, self)._config_changed(name, old, new)
 
     # The typical default figure size is too large for inline use,
     # so we shrink the figure size to 6x4, and tweak fonts to
     # make that fit.
     rc = Dict({'figure.figsize': (6.0,4.0),
         # play nicely with white background in the Qt and notebook frontend
         'figure.facecolor': 'white',
         'figure.edgecolor': 'white',
         # 12pt labels get cutoff on 6x4 logplots, so use 10pt.
         'font.size': 10,
         # 72 dpi matches SVG/qtconsole
         # this only affects PNG export, as SVG has no dpi setting
         'savefig.dpi': 72,
         # 10pt still needs a little more room on the xlabel:
         'figure.subplot.bottom' : .125
         }, config=True,
         help="""Subset of matplotlib rcParams that should be different for the
         inline backend."""
     )
 
     figure_format = CaselessStrEnum(['svg', 'png', 'retina'], default_value='png', config=True,
         help="The image format for figures with the inline backend.")
 
     def _figure_format_changed(self, name, old, new):
         if self.shell is None:
             return
         else:
             select_figure_format(self.shell, new)
     
     close_figures = Bool(True, config=True,
         help="""Close all figures at the end of each cell.
         
         When True, ensures that each cell starts with no active figures, but it
         also means that one must keep track of references in order to edit or
         redraw figures in subsequent cells. This mode is ideal for the notebook,
         where residual plots from other cells might be surprising.
         
         When False, one must call figure() to create new figures. This means
         that gcf() and getfigs() can reference figures created in other cells,
         and the active figure can continue to be edited with pylab/pyplot
         methods that reference the current active figure. This mode facilitates
         iterative editing of figures, and behaves most consistently with
         other matplotlib backends, but figure barriers between cells must
         be explicit.
         """)
 
     shell = Instance('IPython.core.interactiveshell.InteractiveShellABC')
 
 
 #-----------------------------------------------------------------------------
 # Functions
 #-----------------------------------------------------------------------------
 
 def show(close=None):
     """Show all figures as SVG/PNG payloads sent to the IPython clients.
 
     Parameters
     ----------
     close : bool, optional
       If true, a ``plt.close('all')`` call is automatically issued after
       sending all the figures. If this is set, the figures will entirely
       removed from the internal list of figures.
     """
     if close is None:
         close = InlineBackend.instance().close_figures
     try:
         for figure_manager in Gcf.get_all_fig_managers():
             display(figure_manager.canvas.figure)
     finally:
         show._to_draw = []
         if close:
             matplotlib.pyplot.close('all')
 
 
 
 # This flag will be reset by draw_if_interactive when called
 show._draw_called = False
 # list of figures to draw when flush_figures is called
 show._to_draw = []
 
 
 def draw_if_interactive():
     """
     Is called after every pylab drawing command
     """
     # signal that the current active figure should be sent at the end of
     # execution.  Also sets the _draw_called flag, signaling that there will be
     # something to send.  At the end of the code execution, a separate call to
     # flush_figures() will act upon these values
 
     fig = Gcf.get_active().canvas.figure
 
     # Hack: matplotlib FigureManager objects in interacive backends (at least
     # in some of them) monkeypatch the figure object and add a .show() method
     # to it.  This applies the same monkeypatch in order to support user code
     # that might expect `.show()` to be part of the official API of figure
     # objects.
     # For further reference:
     # https://github.com/ipython/ipython/issues/1612
     # https://github.com/matplotlib/matplotlib/issues/835
     
     if not hasattr(fig, 'show'):
         # Queue up `fig` for display
         fig.show = lambda *a: display(fig)
 
     # If matplotlib was manually set to non-interactive mode, this function
     # should be a no-op (otherwise we'll generate duplicate plots, since a user
     # who set ioff() manually expects to make separate draw/show calls).
     if not matplotlib.is_interactive():
         return
 
     # ensure current figure will be drawn, and each subsequent call
     # of draw_if_interactive() moves the active figure to ensure it is
     # drawn last
     try:
         show._to_draw.remove(fig)
     except ValueError:
         # ensure it only appears in the draw list once
         pass
     # Queue up the figure for drawing in next show() call
     show._to_draw.append(fig)
     show._draw_called = True
 
 
 def flush_figures():
     """Send all figures that changed
 
     This is meant to be called automatically and will call show() if, during
     prior code execution, there had been any calls to draw_if_interactive.
     
     This function is meant to be used as a post_execute callback in IPython,
     so user-caused errors are handled with showtraceback() instead of being
     allowed to raise.  If this function is not called from within IPython,
     then these exceptions will raise.
     """
     if not show._draw_called:
         return
     
     if InlineBackend.instance().close_figures:
         # ignore the tracking, just draw and close all figures
         try:
             return show(True)
         except Exception as e:
             # safely show traceback if in IPython, else raise
             try:
                 get_ipython
             except NameError:
                 raise e
             else:
                 get_ipython().showtraceback()
                 return
     try:
         # exclude any figures that were closed:
         active = set([fm.canvas.figure for fm in Gcf.get_all_fig_managers()])
         for fig in [ fig for fig in show._to_draw if fig in active ]:
             try:
                 display(fig)
             except Exception as e:
                 # safely show traceback if in IPython, else raise
                 try:
                     get_ipython
                 except NameError:
                     raise e
                 else:
                     get_ipython().showtraceback()
                     break
     finally:
         # clear flags for next round
         show._to_draw = []
         show._draw_called = False
 
 
 # Changes to matplotlib in version 1.2 requires a mpl backend to supply a default
 # figurecanvas. This is set here to a Agg canvas
 # See https://github.com/matplotlib/matplotlib/pull/1125
 FigureCanvas = FigureCanvasAgg
 

IPython/kernel/zmq/serialize.py

@@ -1,205 +1,198 b''
 """serialization utilities for apply messages
 
 Authors:
 
 * Min RK
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
-# Standard library imports
-import logging
-import os
-import re
-import socket
-import sys
-
 try:
     import cPickle
     pickle = cPickle
 except:
     cPickle = None
     import pickle
 
 
 # IPython imports
 from IPython.utils import py3compat
 from IPython.utils.data import flatten
 from IPython.utils.pickleutil import (
     can, uncan, can_sequence, uncan_sequence, CannedObject,
     istype, sequence_types,
 )
 
 if py3compat.PY3:
     buffer = memoryview
 
 #-----------------------------------------------------------------------------
 # Serialization Functions
 #-----------------------------------------------------------------------------
 
 # default values for the thresholds:
 MAX_ITEMS = 64
 MAX_BYTES = 1024
 
 def _extract_buffers(obj, threshold=MAX_BYTES):
     """extract buffers larger than a certain threshold"""
     buffers = []
     if isinstance(obj, CannedObject) and obj.buffers:
         for i,buf in enumerate(obj.buffers):
             if len(buf) > threshold:
                 # buffer larger than threshold, prevent pickling
                 obj.buffers[i] = None
                 buffers.append(buf)
             elif isinstance(buf, buffer):
                 # buffer too small for separate send, coerce to bytes
                 # because pickling buffer objects just results in broken pointers
                 obj.buffers[i] = bytes(buf)
     return buffers
 
 def _restore_buffers(obj, buffers):
     """restore buffers extracted by """
     if isinstance(obj, CannedObject) and obj.buffers:
         for i,buf in enumerate(obj.buffers):
             if buf is None:
                 obj.buffers[i] = buffers.pop(0)
 
 def serialize_object(obj, buffer_threshold=MAX_BYTES, item_threshold=MAX_ITEMS):
     """Serialize an object into a list of sendable buffers.
     
     Parameters
     ----------
     
     obj : object
         The object to be serialized
     buffer_threshold : int
         The threshold (in bytes) for pulling out data buffers
         to avoid pickling them.
     item_threshold : int
         The maximum number of items over which canning will iterate.
         Containers (lists, dicts) larger than this will be pickled without
         introspection.
     
     Returns
     -------
     [bufs] : list of buffers representing the serialized object.
     """
     buffers = []
     if istype(obj, sequence_types) and len(obj) < item_threshold:
         cobj = can_sequence(obj)
         for c in cobj:
             buffers.extend(_extract_buffers(c, buffer_threshold))
     elif istype(obj, dict) and len(obj) < item_threshold:
         cobj = {}
         for k in sorted(obj.iterkeys()):
             c = can(obj[k])
             buffers.extend(_extract_buffers(c, buffer_threshold))
             cobj[k] = c
     else:
         cobj = can(obj)
         buffers.extend(_extract_buffers(cobj, buffer_threshold))
 
     buffers.insert(0, pickle.dumps(cobj,-1))
     return buffers
 
 def unserialize_object(buffers, g=None):
     """reconstruct an object serialized by serialize_object from data buffers.
     
     Parameters
     ----------
     
     bufs : list of buffers/bytes
     
     g : globals to be used when uncanning
     
     Returns
     -------
     
     (newobj, bufs) : unpacked object, and the list of remaining unused buffers.
     """
     bufs = list(buffers)
     pobj = bufs.pop(0)
     if not isinstance(pobj, bytes):
         # a zmq message
         pobj = bytes(pobj)
     canned = pickle.loads(pobj)
     if istype(canned, sequence_types) and len(canned) < MAX_ITEMS:
         for c in canned:
             _restore_buffers(c, bufs)
         newobj = uncan_sequence(canned, g)
     elif isinstance(canned, dict) and len(canned) < MAX_ITEMS:
         newobj = {}
         for k in sorted(canned.iterkeys()):
             c = canned[k]
             _restore_buffers(c, bufs)
             newobj[k] = uncan(c, g)
     else:
         _restore_buffers(canned, bufs)
         newobj = uncan(canned, g)
     
     return newobj, bufs
 
 def pack_apply_message(f, args, kwargs, buffer_threshold=MAX_BYTES, item_threshold=MAX_ITEMS):
     """pack up a function, args, and kwargs to be sent over the wire
     
     Each element of args/kwargs will be canned for special treatment,
     but inspection will not go any deeper than that.
     
     Any object whose data is larger than `threshold`  will not have their data copied
     (only numpy arrays and bytes/buffers support zero-copy)
     
     Message will be a list of bytes/buffers of the format:
     
     [ cf, pinfo, <arg_bufs>, <kwarg_bufs> ]
     
     With length at least two + len(args) + len(kwargs)
     """
     
     arg_bufs = flatten(serialize_object(arg, buffer_threshold, item_threshold) for arg in args)
     
     kw_keys = sorted(kwargs.keys())
     kwarg_bufs = flatten(serialize_object(kwargs[key], buffer_threshold, item_threshold) for key in kw_keys)
     
     info = dict(nargs=len(args), narg_bufs=len(arg_bufs), kw_keys=kw_keys)
     
     msg = [pickle.dumps(can(f),-1)]
     msg.append(pickle.dumps(info, -1))
     msg.extend(arg_bufs)
     msg.extend(kwarg_bufs)
     
     return msg
 
 def unpack_apply_message(bufs, g=None, copy=True):
     """unpack f,args,kwargs from buffers packed by pack_apply_message()
     Returns: original f,args,kwargs"""
     bufs = list(bufs) # allow us to pop
     assert len(bufs) >= 2, "not enough buffers!"
     if not copy:
         for i in range(2):
             bufs[i] = bufs[i].bytes
     f = uncan(pickle.loads(bufs.pop(0)), g)
     info = pickle.loads(bufs.pop(0))
     arg_bufs, kwarg_bufs = bufs[:info['narg_bufs']], bufs[info['narg_bufs']:]
     
     args = []
     for i in range(info['nargs']):
         arg, arg_bufs = unserialize_object(arg_bufs, g)
         args.append(arg)
     args = tuple(args)
     assert not arg_bufs, "Shouldn't be any arg bufs left over"
     
     kwargs = {}
     for key in info['kw_keys']:
         kwarg, kwarg_bufs = unserialize_object(kwarg_bufs, g)
         kwargs[key] = kwarg
     assert not kwarg_bufs, "Shouldn't be any kwarg bufs left over"
     
     return f,args,kwargs
 

IPython/kernel/zmq/session.py

@@ -1,807 +1,806 b''
 """Session object for building, serializing, sending, and receiving messages in
 IPython. The Session object supports serialization, HMAC signatures, and
 metadata on messages.
 
 Also defined here are utilities for working with Sessions:
 * A SessionFactory to be used as a base class for configurables that work with
 Sessions.
 * A Message object for convenience that allows attribute-access to the msg dict.
 
 Authors:
 
 * Min RK
 * Brian Granger
 * Fernando Perez
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import hmac
 import logging
 import os
 import pprint
 import random
 import uuid
 from datetime import datetime
 
 try:
     import cPickle
     pickle = cPickle
 except:
     cPickle = None
     import pickle
 
 import zmq
 from zmq.utils import jsonapi
 from zmq.eventloop.ioloop import IOLoop
 from zmq.eventloop.zmqstream import ZMQStream
 
-from IPython.config.application import Application, boolean_flag
 from IPython.config.configurable import Configurable, LoggingConfigurable
 from IPython.utils import io
 from IPython.utils.importstring import import_item
 from IPython.utils.jsonutil import extract_dates, squash_dates, date_default
 from IPython.utils.py3compat import str_to_bytes, str_to_unicode
 from IPython.utils.traitlets import (CBytes, Unicode, Bool, Any, Instance, Set,
                                         DottedObjectName, CUnicode, Dict, Integer)
 from IPython.kernel.zmq.serialize import MAX_ITEMS, MAX_BYTES
 
 #-----------------------------------------------------------------------------
 # utility functions
 #-----------------------------------------------------------------------------
 
 def squash_unicode(obj):
     """coerce unicode back to bytestrings."""
     if isinstance(obj,dict):
         for key in obj.keys():
             obj[key] = squash_unicode(obj[key])
             if isinstance(key, unicode):
                 obj[squash_unicode(key)] = obj.pop(key)
     elif isinstance(obj, list):
         for i,v in enumerate(obj):
             obj[i] = squash_unicode(v)
     elif isinstance(obj, unicode):
         obj = obj.encode('utf8')
     return obj
 
 #-----------------------------------------------------------------------------
 # globals and defaults
 #-----------------------------------------------------------------------------
 
 # ISO8601-ify datetime objects
 json_packer = lambda obj: jsonapi.dumps(obj, default=date_default)
 json_unpacker = lambda s: extract_dates(jsonapi.loads(s))
 
 pickle_packer = lambda o: pickle.dumps(o,-1)
 pickle_unpacker = pickle.loads
 
 default_packer = json_packer
 default_unpacker = json_unpacker
 
 DELIM = b"<IDS|MSG>"
 # singleton dummy tracker, which will always report as done
 DONE = zmq.MessageTracker()
 
 #-----------------------------------------------------------------------------
 # Mixin tools for apps that use Sessions
 #-----------------------------------------------------------------------------
 
 session_aliases = dict(
     ident = 'Session.session',
     user = 'Session.username',
     keyfile = 'Session.keyfile',
 )
 
 session_flags  = {
     'secure' : ({'Session' : { 'key' : str_to_bytes(str(uuid.uuid4())),
                             'keyfile' : '' }},
         """Use HMAC digests for authentication of messages.
         Setting this flag will generate a new UUID to use as the HMAC key.
         """),
     'no-secure' : ({'Session' : { 'key' : b'', 'keyfile' : '' }},
         """Don't authenticate messages."""),
 }
 
 def default_secure(cfg):
     """Set the default behavior for a config environment to be secure.
     
     If Session.key/keyfile have not been set, set Session.key to
     a new random UUID.
     """
     
     if 'Session' in cfg:
         if 'key' in cfg.Session or 'keyfile' in cfg.Session:
             return
     # key/keyfile not specified, generate new UUID:
     cfg.Session.key = str_to_bytes(str(uuid.uuid4()))
 
 
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 class SessionFactory(LoggingConfigurable):
     """The Base class for configurables that have a Session, Context, logger,
     and IOLoop.
     """
 
     logname = Unicode('')
     def _logname_changed(self, name, old, new):
         self.log = logging.getLogger(new)
 
     # not configurable:
     context = Instance('zmq.Context')
     def _context_default(self):
         return zmq.Context.instance()
 
     session = Instance('IPython.kernel.zmq.session.Session')
 
     loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
     def _loop_default(self):
         return IOLoop.instance()
 
     def __init__(self, **kwargs):
         super(SessionFactory, self).__init__(**kwargs)
 
         if self.session is None:
             # construct the session
             self.session = Session(**kwargs)
 
 
 class Message(object):
     """A simple message object that maps dict keys to attributes.
 
     A Message can be created from a dict and a dict from a Message instance
     simply by calling dict(msg_obj)."""
 
     def __init__(self, msg_dict):
         dct = self.__dict__
         for k, v in dict(msg_dict).iteritems():
             if isinstance(v, dict):
                 v = Message(v)
             dct[k] = v
 
     # Having this iterator lets dict(msg_obj) work out of the box.
     def __iter__(self):
         return iter(self.__dict__.iteritems())
 
     def __repr__(self):
         return repr(self.__dict__)
 
     def __str__(self):
         return pprint.pformat(self.__dict__)
 
     def __contains__(self, k):
         return k in self.__dict__
 
     def __getitem__(self, k):
         return self.__dict__[k]
 
 
 def msg_header(msg_id, msg_type, username, session):
     date = datetime.now()
     return locals()
 
 def extract_header(msg_or_header):
     """Given a message or header, return the header."""
     if not msg_or_header:
         return {}
     try:
         # See if msg_or_header is the entire message.
         h = msg_or_header['header']
     except KeyError:
         try:
             # See if msg_or_header is just the header
             h = msg_or_header['msg_id']
         except KeyError:
             raise
         else:
             h = msg_or_header
     if not isinstance(h, dict):
         h = dict(h)
     return h
 
 class Session(Configurable):
     """Object for handling serialization and sending of messages.
 
     The Session object handles building messages and sending them
     with ZMQ sockets or ZMQStream objects.  Objects can communicate with each
     other over the network via Session objects, and only need to work with the
     dict-based IPython message spec. The Session will handle
     serialization/deserialization, security, and metadata.
 
     Sessions support configurable serialiization via packer/unpacker traits,
     and signing with HMAC digests via the key/keyfile traits.
 
     Parameters
     ----------
 
     debug : bool
         whether to trigger extra debugging statements
     packer/unpacker : str : 'json', 'pickle' or import_string
         importstrings for methods to serialize message parts.  If just
         'json' or 'pickle', predefined JSON and pickle packers will be used.
         Otherwise, the entire importstring must be used.
 
         The functions must accept at least valid JSON input, and output *bytes*.
 
         For example, to use msgpack:
         packer = 'msgpack.packb', unpacker='msgpack.unpackb'
     pack/unpack : callables
         You can also set the pack/unpack callables for serialization directly.
     session : bytes
         the ID of this Session object.  The default is to generate a new UUID.
     username : unicode
         username added to message headers.  The default is to ask the OS.
     key : bytes
         The key used to initialize an HMAC signature.  If unset, messages
         will not be signed or checked.
     keyfile : filepath
         The file containing a key.  If this is set, `key` will be initialized
         to the contents of the file.
 
     """
 
     debug=Bool(False, config=True, help="""Debug output in the Session""")
 
     packer = DottedObjectName('json',config=True,
             help="""The name of the packer for serializing messages.
             Should be one of 'json', 'pickle', or an import name
             for a custom callable serializer.""")
     def _packer_changed(self, name, old, new):
         if new.lower() == 'json':
             self.pack = json_packer
             self.unpack = json_unpacker
             self.unpacker = new
         elif new.lower() == 'pickle':
             self.pack = pickle_packer
             self.unpack = pickle_unpacker
             self.unpacker = new
         else:
             self.pack = import_item(str(new))
 
     unpacker = DottedObjectName('json', config=True,
         help="""The name of the unpacker for unserializing messages.
         Only used with custom functions for `packer`.""")
     def _unpacker_changed(self, name, old, new):
         if new.lower() == 'json':
             self.pack = json_packer
             self.unpack = json_unpacker
             self.packer = new
         elif new.lower() == 'pickle':
             self.pack = pickle_packer
             self.unpack = pickle_unpacker
             self.packer = new
         else:
             self.unpack = import_item(str(new))
 
     session = CUnicode(u'', config=True,
         help="""The UUID identifying this session.""")
     def _session_default(self):
         u = unicode(uuid.uuid4())
         self.bsession = u.encode('ascii')
         return u
 
     def _session_changed(self, name, old, new):
         self.bsession = self.session.encode('ascii')
 
     # bsession is the session as bytes
     bsession = CBytes(b'')
 
     username = Unicode(str_to_unicode(os.environ.get('USER', 'username')),
         help="""Username for the Session. Default is your system username.""",
         config=True)
 
     metadata = Dict({}, config=True,
         help="""Metadata dictionary, which serves as the default top-level metadata dict for each message.""")
 
     # message signature related traits:
     
     key = CBytes(b'', config=True,
         help="""execution key, for extra authentication.""")
     def _key_changed(self, name, old, new):
         if new:
             self.auth = hmac.HMAC(new)
         else:
             self.auth = None
     
     auth = Instance(hmac.HMAC)
     
     digest_history = Set()
     digest_history_size = Integer(2**16, config=True,
         help="""The maximum number of digests to remember.
         
         The digest history will be culled when it exceeds this value.
         """
     )
 
     keyfile = Unicode('', config=True,
         help="""path to file containing execution key.""")
     def _keyfile_changed(self, name, old, new):
         with open(new, 'rb') as f:
             self.key = f.read().strip()
 
     # for protecting against sends from forks
     pid = Integer()
     
     # serialization traits:
     
     pack = Any(default_packer) # the actual packer function
     def _pack_changed(self, name, old, new):
         if not callable(new):
             raise TypeError("packer must be callable, not %s"%type(new))
 
     unpack = Any(default_unpacker) # the actual packer function
     def _unpack_changed(self, name, old, new):
         # unpacker is not checked - it is assumed to be
         if not callable(new):
             raise TypeError("unpacker must be callable, not %s"%type(new))
     
     # thresholds:
     copy_threshold = Integer(2**16, config=True,
         help="Threshold (in bytes) beyond which a buffer should be sent without copying.")
     buffer_threshold = Integer(MAX_BYTES, config=True,
         help="Threshold (in bytes) beyond which an object's buffer should be extracted to avoid pickling.")
     item_threshold = Integer(MAX_ITEMS, config=True,
         help="""The maximum number of items for a container to be introspected for custom serialization.
         Containers larger than this are pickled outright.
         """
     )
 
     
     def __init__(self, **kwargs):
         """create a Session object
 
         Parameters
         ----------
 
         debug : bool
             whether to trigger extra debugging statements
         packer/unpacker : str : 'json', 'pickle' or import_string
             importstrings for methods to serialize message parts.  If just
             'json' or 'pickle', predefined JSON and pickle packers will be used.
             Otherwise, the entire importstring must be used.
 
             The functions must accept at least valid JSON input, and output
             *bytes*.
 
             For example, to use msgpack:
             packer = 'msgpack.packb', unpacker='msgpack.unpackb'
         pack/unpack : callables
             You can also set the pack/unpack callables for serialization
             directly.
         session : unicode (must be ascii)
             the ID of this Session object.  The default is to generate a new
             UUID.
         bsession : bytes
             The session as bytes
         username : unicode
             username added to message headers.  The default is to ask the OS.
         key : bytes
             The key used to initialize an HMAC signature.  If unset, messages
             will not be signed or checked.
         keyfile : filepath
             The file containing a key.  If this is set, `key` will be
             initialized to the contents of the file.
         """
         super(Session, self).__init__(**kwargs)
         self._check_packers()
         self.none = self.pack({})
         # ensure self._session_default() if necessary, so bsession is defined:
         self.session
         self.pid = os.getpid()
 
     @property
     def msg_id(self):
         """always return new uuid"""
         return str(uuid.uuid4())
 
     def _check_packers(self):
         """check packers for binary data and datetime support."""
         pack = self.pack
         unpack = self.unpack
 
         # check simple serialization
         msg = dict(a=[1,'hi'])
         try:
             packed = pack(msg)
         except Exception:
             raise ValueError("packer could not serialize a simple message")
 
         # ensure packed message is bytes
         if not isinstance(packed, bytes):
             raise ValueError("message packed to %r, but bytes are required"%type(packed))
 
         # check that unpack is pack's inverse
         try:
             unpacked = unpack(packed)
         except Exception:
             raise ValueError("unpacker could not handle the packer's output")
 
         # check datetime support
         msg = dict(t=datetime.now())
         try:
             unpacked = unpack(pack(msg))
         except Exception:
             self.pack = lambda o: pack(squash_dates(o))
             self.unpack = lambda s: extract_dates(unpack(s))
 
     def msg_header(self, msg_type):
         return msg_header(self.msg_id, msg_type, self.username, self.session)
 
     def msg(self, msg_type, content=None, parent=None, header=None, metadata=None):
         """Return the nested message dict.
 
         This format is different from what is sent over the wire. The
         serialize/unserialize methods converts this nested message dict to the wire
         format, which is a list of message parts.
         """
         msg = {}
         header = self.msg_header(msg_type) if header is None else header
         msg['header'] = header
         msg['msg_id'] = header['msg_id']
         msg['msg_type'] = header['msg_type']
         msg['parent_header'] = {} if parent is None else extract_header(parent)
         msg['content'] = {} if content is None else content
         msg['metadata'] = self.metadata.copy()
         if metadata is not None:
             msg['metadata'].update(metadata)
         return msg
 
     def sign(self, msg_list):
         """Sign a message with HMAC digest. If no auth, return b''.
 
         Parameters
         ----------
         msg_list : list
             The [p_header,p_parent,p_content] part of the message list.
         """
         if self.auth is None:
             return b''
         h = self.auth.copy()
         for m in msg_list:
             h.update(m)
         return str_to_bytes(h.hexdigest())
 
     def serialize(self, msg, ident=None):
         """Serialize the message components to bytes.
 
         This is roughly the inverse of unserialize. The serialize/unserialize
         methods work with full message lists, whereas pack/unpack work with
         the individual message parts in the message list.
 
         Parameters
         ----------
         msg : dict or Message
             The nexted message dict as returned by the self.msg method.
 
         Returns
         -------
         msg_list : list
             The list of bytes objects to be sent with the format:
             [ident1,ident2,...,DELIM,HMAC,p_header,p_parent,p_metadata,p_content,
              buffer1,buffer2,...]. In this list, the p_* entities are
             the packed or serialized versions, so if JSON is used, these
             are utf8 encoded JSON strings.
         """
         content = msg.get('content', {})
         if content is None:
             content = self.none
         elif isinstance(content, dict):
             content = self.pack(content)
         elif isinstance(content, bytes):
             # content is already packed, as in a relayed message
             pass
         elif isinstance(content, unicode):
             # should be bytes, but JSON often spits out unicode
             content = content.encode('utf8')
         else:
             raise TypeError("Content incorrect type: %s"%type(content))
 
         real_message = [self.pack(msg['header']),
                         self.pack(msg['parent_header']),
                         self.pack(msg['metadata']),
                         content,
         ]
 
         to_send = []
 
         if isinstance(ident, list):
             # accept list of idents
             to_send.extend(ident)
         elif ident is not None:
             to_send.append(ident)
         to_send.append(DELIM)
 
         signature = self.sign(real_message)
         to_send.append(signature)
 
         to_send.extend(real_message)
 
         return to_send
 
     def send(self, stream, msg_or_type, content=None, parent=None, ident=None,
              buffers=None, track=False, header=None, metadata=None):
         """Build and send a message via stream or socket.
 
         The message format used by this function internally is as follows:
 
         [ident1,ident2,...,DELIM,HMAC,p_header,p_parent,p_content,
          buffer1,buffer2,...]
 
         The serialize/unserialize methods convert the nested message dict into this
         format.
 
         Parameters
         ----------
 
         stream : zmq.Socket or ZMQStream
             The socket-like object used to send the data.
         msg_or_type : str or Message/dict
             Normally, msg_or_type will be a msg_type unless a message is being
             sent more than once. If a header is supplied, this can be set to
             None and the msg_type will be pulled from the header.
 
         content : dict or None
             The content of the message (ignored if msg_or_type is a message).
         header : dict or None
             The header dict for the message (ignored if msg_to_type is a message).
         parent : Message or dict or None
             The parent or parent header describing the parent of this message
             (ignored if msg_or_type is a message).
         ident : bytes or list of bytes
             The zmq.IDENTITY routing path.
         metadata : dict or None
             The metadata describing the message
         buffers : list or None
             The already-serialized buffers to be appended to the message.
         track : bool
             Whether to track.  Only for use with Sockets, because ZMQStream
             objects cannot track messages.
             
 
         Returns
         -------
         msg : dict
             The constructed message.
         """
         if not isinstance(stream, zmq.Socket):
             # ZMQStreams and dummy sockets do not support tracking.
             track = False
 
         if isinstance(msg_or_type, (Message, dict)):
             # We got a Message or message dict, not a msg_type so don't
             # build a new Message.
             msg = msg_or_type
         else:
             msg = self.msg(msg_or_type, content=content, parent=parent,
                            header=header, metadata=metadata)
         if not os.getpid() == self.pid:
             io.rprint("WARNING: attempted to send message from fork")
             io.rprint(msg)
             return
         buffers = [] if buffers is None else buffers
         to_send = self.serialize(msg, ident)
         to_send.extend(buffers)
         longest = max([ len(s) for s in to_send ])
         copy = (longest < self.copy_threshold)
         
         if buffers and track and not copy:
             # only really track when we are doing zero-copy buffers
             tracker = stream.send_multipart(to_send, copy=False, track=True)
         else:
             # use dummy tracker, which will be done immediately
             tracker = DONE
             stream.send_multipart(to_send, copy=copy)
 
         if self.debug:
             pprint.pprint(msg)
             pprint.pprint(to_send)
             pprint.pprint(buffers)
 
         msg['tracker'] = tracker
 
         return msg
 
     def send_raw(self, stream, msg_list, flags=0, copy=True, ident=None):
         """Send a raw message via ident path.
 
         This method is used to send a already serialized message.
 
         Parameters
         ----------
         stream : ZMQStream or Socket
             The ZMQ stream or socket to use for sending the message.
         msg_list : list
             The serialized list of messages to send. This only includes the
             [p_header,p_parent,p_metadata,p_content,buffer1,buffer2,...] portion of
             the message.
         ident : ident or list
             A single ident or a list of idents to use in sending.
         """
         to_send = []
         if isinstance(ident, bytes):
             ident = [ident]
         if ident is not None:
             to_send.extend(ident)
 
         to_send.append(DELIM)
         to_send.append(self.sign(msg_list))
         to_send.extend(msg_list)
         stream.send_multipart(msg_list, flags, copy=copy)
 
     def recv(self, socket, mode=zmq.NOBLOCK, content=True, copy=True):
         """Receive and unpack a message.
 
         Parameters
         ----------
         socket : ZMQStream or Socket
             The socket or stream to use in receiving.
 
         Returns
         -------
         [idents], msg
             [idents] is a list of idents and msg is a nested message dict of
             same format as self.msg returns.
         """
         if isinstance(socket, ZMQStream):
             socket = socket.socket
         try:
             msg_list = socket.recv_multipart(mode, copy=copy)
         except zmq.ZMQError as e:
             if e.errno == zmq.EAGAIN:
                 # We can convert EAGAIN to None as we know in this case
                 # recv_multipart won't return None.
                 return None,None
             else:
                 raise
         # split multipart message into identity list and message dict
         # invalid large messages can cause very expensive string comparisons
         idents, msg_list = self.feed_identities(msg_list, copy)
         try:
             return idents, self.unserialize(msg_list, content=content, copy=copy)
         except Exception as e:
             # TODO: handle it
             raise e
 
     def feed_identities(self, msg_list, copy=True):
         """Split the identities from the rest of the message.
 
         Feed until DELIM is reached, then return the prefix as idents and
         remainder as msg_list. This is easily broken by setting an IDENT to DELIM,
         but that would be silly.
 
         Parameters
         ----------
         msg_list : a list of Message or bytes objects
             The message to be split.
         copy : bool
             flag determining whether the arguments are bytes or Messages
 
         Returns
         -------
         (idents, msg_list) : two lists
             idents will always be a list of bytes, each of which is a ZMQ
             identity. msg_list will be a list of bytes or zmq.Messages of the
             form [HMAC,p_header,p_parent,p_content,buffer1,buffer2,...] and
             should be unpackable/unserializable via self.unserialize at this
             point.
         """
         if copy:
             idx = msg_list.index(DELIM)
             return msg_list[:idx], msg_list[idx+1:]
         else:
             failed = True
             for idx,m in enumerate(msg_list):
                 if m.bytes == DELIM:
                     failed = False
                     break
             if failed:
                 raise ValueError("DELIM not in msg_list")
             idents, msg_list = msg_list[:idx], msg_list[idx+1:]
             return [m.bytes for m in idents], msg_list
 
     def _add_digest(self, signature):
         """add a digest to history to protect against replay attacks"""
         if self.digest_history_size == 0:
             # no history, never add digests
             return
         
         self.digest_history.add(signature)
         if len(self.digest_history) > self.digest_history_size:
             # threshold reached, cull 10%
             self._cull_digest_history()
     
     def _cull_digest_history(self):
         """cull the digest history
         
         Removes a randomly selected 10% of the digest history
         """
         current = len(self.digest_history)
         n_to_cull = max(int(current // 10), current - self.digest_history_size)
         if n_to_cull >= current:
             self.digest_history = set()
             return
         to_cull = random.sample(self.digest_history, n_to_cull)
         self.digest_history.difference_update(to_cull)
     
     def unserialize(self, msg_list, content=True, copy=True):
         """Unserialize a msg_list to a nested message dict.
 
         This is roughly the inverse of serialize. The serialize/unserialize
         methods work with full message lists, whereas pack/unpack work with
         the individual message parts in the message list.
 
         Parameters:
         -----------
         msg_list : list of bytes or Message objects
             The list of message parts of the form [HMAC,p_header,p_parent,
             p_metadata,p_content,buffer1,buffer2,...].
         content : bool (True)
             Whether to unpack the content dict (True), or leave it packed
             (False).
         copy : bool (True)
             Whether to return the bytes (True), or the non-copying Message
             object in each place (False).
 
         Returns
         -------
         msg : dict
             The nested message dict with top-level keys [header, parent_header,
             content, buffers].
         """
         minlen = 5
         message = {}
         if not copy:
             for i in range(minlen):
                 msg_list[i] = msg_list[i].bytes
         if self.auth is not None:
             signature = msg_list[0]
             if not signature:
                 raise ValueError("Unsigned Message")
             if signature in self.digest_history:
                 raise ValueError("Duplicate Signature: %r" % signature)
             self._add_digest(signature)
             check = self.sign(msg_list[1:5])
             if not signature == check:
                 raise ValueError("Invalid Signature: %r" % signature)
         if not len(msg_list) >= minlen:
             raise TypeError("malformed message, must have at least %i elements"%minlen)
         header = self.unpack(msg_list[1])
         message['header'] = header
         message['msg_id'] = header['msg_id']
         message['msg_type'] = header['msg_type']
         message['parent_header'] = self.unpack(msg_list[2])
         message['metadata'] = self.unpack(msg_list[3])
         if content:
             message['content'] = self.unpack(msg_list[4])
         else:
             message['content'] = msg_list[4]
 
         message['buffers'] = msg_list[5:]
         return message
 
 def test_msg2obj():
     am = dict(x=1)
     ao = Message(am)
     assert ao.x == am['x']
 
     am['y'] = dict(z=1)
     ao = Message(am)
     assert ao.y.z == am['y']['z']
 
     k1, k2 = 'y', 'z'
     assert ao[k1][k2] == am[k1][k2]
 
     am2 = dict(ao)
     assert am['x'] == am2['x']
     assert am['y']['z'] == am2['y']['z']
 

IPython/kernel/zmq/zmqshell.py

@@ -1,597 +1,597 b''
 """A ZMQ-based subclass of InteractiveShell.
 
 This code is meant to ease the refactoring of the base InteractiveShell into
 something with a cleaner architecture for 2-process use, without actually
 breaking InteractiveShell itself.  So we're doing something a bit ugly, where
 we subclass and override what we want to fix.  Once this is working well, we
 can go back to the base class and refactor the code for a cleaner inheritance
 implementation that doesn't rely on so much monkeypatching.
 
 But this lets us maintain a fully working IPython as we develop the new
 machinery.  This should thus be thought of as scaffolding.
 """
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib
 import os
 import sys
 import time
 
 # System library imports
 from zmq.eventloop import ioloop
 
 # Our own
 from IPython.core.interactiveshell import (
     InteractiveShell, InteractiveShellABC
 )
 from IPython.core import page
 from IPython.core.autocall import ZMQExitAutocall
 from IPython.core.displaypub import DisplayPublisher
 from IPython.core.error import UsageError
 from IPython.core.magics import MacroToEdit, CodeMagics
 from IPython.core.magic import magics_class, line_magic, Magics
 from IPython.core.payloadpage import install_payload_page
 from IPython.display import display, Javascript
 from IPython.kernel.inprocess.socket import SocketABC
 from IPython.kernel import (
     get_connection_file, get_connection_info, connect_qtconsole
 )
 from IPython.testing.skipdoctest import skip_doctest
-from IPython.utils import io, openpy
+from IPython.utils import openpy
 from IPython.utils.jsonutil import json_clean, encode_images
 from IPython.utils.process import arg_split
 from IPython.utils import py3compat
 from IPython.utils.traitlets import Instance, Type, Dict, CBool, CBytes
-from IPython.utils.warn import warn, error
+from IPython.utils.warn import error
 from IPython.kernel.zmq.displayhook import ZMQShellDisplayHook
 from IPython.kernel.zmq.datapub import ZMQDataPublisher
 from IPython.kernel.zmq.session import extract_header
 from session import Session
 
 #-----------------------------------------------------------------------------
 # Functions and classes
 #-----------------------------------------------------------------------------
 
 class ZMQDisplayPublisher(DisplayPublisher):
     """A display publisher that publishes data using a ZeroMQ PUB socket."""
 
     session = Instance(Session)
     pub_socket = Instance(SocketABC)
     parent_header = Dict({})
     topic = CBytes(b'displaypub')
 
     def set_parent(self, parent):
         """Set the parent for outbound messages."""
         self.parent_header = extract_header(parent)
     
     def _flush_streams(self):
         """flush IO Streams prior to display"""
         sys.stdout.flush()
         sys.stderr.flush()
 
     def publish(self, source, data, metadata=None):
         self._flush_streams()
         if metadata is None:
             metadata = {}
         self._validate_data(source, data, metadata)
         content = {}
         content['source'] = source
         content['data'] = encode_images(data)
         content['metadata'] = metadata
         self.session.send(
             self.pub_socket, u'display_data', json_clean(content),
             parent=self.parent_header, ident=self.topic,
         )
 
     def clear_output(self, stdout=True, stderr=True, other=True):
         content = dict(stdout=stdout, stderr=stderr, other=other)
         
         if stdout:
             print('\r', file=sys.stdout, end='')
         if stderr:
             print('\r', file=sys.stderr, end='')
         
         self._flush_streams()
         
         self.session.send(
             self.pub_socket, u'clear_output', content,
             parent=self.parent_header, ident=self.topic,
         )
 
 @magics_class
 class KernelMagics(Magics):
     #------------------------------------------------------------------------
     # Magic overrides
     #------------------------------------------------------------------------
     # Once the base class stops inheriting from magic, this code needs to be
     # moved into a separate machinery as well.  For now, at least isolate here
     # the magics which this class needs to implement differently from the base
     # class, or that are unique to it.
 
     @line_magic
     def doctest_mode(self, parameter_s=''):
         """Toggle doctest mode on and off.
 
         This mode is intended to make IPython behave as much as possible like a
         plain Python shell, from the perspective of how its prompts, exceptions
         and output look.  This makes it easy to copy and paste parts of a
         session into doctests.  It does so by:
 
         - Changing the prompts to the classic ``>>>`` ones.
         - Changing the exception reporting mode to 'Plain'.
         - Disabling pretty-printing of output.
 
         Note that IPython also supports the pasting of code snippets that have
         leading '>>>' and '...' prompts in them.  This means that you can paste
         doctests from files or docstrings (even if they have leading
         whitespace), and the code will execute correctly.  You can then use
         '%history -t' to see the translated history; this will give you the
         input after removal of all the leading prompts and whitespace, which
         can be pasted back into an editor.
 
         With these features, you can switch into this mode easily whenever you
         need to do testing and changes to doctests, without having to leave
         your existing IPython session.
         """
 
         from IPython.utils.ipstruct import Struct
 
         # Shorthands
         shell = self.shell
         disp_formatter = self.shell.display_formatter
         ptformatter = disp_formatter.formatters['text/plain']
         # dstore is a data store kept in the instance metadata bag to track any
         # changes we make, so we can undo them later.
         dstore = shell.meta.setdefault('doctest_mode', Struct())
         save_dstore = dstore.setdefault
 
         # save a few values we'll need to recover later
         mode = save_dstore('mode', False)
         save_dstore('rc_pprint', ptformatter.pprint)
         save_dstore('rc_active_types',disp_formatter.active_types)
         save_dstore('xmode', shell.InteractiveTB.mode)
 
         if mode == False:
             # turn on
             ptformatter.pprint = False
             disp_formatter.active_types = ['text/plain']
             shell.magic('xmode Plain')
         else:
             # turn off
             ptformatter.pprint = dstore.rc_pprint
             disp_formatter.active_types = dstore.rc_active_types
             shell.magic("xmode " + dstore.xmode)
 
         # Store new mode and inform on console
         dstore.mode = bool(1-int(mode))
         mode_label = ['OFF','ON'][dstore.mode]
         print('Doctest mode is:', mode_label)
 
         # Send the payload back so that clients can modify their prompt display
         payload = dict(
             source='IPython.kernel.zmq.zmqshell.ZMQInteractiveShell.doctest_mode',
             mode=dstore.mode)
         shell.payload_manager.write_payload(payload)
         
     
     _find_edit_target = CodeMagics._find_edit_target
 
     @skip_doctest
     @line_magic
     def edit(self, parameter_s='', last_call=['','']):
         """Bring up an editor and execute the resulting code.
 
         Usage:
           %edit [options] [args]
 
         %edit runs an external text editor. You will need to set the command for
         this editor via the ``TerminalInteractiveShell.editor`` option in your
         configuration file before it will work.
 
         This command allows you to conveniently edit multi-line code right in
         your IPython session.
 
         If called without arguments, %edit opens up an empty editor with a
         temporary file and will execute the contents of this file when you
         close it (don't forget to save it!).
 
 
         Options:
 
         -n <number>: open the editor at a specified line number.  By default,
         the IPython editor hook uses the unix syntax 'editor +N filename', but
         you can configure this by providing your own modified hook if your
         favorite editor supports line-number specifications with a different
         syntax.
 
         -p: this will call the editor with the same data as the previous time
         it was used, regardless of how long ago (in your current session) it
         was.
 
         -r: use 'raw' input.  This option only applies to input taken from the
         user's history.  By default, the 'processed' history is used, so that
         magics are loaded in their transformed version to valid Python.  If
         this option is given, the raw input as typed as the command line is
         used instead.  When you exit the editor, it will be executed by
         IPython's own processor.
 
         -x: do not execute the edited code immediately upon exit. This is
         mainly useful if you are editing programs which need to be called with
         command line arguments, which you can then do using %run.
 
 
         Arguments:
 
         If arguments are given, the following possibilites exist:
 
         - The arguments are numbers or pairs of colon-separated numbers (like
         1 4:8 9). These are interpreted as lines of previous input to be
         loaded into the editor. The syntax is the same of the %macro command.
 
         - If the argument doesn't start with a number, it is evaluated as a
         variable and its contents loaded into the editor. You can thus edit
         any string which contains python code (including the result of
         previous edits).
 
         - If the argument is the name of an object (other than a string),
         IPython will try to locate the file where it was defined and open the
         editor at the point where it is defined. You can use `%edit function`
         to load an editor exactly at the point where 'function' is defined,
         edit it and have the file be executed automatically.
 
         If the object is a macro (see %macro for details), this opens up your
         specified editor with a temporary file containing the macro's data.
         Upon exit, the macro is reloaded with the contents of the file.
 
         Note: opening at an exact line is only supported under Unix, and some
         editors (like kedit and gedit up to Gnome 2.8) do not understand the
         '+NUMBER' parameter necessary for this feature. Good editors like
         (X)Emacs, vi, jed, pico and joe all do.
 
         - If the argument is not found as a variable, IPython will look for a
         file with that name (adding .py if necessary) and load it into the
         editor. It will execute its contents with execfile() when you exit,
         loading any code in the file into your interactive namespace.
 
         After executing your code, %edit will return as output the code you
         typed in the editor (except when it was an existing file). This way
         you can reload the code in further invocations of %edit as a variable,
         via _<NUMBER> or Out[<NUMBER>], where <NUMBER> is the prompt number of
         the output.
 
         Note that %edit is also available through the alias %ed.
 
         This is an example of creating a simple function inside the editor and
         then modifying it. First, start up the editor:
 
         In [1]: ed
         Editing... done. Executing edited code...
         Out[1]: 'def foo():n    print "foo() was defined in an editing session"n'
 
         We can then call the function foo():
 
         In [2]: foo()
         foo() was defined in an editing session
 
         Now we edit foo.  IPython automatically loads the editor with the
         (temporary) file where foo() was previously defined:
 
         In [3]: ed foo
         Editing... done. Executing edited code...
 
         And if we call foo() again we get the modified version:
 
         In [4]: foo()
         foo() has now been changed!
 
         Here is an example of how to edit a code snippet successive
         times. First we call the editor:
 
         In [5]: ed
         Editing... done. Executing edited code...
         hello
         Out[5]: "print 'hello'n"
 
         Now we call it again with the previous output (stored in _):
 
         In [6]: ed _
         Editing... done. Executing edited code...
         hello world
         Out[6]: "print 'hello world'n"
 
         Now we call it with the output #8 (stored in _8, also as Out[8]):
 
         In [7]: ed _8
         Editing... done. Executing edited code...
         hello again
         Out[7]: "print 'hello again'n"
         """
 
         opts,args = self.parse_options(parameter_s,'prn:')
 
         try:
             filename, lineno, _ = CodeMagics._find_edit_target(self.shell, args, opts, last_call)
         except MacroToEdit as e:
             # TODO: Implement macro editing over 2 processes.
             print("Macro editing not yet implemented in 2-process model.")
             return
 
         # Make sure we send to the client an absolute path, in case the working
         # directory of client and kernel don't match
         filename = os.path.abspath(filename)
 
         payload = {
             'source' : 'IPython.kernel.zmq.zmqshell.ZMQInteractiveShell.edit_magic',
             'filename' : filename,
             'line_number' : lineno
         }
         self.shell.payload_manager.write_payload(payload)
 
     # A few magics that are adapted to the specifics of using pexpect and a
     # remote terminal
 
     @line_magic
     def clear(self, arg_s):
         """Clear the terminal."""
         if os.name == 'posix':
             self.shell.system("clear")
         else:
             self.shell.system("cls")
 
     if os.name == 'nt':
         # This is the usual name in windows
         cls = line_magic('cls')(clear)
 
     # Terminal pagers won't work over pexpect, but we do have our own pager
 
     @line_magic
     def less(self, arg_s):
         """Show a file through the pager.
 
         Files ending in .py are syntax-highlighted."""
         if not arg_s:
             raise UsageError('Missing filename.')
 
         cont = open(arg_s).read()
         if arg_s.endswith('.py'):
             cont = self.shell.pycolorize(openpy.read_py_file(arg_s, skip_encoding_cookie=False))
         else:
             cont = open(arg_s).read()
         page.page(cont)
 
     more = line_magic('more')(less)
 
     # Man calls a pager, so we also need to redefine it
     if os.name == 'posix':
         @line_magic
         def man(self, arg_s):
             """Find the man page for the given command and display in pager."""
             page.page(self.shell.getoutput('man %s | col -b' % arg_s,
                                            split=False))
 
     @line_magic
     def connect_info(self, arg_s):
         """Print information for connecting other clients to this kernel
         
         It will print the contents of this session's connection file, as well as
         shortcuts for local clients.
         
         In the simplest case, when called from the most recently launched kernel,
         secondary clients can be connected, simply with:
         
         $> ipython <app> --existing
         
         """
         
         from IPython.core.application import BaseIPythonApplication as BaseIPApp
         
         if BaseIPApp.initialized():
             app = BaseIPApp.instance()
             security_dir = app.profile_dir.security_dir
             profile = app.profile
         else:
             profile = 'default'
             security_dir = ''
         
         try:
             connection_file = get_connection_file()
             info = get_connection_info(unpack=False)
         except Exception as e:
             error("Could not get connection info: %r" % e)
             return
         
         # add profile flag for non-default profile
         profile_flag = "--profile %s" % profile if profile != 'default' else ""
         
         # if it's in the security dir, truncate to basename
         if security_dir == os.path.dirname(connection_file):
             connection_file = os.path.basename(connection_file)
         
         
         print (info + '\n')
         print ("Paste the above JSON into a file, and connect with:\n"
             "    $> ipython <app> --existing <file>\n"
             "or, if you are local, you can connect with just:\n"
             "    $> ipython <app> --existing {0} {1}\n"
             "or even just:\n"
             "    $> ipython <app> --existing {1}\n"
             "if this is the most recent IPython session you have started.".format(
             connection_file, profile_flag
             )
         )
 
     @line_magic
     def qtconsole(self, arg_s):
         """Open a qtconsole connected to this kernel.
         
         Useful for connecting a qtconsole to running notebooks, for better
         debugging.
         """
         
         # %qtconsole should imply bind_kernel for engines:
         try:
             from IPython.parallel import bind_kernel
         except ImportError:
             # technically possible, because parallel has higher pyzmq min-version
             pass
         else:
             bind_kernel()
         
         try:
             p = connect_qtconsole(argv=arg_split(arg_s, os.name=='posix'))
         except Exception as e:
             error("Could not start qtconsole: %r" % e)
             return
     
     @line_magic
     def autosave(self, arg_s):
         """Set the autosave interval in the notebook (in seconds).
         
         The default value is 120, or two minutes.
         ``%autosave 0`` will disable autosave.
         
         This magic only has an effect when called from the notebook interface.
         It has no effect when called in a startup file.
         """
         
         try:
             interval = int(arg_s)
         except ValueError:
             raise UsageError("%%autosave requires an integer, got %r" % arg_s)
         
         # javascript wants milliseconds
         milliseconds = 1000 * interval
         display(Javascript("IPython.notebook.set_autosave_interval(%i)" % milliseconds),
             include=['application/javascript']
         )
         if interval:
             print("Autosaving every %i seconds" % interval)
         else:
             print("Autosave disabled")
 
 
 class ZMQInteractiveShell(InteractiveShell):
     """A subclass of InteractiveShell for ZMQ."""
 
     displayhook_class = Type(ZMQShellDisplayHook)
     display_pub_class = Type(ZMQDisplayPublisher)
     data_pub_class = Type(ZMQDataPublisher)
 
     # Override the traitlet in the parent class, because there's no point using
     # readline for the kernel. Can be removed when the readline code is moved
     # to the terminal frontend.
     colors_force = CBool(True)
     readline_use = CBool(False)
     # autoindent has no meaning in a zmqshell, and attempting to enable it
     # will print a warning in the absence of readline.
     autoindent = CBool(False)
 
     exiter = Instance(ZMQExitAutocall)
     def _exiter_default(self):
         return ZMQExitAutocall(self)
     
     def _exit_now_changed(self, name, old, new):
         """stop eventloop when exit_now fires"""
         if new:
             loop = ioloop.IOLoop.instance()
             loop.add_timeout(time.time()+0.1, loop.stop)
 
     keepkernel_on_exit = None
 
     # Over ZeroMQ, GUI control isn't done with PyOS_InputHook as there is no
     # interactive input being read; we provide event loop support in ipkernel
     from .eventloops import enable_gui
     enable_gui = staticmethod(enable_gui)
 
     def init_environment(self):
         """Configure the user's environment.
 
         """
         env = os.environ
         # These two ensure 'ls' produces nice coloring on BSD-derived systems
         env['TERM'] = 'xterm-color'
         env['CLICOLOR'] = '1'
         # Since normal pagers don't work at all (over pexpect we don't have
         # single-key control of the subprocess), try to disable paging in
         # subprocesses as much as possible.
         env['PAGER'] = 'cat'
         env['GIT_PAGER'] = 'cat'
         
         # And install the payload version of page.
         install_payload_page()
 
     def auto_rewrite_input(self, cmd):
         """Called to show the auto-rewritten input for autocall and friends.
 
         FIXME: this payload is currently not correctly processed by the
         frontend.
         """
         new = self.prompt_manager.render('rewrite') + cmd
         payload = dict(
             source='IPython.kernel.zmq.zmqshell.ZMQInteractiveShell.auto_rewrite_input',
             transformed_input=new,
             )
         self.payload_manager.write_payload(payload)
 
     def ask_exit(self):
         """Engage the exit actions."""
         self.exit_now = True
         payload = dict(
             source='IPython.kernel.zmq.zmqshell.ZMQInteractiveShell.ask_exit',
             exit=True,
             keepkernel=self.keepkernel_on_exit,
             )
         self.payload_manager.write_payload(payload)
 
     def _showtraceback(self, etype, evalue, stb):
 
         exc_content = {
             u'traceback' : stb,
             u'ename' : unicode(etype.__name__),
             u'evalue' : py3compat.safe_unicode(evalue),
         }
 
         dh = self.displayhook
         # Send exception info over pub socket for other clients than the caller
         # to pick up
         topic = None
         if dh.topic:
             topic = dh.topic.replace(b'pyout', b'pyerr')
         
         exc_msg = dh.session.send(dh.pub_socket, u'pyerr', json_clean(exc_content), dh.parent_header, ident=topic)
 
         # FIXME - Hack: store exception info in shell object.  Right now, the
         # caller is reading this info after the fact, we need to fix this logic
         # to remove this hack.  Even uglier, we need to store the error status
         # here, because in the main loop, the logic that sets it is being
         # skipped because runlines swallows the exceptions.
         exc_content[u'status'] = u'error'
         self._reply_content = exc_content
         # /FIXME
 
         return exc_content
 
     def set_next_input(self, text):
         """Send the specified text to the frontend to be presented at the next
         input cell."""
         payload = dict(
             source='IPython.kernel.zmq.zmqshell.ZMQInteractiveShell.set_next_input',
             text=text
         )
         self.payload_manager.write_payload(payload)
     
     #-------------------------------------------------------------------------
     # Things related to magics
     #-------------------------------------------------------------------------
 
     def init_magics(self):
         super(ZMQInteractiveShell, self).init_magics()
         self.register_magics(KernelMagics)
         self.magics_manager.register_alias('ed', 'edit')
 
 
 
 InteractiveShellABC.register(ZMQInteractiveShell)

IPython/utils/pickleutil.py

@@ -1,353 +1,354 b''
 # encoding: utf-8
 
 """Pickle related utilities. Perhaps this should be called 'can'."""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-------------------------------------------------------------------------------
 
 #-------------------------------------------------------------------------------
 # Imports
 #-------------------------------------------------------------------------------
 
 import copy
 import logging
 import sys
 from types import FunctionType
 
 try:
     import cPickle as pickle
 except ImportError:
     import pickle
 
 try:
     import numpy
 except:
     numpy = None
 
+import codeutil  # This registers a hook when it's imported
 import py3compat
 from importstring import import_item
 
 from IPython.config import Application
 
 if py3compat.PY3:
     buffer = memoryview
     class_type = type
 else:
     from types import ClassType
     class_type = (type, ClassType)
 
 #-------------------------------------------------------------------------------
 # Classes
 #-------------------------------------------------------------------------------
 
 
 class CannedObject(object):
     def __init__(self, obj, keys=[], hook=None):
         """can an object for safe pickling
         
         Parameters
         ==========
         
         obj:
             The object to be canned
         keys: list (optional)
             list of attribute names that will be explicitly canned / uncanned
         hook: callable (optional)
             An optional extra callable,
             which can do additional processing of the uncanned object.
         
         large data may be offloaded into the buffers list,
         used for zero-copy transfers.
         """
         self.keys = keys
         self.obj = copy.copy(obj)
         self.hook = can(hook)
         for key in keys:
             setattr(self.obj, key, can(getattr(obj, key)))
         
         self.buffers = []
 
     def get_object(self, g=None):
         if g is None:
             g = {}
         obj = self.obj
         for key in self.keys:
             setattr(obj, key, uncan(getattr(obj, key), g))
         
         if self.hook:
             self.hook = uncan(self.hook, g)
             self.hook(obj, g)
         return self.obj
     
 
 class Reference(CannedObject):
     """object for wrapping a remote reference by name."""
     def __init__(self, name):
         if not isinstance(name, basestring):
             raise TypeError("illegal name: %r"%name)
         self.name = name
         self.buffers = []
 
     def __repr__(self):
         return "<Reference: %r>"%self.name
 
     def get_object(self, g=None):
         if g is None:
             g = {}
         
         return eval(self.name, g)
 
 
 class CannedFunction(CannedObject):
 
     def __init__(self, f):
         self._check_type(f)
         self.code = f.func_code
         if f.func_defaults:
             self.defaults = [ can(fd) for fd in f.func_defaults ]
         else:
             self.defaults = None
         self.module = f.__module__ or '__main__'
         self.__name__ = f.__name__
         self.buffers = []
 
     def _check_type(self, obj):
         assert isinstance(obj, FunctionType), "Not a function type"
 
     def get_object(self, g=None):
         # try to load function back into its module:
         if not self.module.startswith('__'):
             __import__(self.module)
             g = sys.modules[self.module].__dict__
 
         if g is None:
             g = {}
         if self.defaults:
             defaults = tuple(uncan(cfd, g) for cfd in self.defaults)
         else:
             defaults = None
         newFunc = FunctionType(self.code, g, self.__name__, defaults)
         return newFunc
 
 class CannedClass(CannedObject):
 
     def __init__(self, cls):
         self._check_type(cls)
         self.name = cls.__name__
         self.old_style = not isinstance(cls, type)
         self._canned_dict = {}
         for k,v in cls.__dict__.items():
             if k not in ('__weakref__', '__dict__'):
                 self._canned_dict[k] = can(v)
         if self.old_style:
             mro = []
         else:
             mro = cls.mro()
         
         self.parents = [ can(c) for c in mro[1:] ]
         self.buffers = []
 
     def _check_type(self, obj):
         assert isinstance(obj, class_type), "Not a class type"
 
     def get_object(self, g=None):
         parents = tuple(uncan(p, g) for p in self.parents)
         return type(self.name, parents, uncan_dict(self._canned_dict, g=g))
 
 class CannedArray(CannedObject):
     def __init__(self, obj):
         self.shape = obj.shape
         self.dtype = obj.dtype.descr if obj.dtype.fields else obj.dtype.str
         if sum(obj.shape) == 0:
             # just pickle it
             self.buffers = [pickle.dumps(obj, -1)]
         else:
             # ensure contiguous
             obj = numpy.ascontiguousarray(obj, dtype=None)
             self.buffers = [buffer(obj)]
     
     def get_object(self, g=None):
         data = self.buffers[0]
         if sum(self.shape) == 0:
             # no shape, we just pickled it
             return pickle.loads(data)
         else:
             return numpy.frombuffer(data, dtype=self.dtype).reshape(self.shape)
 
 
 class CannedBytes(CannedObject):
     wrap = bytes
     def __init__(self, obj):
         self.buffers = [obj]
     
     def get_object(self, g=None):
         data = self.buffers[0]
         return self.wrap(data)
 
 def CannedBuffer(CannedBytes):
     wrap = buffer
 
 #-------------------------------------------------------------------------------
 # Functions
 #-------------------------------------------------------------------------------
 
 def _logger():
     """get the logger for the current Application
     
     the root logger will be used if no Application is running
     """
     if Application.initialized():
         logger = Application.instance().log
     else:
         logger = logging.getLogger()
         if not logger.handlers:
             logging.basicConfig()
     
     return logger
 
 def _import_mapping(mapping, original=None):
     """import any string-keys in a type mapping
     
     """
     log = _logger()
     log.debug("Importing canning map")
     for key,value in mapping.items():
         if isinstance(key, basestring):
             try:
                 cls = import_item(key)
             except Exception:
                 if original and key not in original:
                     # only message on user-added classes
                     log.error("cannning class not importable: %r", key, exc_info=True)
                 mapping.pop(key)
             else:
                 mapping[cls] = mapping.pop(key)
 
 def istype(obj, check):
     """like isinstance(obj, check), but strict
     
     This won't catch subclasses.
     """
     if isinstance(check, tuple):
         for cls in check:
             if type(obj) is cls:
                 return True
         return False
     else:
         return type(obj) is check
 
 def can(obj):
     """prepare an object for pickling"""
     
     import_needed = False
     
     for cls,canner in can_map.iteritems():
         if isinstance(cls, basestring):
             import_needed = True
             break
         elif istype(obj, cls):
             return canner(obj)
     
     if import_needed:
         # perform can_map imports, then try again
         # this will usually only happen once
         _import_mapping(can_map, _original_can_map)
         return can(obj)
     
     return obj
 
 def can_class(obj):
     if isinstance(obj, class_type) and obj.__module__ == '__main__':
         return CannedClass(obj)
     else:
         return obj
 
 def can_dict(obj):
     """can the *values* of a dict"""
     if istype(obj, dict):
         newobj = {}
         for k, v in obj.iteritems():
             newobj[k] = can(v)
         return newobj
     else:
         return obj
 
 sequence_types = (list, tuple, set)
 
 def can_sequence(obj):
     """can the elements of a sequence"""
     if istype(obj, sequence_types):
         t = type(obj)
         return t([can(i) for i in obj])
     else:
         return obj
 
 def uncan(obj, g=None):
     """invert canning"""
     
     import_needed = False
     for cls,uncanner in uncan_map.iteritems():
         if isinstance(cls, basestring):
             import_needed = True
             break
         elif isinstance(obj, cls):
             return uncanner(obj, g)
     
     if import_needed:
         # perform uncan_map imports, then try again
         # this will usually only happen once
         _import_mapping(uncan_map, _original_uncan_map)
         return uncan(obj, g)
     
     return obj
 
 def uncan_dict(obj, g=None):
     if istype(obj, dict):
         newobj = {}
         for k, v in obj.iteritems():
             newobj[k] = uncan(v,g)
         return newobj
     else:
         return obj
 
 def uncan_sequence(obj, g=None):
     if istype(obj, sequence_types):
         t = type(obj)
         return t([uncan(i,g) for i in obj])
     else:
         return obj
 
 def _uncan_dependent_hook(dep, g=None):
     dep.check_dependency()
     
 def can_dependent(obj):
     return CannedObject(obj, keys=('f', 'df'), hook=_uncan_dependent_hook)
 
 #-------------------------------------------------------------------------------
 # API dictionaries
 #-------------------------------------------------------------------------------
 
 # These dicts can be extended for custom serialization of new objects
 
 can_map = {
     'IPython.parallel.dependent' : can_dependent,
     'numpy.ndarray' : CannedArray,
     FunctionType : CannedFunction,
     bytes : CannedBytes,
     buffer : CannedBuffer,
     class_type : can_class,
 }
 
 uncan_map = {
     CannedObject : lambda obj, g: obj.get_object(g),
 }
 
 # for use in _import_mapping:
 _original_can_map = can_map.copy()
 _original_uncan_map = uncan_map.copy()