kernelmanager.py
243 lines
| 8.3 KiB
| text/x-python
|
PythonLexer
|
r2643 | """ Defines a KernelManager that provides signals and slots. | ||
|
r2609 | """ | ||
# System library imports. | ||||
|
r3304 | from IPython.external.qt import QtCore | ||
|
r2609 | |||
# IPython imports. | ||||
|
r2786 | from IPython.utils.traitlets import Type | ||
|
r2611 | from IPython.zmq.kernelmanager import KernelManager, SubSocketChannel, \ | ||
|
r3974 | ShellSocketChannel, StdInSocketChannel, HBSocketChannel | ||
|
r2994 | from util import MetaQObjectHasTraits, SuperQObject | ||
|
r2609 | |||
|
r2695 | |||
|
r2994 | class SocketChannelQObject(SuperQObject): | ||
# Emitted when the channel is started. | ||||
|
r3304 | started = QtCore.Signal() | ||
|
r2994 | |||
# Emitted when the channel is stopped. | ||||
|
r3304 | stopped = QtCore.Signal() | ||
|
r2994 | |||
#--------------------------------------------------------------------------- | ||||
|
r3974 | # 'ZMQSocketChannel' interface | ||
|
r2994 | #--------------------------------------------------------------------------- | ||
def start(self): | ||||
""" Reimplemented to emit signal. | ||||
""" | ||||
super(SocketChannelQObject, self).start() | ||||
self.started.emit() | ||||
def stop(self): | ||||
""" Reimplemented to emit signal. | ||||
""" | ||||
super(SocketChannelQObject, self).stop() | ||||
self.stopped.emit() | ||||
|
r3974 | class QtShellSocketChannel(SocketChannelQObject, ShellSocketChannel): | ||
|
r2994 | |||
# Emitted when any message is received. | ||||
|
r3304 | message_received = QtCore.Signal(object) | ||
|
r2994 | |||
|
r3024 | # Emitted when a reply has been received for the corresponding request | ||
# type. | ||||
|
r3304 | execute_reply = QtCore.Signal(object) | ||
complete_reply = QtCore.Signal(object) | ||||
object_info_reply = QtCore.Signal(object) | ||||
|
r3820 | history_reply = QtCore.Signal(object) | ||
|
r2994 | |||
|
r3032 | # Emitted when the first reply comes back. | ||
|
r3304 | first_reply = QtCore.Signal() | ||
|
r3024 | |||
# Used by the first_reply signal logic to determine if a reply is the | ||||
# first. | ||||
_handlers_called = False | ||||
|
r2994 | #--------------------------------------------------------------------------- | ||
|
r3974 | # 'ShellSocketChannel' interface | ||
|
r2994 | #--------------------------------------------------------------------------- | ||
def call_handlers(self, msg): | ||||
""" Reimplemented to emit signals instead of making callbacks. | ||||
""" | ||||
# Emit the generic signal. | ||||
self.message_received.emit(msg) | ||||
# Emit signals for specialized message types. | ||||
|
r4282 | msg_type = msg['header']['msg_type'] | ||
|
r2994 | signal = getattr(self, msg_type, None) | ||
if signal: | ||||
signal.emit(msg) | ||||
|
r3024 | if not self._handlers_called: | ||
self.first_reply.emit() | ||||
|
r3032 | self._handlers_called = True | ||
|
r3024 | |||
|
r3032 | #--------------------------------------------------------------------------- | ||
|
r3974 | # 'QtShellSocketChannel' interface | ||
|
r3032 | #--------------------------------------------------------------------------- | ||
|
r3024 | |||
def reset_first_reply(self): | ||||
""" Reset the first_reply signal to fire again on the next reply. | ||||
""" | ||||
self._handlers_called = False | ||||
|
r2994 | |||
class QtSubSocketChannel(SocketChannelQObject, SubSocketChannel): | ||||
|
r2609 | |||
# Emitted when any message is received. | ||||
|
r3304 | message_received = QtCore.Signal(object) | ||
|
r2609 | |||
|
r2770 | # Emitted when a message of type 'stream' is received. | ||
|
r3304 | stream_received = QtCore.Signal(object) | ||
|
r2609 | |||
|
r2770 | # Emitted when a message of type 'pyin' is received. | ||
|
r3304 | pyin_received = QtCore.Signal(object) | ||
|
r2770 | |||
# Emitted when a message of type 'pyout' is received. | ||||
|
r3304 | pyout_received = QtCore.Signal(object) | ||
|
r2770 | |||
# Emitted when a message of type 'pyerr' is received. | ||||
|
r3304 | pyerr_received = QtCore.Signal(object) | ||
|
r2770 | |||
|
r3277 | # Emitted when a message of type 'display_data' is received | ||
|
r3307 | display_data_received = QtCore.Signal(object) | ||
|
r3277 | |||
|
r2770 | # Emitted when a crash report message is received from the kernel's | ||
# last-resort sys.excepthook. | ||||
|
r3304 | crash_received = QtCore.Signal(object) | ||
|
r2611 | |||
|
r3090 | # Emitted when a shutdown is noticed. | ||
|
r3304 | shutdown_reply_received = QtCore.Signal(object) | ||
|
r3090 | |||
|
r2611 | #--------------------------------------------------------------------------- | ||
|
r2609 | # 'SubSocketChannel' interface | ||
#--------------------------------------------------------------------------- | ||||
def call_handlers(self, msg): | ||||
""" Reimplemented to emit signals instead of making callbacks. | ||||
""" | ||||
# Emit the generic signal. | ||||
self.message_received.emit(msg) | ||||
# Emit signals for specialized message types. | ||||
|
r4282 | msg_type = msg['header']['msg_type'] | ||
|
r2770 | signal = getattr(self, msg_type + '_received', None) | ||
if signal: | ||||
signal.emit(msg) | ||||
elif msg_type in ('stdout', 'stderr'): | ||||
self.stream_received.emit(msg) | ||||
|
r2609 | |||
|
r2614 | def flush(self): | ||
""" Reimplemented to ensure that signals are dispatched immediately. | ||||
""" | ||||
super(QtSubSocketChannel, self).flush() | ||||
QtCore.QCoreApplication.instance().processEvents() | ||||
|
r2609 | |||
|
r3974 | class QtStdInSocketChannel(SocketChannelQObject, StdInSocketChannel): | ||
|
r2609 | |||
|
r2707 | # Emitted when any message is received. | ||
|
r3304 | message_received = QtCore.Signal(object) | ||
|
r2707 | |||
|
r2730 | # Emitted when an input request is received. | ||
|
r3304 | input_requested = QtCore.Signal(object) | ||
|
r2707 | |||
|
r2611 | #--------------------------------------------------------------------------- | ||
|
r3974 | # 'StdInSocketChannel' interface | ||
|
r2707 | #--------------------------------------------------------------------------- | ||
def call_handlers(self, msg): | ||||
""" Reimplemented to emit signals instead of making callbacks. | ||||
""" | ||||
# Emit the generic signal. | ||||
self.message_received.emit(msg) | ||||
# Emit signals for specialized message types. | ||||
|
r4282 | msg_type = msg['header']['msg_type'] | ||
|
r2730 | if msg_type == 'input_request': | ||
self.input_requested.emit(msg) | ||||
|
r2611 | |||
|
r2770 | |||
|
r2994 | class QtHBSocketChannel(SocketChannelQObject, HBSocketChannel): | ||
|
r2910 | |||
# Emitted when the kernel has died. | ||||
|
r3304 | kernel_died = QtCore.Signal(object) | ||
|
r2910 | |||
#--------------------------------------------------------------------------- | ||||
|
r2994 | # 'HBSocketChannel' interface | ||
|
r2910 | #--------------------------------------------------------------------------- | ||
def call_handlers(self, since_last_heartbeat): | ||||
""" Reimplemented to emit signals instead of making callbacks. | ||||
""" | ||||
# Emit the generic signal. | ||||
self.kernel_died.emit(since_last_heartbeat) | ||||
|
r2994 | class QtKernelManager(KernelManager, SuperQObject): | ||
|
r2643 | """ A KernelManager that provides signals and slots. | ||
|
r2611 | """ | ||
|
r2643 | __metaclass__ = MetaQObjectHasTraits | ||
# Emitted when the kernel manager has started listening. | ||||
|
r3304 | started_channels = QtCore.Signal() | ||
|
r2643 | |||
# Emitted when the kernel manager has stopped listening. | ||||
|
r3304 | stopped_channels = QtCore.Signal() | ||
|
r2643 | |||
# Use Qt-specific channel classes that emit signals. | ||||
|
r2786 | sub_channel_class = Type(QtSubSocketChannel) | ||
|
r3974 | shell_channel_class = Type(QtShellSocketChannel) | ||
stdin_channel_class = Type(QtStdInSocketChannel) | ||||
|
r2910 | hb_channel_class = Type(QtHBSocketChannel) | ||
|
r2643 | |||
#--------------------------------------------------------------------------- | ||||
# 'KernelManager' interface | ||||
#--------------------------------------------------------------------------- | ||||
|
r3032 | |||
#------ Kernel process management ------------------------------------------ | ||||
def start_kernel(self, *args, **kw): | ||||
""" Reimplemented for proper heartbeat management. | ||||
""" | ||||
|
r3974 | if self._shell_channel is not None: | ||
self._shell_channel.reset_first_reply() | ||||
|
r3032 | super(QtKernelManager, self).start_kernel(*args, **kw) | ||
#------ Channel management ------------------------------------------------- | ||||
|
r2643 | |||
|
r2994 | def start_channels(self, *args, **kw): | ||
|
r2643 | """ Reimplemented to emit signal. | ||
""" | ||||
|
r2994 | super(QtKernelManager, self).start_channels(*args, **kw) | ||
|
r2701 | self.started_channels.emit() | ||
|
r2643 | |||
|
r2699 | def stop_channels(self): | ||
|
r2643 | """ Reimplemented to emit signal. | ||
""" | ||||
|
r2699 | super(QtKernelManager, self).stop_channels() | ||
|
r2701 | self.stopped_channels.emit() | ||
|
r3032 | |||
@property | ||||
|
r3974 | def shell_channel(self): | ||
|
r3032 | """ Reimplemented for proper heartbeat management. | ||
""" | ||||
|
r3974 | if self._shell_channel is None: | ||
self._shell_channel = super(QtKernelManager, self).shell_channel | ||||
self._shell_channel.first_reply.connect(self._first_reply) | ||||
return self._shell_channel | ||||
|
r3032 | |||
#--------------------------------------------------------------------------- | ||||
# Protected interface | ||||
#--------------------------------------------------------------------------- | ||||
def _first_reply(self): | ||||
""" Unpauses the heartbeat channel when the first reply is received on | ||||
the execute channel. Note that this will *not* start the heartbeat | ||||
channel if it is not already running! | ||||
""" | ||||
if self._hb_channel is not None: | ||||
self._hb_channel.unpause() | ||||