client.py
563 lines
| 18.7 KiB
| text/x-python
|
PythonLexer
MinRK
|
r10288 | """ Defines a KernelClient that provides signals and slots. | ||
epatters
|
r2609 | """ | ||
Thomas Kluyver
|
r19209 | import atexit | ||
import errno | ||||
from threading import Thread | ||||
import time | ||||
import zmq | ||||
# import ZMQError in top-level namespace, to avoid ugly attribute-error messages | ||||
# during garbage collection of threads at exit: | ||||
from zmq import ZMQError | ||||
from zmq.eventloop import ioloop, zmqstream | ||||
from IPython.external.qt import QtCore | ||||
epatters
|
r2609 | |||
MinRK
|
r10288 | # Local imports | ||
Brian Granger
|
r2786 | from IPython.utils.traitlets import Type | ||
Thomas Kluyver
|
r19211 | from IPython.kernel.channels import HBChannel,\ | ||
make_shell_socket, make_iopub_socket, make_stdin_socket | ||||
MinRK
|
r10288 | from IPython.kernel import KernelClient | ||
epatters
|
r2609 | |||
Thomas Kluyver
|
r19209 | from .kernel_mixins import (QtHBChannelMixin, QtKernelClientMixin) | ||
from .util import SuperQObject | ||||
Brian Granger
|
r2695 | |||
Thomas Kluyver
|
r19209 | class QtHBChannel(QtHBChannelMixin, HBChannel): | ||
epatters
|
r8408 | pass | ||
Brian Granger
|
r3277 | |||
Thomas Kluyver
|
r19209 | from IPython.core.release import kernel_protocol_version_info | ||
epatters
|
r2611 | |||
Thomas Kluyver
|
r19209 | from IPython.kernel.channelsabc import ( | ||
ShellChannelABC, IOPubChannelABC, StdInChannelABC, | ||||
) | ||||
from IPython.utils.py3compat import string_types, iteritems | ||||
MinRK
|
r3090 | |||
Thomas Kluyver
|
r19209 | major_protocol_version = kernel_protocol_version_info[0] | ||
class InvalidPortNumber(Exception): | ||||
epatters
|
r8408 | pass | ||
Bernardo B. Marques
|
r4872 | |||
Thomas Kluyver
|
r19209 | # some utilities to validate message structure, these might get moved elsewhere | ||
# if they prove to have more generic utility | ||||
def validate_string_dict(dct): | ||||
"""Validate that the input is a dict with string keys and values. | ||||
Raises ValueError if not.""" | ||||
for k,v in iteritems(dct): | ||||
if not isinstance(k, string_types): | ||||
raise ValueError('key %r in dict must be a string' % k) | ||||
if not isinstance(v, string_types): | ||||
raise ValueError('value %r in dict must be a string' % v) | ||||
class QtZMQSocketChannel(SuperQObject, Thread): | ||||
"""The base class for the channels that use ZMQ sockets.""" | ||||
context = None | ||||
session = None | ||||
socket = None | ||||
ioloop = None | ||||
stream = None | ||||
_address = None | ||||
_exiting = False | ||||
proxy_methods = [] | ||||
# Emitted when the channel is started. | ||||
started = QtCore.Signal() | ||||
# Emitted when the channel is stopped. | ||||
stopped = QtCore.Signal() | ||||
message_received = QtCore.Signal(object) | ||||
#--------------------------------------------------------------------------- | ||||
# InProcessChannel interface | ||||
#--------------------------------------------------------------------------- | ||||
def call_handlers_later(self, *args, **kwds): | ||||
""" Call the message handlers later. | ||||
""" | ||||
do_later = lambda: self.call_handlers(*args, **kwds) | ||||
QtCore.QTimer.singleShot(0, do_later) | ||||
def process_events(self): | ||||
""" Process any pending GUI events. | ||||
""" | ||||
QtCore.QCoreApplication.instance().processEvents() | ||||
def __init__(self, context, session, address): | ||||
"""Create a channel. | ||||
Parameters | ||||
---------- | ||||
context : :class:`zmq.Context` | ||||
The ZMQ context to use. | ||||
session : :class:`session.Session` | ||||
The session to use. | ||||
address : zmq url | ||||
Standard (ip, port) tuple that the kernel is listening on. | ||||
""" | ||||
super(QtZMQSocketChannel, self).__init__() | ||||
self.daemon = True | ||||
self.context = context | ||||
self.session = session | ||||
if isinstance(address, tuple): | ||||
if address[1] == 0: | ||||
message = 'The port number for a channel cannot be 0.' | ||||
raise InvalidPortNumber(message) | ||||
address = "tcp://%s:%i" % address | ||||
self._address = address | ||||
atexit.register(self._notice_exit) | ||||
def _notice_exit(self): | ||||
self._exiting = True | ||||
def _run_loop(self): | ||||
"""Run my loop, ignoring EINTR events in the poller""" | ||||
while True: | ||||
try: | ||||
self.ioloop.start() | ||||
except ZMQError as e: | ||||
if e.errno == errno.EINTR: | ||||
continue | ||||
else: | ||||
raise | ||||
except Exception: | ||||
if self._exiting: | ||||
break | ||||
else: | ||||
raise | ||||
else: | ||||
break | ||||
def start(self): | ||||
""" Reimplemented to emit signal. | ||||
""" | ||||
super(QtZMQSocketChannel, self).start() | ||||
self.started.emit() | ||||
def stop(self): | ||||
"""Stop the channel's event loop and join its thread. | ||||
This calls :meth:`~threading.Thread.join` and returns when the thread | ||||
terminates. :class:`RuntimeError` will be raised if | ||||
:meth:`~threading.Thread.start` is called again. | ||||
""" | ||||
if self.ioloop is not None: | ||||
self.ioloop.stop() | ||||
self.join() | ||||
self.close() | ||||
self.stopped.emit() | ||||
def close(self): | ||||
if self.ioloop is not None: | ||||
try: | ||||
self.ioloop.close(all_fds=True) | ||||
except Exception: | ||||
pass | ||||
if self.socket is not None: | ||||
try: | ||||
self.socket.close(linger=0) | ||||
except Exception: | ||||
pass | ||||
self.socket = None | ||||
@property | ||||
def address(self): | ||||
"""Get the channel's address as a zmq url string. | ||||
These URLS have the form: 'tcp://127.0.0.1:5555'. | ||||
""" | ||||
return self._address | ||||
def _queue_send(self, msg): | ||||
"""Queue a message to be sent from the IOLoop's thread. | ||||
Parameters | ||||
---------- | ||||
msg : message to send | ||||
This is threadsafe, as it uses IOLoop.add_callback to give the loop's | ||||
thread control of the action. | ||||
""" | ||||
def thread_send(): | ||||
self.session.send(self.stream, msg) | ||||
self.ioloop.add_callback(thread_send) | ||||
def _handle_recv(self, msg): | ||||
"""Callback for stream.on_recv. | ||||
Unpacks message, and calls handlers with it. | ||||
""" | ||||
ident,smsg = self.session.feed_identities(msg) | ||||
msg = self.session.deserialize(smsg) | ||||
self.call_handlers(msg) | ||||
def call_handlers(self, msg): | ||||
"""This method is called in the ioloop thread when a message arrives. | ||||
Subclasses should override this method to handle incoming messages. | ||||
It is important to remember that this method is called in the thread | ||||
so that some logic must be done to ensure that the application level | ||||
handlers are called in the application thread. | ||||
""" | ||||
# Emit the generic signal. | ||||
self.message_received.emit(msg) | ||||
class QtShellChannel(QtZMQSocketChannel): | ||||
"""The shell channel for issuing request/replies to the kernel.""" | ||||
command_queue = None | ||||
# flag for whether execute requests should be allowed to call raw_input: | ||||
allow_stdin = True | ||||
proxy_methods = [ | ||||
'execute', | ||||
'complete', | ||||
'inspect', | ||||
'history', | ||||
'kernel_info', | ||||
'shutdown', | ||||
'is_complete', | ||||
] | ||||
# Emitted when a reply has been received for the corresponding request type. | ||||
execute_reply = QtCore.Signal(object) | ||||
complete_reply = QtCore.Signal(object) | ||||
inspect_reply = QtCore.Signal(object) | ||||
history_reply = QtCore.Signal(object) | ||||
kernel_info_reply = QtCore.Signal(object) | ||||
def __init__(self, context, session, address): | ||||
super(QtShellChannel, self).__init__(context, session, address) | ||||
self.ioloop = ioloop.IOLoop() | ||||
def run(self): | ||||
"""The thread's main activity. Call start() instead.""" | ||||
Thomas Kluyver
|
r19211 | self.socket = make_shell_socket(self.context, self.session.bsession, self.address) | ||
Thomas Kluyver
|
r19209 | self.stream = zmqstream.ZMQStream(self.socket, self.ioloop) | ||
self.stream.on_recv(self._handle_recv) | ||||
self._run_loop() | ||||
def call_handlers(self, msg): | ||||
super(QtShellChannel, self).call_handlers(msg) | ||||
# Catch kernel_info_reply for message spec adaptation | ||||
msg_type = msg['header']['msg_type'] | ||||
if msg_type == 'kernel_info_reply': | ||||
self._handle_kernel_info_reply(msg) | ||||
# Emit specific signals | ||||
signal = getattr(self, msg_type, None) | ||||
if signal: | ||||
signal.emit(msg) | ||||
def execute(self, code, silent=False, store_history=True, | ||||
user_expressions=None, allow_stdin=None): | ||||
"""Execute code in the kernel. | ||||
Parameters | ||||
---------- | ||||
code : str | ||||
A string of Python code. | ||||
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. | ||||
Returns | ||||
------- | ||||
The msg_id of the message sent. | ||||
""" | ||||
if user_expressions is None: | ||||
user_expressions = {} | ||||
if allow_stdin is None: | ||||
allow_stdin = self.allow_stdin | ||||
# Don't waste network traffic if inputs are invalid | ||||
if not isinstance(code, string_types): | ||||
raise ValueError('code %r must be a string' % code) | ||||
validate_string_dict(user_expressions) | ||||
# Create class for content/msg creation. Related to, but possibly | ||||
# not in Session. | ||||
content = dict(code=code, silent=silent, store_history=store_history, | ||||
user_expressions=user_expressions, | ||||
allow_stdin=allow_stdin, | ||||
) | ||||
msg = self.session.msg('execute_request', content) | ||||
self._queue_send(msg) | ||||
return msg['header']['msg_id'] | ||||
def complete(self, code, cursor_pos=None): | ||||
"""Tab complete text in the kernel's namespace. | ||||
Parameters | ||||
---------- | ||||
code : str | ||||
The context in which completion is requested. | ||||
Can be anything between a variable name and an entire cell. | ||||
cursor_pos : int, optional | ||||
The position of the cursor in the block of code where the completion was requested. | ||||
Default: ``len(code)`` | ||||
Returns | ||||
------- | ||||
The msg_id of the message sent. | ||||
""" | ||||
if cursor_pos is None: | ||||
cursor_pos = len(code) | ||||
content = dict(code=code, cursor_pos=cursor_pos) | ||||
msg = self.session.msg('complete_request', content) | ||||
self._queue_send(msg) | ||||
return msg['header']['msg_id'] | ||||
def inspect(self, code, cursor_pos=None, detail_level=0): | ||||
"""Get metadata information about an object in the kernel's namespace. | ||||
It is up to the kernel to determine the appropriate object to inspect. | ||||
Parameters | ||||
---------- | ||||
code : str | ||||
The context in which info is requested. | ||||
Can be anything between a variable name and an entire cell. | ||||
cursor_pos : int, optional | ||||
The position of the cursor in the block of code where the info was requested. | ||||
Default: ``len(code)`` | ||||
detail_level : int, optional | ||||
The level of detail for the introspection (0-2) | ||||
Returns | ||||
------- | ||||
The msg_id of the message sent. | ||||
""" | ||||
if cursor_pos is None: | ||||
cursor_pos = len(code) | ||||
content = dict(code=code, cursor_pos=cursor_pos, | ||||
detail_level=detail_level, | ||||
) | ||||
msg = self.session.msg('inspect_request', content) | ||||
self._queue_send(msg) | ||||
return msg['header']['msg_id'] | ||||
def history(self, raw=True, output=False, hist_access_type='range', **kwargs): | ||||
"""Get entries from the kernel's history list. | ||||
Parameters | ||||
---------- | ||||
raw : bool | ||||
If True, return the raw input. | ||||
output : bool | ||||
If True, then return the output as well. | ||||
hist_access_type : str | ||||
'range' (fill in session, start and stop params), 'tail' (fill in n) | ||||
or 'search' (fill in pattern param). | ||||
session : int | ||||
For a range request, the session from which to get lines. Session | ||||
numbers are positive integers; negative ones count back from the | ||||
current session. | ||||
start : int | ||||
The first line number of a history range. | ||||
stop : int | ||||
The final (excluded) line number of a history range. | ||||
n : int | ||||
The number of lines of history to get for a tail request. | ||||
pattern : str | ||||
The glob-syntax pattern for a search request. | ||||
Returns | ||||
------- | ||||
The msg_id of the message sent. | ||||
""" | ||||
content = dict(raw=raw, output=output, hist_access_type=hist_access_type, | ||||
**kwargs) | ||||
msg = self.session.msg('history_request', content) | ||||
self._queue_send(msg) | ||||
return msg['header']['msg_id'] | ||||
def kernel_info(self): | ||||
"""Request kernel info.""" | ||||
msg = self.session.msg('kernel_info_request') | ||||
self._queue_send(msg) | ||||
return msg['header']['msg_id'] | ||||
def _handle_kernel_info_reply(self, msg): | ||||
"""handle kernel info reply | ||||
sets protocol adaptation version | ||||
""" | ||||
adapt_version = int(msg['content']['protocol_version'].split('.')[0]) | ||||
if adapt_version != major_protocol_version: | ||||
self.session.adapt_version = adapt_version | ||||
def shutdown(self, restart=False): | ||||
"""Request an immediate kernel shutdown. | ||||
Upon receipt of the (empty) reply, client code can safely assume that | ||||
the kernel has shut down and it's safe to forcefully terminate it if | ||||
it's still alive. | ||||
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. | ||||
""" | ||||
# Send quit message to kernel. Once we implement kernel-side setattr, | ||||
# this should probably be done that way, but for now this will do. | ||||
msg = self.session.msg('shutdown_request', {'restart':restart}) | ||||
self._queue_send(msg) | ||||
return msg['header']['msg_id'] | ||||
def is_complete(self, code): | ||||
msg = self.session.msg('is_complete_request', {'code': code}) | ||||
self._queue_send(msg) | ||||
return msg['header']['msg_id'] | ||||
class QtIOPubChannel(QtZMQSocketChannel): | ||||
"""The iopub channel which listens for messages that the kernel publishes. | ||||
This channel is where all output is published to frontends. | ||||
""" | ||||
# Emitted when a message of type 'stream' is received. | ||||
stream_received = QtCore.Signal(object) | ||||
# Emitted when a message of type 'execute_input' is received. | ||||
execute_input_received = QtCore.Signal(object) | ||||
# Emitted when a message of type 'execute_result' is received. | ||||
execute_result_received = QtCore.Signal(object) | ||||
# Emitted when a message of type 'error' is received. | ||||
error_received = QtCore.Signal(object) | ||||
# Emitted when a message of type 'display_data' is received | ||||
display_data_received = QtCore.Signal(object) | ||||
# Emitted when a crash report message is received from the kernel's | ||||
# last-resort sys.excepthook. | ||||
crash_received = QtCore.Signal(object) | ||||
# Emitted when a shutdown is noticed. | ||||
shutdown_reply_received = QtCore.Signal(object) | ||||
def __init__(self, context, session, address): | ||||
super(QtIOPubChannel, self).__init__(context, session, address) | ||||
self.ioloop = ioloop.IOLoop() | ||||
def run(self): | ||||
"""The thread's main activity. Call start() instead.""" | ||||
Thomas Kluyver
|
r19211 | self.socket = make_iopub_socket(self.context, self.session.bsession, self.address) | ||
Thomas Kluyver
|
r19209 | self.stream = zmqstream.ZMQStream(self.socket, self.ioloop) | ||
self.stream.on_recv(self._handle_recv) | ||||
self._run_loop() | ||||
def call_handlers(self, msg): | ||||
super(QtIOPubChannel, self).call_handlers(msg) | ||||
# Emit signals for specialized message types. | ||||
msg_type = msg['header']['msg_type'] | ||||
signal = getattr(self, msg_type + '_received', None) | ||||
if signal: | ||||
signal.emit(msg) | ||||
def flush(self, timeout=1.0): | ||||
"""Immediately processes all pending messages on the iopub channel. | ||||
Callers should use this method to ensure that :meth:`call_handlers` | ||||
has been called for all messages that have been received on the | ||||
0MQ SUB socket of this channel. | ||||
This method is thread safe. | ||||
Parameters | ||||
---------- | ||||
timeout : float, optional | ||||
The maximum amount of time to spend flushing, in seconds. The | ||||
default is one second. | ||||
""" | ||||
# We do the IOLoop callback process twice to ensure that the IOLoop | ||||
# gets to perform at least one full poll. | ||||
stop_time = time.time() + timeout | ||||
for i in range(2): | ||||
self._flushed = False | ||||
self.ioloop.add_callback(self._flush) | ||||
while not self._flushed and time.time() < stop_time: | ||||
time.sleep(0.01) | ||||
def _flush(self): | ||||
"""Callback for :method:`self.flush`.""" | ||||
self.stream.flush() | ||||
self._flushed = True | ||||
class QtStdInChannel(QtZMQSocketChannel): | ||||
"""The stdin channel to handle raw_input requests that the kernel makes.""" | ||||
msg_queue = None | ||||
proxy_methods = ['input'] | ||||
# Emitted when an input request is received. | ||||
input_requested = QtCore.Signal(object) | ||||
def __init__(self, context, session, address): | ||||
super(QtStdInChannel, self).__init__(context, session, address) | ||||
self.ioloop = ioloop.IOLoop() | ||||
def run(self): | ||||
"""The thread's main activity. Call start() instead.""" | ||||
Thomas Kluyver
|
r19211 | self.socket = make_stdin_socket(self.context, self.session.bsession, self.address) | ||
Thomas Kluyver
|
r19209 | self.stream = zmqstream.ZMQStream(self.socket, self.ioloop) | ||
self.stream.on_recv(self._handle_recv) | ||||
self._run_loop() | ||||
def call_handlers(self, msg): | ||||
super(QtStdInChannel, self).call_handlers(msg) | ||||
# Emit signals for specialized message types. | ||||
msg_type = msg['header']['msg_type'] | ||||
if msg_type == 'input_request': | ||||
self.input_requested.emit(msg) | ||||
def input(self, string): | ||||
"""Send a string of raw input to the kernel.""" | ||||
content = dict(value=string) | ||||
msg = self.session.msg('input_reply', content) | ||||
self._queue_send(msg) | ||||
ShellChannelABC.register(QtShellChannel) | ||||
IOPubChannelABC.register(QtIOPubChannel) | ||||
StdInChannelABC.register(QtStdInChannel) | ||||
epatters
|
r2609 | |||
MinRK
|
r10288 | class QtKernelClient(QtKernelClientMixin, KernelClient): | ||
""" A KernelClient that provides signals and slots. | ||||
epatters
|
r2611 | """ | ||
Brian Granger
|
r9120 | iopub_channel_class = Type(QtIOPubChannel) | ||
shell_channel_class = Type(QtShellChannel) | ||||
stdin_channel_class = Type(QtStdInChannel) | ||||
hb_channel_class = Type(QtHBChannel) | ||||