Created
August 30, 2012 16:18
-
-
Save ChrisBeaumont/3532163 to your computer and use it in GitHub Desktop.
In-process IPython Qt Widget. Take 2
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
An attempt to embed an IPython Qt Terminal widget in-process | |
Current Behavior | |
---------------- | |
1) Multiple widgets can be instantiated | |
2) Each widget shares the same namespace (and underlying kernel) | |
3) stdout and stderr events generated within the widget get printed in | |
the widget | |
4) stdout, stderr events generated outisde the widget get printed to | |
the stdout or stderr of the terminal which started python | |
""" | |
import sys | |
import atexit | |
from zmq import ZMQError | |
from zmq.eventloop.zmqstream import ZMQStream | |
from IPython.zmq.ipkernel import IPKernelApp, Kernel | |
from IPython.zmq.iostream import OutStream | |
from IPython.lib.kernel import find_connection_file | |
from IPython.frontend.qt.kernelmanager import QtKernelManager | |
from IPython.frontend.qt.console.rich_ipython_widget import RichIPythonWidget | |
from PyQt4 import QtGui, QtCore | |
from contextlib import contextmanager | |
import logging | |
from zmq.eventloop import ioloop | |
@contextmanager | |
def redirect_output(session, pub_socket): | |
"""Prevent any of the widgets from permanently hijacking stdout or stderr""" | |
sys.stdout = OutStream(session, pub_socket, u'stdout') | |
sys.stderr = OutStream(session, pub_socket, u'stderr') | |
try: | |
yield | |
finally: | |
sys.stdout = sys.__stdout__ | |
sys.stderr = sys.__stderr__ | |
def non_blocking_eventloop(kernel): | |
kernel.timer = QtCore.QTimer() | |
kernel.timer.timeout.connect(kernel.do_one_iteration) | |
kernel.timer.start(1000 * kernel._poll_interval) | |
class EmbeddedQtKernel(Kernel): | |
def __init__(self, *args, **kwargs): | |
super(EmbeddedQtKernel, self).__init__(*args, **kwargs) | |
self.eventloop = non_blocking_eventloop | |
def do_one_iteration(self): | |
with redirect_output(self.session, self.iopub_socket): | |
super(EmbeddedQtKernel, self).do_one_iteration() | |
class EmbeddedQtKernelApp(IPKernelApp): | |
def init_kernel(self): | |
shell_stream = ZMQStream(self.shell_socket) | |
kernel = EmbeddedQtKernel(config=self.config, session=self.session, | |
shell_streams=[shell_stream], | |
iopub_socket=self.iopub_socket, | |
stdin_socket=self.stdin_socket, | |
log=self.log, | |
profile_dir=self.profile_dir, | |
) | |
self.kernel = kernel | |
kernel.record_ports(self.ports) | |
def start(self): | |
#handoff between IOLoop and QApplication event loops | |
loop = ioloop.IOLoop.instance() | |
stopper = ioloop.PeriodicCallback(loop.stop, 0, loop) | |
self.timer = QtCore.QTimer() | |
self.timer.timeout.connect(loop.start) | |
self.timer.start(0) | |
stopper.start() | |
super(EmbeddedQtKernelApp, self).start() | |
class EmbeddedIPythonWidget(RichIPythonWidget): | |
gui_completion = 'droplist' | |
def __init__(self, **kwargs): | |
super(EmbeddedIPythonWidget, self).__init__(**kwargs) | |
self._init_kernel_app() | |
self._init_kernel_manager() | |
self.update_namespace(kwargs) | |
def _init_kernel_app(self): | |
app = EmbeddedQtKernelApp.instance() | |
try: | |
app.initialize([]) | |
except ZMQError: | |
pass # already set up | |
try: | |
app.start() | |
except RuntimeError: # already started | |
pass | |
self.app = app | |
def _init_kernel_manager(self): | |
connection_file = find_connection_file(self.app.connection_file) | |
manager = QtKernelManager(connection_file=connection_file) | |
manager.load_connection_file() | |
manager.start_channels() | |
atexit.register(manager.cleanup_connection_file) | |
self.kernel_manager = manager | |
def update_namespace(self, ns): | |
self.app.shell.user_ns.update(ns) | |
def main(): | |
qtapp = QtGui.QApplication([]) | |
w = QtGui.QWidget() | |
layout = QtGui.QVBoxLayout() | |
button_layout = QtGui.QHBoxLayout() | |
b1 = QtGui.QPushButton("Throw an Exception") | |
b2 = QtGui.QPushButton("Print 'testing'") | |
b3 = QtGui.QPushButton("Log 'testing'") | |
b4 = QtGui.QPushButton("Sleep") | |
b5 = QtGui.QPushButton("Show a QFileDialog") | |
def print_testing(): | |
print 'testing' | |
def throw_exception(): | |
1 / 0 | |
def log_testing(): | |
logging.critical("testing") | |
def wait(): | |
from time import sleep | |
sleep(2) | |
def add_dialog(): | |
dialog = QtGui.QFileDialog(w) | |
dialog.show() | |
b1.pressed.connect(throw_exception) | |
b2.pressed.connect(print_testing) | |
b3.pressed.connect(log_testing) | |
b4.pressed.connect(wait) | |
b5.pressed.connect(add_dialog) | |
w.setLayout(layout) | |
layout.addWidget(EmbeddedIPythonWidget(testing=123)) | |
layout.addLayout(button_layout) | |
button_layout.addWidget(b1) | |
button_layout.addWidget(b2) | |
button_layout.addWidget(b3) | |
button_layout.addWidget(b4) | |
button_layout.addWidget(b5) | |
layout.addWidget(EmbeddedIPythonWidget()) | |
w.show() | |
qtapp.exec_() | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hi.
I have problem with the module IPython.zmq. I've been installed this and all related modules but I have module error.
I work with python 3.7.