应用程序
我正在尝试使用Python标准库InteractiveConsole构建PyQt5应用程序的Python shell，以便让用户脚本实时绘图。我使用QTextEdit来显示shell的stdout。
问题
当我在shell中进行for循环时，应用程序会冻结，因为将内容insertPlainText()到QTextEdit太快。因此，我编写了一个缓冲区，以延迟插入几毫秒。但是，我注意到，只要在for循环中运行任何阻塞函数，例如time.sleep()，它就会冻结。因此，在for循环内部的打印仅在循环完成后显示。如果禁用缓冲区，则不会发生这种情况。
例如，如果我在shell中执行以下操作：
>>>for i in range(10):
... time.sleep(1)
... print(i)
...
只有等待10秒后才会打印此内容。
代码
根据MVCE指南，以下是我编写的最简版本。
这是main.ui文件：
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>main_window</class>
<widget class="QMainWindow" name="main_window">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>800</width>
<height>600</height>
</rect>
</property>
<property name="sizePolicy">
<sizepolicy hsizetype="Preferred" vsizetype="Preferred">
<horstretch>0</horstretch>
<verstretch>0</verstretch>
</sizepolicy>
</property>
<property name="windowTitle">
<string>MainWindow</string>
</property>
<property name="tabShape">
<enum>QTabWidget::Rounded</enum>
</property>
<widget class="QWidget" name="central_widget">
<layout class="QHBoxLayout" name="horizontalLayout">
<item>
<layout class="QVBoxLayout" name="console_layout">
<item>
<widget class="QTextEdit" name="console_log">
<property name="undoRedoEnabled">
<bool>false</bool>
</property>
</widget>
</item>
<item>
<layout class="QHBoxLayout" name="horizontalLayout_4">
<item>
<widget class="QLabel" name="console_prompt">
<property name="text">
<string/>
</property>
</widget>
</item>
<item>
<widget class="QLineEdit" name="console_input">
<property name="frame">
<bool>true</bool>
</property>
</widget>
</item>
</layout>
</item>
</layout>
</item>
</layout>
</widget>
<widget class="QMenuBar" name="menu_bar">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>800</width>
<height>26</height>
</rect>
</property>
</widget>
<widget class="QStatusBar" name="status_bar"/>
</widget>
<resources/>
<connections/>
</ui>
这是main.py文件：
import sys
from code import InteractiveConsole
from io import StringIO
from queue import Queue, Empty
from PyQt5 import uic
from PyQt5.QtCore import pyqtSlot, QThread, QObject, pyqtSignal, QTimer
from PyQt5.QtGui import QTextOption, QTextCursor
from PyQt5.QtWidgets import QApplication
__author__ = "daegontaven"
__copyright__ = "daegontaven"
__license__ = "gpl3"
class BaseSignals(QObject):
"""
Standard set of pyqtSignals.
"""
signal_str = pyqtSignal(str)
signal_int = pyqtSignal(int)
signal_float = pyqtSignal(float)
signal_list = pyqtSignal(list)
signal_tuple = pyqtSignal(tuple)
signal_dict = pyqtSignal(dict)
signal_object = pyqtSignal(object)
def __init__(self):
QObject.__init__(self)
class DelayedBuffer(QObject):
"""
A buffer that uses a queue to store strings. It removes the
first appended string first in a constant interval.
"""
written = pyqtSignal(str)
def __init__(self, output, delay):
"""
:param output: used to access BaseSignals
:param delay: delay for emitting
"""
super().__init__()
self.output = output
# Set Delay
self.delay = delay
self.queue = Queue()
self.timer = QTimer()
self.timer.timeout.connect(self.process)
self.timer.start(self.delay)
def write(self, string):
self.queue.put(string)
def process(self):
"""
Try to send the data to the stream
"""
try:
data = self.queue.get(block=False)
self.written.emit(data)
except Empty:
pass
def emit(self, string):
"""
Force emit of string.
"""
self.output.signal_str.emit(string)
class ConsoleStream(StringIO):
"""
Custom StreamIO class that emits a signal on each write.
"""
def __init__(self, enabled=True, *args, **kwargs):
"""
Starts a delayed buffer to store writes due to UI
refresh limitations.
:param enabled: set False to bypass the buffer
"""
StringIO.__init__(self, *args, **kwargs)
self.enabled = enabled
self.output = BaseSignals()
# Buffer
self.thread = QThread()
self.buffer = DelayedBuffer(self.output, delay=5)
self.buffer.moveToThread(self.thread)
self.buffer.written.connect(self.get)
self.thread.start()
def write(self, string):
"""
Overrides the parent write method and emits a signal
meant to be received by interpreters.
:param string: single write output from stdout
"""
if self.enabled:
self.buffer.write(string)
else:
self.output.signal_str.emit(string)
def get(self, string):
self.output.signal_str.emit(string)
class PythonInterpreter(QObject, InteractiveConsole):
"""
A reimplementation of the builtin InteractiveConsole to
work with threads.
"""
output = pyqtSignal(str)
push_command = pyqtSignal(str)
multi_line = pyqtSignal(bool)
def __init__(self):
QObject.__init__(self)
self.l = {}
InteractiveConsole.__init__(self, self.l)
self.stream = ConsoleStream()
self.stream.output.signal_str.connect(self.console)
self.push_command.connect(self.command)
def write(self, string):
self.output.emit(string)
def runcode(self, code):
"""
Overrides and captures stdout and stdin from
InteractiveConsole.
"""
sys.stdout = self.stream
sys.stderr = self.stream
sys.excepthook = sys.__excepthook__
result = InteractiveConsole.runcode(self, code)
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__
return result
@pyqtSlot(str)
def command(self, command):
"""
:param command: line retrieved from console_input on
returnPressed
"""
result = self.push(command)
self.multi_line.emit(result)
@pyqtSlot(str)
def console(self, string):
"""
:param string: processed output from a stream
"""
self.output.emit(string)
class MainWindow:
"""
The main GUI window. Opens maximized.
"""
def __init__(self):
self.ui = uic.loadUi("main.ui")
self.ui.showMaximized()
# Console Properties
self.ui.console_log.document().setMaximumBlockCount(1000)
self.ui.console_log.setWordWrapMode(QTextOption.WrapAnywhere)
self.ps1 = '>>>'
self.ps2 = '...'
self.ui.console_prompt.setText(self.ps1)
# Spawn Interpreter
self.thread = QThread()
self.thread.start()
self.interpreter = PythonInterpreter()
self.interpreter.moveToThread(self.thread)
# Interpreter Signals
self.ui.console_input.returnPressed.connect(self.send_console_input)
self.interpreter.output.connect(self.send_console_log)
self.interpreter.multi_line.connect(self.prompt)
def prompt(self, multi_line):
"""
Sets what prompt to use.
"""
if multi_line:
self.ui.console_prompt.setText(self.ps2)
else:
self.ui.console_prompt.setText(self.ps1)
def send_console_input(self):
"""
Send input grabbed from the QLineEdit prompt to the console.
"""
command = self.ui.console_input.text()
self.ui.console_input.clear()
self.interpreter.push_command.emit(str(command))
def send_console_log(self, command):
"""
Set the output from InteractiveConsole in the QTextEdit.
Auto scroll scrollbar.
"""
# Checks if scrolled
old_cursor = self.ui.console_log.textCursor()
old_scrollbar = self.ui.console_log.verticalScrollBar().value()
new_scrollbar = self.ui.console_log.verticalScrollBar().maximum()
if old_scrollbar == new_scrollbar:
scrolled = True
else:
scrolled = False
# Sets the text
self.ui.console_log.insertPlainText(command)
# Scrolls/Moves cursor based on available data
if old_cursor.hasSelection() or not scrolled:
self.ui.console_log.setTextCursor(old_cursor)
self.ui.console_log.verticalScrollBar().setValue(old_scrollbar)
else:
self.ui.console_log.moveCursor(QTextCursor.End)
self.ui.console_log.verticalScrollBar().setValue(
self.ui.console_log.verticalScrollBar().maximum()
)
def main():
app = QApplication(sys.argv)
window = MainWindow()
sys.exit(app.exec_())
if __name__ == "__main__":
main()
需要使用 BaseSignals 类来实现主线程和解释器之间的通信。这里有一个记录，详细说明了为什么要实现这个类。
我的理解
这行代码负责插入纯文本 self.output.signal_str.emit(data)。这个 emit() 是在一个 QThread 中发生的。因此，在多个 self.buffer.write() 完成之前，emit() 不会被处理。我认为在 DelayedBuffer.process() 中添加 QApplication.processEvents() 可以帮助解决问题，但事实上并没有起到作用。不过我承认我可能是错的。
感谢您提供的任何帮助。
data = self.queue.get(block=False)是非阻塞的（我认为）？这将导致该线程本质上成为一个忙等待循环 - 这肯定不会有所帮助。 - G.M.