在 PyQt 应用程序中嵌入 IPython Qt 控制台

Question

我想在我正在开发的 PyQt 应用程序中嵌入一个 IPython qt 控制台小部件。下面提供的代码（并改编自 https://stackoverflow.com/a/9796491/1332492）为 IPython v0.12 完成了此任务。但是，这会在 IPython v0.13 中的 self.heartbeat.start() 和 RuntimeError: threads can only be started once 行崩溃。注释掉这一行会打开小部件，但不会响应用户输入。

有谁知道如何实现 IPython v0.13 的等效功能？

"""
Adapted from
https://stackoverflow.com/a/9796491/1332492
"""
import os
import atexit

from IPython.zmq.ipkernel import IPKernelApp
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 IPython.config.application import catch_config_error
from PyQt4 import QtCore


class IPythonLocalKernelApp(IPKernelApp):
    DEFAULT_INSTANCE_ARGS = ['']

    @catch_config_error
    def initialize(self, argv=None):
        super(IPythonLocalKernelApp, self).initialize(argv)
        self.kernel.eventloop = self.loop_qt4_nonblocking

    def loop_qt4_nonblocking(self, kernel):
        """Non-blocking version of the ipython qt4 kernel loop"""
        kernel.timer = QtCore.QTimer()
        kernel.timer.timeout.connect(kernel.do_one_iteration)
        kernel.timer.start(1000*kernel._poll_interval)

    def start(self, argv=DEFAULT_INSTANCE_ARGS):
        """Starts IPython kernel app
        argv: arguments passed to kernel
        """
        self.initialize(argv)
        self.heartbeat.start()

        if self.poller is not None:
            self.poller.start()

        self.kernel.start()


class IPythonConsoleQtWidget(RichIPythonWidget):
    _connection_file = None

    def __init__(self, *args, **kw):
        RichIPythonWidget.__init__(self, *args, **kw)
        self._existing = True
        self._may_close = False
        self._confirm_exit = False

    def _init_kernel_manager(self):
        km = QtKernelManager(connection_file=self._connection_file, config=self.config)
        km.load_connection_file()
        km.start_channels(hb=self._heartbeat)
        self.kernel_manager = km
        atexit.register(self.kernel_manager.cleanup_connection_file)

    def connect_kernel(self, connection_file, heartbeat=False):
        self._heartbeat = heartbeat
        if os.path.exists(connection_file):
            self._connection_file = connection_file
        else:
            self._connection_file = find_connection_file(connection_file)

        self._init_kernel_manager()


def main(**kwargs):
    kernelapp = IPythonLocalKernelApp.instance()
    kernelapp.start()

    widget = IPythonConsoleQtWidget()
    widget.connect_kernel(connection_file=kernelapp.connection_file)
    widget.show()

    return widget

if __name__ == "__main__":
    from PyQt4.QtGui import QApplication
    app = QApplication([''])
    main()
    app.exec_()

v0.13 的回溯

RuntimeError                              Traceback (most recent call last)
/Users/beaumont/terminal.py in <module>()
     80     from PyQt4.QtGui import QApplication
     81     app = QApplication([''])
---> 82     main()
        global main = <function main at 0x106d0c848>
     83     app.exec_()

/Users/beaumont/terminal.py in main(**kwargs={})
     69 def main(**kwargs):
     70     kernelapp = IPythonLocalKernelApp.instance()
---> 71     kernelapp.start()
        kernelapp.start = <bound method IPythonLocalKernelApp.start of     <__main__.IPythonLocalKernelApp object at 0x106d10590>>
     72 
     73     widget = IPythonConsoleQtWidget()

