如何杀死 PyQt5 中的 QRunnable?
How to kill the QRunnable in PyQt5?
我有一个带有两个按钮 start
和 end
的应用程序。开始按钮将启动一个线程,该线程运行录音功能。此函数是使用 sounddevice
和 soundfile
库编写的。录音可以持续任意时间,用户可以随时按 ctrl+c
停止。
所以,现在我想为 end
按钮实现一个函数来停止通过按下 start
按钮启动的线程,或者该函数可以向线程发送 ctrl+c
信号.因此,当前录制将停止。我不确定如何实现这一目标。感谢您的帮助。
由两个.py组成的代码如下:
audio_record.py
import os
import signal
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
import time
import queue
from PyQt5 import QtCore, QtGui, QtWidgets
import soundfile as sf
import sounddevice as sd
import mythreading
class Ui_MainWindow(object):
def __init__(self):
self.threadpool = QThreadPool()
print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())
def setupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(640, 480)
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.pushButton = QtWidgets.QPushButton(self.centralwidget)
self.pushButton.setGeometry(QtCore.QRect(280, 190, 75, 23))
self.pushButton.setObjectName("pushButton")
self.pushButton.clicked.connect(self.start_button_func)
self.pushButton_1 = QtWidgets.QPushButton(self.centralwidget)
self.pushButton_1.setGeometry(QtCore.QRect(380, 190, 75, 23))
self.pushButton_1.setObjectName("pushButton")
self.pushButton_1.clicked.connect(self.end_button_func)
MainWindow.setCentralWidget(self.centralwidget)
self.menubar = QtWidgets.QMenuBar(MainWindow)
self.menubar.setGeometry(QtCore.QRect(0, 0, 640, 21))
self.menubar.setObjectName("menubar")
MainWindow.setMenuBar(self.menubar)
self.statusbar = QtWidgets.QStatusBar(MainWindow)
self.statusbar.setObjectName("statusbar")
MainWindow.setStatusBar(self.statusbar)
self.retranslateUi(MainWindow)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
def retranslateUi(self, MainWindow):
_translate = QtCore.QCoreApplication.translate
MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
self.pushButton.setText(_translate("MainWindow", "Start"))
self.pushButton_1.setText(_translate("MainWindow", "End"))
def record(self):
self.pid = os.getpid()
self.q = queue.Queue()
self.s = sd.InputStream(samplerate=48000, channels=2, callback=self.callback)
try:
# Make sure the file is open before recording begins
with sf.SoundFile('check.wav', mode='x', samplerate=48000, channels=2, subtype="PCM_16") as file:
with self.s:
# 1 second silence before the recording begins
time.sleep(1)
print('START')
print('#' * 80)
print('press Ctrl+C to stop the recording')
while True:
file.write(self.q.get())
except OSError:
print('The file to be recorded already exists.')
sys.exit(1)
def callback(self, indata, frames, time, status):
"""
This function is called for each audio block from the record function.
"""
if status:
print(status, file=sys.stderr)
self.q.put(indata.copy())
def start_button_func(self):
self.worker = mythreading.Worker(self.record)
self.threadpool.start(self.worker)
def end_button_func(self):
print('how to stop?')
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
MainWindow = QtWidgets.QMainWindow()
ui = Ui_MainWindow()
ui.setupUi(MainWindow)
MainWindow.show()
sys.exit(app.exec_())
mythreading.py如下:
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
class Worker(QRunnable):
def __init__(self, fn, *args, **kwargs):
super(Worker, self).__init__()
self.fn = fn
@pyqtSlot()
def run(self):
self.fn()
您必须使用标志,在本例中为 threading.Event()
以指示不应再执行该线程。对于 Ctrl + C
你必须使用 QShortcut
import os
import queue
from PyQt5 import QtCore, QtGui, QtWidgets
import soundfile as sf
import sounddevice as sd
import mythreading
import threading
class Ui_MainWindow(object):
def setupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(640, 480)
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.pushButton = QtWidgets.QPushButton(self.centralwidget)
self.pushButton.setGeometry(QtCore.QRect(280, 190, 75, 23))
self.pushButton.setObjectName("pushButton")
self.pushButton_1 = QtWidgets.QPushButton(self.centralwidget)
self.pushButton_1.setGeometry(QtCore.QRect(380, 190, 75, 23))
self.pushButton_1.setObjectName("pushButton")
MainWindow.setCentralWidget(self.centralwidget)
self.menubar = QtWidgets.QMenuBar(MainWindow)
self.menubar.setGeometry(QtCore.QRect(0, 0, 640, 21))
self.menubar.setObjectName("menubar")
MainWindow.setMenuBar(self.menubar)
self.statusbar = QtWidgets.QStatusBar(MainWindow)
self.statusbar.setObjectName("statusbar")
MainWindow.setStatusBar(self.statusbar)
self.retranslateUi(MainWindow)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
def retranslateUi(self, MainWindow):
_translate = QtCore.QCoreApplication.translate
MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
self.pushButton.setText(_translate("MainWindow", "Start"))
self.pushButton_1.setText(_translate("MainWindow", "End"))
class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
def __init__(self, parent=None):
super(MainWindow, self).__init__(parent)
self.setupUi(self)
self.threadpool = QtCore.QThreadPool()
print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())
self.pushButton.clicked.connect(self.start_button_func)
self.pushButton_1.clicked.connect(self.end_button_func)
self.event_stop = threading.Event()
QtWidgets.QShortcut("Ctrl+C", self, activated=self.end_button_func)
def record(self):
self.pid = os.getpid()
self.q = queue.Queue()
self.s = sd.InputStream(samplerate=48000, channels=2, callback=self.callback)
try:
# Make sure the file is open before recording begins
with sf.SoundFile('check.wav', mode='x', samplerate=48000, channels=2, subtype="PCM_16") as file:
with self.s:
# 1 second silence before the recording begins
QtCore.QThread.sleep(1)
print('START')
print('#' * 80)
print('press Ctrl+C to stop the recording')
while not self.event_stop.is_set():
file.write(self.q.get())
print("STOP")
except OSError:
print('The file to be recorded already exists.')
sys.exit(1)
def callback(self, indata, frames, time, status):
if status:
print(status, file=sys.stderr)
self.q.put(indata.copy())
@QtCore.pyqtSlot()
def start_button_func(self):
print("start")
self.worker = mythreading.Worker(self.record)
self.threadpool.start(self.worker)
@QtCore.pyqtSlot()
def end_button_func(self):
print('how to stop?')
self.event_stop.set()
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
w = MainWindow()
w.show()
sys.exit(app.exec_())
我有一个带有两个按钮 start
和 end
的应用程序。开始按钮将启动一个线程,该线程运行录音功能。此函数是使用 sounddevice
和 soundfile
库编写的。录音可以持续任意时间,用户可以随时按 ctrl+c
停止。
所以,现在我想为 end
按钮实现一个函数来停止通过按下 start
按钮启动的线程,或者该函数可以向线程发送 ctrl+c
信号.因此,当前录制将停止。我不确定如何实现这一目标。感谢您的帮助。
由两个.py组成的代码如下:
audio_record.py
import os
import signal
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
import time
import queue
from PyQt5 import QtCore, QtGui, QtWidgets
import soundfile as sf
import sounddevice as sd
import mythreading
class Ui_MainWindow(object):
def __init__(self):
self.threadpool = QThreadPool()
print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())
def setupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(640, 480)
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.pushButton = QtWidgets.QPushButton(self.centralwidget)
self.pushButton.setGeometry(QtCore.QRect(280, 190, 75, 23))
self.pushButton.setObjectName("pushButton")
self.pushButton.clicked.connect(self.start_button_func)
self.pushButton_1 = QtWidgets.QPushButton(self.centralwidget)
self.pushButton_1.setGeometry(QtCore.QRect(380, 190, 75, 23))
self.pushButton_1.setObjectName("pushButton")
self.pushButton_1.clicked.connect(self.end_button_func)
MainWindow.setCentralWidget(self.centralwidget)
self.menubar = QtWidgets.QMenuBar(MainWindow)
self.menubar.setGeometry(QtCore.QRect(0, 0, 640, 21))
self.menubar.setObjectName("menubar")
MainWindow.setMenuBar(self.menubar)
self.statusbar = QtWidgets.QStatusBar(MainWindow)
self.statusbar.setObjectName("statusbar")
MainWindow.setStatusBar(self.statusbar)
self.retranslateUi(MainWindow)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
def retranslateUi(self, MainWindow):
_translate = QtCore.QCoreApplication.translate
MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
self.pushButton.setText(_translate("MainWindow", "Start"))
self.pushButton_1.setText(_translate("MainWindow", "End"))
def record(self):
self.pid = os.getpid()
self.q = queue.Queue()
self.s = sd.InputStream(samplerate=48000, channels=2, callback=self.callback)
try:
# Make sure the file is open before recording begins
with sf.SoundFile('check.wav', mode='x', samplerate=48000, channels=2, subtype="PCM_16") as file:
with self.s:
# 1 second silence before the recording begins
time.sleep(1)
print('START')
print('#' * 80)
print('press Ctrl+C to stop the recording')
while True:
file.write(self.q.get())
except OSError:
print('The file to be recorded already exists.')
sys.exit(1)
def callback(self, indata, frames, time, status):
"""
This function is called for each audio block from the record function.
"""
if status:
print(status, file=sys.stderr)
self.q.put(indata.copy())
def start_button_func(self):
self.worker = mythreading.Worker(self.record)
self.threadpool.start(self.worker)
def end_button_func(self):
print('how to stop?')
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
MainWindow = QtWidgets.QMainWindow()
ui = Ui_MainWindow()
ui.setupUi(MainWindow)
MainWindow.show()
sys.exit(app.exec_())
mythreading.py如下:
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
class Worker(QRunnable):
def __init__(self, fn, *args, **kwargs):
super(Worker, self).__init__()
self.fn = fn
@pyqtSlot()
def run(self):
self.fn()
您必须使用标志,在本例中为 threading.Event()
以指示不应再执行该线程。对于 Ctrl + C
你必须使用 QShortcut
import os
import queue
from PyQt5 import QtCore, QtGui, QtWidgets
import soundfile as sf
import sounddevice as sd
import mythreading
import threading
class Ui_MainWindow(object):
def setupUi(self, MainWindow):
MainWindow.setObjectName("MainWindow")
MainWindow.resize(640, 480)
self.centralwidget = QtWidgets.QWidget(MainWindow)
self.centralwidget.setObjectName("centralwidget")
self.pushButton = QtWidgets.QPushButton(self.centralwidget)
self.pushButton.setGeometry(QtCore.QRect(280, 190, 75, 23))
self.pushButton.setObjectName("pushButton")
self.pushButton_1 = QtWidgets.QPushButton(self.centralwidget)
self.pushButton_1.setGeometry(QtCore.QRect(380, 190, 75, 23))
self.pushButton_1.setObjectName("pushButton")
MainWindow.setCentralWidget(self.centralwidget)
self.menubar = QtWidgets.QMenuBar(MainWindow)
self.menubar.setGeometry(QtCore.QRect(0, 0, 640, 21))
self.menubar.setObjectName("menubar")
MainWindow.setMenuBar(self.menubar)
self.statusbar = QtWidgets.QStatusBar(MainWindow)
self.statusbar.setObjectName("statusbar")
MainWindow.setStatusBar(self.statusbar)
self.retranslateUi(MainWindow)
QtCore.QMetaObject.connectSlotsByName(MainWindow)
def retranslateUi(self, MainWindow):
_translate = QtCore.QCoreApplication.translate
MainWindow.setWindowTitle(_translate("MainWindow", "MainWindow"))
self.pushButton.setText(_translate("MainWindow", "Start"))
self.pushButton_1.setText(_translate("MainWindow", "End"))
class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
def __init__(self, parent=None):
super(MainWindow, self).__init__(parent)
self.setupUi(self)
self.threadpool = QtCore.QThreadPool()
print("Multithreading with maximum %d threads" % self.threadpool.maxThreadCount())
self.pushButton.clicked.connect(self.start_button_func)
self.pushButton_1.clicked.connect(self.end_button_func)
self.event_stop = threading.Event()
QtWidgets.QShortcut("Ctrl+C", self, activated=self.end_button_func)
def record(self):
self.pid = os.getpid()
self.q = queue.Queue()
self.s = sd.InputStream(samplerate=48000, channels=2, callback=self.callback)
try:
# Make sure the file is open before recording begins
with sf.SoundFile('check.wav', mode='x', samplerate=48000, channels=2, subtype="PCM_16") as file:
with self.s:
# 1 second silence before the recording begins
QtCore.QThread.sleep(1)
print('START')
print('#' * 80)
print('press Ctrl+C to stop the recording')
while not self.event_stop.is_set():
file.write(self.q.get())
print("STOP")
except OSError:
print('The file to be recorded already exists.')
sys.exit(1)
def callback(self, indata, frames, time, status):
if status:
print(status, file=sys.stderr)
self.q.put(indata.copy())
@QtCore.pyqtSlot()
def start_button_func(self):
print("start")
self.worker = mythreading.Worker(self.record)
self.threadpool.start(self.worker)
@QtCore.pyqtSlot()
def end_button_func(self):
print('how to stop?')
self.event_stop.set()
if __name__ == "__main__":
import sys
app = QtWidgets.QApplication(sys.argv)
w = MainWindow()
w.show()
sys.exit(app.exec_())