python - 为什么信号不发出？

Question

应用

我正在尝试使用标准库 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() 会有所帮助。它没有。不过，我承认我对此可能是错误的。

感谢任何帮助。提前致谢。

Answer 1

您的解释器线程正在阻塞 InteractiveConsole.runco​​de() 调用。在此调用完成之前，它将无法处理任何信号。这就是您看到延迟输出的原因。

你可以通过改变来得到你想要的效果

self.interpreter.output.connect(self.send_console_log)

到

self.interpreter.stream.output.signal_str.connect(self.send_console_log)

对于一些老派的调试，断开你正在处理的 stderr 并在周围散布一些打印语句，比如......

print('runcode after', file=sys.stderr)