PyQt5 и threading - запуск и остановка потоков кнопкой
Пример: имеется программа, которая при нажатии на кнопку "старт" начинает выполнять несколько непрерывных циклов одновременно в отдельных потоках, которые при нажатии на ту же кнопку останавливаются.
Как реализовать остановку в виде:
1) Резкого прерывания
2) "Безопасного" прерывания (цикл доходит до конца и останавливается)
3) Прерывания с "чекпоинтами" (в цикле несколько заданных возможных мест для выхода, которое выбирается исходя из места в коде, где была нажата кнопка)
main.py:
from PyQt5 import QtCore, QtWidgets
from funs import function1, function2, function3
class Ui_Form(object):
def setupUi(self, Form):
Form.setObjectName("Form")
Form.resize(400, 300)
self.pushButton = QtWidgets.QPushButton(Form)
self.pushButton.setGeometry(QtCore.QRect(140, 160, 75, 23))
self.pushButton.setObjectName("pushButton")
self.plainTextEdit = QtWidgets.QPlainTextEdit(Form)
self.plainTextEdit.setGeometry(QtCore.QRect(80, 30, 261, 91))
self.plainTextEdit.setObjectName("plainTextEdit")
self.retranslateUi(Form)
QtCore.QMetaObject.connectSlotsByName(Form)
def retranslateUi(self, Form):
_translate = QtCore.QCoreApplication.translate
Form.setWindowTitle(_translate("Form", "Form"))
self.pushButton.setText(_translate("Form", "Начать"))
class MyWin(QtWidgets.QWidget, Ui_Form):
def __init__(self):
super().__init__()
self.setupUi(self)
self.pushButton.clicked.connect(self.start)
def start(self):
global working
if self.pushButton.text() == 'Начать':
self.pushButton.setText('Стоп')
working = True
function1().start()
function2().start()
function3().start()
else:
self.pushButton.setText('Начать')
working = False
if __name__ == "__main__":
import sys
working = False
app = QtWidgets.QApplication(sys.argv)
w = MyWin()
w.show()
sys.exit(app.exec_())
funs.py:
import threading
import time
class function1(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
while True:
for n in range(6):
print(n)
time.sleep(1)
print()
class function2(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
while True:
for n in range(10, 16):
print(n)
time.sleep(2)
print()
class function3(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
while True:
print('Привет!')
print('Я')
time.sleep(5)
# Чекпоинт
print('вывожу')
print('слова')
time.sleep(5)
# Чекпоинт
print('через')
print('время')
time.sleep(5)
# Чекпоинт
Ответы (1 шт):
Автор решения: S. Nick
→ Ссылка
Предлагаю вам попробовать следующий вариант:
from PyQt5 import QtCore, QtWidgets
#import threading
#import time
#from funs import function1, function2, function3
class Function1(QtCore.QThread):
dataChanged = QtCore.pyqtSignal(str)
finished = QtCore.pyqtSignal(str)
def __init__(self):
super().__init__()
self.working = True
def run(self):
self.dataChanged.emit("Start Function1")
while self.working:
for n in range(6):
self.dataChanged.emit(f"Data Function1: {n}")
self.msleep(1000)
if not self.stoped():
break
self.finished.emit(f"finished Function1: {n} -------")
def stoped(self):
return self.working
class Function2(QtCore.QThread):
dataChanged = QtCore.pyqtSignal(str)
finished = QtCore.pyqtSignal(str)
def __init__(self):
super().__init__()
self.working = True
def run(self):
while True:
for n in range(10, 16):
print(n)
time.sleep(2)
def run(self):
self.dataChanged.emit("Start Function 22")
while self.working:
for n in range(10, 16):
self.dataChanged.emit(f"Data Function 22: {n}")
self.msleep(1000)
if not self.stoped():
break
self.finished.emit(f"finished Function 22: {n} -------")
def stoped(self):
return self.working
class Function3(QtCore.QThread):
dataChanged = QtCore.pyqtSignal(str)
finished = QtCore.pyqtSignal(str)
def __init__(self):
super().__init__()
self.working = True
def run(self):
while True:
n = 'Привет! Я '
self.dataChanged.emit(f"Data Function 333: {n}")
self.msleep(3000)
# Чекпоинт
if not self.stoped():
break
n = 'вывожу слова '
self.dataChanged.emit(f"Data Function 333: {n}")
self.msleep(3000)
# Чекпоинт
if not self.stoped():
break
n = 'через время'
self.dataChanged.emit(f"Data Function 333: {n}")
self.msleep(3000)
# Чекпоинт
if not self.stoped():
break
self.finished.emit(f"finished Function 333: {n} -------")
def stoped(self):
return self.working
class Ui_Form(object):
def setupUi(self, Form):
Form.setObjectName("Form")
Form.resize(400, 300)
self.pushButton = QtWidgets.QPushButton(Form)
self.pushButton.setGeometry(QtCore.QRect(165, 270, 70, 23))
self.pushButton.setObjectName("pushButton")
self.plainTextEdit = QtWidgets.QPlainTextEdit(Form)
self.plainTextEdit.setGeometry(QtCore.QRect(10, 10, 380, 250))
self.plainTextEdit.setObjectName("plainTextEdit")
self.retranslateUi(Form)
QtCore.QMetaObject.connectSlotsByName(Form)
def retranslateUi(self, Form):
_translate = QtCore.QCoreApplication.translate
Form.setWindowTitle(_translate("Form", "Form"))
self.pushButton.setText(_translate("Form", "Начать"))
class MyWin(QtWidgets.QWidget, Ui_Form):
def __init__(self):
super().__init__()
self.setupUi(self)
self.pushButton.clicked.connect(self.start)
self.function1 = Function1()
self.function1.dataChanged.connect(self.dataThreads)
self.function1.finished.connect(self.finishThreads)
self.function2 = Function2()
self.function2.dataChanged.connect(self.dataThreads)
self.function2.finished.connect(self.finishThreads)
self.function3 = Function3()
self.function3.dataChanged.connect(self.dataThreads)
self.function3.finished.connect(self.finishThreads)
def start(self):
if self.pushButton.text() == 'Начать':
self.pushButton.setText('Стоп')
self.function1.working = True
self.function1.start()
self.function2.working = True
self.function2.start()
self.function3.working = True
self.function3.start()
else:
self.pushButton.setText('Начать')
self.function1.working = False
self.function2.working = False
self.function3.working = False
def dataThreads(self, text):
self.plainTextEdit.appendPlainText(text)
def finishThreads(self, text):
self.plainTextEdit.appendPlainText(text)
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
w = MyWin()
w.show()
sys.exit(app.exec_())
