Merge pull request #185 from minrk/KernelClient.run

add BlockingKernelClient.execute_interactive

.travis.yml

@@ -13,6 +13,6 @@ install:
  - pip install -f travis-wheels/wheelhouse -e .[test] codecov
  - python -c 'import ipykernel.kernelspec; ipykernel.kernelspec.install(user=True)'
 script:
- - nosetests --with-coverage --cover-package jupyter_client jupyter_client
+ - nosetests -v --with-coverage --cover-package jupyter_client jupyter_client
 after_success:
  - codecov

jupyter_client/blocking/client.py

@@ -5,17 +5,69 @@
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
 
+from __future__ import print_function
+
+from functools import partial
+from getpass import getpass
 try:
  from queue import Empty # Python 3
 except ImportError:
  from Queue import Empty # Python 2
+import sys
 import time
 
+import zmq
+
 from traitlets import Type
 from jupyter_client.channels import HBChannel
 from jupyter_client.client import KernelClient
 from .channels import ZMQSocketChannel
 
+try:
+ monotonic = time.monotonic
+except AttributeError:
+ # py2
+ monotonic = time.time # close enough
+
+try:
+ TimeoutError
+except NameError:
+ # py2
+ TimeoutError = RuntimeError
+
+
+def reqrep(meth):
+ def wrapped(self, *args, **kwargs):
+ reply = kwargs.pop('reply', False)
+ timeout = kwargs.pop('timeout', None)
+ msg_id = meth(self, *args, **kwargs)
+ if not reply:
+ return msg_id
+
+ return self._recv_reply(msg_id, timeout=timeout)
+
+ basedoc, _ = meth.__doc__.split('Returns\n', 1)
+ parts = [basedoc.strip()]
+ if 'Parameters' not in basedoc:
+ parts.append("""
+ Parameters
+ ----------
+ """)
+ parts.append("""
+ reply: bool (default: False)
+ Whether to wait for and return reply
+ timeout: float or None (default: None)
+ Timeout to use when waiting for a reply
+
+ Returns
+ -------
+ msg_id: str
+ The msg_id of the request sent, if reply=False (default)
+ reply: dict
+ The reply message for this request, if reply=True
+ """)
+ wrapped.__doc__ = '\n'.join(parts)
+ return wrapped
 
 class BlockingKernelClient(KernelClient):
  """A BlockingKernelClient """
@@ -74,3 +126,212 @@ def wait_for_ready(self, timeout=None):
  iopub_channel_class = Type(ZMQSocketChannel)
  stdin_channel_class = Type(ZMQSocketChannel)
  hb_channel_class = Type(HBChannel)
