Как прервать работу потока (moveToThread) по нажатию на кнопку?
В отдельном потоке я открываю окна некоторой программы и симулирую активность.
Для задержек я использую time.sleep(), но это не самое лучшее решение.
Также я не уверен, как правильно прервать работу потока при нажатии кнопки "Stop".
В данный момент остановка происходит только после завершения всех операций.
Упрощённый пример кода:
import sys
import win32gui
import win32con
import win32api
import time
from pynput.keyboard import Key, Controller as KeybController
from PyQt5.QtCore import QObject, QThread
from PyQt5.QtWidgets import (
QApplication,
QLabel,
QMainWindow,
QPushButton,
QVBoxLayout,
QWidget,
)
class Worker(QObject):
def __init__(self, window_name):
super().__init__()
self.window_name = window_name
self.keyboard = KeybController()
def run(self):
handles = self.collect_handles(self.window_name)
for handle in handles:
self.simulate_activity(handle)
def collect_handles(self, name):
handles = []
def callback(handle, _):
if handle and win32gui.GetWindowText(handle) == name:
handles.append(handle)
win32gui.EnumWindows(callback, None)
return handles
def simulate_activity(self, handle):
win32gui.ShowWindow(handle, win32con.SW_NORMAL)
win32gui.SetForegroundWindow(handle)
time.sleep(2)
self.keyboard_press(Key.space, 0.3)
win32gui.ShowWindow(handle, win32con.SW_MINIMIZE)
time.sleep(1)
def keyboard_press(self, key, duration):
self.keyboard.press(key)
time.sleep(duration)
self.keyboard.release(key)
class Window(QMainWindow):
def __init__(self, parent=None):
super().__init__(parent)
self.setupUi()
def setupUi(self):
self.setWindowTitle("GUI")
self.resize(300, 150)
self.centralWidget = QWidget()
self.setCentralWidget(self.centralWidget)
self.startBtn = QPushButton("Start", self)
self.startBtn.clicked.connect(self.runTask)
self.stopBtn = QPushButton("Stop", self)
self.stopBtn.clicked.connect(self.stopTask)
layout = QVBoxLayout()
layout.addWidget(self.startBtn)
layout.addWidget(self.stopBtn)
self.centralWidget.setLayout(layout)
def stopTask(self):
if self.thread.isRunning():
self.thread.quit()
self.worker.deleteLater()
def runTask(self):
self.thread = QThread()
self.worker = Worker("Папка")
self.worker.moveToThread(self.thread)
self.thread.started.connect(self.worker.run)
self.thread.start()
self.startBtn.setEnabled(False)
self.thread.finished.connect(
lambda: self.startBtn.setEnabled(True)
)
app = QApplication(sys.argv)
win = Window()
win.show()
sys.exit(app.exec())
UPD. Нашёл для себя следующее решение:
Функция рекурсивно вызывается при помощи таймеров, при этом поток не блокируется.
class Worker(QObject):
finished = pyqtSignal()
completed = pyqtSignal()
def __init__(self, window_name):
super().__init__()
self.window_name = window_name
self.keyboard = KeybController()
self.completed.connect(self.handle_process_completed)
def run(self):
handles = self.collect_handles(self.window_name)
self.simulate_activity(handles)
def handle_process_completed(self):
#Дополнительные действия, если потребуется
self.finished.emit()
def simulate_activity(self, handles, index=0):
if index + 1 > len(handles):
self.completed.emit()
return
handle = handles[index]
self.show_window(handle)
QTimer.singleShot(2000,
lambda: (
self.keyboard_press(Key.space, 300),
self.hide_window(handle),
QTimer.singleShot(1000, lambda: self.simulate_activity(handles, index + 1))
)
)
При инициализации воркера подключаем сигнал:
self.worker.finished.connect(self.stopTask)