【发布时间】：2018-01-28 21:52:13
【问题描述】：
应用程序
我正在尝试使用 stdlib InteractiveConsole 为我的 PyQt5 应用程序构建一个 python shell，这样我就可以让用户编写实时绘图脚本。我正在使用 QTextEdit 从 shell 显示标准输出。
问题
当我在 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。这是一个transcript，说明为什么要实施。
我知道的
这一行负责插入纯文本self.output.signal_str.emit(data)。这个emit() 发生在QThread 中。因此，在多个 self.buffer.write() 完成之前，emit() 不会被处理。我认为在DelayedBuffer.process() 中添加QApplication.processEvents() 会有所帮助。它没有。但我承认我可能错了。
任何帮助表示赞赏。 提前致谢。
【问题讨论】：
-
与另一个问题一样，这非常确实需要minimal reproducible example。链接到 github 等非现场资源是不可接受的替代品。
-
@ekhumoro 我可以让这个更简单吗？最简单的版本是 160 行，没有缓冲。相信我，这很简单。除非你想让我把整个源代码放在这里。上次我为这个主题尝试最小版本时，关键想法丢失了。如果我别无选择，我不会这样做。
-
如果您有 MCVE，请将完整代码放在问题本身中，而不是链接到场外资源。
-
可能不相关但是......为什么调用
data = self.queue.get(block=False)是非阻塞的（我认为）？这将导致该线程本质上是一个忙于等待的循环——这无济于事。 -
@daegontaven。您的问题是：“应用程序冻结，因为 QTextEdit 的 insertPlainText() 太快了。所以我写了一个缓冲区来延迟插入”。它没有提到性能问题或运行长循环的问题。更重要的是，您的示例代码也没有演示任何这些内容。如果您真的需要帮助，请通过向他们提供所需的所有信息来帮助他们。
标签： python python-3.x qt pyqt pyqt5