+
+
+ def _recv_reply(self, msg_id, timeout=None):
+ """Receive and return the reply for a given request"""
+ if timeout is not None:
+ deadline = monotonic() + timeout
+ while True:
+ if timeout is not None:
+ timeout = max(0, deadline - monotonic())
+ try:
+ reply = self.get_shell_msg(timeout=timeout)
+ except Empty:
+ raise TimeoutError("Timeout waiting for reply")
+ if reply['parent_header'].get('msg_id') != msg_id:
+ # not my reply, someone may have forgotten to retrieve theirs
+ continue
+ return reply
+
+
+ execute = reqrep(KernelClient.execute)
+ history = reqrep(KernelClient.history)
+ complete = reqrep(KernelClient.complete)
+ inspect = reqrep(KernelClient.inspect)
+ kernel_info = reqrep(KernelClient.kernel_info)
+ comm_info = reqrep(KernelClient.comm_info)
+ shutdown = reqrep(KernelClient.shutdown)
+
+
+ def _stdin_hook_default(self, msg):
+ """Handle an input request"""
+ content = msg['content']
+ if content.get('password', False):
+ prompt = getpass
+ elif sys.version_info < (3,):
+ prompt = raw_input
+ else:
+ prompt = input
+
+ try:
+ raw_data = prompt(content["prompt"])
+ except EOFError:
+ # turn EOFError into EOF character
+ raw_data = '\x04'
+ except KeyboardInterrupt:
+ sys.stdout.write('\n')
+ return
+
+ # only send stdin reply if there *was not* another request
+ # or execution finished while we were reading.
+ if not (self.stdin_channel.msg_ready() or self.shell_channel.msg_ready()):
+ self.input(raw_data)
+
+ def _output_hook_default(self, msg):
+ """Default hook for redisplaying plain-text output"""
+ msg_type = msg['header']['msg_type']
+ content = msg['content']
+ if msg_type == 'stream':
+ stream = getattr(sys, content['name'])
+ stream.write(content['text'])
+ elif msg_type in ('display_data', 'execute_result'):
+ sys.stdout.write(content['data'].get('text/plain', ''))
+ elif msg_type == 'error':
+ print('\n'.join(content['traceback']), file=sys.stderr)
+
+ def _output_hook_kernel(self, session, socket, parent_header, msg):
+ """Output hook when running inside an IPython kernel
+
+ adds rich output support.
+ """
+ msg_type = msg['header']['msg_type']
+ if msg_type in ('display_data', 'execute_result', 'error'):
+ session.send(socket, msg_type, msg['content'], parent=parent_header)
+ else:
+ self._output_hook_default(msg)
+
+ def execute_interactive(self, code, silent=False, store_history=True,
+ user_expressions=None, allow_stdin=None, stop_on_error=True,
+ timeout=None, output_hook=None, stdin_hook=None,
+ ):
+ """Execute code in the kernel interactively
+
+ Output will be redisplayed, and stdin prompts will be relayed as well.
+ If an IPython kernel is detected, rich output will be displayed.
+
+ You can pass a custom output_hook callable that will be called
+ with every IOPub message that is produced instead of the default redisplay.
+
+ Parameters
+ ----------
+ code : str
+ A string of code in the kernel's language.
+
+ silent : bool, optional (default False)
+ If set, the kernel will execute the code as quietly possible, and
+ will force store_history to be False.
+
+ store_history : bool, optional (default True)
+ If set, the kernel will store command history. This is forced
+ to be False if silent is True.
+
+ user_expressions : dict, optional
+ A dict mapping names to expressions to be evaluated in the user's
+ dict. The expression values are returned as strings formatted using
+ :func:`repr`.
+
+ allow_stdin : bool, optional (default self.allow_stdin)
+ Flag for whether the kernel can send stdin requests to frontends.
+
+ Some frontends (e.g. the Notebook) do not support stdin requests.
+ If raw_input is called from code executed from such a frontend, a
+ StdinNotImplementedError will be raised.
+
+ stop_on_error: bool, optional (default True)
+ Flag whether to abort the execution queue, if an exception is encountered.
+
+ timeout: float or None (default: None)
+ Timeout to use when waiting for a reply
+
+ output_hook: callable(msg)
+ Function to be called with output messages.
+ If not specified, output will be redisplayed.
+
+ stdin_hook: callable(msg)
+ Function to be called with stdin_request messages.
+ If not specified, input/getpass will be called.
+
+ Returns
+ -------
+ reply: dict
+ The reply message for this request
+ """
+ if not self.iopub_channel.is_alive():
+ raise RuntimeError("IOPub channel must be running to receive output")
+ if allow_stdin is None:
+ allow_stdin = self.allow_stdin
+ if allow_stdin and not self.stdin_channel.is_alive():
+ raise RuntimeError("stdin channel must be running to allow input")
+ msg_id = self.execute(code,
+ silent=silent,
+ store_history=store_history,
+ user_expressions=user_expressions,
+ allow_stdin=allow_stdin,
+ stop_on_error=stop_on_error,
+ )
+ if stdin_hook is None:
+ stdin_hook = self._stdin_hook_default
+ if output_hook is None:
+ # detect IPython kernel
+ if 'IPython' in sys.modules:
+ from IPython import get_ipython
+ ip = get_ipython()
+ in_kernel = getattr(ip, 'kernel', False)
+ if in_kernel:
+ output_hook = partial(
+ self._output_hook_kernel,
+ ip.display_pub.session,
+ ip.display_pub.pub_socket,
+ ip.display_pub.parent_header,
+ )
+ if output_hook is None:
+ # default: redisplay plain-text outputs
+ output_hook = self._output_hook_default
+
+ # set deadline based on timeout
+ if timeout is not None:
+ deadline = monotonic() + timeout
+ else:
+ timeout_ms = None
+
+ poller = zmq.Poller()
+ iopub_socket = self.iopub_channel.socket
+ poller.register(iopub_socket, zmq.POLLIN)
+ if allow_stdin:
+ stdin_socket = self.stdin_channel.socket
+ poller.register(stdin_socket, zmq.POLLIN)
+ else:
+ stdin_socket = None
+
+ # wait for output and redisplay it
+ while True:
+ if timeout is not None:
+ timeout = max(0, deadline - monotonic())
+ timeout_ms = 1e3 * timeout
+ events = dict(poller.poll(timeout_ms))
+ if not events:
+ raise TimeoutError("Timeout waiting for output")
+ if stdin_socket in events:
+ req = self.stdin_channel.get_msg(timeout=0)
+ stdin_hook(req)
+ continue
+ if iopub_socket not in events:
+ continue
+
+ msg = self.iopub_channel.get_msg(timeout=0)
+
+ if msg['parent_header'].get('msg_id') != msg_id:
+ # not from my request
+ continue
+ output_hook(msg)
+
+ # stop on idle
+ if msg['header']['msg_type'] == 'status' and \
+ msg['content']['execution_state'] == 'idle':
+ break
+
+ # output is done, get the reply
+ if timeout is not None:
+ timeout = max(0, deadline - monotonic())
+ return self._recv_reply(msg_id, timeout=timeout)

jupyter_client/client.py

@@ -338,20 +338,33 @@ def history(self, raw=True, output=False, hist_access_type='range', **kwargs):
  -------
  The ID of the message sent.
  """
+ if hist_access_type == 'range':
+ kwargs.setdefault('session', 0)
+ kwargs.setdefault('start', 0)
  content = dict(raw=raw, output=output, hist_access_type=hist_access_type,
  **kwargs)
  msg = self.session.msg('history_request', content)
  self.shell_channel.send(msg)
  return msg['header']['msg_id']
 
  def kernel_info(self):
- """Request kernel info."""
+ """Request kernel info
+
+ Returns
+ -------
+ The msg_id of the message sent
+ """
  msg = self.session.msg('kernel_info_request')
  self.shell_channel.send(msg)
  return msg['header']['msg_id']
 
  def comm_info(self, target_name=None):
- """Request comm info."""
+ """Request comm info
+ 
+ Returns
+ -------
+ The msg_id of the message sent
+ """
  if target_name is None:
  content = {}
  else:
@@ -380,6 +393,10 @@ def shutdown(self, restart=False):
  The kernel will send the reply via a function registered with Python's
  atexit module, ensuring it's truly done as the kernel is done with all
  normal operation.
+
+ Returns
+ -------
+ The msg_id of the message sent
  """
  # Send quit message to kernel. Once we implement kernel-side setattr,
  # this should probably be done that way, but for now this will do.