I have an app with two buttons start and end. The start button will start a thread, which runs the audio recording function. This function is written using sounddevice and soundfile libraries. The audio recording can take place for an arbitary duration and the user can stop anytime by pressing ctrl+c.
So, now I want to implement a function for the end button to stop the thread which started by pressing start button or the function can send ctrl+c signal to the thread. So, that the current recording will be stopped. I am not sure how to achieve this. Any help is appreciated.
The code consisting in two .py is as it follows:
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 is as follows:
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()
You have to use a flag, in this case threading.Event() to indicate that the thread should no longer be executed. For the case of Ctrl + C you must use 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_())