/usr/lib/python3/dist-packages/qtconsole/inprocess.py is in python3-qtconsole 4.3.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """ Defines an in-process KernelManager with signals and slots.
"""
from qtconsole.qt import QtCore
from ipykernel.inprocess import (
InProcessHBChannel, InProcessKernelClient, InProcessKernelManager,
)
from ipykernel.inprocess.channels import InProcessChannel
from traitlets import Type
from .util import SuperQObject
from .kernel_mixins import (
QtKernelClientMixin, QtKernelManagerMixin,
)
from .rich_jupyter_widget import RichJupyterWidget
class QtInProcessChannel(SuperQObject, InProcessChannel):
# Emitted when the channel is started.
started = QtCore.Signal()
# Emitted when the channel is stopped.
stopped = QtCore.Signal()
# Emitted when any message is received.
message_received = QtCore.Signal(object)
def start(self):
""" Reimplemented to emit signal.
"""
super(QtInProcessChannel, self).start()
self.started.emit()
def stop(self):
""" Reimplemented to emit signal.
"""
super(QtInProcessChannel, self).stop()
self.stopped.emit()
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 call_handlers(self, msg):
self.message_received.emit(msg)
def process_events(self):
""" Process any pending GUI events.
"""
QtCore.QCoreApplication.instance().processEvents()
def flush(self, timeout=1.0):
""" Reimplemented to ensure that signals are dispatched immediately.
"""
super(QtInProcessChannel, self).flush()
self.process_events()
class QtInProcessHBChannel(SuperQObject, InProcessHBChannel):
# This signal will never be fired, but it needs to exist
kernel_died = QtCore.Signal()
class QtInProcessKernelClient(QtKernelClientMixin, InProcessKernelClient):
""" An in-process KernelManager with signals and slots.
"""
iopub_channel_class = Type(QtInProcessChannel)
shell_channel_class = Type(QtInProcessChannel)
stdin_channel_class = Type(QtInProcessChannel)
hb_channel_class = Type(QtInProcessHBChannel)
class QtInProcessKernelManager(QtKernelManagerMixin, InProcessKernelManager):
client_class = __module__ + '.QtInProcessKernelClient'
class QtInProcessRichJupyterWidget(RichJupyterWidget):
""" An in-process Jupyter Widget that enables multiline editing
"""
def _is_complete(self, source, interactive=True):
shell = self.kernel_manager.kernel.shell
status, indent_spaces = \
shell.input_transformer_manager.check_complete(source)
if indent_spaces is None:
indent = ''
else:
indent = ' ' * indent_spaces
return status != 'incomplete', indent
|