Skip to content

Instantly share code, notes, and snippets.

@ChrisBeaumont
Created August 30, 2012 16:18
Show Gist options
  • Save ChrisBeaumont/3532163 to your computer and use it in GitHub Desktop.
Save ChrisBeaumont/3532163 to your computer and use it in GitHub Desktop.
In-process IPython Qt Widget. Take 2
"""
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()
@kimia-cvengineer
Copy link

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.

from IPython.zmq.ipkernel import IPKernelApp
ModuleNotFoundError: No module named 'IPython.zmq'

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment