使用QProcess后重新运行 python脚本
Re-running the python script after using it QProcess
我想在PyQt5中根据spinBox的值对QProcess进行排队，并在textEdit中使用readAll()显示文本，但是无论我在spinBox中指定什么值，脚本只运行1次，其结果不是显示在文本编辑中。请告诉我解决方案。
我最近才开始学习 python。可能对我来说还是太难了，但我想整理一下。
test.py
from functools import cached_property
from pathlib import Path
from PyQt5 import QtGui, QtWidgets, QtCore, uic
import queue
import os
import sys
CURRENT_DIRECTORY = Path(__file__).resolve().parent
class MainWindow(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
QtWidgets.QMainWindow.__init__(self)
self.ui = uic.loadUi(os.fspath(CURRENT_DIRECTORY / "test.ui"), self)
self.resize(820, 300)
self.setFixedSize(self.size())
managers = TaskManager(self)
task_list = [os.fspath(CURRENT_DIRECTORY / "test2.py")]
for task in task_list:
managers.appendTask(task)
self.ui.pushButtonStart.clicked.connect(managers.start)
self.ui.pushButtonStop.clicked.connect(managers.stop)
managers.messageChanged.connect(self.textEdit.append)
@cached_property
def manager(self):
return TaskManager()
class TaskManager(QtCore.QObject):
messageChanged = QtCore.pyqtSignal(str)
numbersTaskRunningChanged = QtCore.pyqtSignal(int)
def __init__(self, parent=None):
super(TaskManager, self).__init__(parent)
self._max_task = 1
self._queue = queue.Queue()
self._numbers_task_running = 0
self._running = False
@cached_property
def processes(self):
return list()
def execute(self, script, metadata=None):
process = QtCore.QProcess()
process.setProperty("metadata", metadata or dict())
process.finished.connect(self.handle_finished)
process.setProgram(sys.executable)
process.setArguments([script])
process.start()
self.processes.append(process)
def handle_finished(self):
process = self.sender()
self.processes.remove(process)
metadata = process.property("metadata")
print(f"{metadata} finished")
def setMaxTask(self, max_task):
self._max_task = max_task
if self._running:
self.call_task()
def maxTask(self):
return self._max_task
def appendTask(self, task):
self._queue.put(task)
self.call_task()
def start(self):
interface = MainWindow()
interface.pushButtonStart.setEnabled(False)
interface.pushButtonStop.setEnabled(True)
self._running = True
self.call_task()
number_of_processes = interface.ui.spinBox.value()
script = os.fspath(CURRENT_DIRECTORY / "test2.py")
for i in range(number_of_processes):
interface.manager.execute(script, dict(i=i))
def stop(self):
interface = MainWindow()
interface.pushButtonStart.setEnabled(True)
interface.pushButtonStop.setEnabled(False)
self._running = False
def call_task(self):
process = QtCore.QProcess(self)
process.setProcessChannelMode(QtCore.QProcess.MergedChannels)
process.readyReadStandardOutput.connect(self.on_readyReadStandardOutput)
process.finished.connect(self.on_finished)
process.started.connect(self.on_started)
process.errorOccurred.connect(self.on_errorOccurred)
def on_readyReadStandardOutput(self):
codec = QtCore.QTextCodec.codecForLocale()
decoder_stdout = codec.makeDecoder()
process = self.sender()
text = decoder_stdout.toUnicode(process.readAllStandardOutput())
self.messageChanged.emit(text)
def on_errorOccurred(self, error):
process = self.sender()
print("error: ", error, "-", " ".join([process.program()] + process.arguments()))
self.call_task()
def on_finished(self):
process = self.sender()
self._numbers_task_running -= 1
self.numbersTaskRunningChanged.emit(self._numbers_task_running)
self.call_task()
def on_started(self):
process = self.sender()
print("started: ", " ".join([process.program()] + process.arguments()))
self._numbers_task_running += 1
self.numbersTaskRunningChanged.emit(self._numbers_task_running)
self.call_task()
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
w = MainWindow()
w.show()
sys.exit(app.exec())
test.ui
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>MainWindow</class>
<widget class="QMainWindow" name="MainWindow">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>820</width>
<height>300</height>
</rect>
</property>
<property name="windowTitle">
<string>MainWindow</string>
</property>
<property name="styleSheet">
<string notr="true">background-color: rgb(245, 245, 245);</string>
</property>
<widget class="QWidget" name="centralwidget">
<widget class="QWidget" name="verticalLayoutWidget">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>121</width>
<height>271</height>
</rect>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<property name="sizeConstraint">
<enum>QLayout::SetDefaultConstraint</enum>
</property>
<item>
<widget class="QPushButton" name="pushButtonStart">
<property name="enabled">
<bool>true</bool>
</property>
<property name="font">
<font>
<pointsize>10</pointsize>
</font>
</property>
<property name="styleSheet">
<string notr="true">background-color: rgb(0, 170, 0);</string>
</property>
<property name="text">
<string>Start</string>
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="pushButtonStop">
<property name="enabled">
<bool>false</bool>
</property>
<property name="font">
<font>
<pointsize>10</pointsize>
</font>
</property>
<property name="styleSheet">
<string notr="true">background-color: rgb(255, 0, 0);</string>
</property>
<property name="text">
<string>Stop</string>
</property>
</widget>
</item>
<item>
<widget class="QFrame" name="frame">
<property name="frameShape">
<enum>QFrame::StyledPanel</enum>
</property>
<property name="frameShadow">
<enum>QFrame::Raised</enum>
</property>
<widget class="QWidget" name="layoutWidget">
<property name="geometry">
<rect>
<x>10</x>
<y>10</y>
<width>106</width>
<height>20</height>
</rect>
</property>
<layout class="QGridLayout" name="gridLayout_2">
<item row="0" column="0">
<widget class="QLabel" name="label_6">
<property name="text">
<string>Value</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QSpinBox" name="spinBox">
<property name="minimum">
<number>1</number>
</property>
<property name="maximum">
<number>4000</number>
</property>
<property name="value">
<number>1</number>
</property>
</widget>
</item>
</layout>
</widget>
</widget>
</item>
</layout>
</widget>
<widget class="QWidget" name="verticalLayoutWidget_2">
<property name="geometry">
<rect>
<x>120</x>
<y>0</y>
<width>691</width>
<height>271</height>
</rect>
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<widget class="QTextEdit" name="textEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
</layout>
</widget>
</widget>
<widget class="QStatusBar" name="statusbar"/>
</widget>
<resources/>
<connections/>
</ui>
test2.py
import random, string
listToPrint = set()
listToPrint.add(''.join(random.choice(string.hexdigits) for i in range(16)))
print(listToPrint)
问题在于您不断创建 MainWindow 的新实例，因此用于跟踪进程数的旋转框始终具有默认初始值 1。
您还不止一次地犯了同样的错误，而且您甚至没有为 TaskManager 使用 cached_property。
一个可能的解决方案是将主 window 实例传递给任务管理器，这样它将使用旋转框的 actual 值。考虑到这不是一个非常优雅的解决方案，因为任务管理器可能应该 nothing window，并且所有通信都应该使用更多信号和槽进行。
class MainWindow(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
QtWidgets.QMainWindow.__init__(self)
uic.loadUi("test.ui", self)
self.resize(820, 300)
self.setFixedSize(self.size())
self.manager = TaskManager(self)
task_list = ["test2.py"]
for task in task_list:
self.manager.appendTask(task)
self.pushButtonStart.clicked.connect(self.manager.start)
self.pushButtonStop.clicked.connect(self.manager.stop)
self.manager.messageChanged.connect(self.textEdit.append)
class TaskManager(QtCore.QObject):
messageChanged = QtCore.pyqtSignal(str)
numbersTaskRunningChanged = QtCore.pyqtSignal(int)
def __init__(self, mainWindow):
super(TaskManager, self).__init__(mainWindow)
self.mainWindow = mainWindow
# ...
def start(self):
self._running = True
self.call_task()
number_of_processes = self.mainWindow.spinBox.value()
script = "test2.py"
for i in range(number_of_processes):
self.execute(script, dict(i=i))
def stop(self):
self.mainWindow.pushButtonStart.setEnabled(True)
self.mainWindow.pushButtonStop.setEnabled(False)
self._running = False
请注意，上述更正非常少，并且由于各种原因，代码在面向对象编程的许多方面都不是最佳的。
考虑到，从你的代码来看，你可能尝试过改编别人的代码，但没有完全理解其机制，我强烈建议你仔细研究原始代码和我的修改，并耐心研究classes/instances 和一般的 OOP。
你的请求令人困惑，因为你谈论队列但你从未使用过它，似乎你已经修补了我的代码，认为它们会神奇地加入并工作但现实（不幸的是，幸运的是）并非如此。
您必须明白，一个好的做法是将每个 class 执行的任务分开。 TaskManager的任务是做任务的队列处理，不是直接和GUI交互。而GUI的任务就是获取用户信息并启动TaskManager。
以上述为基础，我创建了一个class，如果您添加任务，它们将按顺序执行，也就是说，当我执行完一个任务时，它只会执行一个挂起的任务。而 stop 方法所做的是停止正在执行的任务并消除挂起的任务。同样 class 将通过信号发出进程的信息。
GUI 必须获取该信息并将其转换为字符串才能在 QTextEdit 中显示。另一方面，TaskManager class 有一个 running_changed 信号指示何时执行任务以启用或禁用停止按钮，因为在没有任务时启用它没有意义被执行。
from functools import cached_property
from pathlib import Path
from PyQt5 import QtGui, QtWidgets, QtCore, uic
from collections import deque
import os
import sys
CURRENT_DIRECTORY = Path(__file__).resolve().parent
class MainWindow(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
QtWidgets.QMainWindow.__init__(self)
self.ui = uic.loadUi(os.fspath(CURRENT_DIRECTORY / "test.ui"), self)
self.setFixedSize(820, 300)
self.ui.pushButtonStop.setEnabled
self.ui.pushButtonStart.clicked.connect(self.handle_start)
self.ui.pushButtonStop.clicked.connect(self.handle_stop)
self.manager.standard_output_changed.connect(self.process_data)
self.manager.standard_error_changed.connect(self.process_data)
self.manager.running_changed.connect(self.handle_running_changed)
def handle_start(self):
number_of_processes = self.ui.spinBox.value()
for _ in range(number_of_processes):
self.manager.append(
sys.executable, [os.fspath(CURRENT_DIRECTORY / "test2.py")]
)
def handle_stop(self):
self.manager.stop()
def handle_running_changed(self):
self.ui.pushButtonStop.setEnabled(self.manager.running)
@cached_property
def manager(self):
return TaskManager()
def process_data(self, data):
codec = QtCore.QTextCodec.codecForLocale()
decoder_stdout = codec.makeDecoder()
text = decoder_stdout.toUnicode(data)
self.textEdit.append(text)
class TaskManager(QtCore.QObject):
standard_output_changed = QtCore.pyqtSignal(QtCore.QByteArray)
standard_error_changed = QtCore.pyqtSignal(QtCore.QByteArray)
running_changed = QtCore.pyqtSignal()
_running = False
@cached_property
def process(self):
process = QtCore.QProcess()
process.readyReadStandardOutput.connect(self.handle_standard_output)
process.readyReadStandardError.connect(self.handle_standard_error)
process.finished.connect(self.try_start)
return process
@cached_property
def queue(self):
return deque()
@property
def running(self):
return self._running
@running.setter
def running(self, running):
if self.running == running:
return
self._running = running
self.running_changed.emit()
def append(self, program, arguments=()):
self.queue.append((program, arguments))
self.try_start()
def try_start(self):
if len(self.queue) == 0:
self.running = False
return
if self.process.state() == QtCore.QProcess.NotRunning:
program, arguments = self.queue.popleft()
self.process.setProgram(program)
self.process.setArguments(arguments)
self.process.start()
self.running = True
def handle_standard_output(self):
self.standard_output_changed.emit(self.process.readAllStandardOutput())
def handle_standard_error(self):
self.standard_error_changed.emit(self.process.readAllStandardError())
def stop(self):
if self.process.state() != QtCore.QProcess.NotRunning:
self.process.kill()
self.queue.clear()
self.running = False
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
w = MainWindow()
w.show()
sys.exit(app.exec())
为了测试我的代码，我修改了 test2.py，通过添加诸如 time.sleep 之类的耗时任务使其变得更加复杂。所以你可以完成 运行 的任务。另一方面，每按一次开始按钮，就会在正在执行的任务中增加n个任务。
import random
import string
import time
listToPrint = set()
for i in range(4):
listToPrint.add("".join(random.sample(string.hexdigits, 16)))
print(i, flush=True)
time.sleep(1)
print(listToPrint, flush=True)
我想在PyQt5中根据spinBox的值对QProcess进行排队，并在textEdit中使用readAll()显示文本，但是无论我在spinBox中指定什么值，脚本只运行1次，其结果不是显示在文本编辑中。请告诉我解决方案。 我最近才开始学习 python。可能对我来说还是太难了，但我想整理一下。
test.py
from functools import cached_property
from pathlib import Path
from PyQt5 import QtGui, QtWidgets, QtCore, uic
import queue
import os
import sys
CURRENT_DIRECTORY = Path(__file__).resolve().parent
class MainWindow(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
QtWidgets.QMainWindow.__init__(self)
self.ui = uic.loadUi(os.fspath(CURRENT_DIRECTORY / "test.ui"), self)
self.resize(820, 300)
self.setFixedSize(self.size())
managers = TaskManager(self)
task_list = [os.fspath(CURRENT_DIRECTORY / "test2.py")]
for task in task_list:
managers.appendTask(task)
self.ui.pushButtonStart.clicked.connect(managers.start)
self.ui.pushButtonStop.clicked.connect(managers.stop)
managers.messageChanged.connect(self.textEdit.append)
@cached_property
def manager(self):
return TaskManager()
class TaskManager(QtCore.QObject):
messageChanged = QtCore.pyqtSignal(str)
numbersTaskRunningChanged = QtCore.pyqtSignal(int)
def __init__(self, parent=None):
super(TaskManager, self).__init__(parent)
self._max_task = 1
self._queue = queue.Queue()
self._numbers_task_running = 0
self._running = False
@cached_property
def processes(self):
return list()
def execute(self, script, metadata=None):
process = QtCore.QProcess()
process.setProperty("metadata", metadata or dict())
process.finished.connect(self.handle_finished)
process.setProgram(sys.executable)
process.setArguments([script])
process.start()
self.processes.append(process)
def handle_finished(self):
process = self.sender()
self.processes.remove(process)
metadata = process.property("metadata")
print(f"{metadata} finished")
def setMaxTask(self, max_task):
self._max_task = max_task
if self._running:
self.call_task()
def maxTask(self):
return self._max_task
def appendTask(self, task):
self._queue.put(task)
self.call_task()
def start(self):
interface = MainWindow()
interface.pushButtonStart.setEnabled(False)
interface.pushButtonStop.setEnabled(True)
self._running = True
self.call_task()
number_of_processes = interface.ui.spinBox.value()
script = os.fspath(CURRENT_DIRECTORY / "test2.py")
for i in range(number_of_processes):
interface.manager.execute(script, dict(i=i))
def stop(self):
interface = MainWindow()
interface.pushButtonStart.setEnabled(True)
interface.pushButtonStop.setEnabled(False)
self._running = False
def call_task(self):
process = QtCore.QProcess(self)
process.setProcessChannelMode(QtCore.QProcess.MergedChannels)
process.readyReadStandardOutput.connect(self.on_readyReadStandardOutput)
process.finished.connect(self.on_finished)
process.started.connect(self.on_started)
process.errorOccurred.connect(self.on_errorOccurred)
def on_readyReadStandardOutput(self):
codec = QtCore.QTextCodec.codecForLocale()
decoder_stdout = codec.makeDecoder()
process = self.sender()
text = decoder_stdout.toUnicode(process.readAllStandardOutput())
self.messageChanged.emit(text)
def on_errorOccurred(self, error):
process = self.sender()
print("error: ", error, "-", " ".join([process.program()] + process.arguments()))
self.call_task()
def on_finished(self):
process = self.sender()
self._numbers_task_running -= 1
self.numbersTaskRunningChanged.emit(self._numbers_task_running)
self.call_task()
def on_started(self):
process = self.sender()
print("started: ", " ".join([process.program()] + process.arguments()))
self._numbers_task_running += 1
self.numbersTaskRunningChanged.emit(self._numbers_task_running)
self.call_task()
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
w = MainWindow()
w.show()
sys.exit(app.exec())
test.ui
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>MainWindow</class>
<widget class="QMainWindow" name="MainWindow">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>820</width>
<height>300</height>
</rect>
</property>
<property name="windowTitle">
<string>MainWindow</string>
</property>
<property name="styleSheet">
<string notr="true">background-color: rgb(245, 245, 245);</string>
</property>
<widget class="QWidget" name="centralwidget">
<widget class="QWidget" name="verticalLayoutWidget">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>121</width>
<height>271</height>
</rect>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<property name="sizeConstraint">
<enum>QLayout::SetDefaultConstraint</enum>
</property>
<item>
<widget class="QPushButton" name="pushButtonStart">
<property name="enabled">
<bool>true</bool>
</property>
<property name="font">
<font>
<pointsize>10</pointsize>
</font>
</property>
<property name="styleSheet">
<string notr="true">background-color: rgb(0, 170, 0);</string>
</property>
<property name="text">
<string>Start</string>
</property>
</widget>
</item>
<item>
<widget class="QPushButton" name="pushButtonStop">
<property name="enabled">
<bool>false</bool>
</property>
<property name="font">
<font>
<pointsize>10</pointsize>
</font>
</property>
<property name="styleSheet">
<string notr="true">background-color: rgb(255, 0, 0);</string>
</property>
<property name="text">
<string>Stop</string>
</property>
</widget>
</item>
<item>
<widget class="QFrame" name="frame">
<property name="frameShape">
<enum>QFrame::StyledPanel</enum>
</property>
<property name="frameShadow">
<enum>QFrame::Raised</enum>
</property>
<widget class="QWidget" name="layoutWidget">
<property name="geometry">
<rect>
<x>10</x>
<y>10</y>
<width>106</width>
<height>20</height>
</rect>
</property>
<layout class="QGridLayout" name="gridLayout_2">
<item row="0" column="0">
<widget class="QLabel" name="label_6">
<property name="text">
<string>Value</string>
</property>
</widget>
</item>
<item row="0" column="1">
<widget class="QSpinBox" name="spinBox">
<property name="minimum">
<number>1</number>
</property>
<property name="maximum">
<number>4000</number>
</property>
<property name="value">
<number>1</number>
</property>
</widget>
</item>
</layout>
</widget>
</widget>
</item>
</layout>
</widget>
<widget class="QWidget" name="verticalLayoutWidget_2">
<property name="geometry">
<rect>
<x>120</x>
<y>0</y>
<width>691</width>
<height>271</height>
</rect>
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<widget class="QTextEdit" name="textEdit">
<property name="readOnly">
<bool>true</bool>
</property>
</widget>
</item>
</layout>
</widget>
</widget>
<widget class="QStatusBar" name="statusbar"/>
</widget>
<resources/>
<connections/>
</ui>
test2.py
import random, string
listToPrint = set()
listToPrint.add(''.join(random.choice(string.hexdigits) for i in range(16)))
print(listToPrint)
问题在于您不断创建 MainWindow 的新实例，因此用于跟踪进程数的旋转框始终具有默认初始值 1。
您还不止一次地犯了同样的错误，而且您甚至没有为 TaskManager 使用 cached_property。
一个可能的解决方案是将主 window 实例传递给任务管理器，这样它将使用旋转框的 actual 值。考虑到这不是一个非常优雅的解决方案，因为任务管理器可能应该 nothing window，并且所有通信都应该使用更多信号和槽进行。
class MainWindow(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
QtWidgets.QMainWindow.__init__(self)
uic.loadUi("test.ui", self)
self.resize(820, 300)
self.setFixedSize(self.size())
self.manager = TaskManager(self)
task_list = ["test2.py"]
for task in task_list:
self.manager.appendTask(task)
self.pushButtonStart.clicked.connect(self.manager.start)
self.pushButtonStop.clicked.connect(self.manager.stop)
self.manager.messageChanged.connect(self.textEdit.append)
class TaskManager(QtCore.QObject):
messageChanged = QtCore.pyqtSignal(str)
numbersTaskRunningChanged = QtCore.pyqtSignal(int)
def __init__(self, mainWindow):
super(TaskManager, self).__init__(mainWindow)
self.mainWindow = mainWindow
# ...
def start(self):
self._running = True
self.call_task()
number_of_processes = self.mainWindow.spinBox.value()
script = "test2.py"
for i in range(number_of_processes):
self.execute(script, dict(i=i))
def stop(self):
self.mainWindow.pushButtonStart.setEnabled(True)
self.mainWindow.pushButtonStop.setEnabled(False)
self._running = False
请注意，上述更正非常少，并且由于各种原因，代码在面向对象编程的许多方面都不是最佳的。
考虑到，从你的代码来看，你可能尝试过改编别人的代码，但没有完全理解其机制，我强烈建议你仔细研究原始代码和我的修改，并耐心研究classes/instances 和一般的 OOP。
你的请求令人困惑，因为你谈论队列但你从未使用过它，似乎你已经修补了我的代码，认为它们会神奇地加入并工作但现实（不幸的是，幸运的是）并非如此。
您必须明白，一个好的做法是将每个 class 执行的任务分开。 TaskManager的任务是做任务的队列处理，不是直接和GUI交互。而GUI的任务就是获取用户信息并启动TaskManager。
以上述为基础，我创建了一个class，如果您添加任务，它们将按顺序执行，也就是说，当我执行完一个任务时，它只会执行一个挂起的任务。而 stop 方法所做的是停止正在执行的任务并消除挂起的任务。同样 class 将通过信号发出进程的信息。
GUI 必须获取该信息并将其转换为字符串才能在 QTextEdit 中显示。另一方面，TaskManager class 有一个 running_changed 信号指示何时执行任务以启用或禁用停止按钮，因为在没有任务时启用它没有意义被执行。
from functools import cached_property
from pathlib import Path
from PyQt5 import QtGui, QtWidgets, QtCore, uic
from collections import deque
import os
import sys
CURRENT_DIRECTORY = Path(__file__).resolve().parent
class MainWindow(QtWidgets.QMainWindow):
def __init__(self):
super().__init__()
QtWidgets.QMainWindow.__init__(self)
self.ui = uic.loadUi(os.fspath(CURRENT_DIRECTORY / "test.ui"), self)
self.setFixedSize(820, 300)
self.ui.pushButtonStop.setEnabled
self.ui.pushButtonStart.clicked.connect(self.handle_start)
self.ui.pushButtonStop.clicked.connect(self.handle_stop)
self.manager.standard_output_changed.connect(self.process_data)
self.manager.standard_error_changed.connect(self.process_data)
self.manager.running_changed.connect(self.handle_running_changed)
def handle_start(self):
number_of_processes = self.ui.spinBox.value()
for _ in range(number_of_processes):
self.manager.append(
sys.executable, [os.fspath(CURRENT_DIRECTORY / "test2.py")]
)
def handle_stop(self):
self.manager.stop()
def handle_running_changed(self):
self.ui.pushButtonStop.setEnabled(self.manager.running)
@cached_property
def manager(self):
return TaskManager()
def process_data(self, data):
codec = QtCore.QTextCodec.codecForLocale()
decoder_stdout = codec.makeDecoder()
text = decoder_stdout.toUnicode(data)
self.textEdit.append(text)
class TaskManager(QtCore.QObject):
standard_output_changed = QtCore.pyqtSignal(QtCore.QByteArray)
standard_error_changed = QtCore.pyqtSignal(QtCore.QByteArray)
running_changed = QtCore.pyqtSignal()
_running = False
@cached_property
def process(self):
process = QtCore.QProcess()
process.readyReadStandardOutput.connect(self.handle_standard_output)
process.readyReadStandardError.connect(self.handle_standard_error)
process.finished.connect(self.try_start)
return process
@cached_property
def queue(self):
return deque()
@property
def running(self):
return self._running
@running.setter
def running(self, running):
if self.running == running:
return
self._running = running
self.running_changed.emit()
def append(self, program, arguments=()):
self.queue.append((program, arguments))
self.try_start()
def try_start(self):
if len(self.queue) == 0:
self.running = False
return
if self.process.state() == QtCore.QProcess.NotRunning:
program, arguments = self.queue.popleft()
self.process.setProgram(program)
self.process.setArguments(arguments)
self.process.start()
self.running = True
def handle_standard_output(self):
self.standard_output_changed.emit(self.process.readAllStandardOutput())
def handle_standard_error(self):
self.standard_error_changed.emit(self.process.readAllStandardError())
def stop(self):
if self.process.state() != QtCore.QProcess.NotRunning:
self.process.kill()
self.queue.clear()
self.running = False
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
w = MainWindow()
w.show()
sys.exit(app.exec())
为了测试我的代码，我修改了 test2.py，通过添加诸如 time.sleep 之类的耗时任务使其变得更加复杂。所以你可以完成 运行 的任务。另一方面，每按一次开始按钮，就会在正在执行的任务中增加n个任务。
import random
import string
import time
listToPrint = set()
for i in range(4):
listToPrint.add("".join(random.sample(string.hexdigits, 16)))
print(i, flush=True)
time.sleep(1)
print(listToPrint, flush=True)