/Users/beaumont/terminal.py in start(self=<__main__.IPythonLocalKernelApp object>, argv=[''])
     33         """
     34         self.initialize(argv)
---> 35         self.heartbeat.start()
        self.heartbeat.start = <bound method Heartbeat.start of <Heartbeat(Thread-1, started daemon 4458577920)>>
     36 
     37         if self.poller is not None:

/opt/local/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.pyc in start(self=<Heartbeat(Thread-1, started daemon 4458577920)>)
    487             raise RuntimeError("thread.__init__() not called")
    488         if self.__started.is_set():
--> 489             raise RuntimeError("threads can only be started once")
        global RuntimeError = undefined
    490         if __debug__:
    491             self._note("%s.start(): starting thread", self)

RuntimeError: threads can only be started once

Answer 1

好的，这段代码似乎可以解决问题（即，它在 Qt 小部件中放置了一个非阻塞 ipython 解释器，可以嵌入到其他小部件中）。传递给 terminal_widget 的关键字被添加到小部件的命名空间中

import atexit

from IPython.zmq.ipkernel import IPKernelApp
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 IPython.utils.traitlets import TraitError
from PyQt4 import QtGui, QtCore

def event_loop(kernel):
    kernel.timer = QtCore.QTimer()
    kernel.timer.timeout.connect(kernel.do_one_iteration)
    kernel.timer.start(1000*kernel._poll_interval)

def default_kernel_app():
    app = IPKernelApp.instance()
    app.initialize(['python', '--pylab=qt'])
    app.kernel.eventloop = event_loop
    return app

def default_manager(kernel):
    connection_file = find_connection_file(kernel.connection_file)
    manager = QtKernelManager(connection_file=connection_file)
    manager.load_connection_file()
    manager.start_channels()
    atexit.register(manager.cleanup_connection_file)
    return manager

def console_widget(manager):
    try: # Ipython v0.13
        widget = RichIPythonWidget(gui_completion='droplist')
    except TraitError:  # IPython v0.12
        widget = RichIPythonWidget(gui_completion=True)
    widget.kernel_manager = manager
    return widget

def terminal_widget(**kwargs):
    kernel_app = default_kernel_app()
    manager = default_manager(kernel_app)
    widget = console_widget(manager)

    #update namespace                                                           
    kernel_app.shell.user_ns.update(kwargs)

    kernel_app.start()
    return widget

app = QtGui.QApplication([])
widget = terminal_widget(testing=123)
widget.show()
app.exec_()

Answer 2

带有一些清理功能的 IPython 0.13 版本：

#coding: utf-8
'''
Updated for IPython 0.13
Created on 18-03-2012
Updated:   11-09-2012
@author: Paweł Jarosz
'''

import atexit

from PySide import QtCore, QtGui

from IPython.zmq.ipkernel import IPKernelApp
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 IPython.config.application import catch_config_error

DEFAULT_INSTANCE_ARGS = ['qtconsole','--pylab=inline', '--colors=linux']

class IPythonLocalKernelApp(IPKernelApp):
    @catch_config_error
    def initialize(self, argv=DEFAULT_INSTANCE_ARGS):
        """
        argv: IPython args

        example:

            app = QtGui.QApplication([])
            kernelapp = IPythonLocalKernelApp.instance()
            kernelapp.initialize()

            widget = IPythonConsoleQtWidget()
            widget.set_default_style(colors='linux')

            widget.connect_kernel(connection_file=kernelapp.get_connection_file())
            # if you won't to connect to remote kernel you don't need kernelapp part, just widget part and:

            # widget.connect_kernel(connection_file='kernel-16098.json')

            # where kernel-16098.json is the kernel name
            widget.show()

            namespace = kernelapp.get_user_namespace()
            nxxx = 12
            namespace["widget"] = widget
            namespace["QtGui"]=QtGui
            namespace["nxxx"]=nxxx

            app.exec_()
        """
        super(IPythonLocalKernelApp, self).initialize(argv)
        self.kernel.eventloop = self.loop_qt4_nonblocking
        self.kernel.start()
        self.start()

    def loop_qt4_nonblocking(self, kernel):
        """Non-blocking version of the ipython qt4 kernel loop"""
        kernel.timer = QtCore.QTimer()
        kernel.timer.timeout.connect(kernel.do_one_iteration)
        kernel.timer.start(1000*kernel._poll_interval)

    def get_connection_file(self):
        """Returne current kernel connection file."""
        return self.connection_file

    def get_user_namespace(self):
        """Returns current kernel userspace dict"""
        return self.kernel.shell.user_ns

class IPythonConsoleQtWidget(RichIPythonWidget):

    def connect_kernel(self, connection_file, heartbeat = False):
        """
        connection_file: str - is the connection file name, for example 'kernel-16098.json'
        heartbeat: bool - workaround, needed for right click/save as ... errors ... i don't know how to 
                          fix this issue. Anyone knows? Anyway it needs more testing
            example1 (standalone):

                    app = QtGui.QApplication([])
                    widget = IPythonConsoleQtWidget()
                    widget.set_default_style(colors='linux')


                    widget.connect_kernel(connection_file='some connection file name')

                    app.exec_()

            example2 (IPythonLocalKernelApp):

                    app = QtGui.QApplication([])

                    kernelapp = IPythonLocalKernelApp.instance()
                    kernelapp.initialize()

                    widget = IPythonConsoleQtWidget()

                    # Green text, black background ;)
                    widget.set_default_style(colors='linux')

                    widget.connect_kernel(connection_file='kernelapp.get_connection_file())

                    app.exec_()

        """
        km = QtKernelManager(connection_file=find_connection_file(connection_file), config=self.config)
        km.load_connection_file()
        km.start_channels(hb=heartbeat)
        self.kernel_manager = km
        atexit.register(self.kernel_manager.cleanup_connection_file)

def main():

    app = QtGui.QApplication([])
    kernelapp = IPythonLocalKernelApp.instance()
    kernelapp.initialize()

    widget = IPythonConsoleQtWidget()
    widget.set_default_style(colors='linux')

    widget.connect_kernel(connection_file=kernelapp.get_connection_file())
    # if you connect to outside app kernel you don't need kernelapp part, 
    # just widget part and:

    # widget.connect_kernel(connection_file='kernel-16098.json')

    # where kernel-16098.json is the kernel name
    widget.show()

    namespace = kernelapp.get_user_namespace()
    nxxx = 12
    namespace["widget"] = widget
    namespace["QtGui"]=QtGui
    namespace["nxxx"]=nxxx

    app.exec_()


if __name__=='__main__':
    main()    

Answer 3

@ChrisB 接受的答案适用于 IPython 0.13 版，但不适用于较新的版本。从github上IPython内核库的examples section来看，this是v1.x+（目前用4.0.1测试）的方式，它的特点是控制台和内核在同一个进程中。

这是一个基于官方示例的示例，它提供了一个可以轻松插入应用程序的便利类。它已设置为在 Python 2.7 上与 pyqt4 和 IPython 4.0.1 一起使用：

（注意：您需要安装 ipykernel 和 qtconsole 软件包）

# Set the QT API to PyQt4
import os
os.environ['QT_API'] = 'pyqt'
import sip
sip.setapi("QString", 2)
sip.setapi("QVariant", 2)
from PyQt4.QtGui  import *
# Import the console machinery from ipython
from qtconsole.rich_ipython_widget import RichIPythonWidget
from qtconsole.inprocess import QtInProcessKernelManager
from IPython.lib import guisupport

class QIPythonWidget(RichIPythonWidget):
    """ Convenience class for a live IPython console widget. We can replace the standard banner using the customBanner argument"""
    def __init__(self,customBanner=None,*args,**kwargs):
        if not customBanner is None: self.banner=customBanner
        super(QIPythonWidget, self).__init__(*args,**kwargs)
        self.kernel_manager = kernel_manager = QtInProcessKernelManager()
        kernel_manager.start_kernel()
        kernel_manager.kernel.gui = 'qt4'
        self.kernel_client = kernel_client = self._kernel_manager.client()
        kernel_client.start_channels()

        def stop():
            kernel_client.stop_channels()
            kernel_manager.shutdown_kernel()
            guisupport.get_app_qt4().exit()            
        self.exit_requested.connect(stop)

    def pushVariables(self,variableDict):
        """ Given a dictionary containing name / value pairs, push those variables to the IPython console widget """
        self.kernel_manager.kernel.shell.push(variableDict)
    def clearTerminal(self):
        """ Clears the terminal """
        self._control.clear()    
    def printText(self,text):
        """ Prints some plain text to the console """
        self._append_plain_text(text)        
    def executeCommand(self,command):
        """ Execute a command in the frame of the console widget """
        self._execute(command,False)


class ExampleWidget(QWidget):
    """ Main GUI Widget including a button and IPython Console widget inside vertical layout """
    def __init__(self, parent=None):
        super(ExampleWidget, self).__init__(parent)
        layout = QVBoxLayout(self)
        self.button = QPushButton('Another widget')
        ipyConsole = QIPythonWidget(customBanner="Welcome to the embedded ipython console\n")
        layout.addWidget(self.button)
        layout.addWidget(ipyConsole)        
        # This allows the variable foo and method print_process_id to be accessed from the ipython console
        ipyConsole.pushVariables({"foo":43,"print_process_id":print_process_id})
        ipyConsole.printText("The variable 'foo' and the method 'print_process_id()' are available. Use the 'whos' command for information.")                           

def print_process_id():
    print 'Process ID is:', os.getpid()        

def main():
    app  = QApplication([])
    widget = ExampleWidget()
    widget.show()
    app.exec_()    

if __name__ == '__main__':
    main()

Answer 4

可能帮助其他人研究这个：我遇到了这个例子：

https://github.com/gpoulin/python-test/blob/master/embedded_qtconsole.py

经过测试并与 PySide、IPython 2.1.0、Python 3.4.1 配合使用。看来我什至可以直接使用 matplotlib。

from IPython.qt.console.rich_ipython_widget import RichIPythonWidget
from IPython.qt.inprocess import QtInProcessKernelManager
from PySide import QtGui, QtCore


class EmbedIPython(RichIPythonWidget):

    def __init__(self, **kwarg):
        super(RichIPythonWidget, self).__init__()
        self.kernel_manager = QtInProcessKernelManager()
        self.kernel_manager.start_kernel()
        self.kernel = self.kernel_manager.kernel
        self.kernel.gui = 'qt4'
        self.kernel.shell.push(kwarg)
        self.kernel_client = self.kernel_manager.client()
        self.kernel_client.start_channels()


class MainWindow(QtGui.QMainWindow):

    def __init__(self, parent=None):
        super(MainWindow, self).__init__(parent)

        self.textEdit = QtGui.QTextEdit()

        but1 = QtGui.QPushButton('write')
        but1.clicked.connect(self.but_write)

        but2 = QtGui.QPushButton('read')
        but2.clicked.connect(self.but_read)

        self.a = {'text': ''}
        self.console = EmbedIPython(testing=123, a=self.a)
        self.console.kernel.shell.run_cell('%pylab qt')

        vbox = QtGui.QVBoxLayout()
        hbox = QtGui.QHBoxLayout()
        vbox.addWidget(self.textEdit)
        vbox.addWidget(self.console)
        hbox.addWidget(but1)
        hbox.addWidget(but2)
        vbox.addLayout(hbox)

        b = QtGui.QWidget()
        b.setLayout(vbox)
        self.setCentralWidget(b)

    def but_read(self):
        self.a['text'] = self.textEdit.toPlainText()
        self.console.execute("print('a[\\\'text\\\'] = \"'+ a['text'] +'\"')")

    def but_write(self):
        self.textEdit.setText(self.a['text'])


if __name__ == '__main__':
    import sys
    app = QtGui.QApplication(sys.argv)
    main = MainWindow()
    main.show()
    sys.exit(app.exec_())

Answer 5

在 PyQt5 中运行的 2016 年更新：

from qtconsole.qt import QtGui
from qtconsole.rich_jupyter_widget import RichJupyterWidget
from qtconsole.inprocess import QtInProcessKernelManager


class ConsoleWidget(RichJupyterWidget):


    def __init__(self, customBanner=None, *args, **kwargs):
        super(ConsoleWidget, self).__init__(*args, **kwargs)

        if customBanner is not None:
            self.banner = customBanner

        self.font_size = 6
        self.kernel_manager = kernel_manager = QtInProcessKernelManager()
        kernel_manager.start_kernel(show_banner=False)
        kernel_manager.kernel.gui = 'qt'
        self.kernel_client = kernel_client = self._kernel_manager.client()
        kernel_client.start_channels()

        def stop():
            kernel_client.stop_channels()
            kernel_manager.shutdown_kernel()
            guisupport.get_app_qt().exit()

        self.exit_requested.connect(stop)

    def push_vars(self, variableDict):
        """
        Given a dictionary containing name / value pairs, push those variables
        to the Jupyter console widget
        """
        self.kernel_manager.kernel.shell.push(variableDict)

    def clear(self):
        """
        Clears the terminal
        """
        self._control.clear()

        # self.kernel_manager

    def print_text(self, text):
        """
        Prints some plain text to the console
        """
        self._append_plain_text(text)

    def execute_command(self, command):
        """
        Execute a command in the frame of the console widget
        """
        self._execute(command, False)


if __name__ == '__main__':
    app = QtGui.QApplication([])
    widget = ConsoleWidget()
    widget.show()
    app.exec_()

Answer 6

更新答案：

• Python 3.8
• IPython 7.22.0
• QtConsole 5.0.3

根据之前的答案修改。

from qtpy.QtWidgets import QApplication
from qtconsole.rich_jupyter_widget import RichJupyterWidget
from qtconsole.inprocess import QtInProcessKernelManager
from IPython.lib import guisupport


class ConsoleWidget(RichJupyterWidget):

    def __init__(self, customBanner=None, *args, **kwargs):
        super(ConsoleWidget, self).__init__(*args, **kwargs)

        if customBanner is not None:
            self.banner = customBanner

        self.font_size = 10
        self.kernel_manager = QtInProcessKernelManager()
        self.kernel_manager.start_kernel(show_banner=False)
        self.kernel_manager.kernel.gui = 'qt'
        self.kernel_client = self._kernel_manager.client()
        self.kernel_client.start_channels()

        def stop():
            self.kernel_client.stop_channels()
            self.kernel_manager.shutdown_kernel()
            guisupport.get_app_qt4().exit()

        self.exit_requested.connect(stop)

    def push_vars(self, variableDict):
        """
        Given a dictionary containing name / value pairs, push those variables
        to the Jupyter console widget
        """
        self.kernel_manager.kernel.shell.push(variableDict)

    def clear(self):
        """
        Clears the terminal
        """
        self._control.clear()

    def print_text(self, text):
        """
        Prints some plain text to the console
        """
        self._append_plain_text(text)

    def execute_command(self, command):
        """
        Execute a command in the frame of the console widget
        """
        self._execute(command, False)


if __name__ == '__main__':
    app = QApplication([])
    widget = ConsoleWidget()
    widget.show()
    app.exec_()