usage of pyside threads for ripping and encoding

gui/__init__.py

@@ -1,6 +1,7 @@
 import logging
 from PySide6.QtWidgets import (
     QMainWindow,
+    QMessageBox,
     QDialog,
     QDialogButtonBox,
     QWidget,
@@ -14,8 +15,16 @@ from PySide6.QtWidgets import (
     QPushButton,
     QLineEdit
 )
-from PySide6.QtCore import Qt, SIGNAL
+from PySide6.QtCore import (
-import task
+    Qt,
+    SIGNAL,
+    QObject,
+    Signal,
+    Slot,
+    QThread
+)
+from queue import Queue
+import time
 
 try:
     from config import APP_NAME as ROOT_LOGGER_NAME
@@ -24,6 +33,42 @@ except ImportError:
 logger = logging.getLogger(ROOT_LOGGER_NAME).getChild(__name__)
 
 
+class QueueWorker(QObject):
+    progress_updated = Signal(int, int)
+    job_finished = Signal(int)
+    all_jobs_finished = Signal()
+
+    def __init__(self, start_val, end_val):
+        super().__init__()
+        self.queue = Queue()
+        self.start_val = start_val
+        self.end_val = end_val
+        self.__error_msg__ = None
+
+    def get_error(self):
+        return self.__error_msg__
+
+    def set_error(self, error_msg):
+        if self.__error_msg__ is None:
+            self.__error_msg__ = error_msg
+
+    @Slot(int)
+    def add_job(self, job_index, job_data):
+        self.queue.put(job_index)
+
+    @Slot()
+    def run(self):
+        while True:
+            job_index = self.queue.get()
+            if job_index is None:
+                break
+            for i in range(self.start_val, self.end_val + 1):
+                time.sleep(0.02)
+                self.progress_updated.emit(i, job_index)
+            self.job_finished.emit(job_index)
+        self.all_jobs_finished.emit()
+
+
 class RowWidget(QWidget):
     """
     Ein Widget, das eine einzelne Zeile in unserer Liste darstellt.
@@ -62,8 +107,6 @@ class MainWindow(QMainWindow):
         super().__init__()
         self.__basepath__ = basepath
         #
-        self.__rip_task__ = task.delayed(0, self.__rip_it__)
-        #
         self.setWindowTitle("PyRip")
         self.setGeometry(100, 100, 900, 600)  # Fensterbreite angepasst
         self.set_status("Ready...")
@@ -109,11 +152,16 @@ class MainWindow(QMainWindow):
 
         self.button_rip = QPushButton("Rip Disc")
         self.button_rip.setDisabled(True)
-        self.button_rip.clicked.connect(self.rip_it)
+        self.button_rip.clicked.connect(self.rip_all)
         button_layout.addWidget(self.button_rip)
 
         self.connect(self.list_widget, SIGNAL("itemDoubleClicked(QListWidgetItem *)"), self.single_track_selected)
 
+    def clear(self):
+        self.list_widget.clear()
+        for key in self.FIELD_NAMES:
+            self.set_album_field(key, "")
+
     def set_album_field(self, key, value):
         i = self.FIELD_NAMES.index(key)
         line = self.__album_lines__[i]
@@ -139,22 +187,74 @@ class MainWindow(QMainWindow):
     def set_status(self, status):
         self.statusBar().showMessage(status)
 
-    def set_title_progress(self, num, value):
+    @Slot(int, int)
-        list_item = self.list_widget.item(num)
+    def set_title_progress(self, value, bar_index):
-        row_widget = self.list_widget.itemWidget(list_item)
+        if 0 <= bar_index < self.list_widget.count():
-        row_widget.progress_bar.setValue(value)
+            list_item = self.list_widget.item(bar_index)
+            row_widget = self.list_widget.itemWidget(list_item)
+            row_widget.progress_bar.setValue(value)
 
     def read_it(self):
-        print("read_it button action not yet implemented...")
+        print("read_it button action not yet implemented, adding example data...")
+        for key in self.FIELD_NAMES:
+            self.set_album_field(key, key)
+        for track in range(0, 5):
+            self.append_title("Track %02d" % (track + 1))
+        self.button_rip.setEnabled(True)
 
-    def rip_it(self):
+    def get_rip_worker(self):
-        self.__rip_task__.run()
+        return QueueWorker(start_val=0, end_val=50)
 
-    def __rip_it__(self):
+    def get_encode_worker(self):
-        print("rip_it button action not yet implemented...")
+        return QueueWorker(start_val=50, end_val=100)
+
+    def rip_all(self, clicked_data: bool = None, single_job: int = None):
+        self.button_read.setEnabled(False)
+        self.button_rip.setEnabled(False)
+
+        self.thread1 = QThread()
+        self.worker1 = self.get_rip_worker()
+        self.worker1.moveToThread(self.thread1)
+
+        self.thread2 = QThread()
+        self.worker2 = self.get_encode_worker()
+        self.worker2.moveToThread(self.thread2)
+
+        # Connect Signals and Slots
+        self.worker1.progress_updated.connect(self.set_title_progress)
+        self.worker2.progress_updated.connect(self.set_title_progress)
+
+        # We route the signal through the main window.
+        self.worker1.job_finished.connect(self.schedule_phase_two_task)
+
+        self.thread1.started.connect(self.worker1.run)
+        self.thread2.started.connect(self.worker2.run)
+
+        self.worker1.all_jobs_finished.connect(self.phase_one_complete)
+        self.worker2.all_jobs_finished.connect(self.all_tasks_complete)
+
+        self.thread1.start()
+        self.thread2.start()
+
+        self.set_status("Start ripping...")
+        if single_job is None:
+            for i in range(0, self.list_widget.count()):
+                self.worker1.add_job(i, self.get_job_data(i))
+        elif 0 <= single_job < self.list_widget.count():
+            self.worker1.add_job(single_job, self.get_job_data(single_job))
+
+        self.worker1.add_job(None, None)
+
+    def get_job_data(self, job_index):
+        # placeholder for disc function
+        return None
 
     def single_track_selected(self, list_item):
-        print("single_track_selected action not yet implemented...")
+        list_index = self.list_widget.row(list_item)
+        if self.button_rip.isEnabled():
+            self.rip_all(single_job=list_index)
+        else:
+            logger.warning("A rip process is running ignoring rip request of track %02d", list_index + 1)
 
     def on_disc_removed(self, device):
         self.clear()
@@ -170,11 +270,48 @@ class MainWindow(QMainWindow):
             if int(device.properties.get('ID_CDROM_MEDIA_TRACK_COUNT_AUDIO', 0)) > 0:
                 self.on_disc_new(device)
 
-    def join(self):
+    @Slot(int)
-        self.__rip_task__.join
+    def schedule_phase_two_task(self, job_index):
+        """
+        This slot receives the 'job_finished' signal from worker1
+        and explicitly adds the job to worker2.
+        """
+        self.worker2.add_job(job_index, self.get_job_data(job_index))
 
-    def stop(self):
+    @Slot()
-        self.__rip_task__.stop()
+    def phase_one_complete(self):
+        self.worker2.add_job(None, None)
+
+    @Slot()
+    def all_tasks_complete(self):
+        self.thread1.quit()
+        self.thread1.wait()
+        self.thread2.quit()
+        self.thread2.wait()
+
+        self.worker1.deleteLater()
+        self.worker2.deleteLater()
+        self.thread1.deleteLater()
+        self.thread2.deleteLater()
+
+        rip_error = self.worker1.get_error()
+        enc_error = self.worker2.get_error()
+        if rip_error is None and enc_error is None:
+            self.set_status("Finished ripping...")
+        else:
+            self.set_status("Ripping FAILED...")
+            if rip_error is not None:
+                msg = rip_error
+                if enc_error is not None:
+                    msg += "\n"
+                    msg += enc_error
+            else:
+                msg = enc_error
+
+            self.set_status(msg)
+            QMessageBox.critical(self, "Error!", msg)
+        self.button_read.setEnabled(True)
+        self.button_rip.setEnabled(True)
 
 
 class SelectionDialog(QDialog):

pyripgui.py

@@ -3,15 +3,19 @@ import fstools
 from gui import MainWindow, SelectionDialog
 import media
 import os
+from PySide6.QtCore import QObject, Signal, Slot
 from PySide6.QtWidgets import QApplication, QMessageBox
 import pyudev
 from pyudev.pyside6 import MonitorObserver
+from queue import Queue
 import report
 import sys
+import time
 
 logger = report.app_logging_config()
 
-# TODO: Usage of QThread instead of task.delayed
+# TODO: Stop / Kill Threads before close: QThread: Destroyed while thread '' is still running
+# TODO: Stop Queues, on Error. Possibly better error handling.
 # TODO: Error handling on cancelling cddb choose Dialog
 
 
@@ -23,15 +27,110 @@ MEDIA_GUI_DICT = {
 }
 
 
-class RipMainWindow(MainWindow):
+class RipQueueWorker(QObject):
+    progress_updated = Signal(int, int)
+    job_finished = Signal(int)
+    all_jobs_finished = Signal()
+
     def __init__(self, basepath):
-        super().__init__(basepath)
+        self.__basepath__ = basepath
+        self.__job_data__ = {}
+        super().__init__()
+        self.queue = Queue()
+        self.__error_msg__ = None
 
-    def clear(self):
+    def get_error(self):
-        self.list_widget.clear()
+        return self.__error_msg__
-        for key in MEDIA_GUI_DICT:
-            self.set_album_field(key, "")
 
+    def set_error(self, error_msg):
+        if self.__error_msg__ is None:
+            self.__error_msg__ = error_msg
+
+    @Slot(int)
+    def add_job(self, job_index, job_data):
+        self.__job_data__[job_index] = job_data
+        self.queue.put(job_index)
+
+    def progress_adaption(self, value, track_num):
+        self.progress_updated.emit(50 * value, track_num - 1)
+
+    @Slot()
+    def run(self):
+        while True:
+            job_index = self.queue.get()
+            track_info = self.__job_data__.get(job_index)
+            if job_index is None or track_info is None:
+                break
+            # Rip track job_index
+            wavfile = media.track_to_targetpath(self.__basepath__, track_info, 'wav')
+            try:
+                fstools.mkdir(os.path.dirname(wavfile))
+            except PermissionError:
+                msg = f"Unable to create ripping target path: {os.path.dirname(wavfile)}"
+                logger.exception(msg)
+                self.set_error(msg)
+                break
+            if media.disc_track_rip(job_index + 1, wavfile, self.progress_adaption) != 0:
+                msg = f"Unable to rip: {wavfile}"
+                logger.exception(msg)
+                self.set_error(msg)
+                break
+            if job_index in self.__job_data__:
+                del self.__job_data__[job_index]
+            self.job_finished.emit(job_index)
+        self.all_jobs_finished.emit()
+
+
+class EncodeQueueWorker(QObject):
+    progress_updated = Signal(int, int)
+    job_finished = Signal(int)
+    all_jobs_finished = Signal()
+
+    def __init__(self, basepath):
+        self.__basepath__ = basepath
+        self.__job_data__ = {}
+        super().__init__()
+        self.queue = Queue()
+        self.__error_msg__ = None
+
+    def get_error(self):
+        return self.__error_msg__
+
+    def set_error(self, error_msg):
+        if self.__error_msg__ is None:
+            self.__error_msg__ = error_msg
+
+    @Slot(int)
+    def add_job(self, job_index, job_data):
+        self.__job_data__[job_index] = job_data
+        self.queue.put(job_index)
+
+    def progress_adaption(self, value, track_num):
+        self.progress_updated.emit(50 + 50 * value, track_num - 1)
+
+    @Slot()
+    def run(self):
+        while True:
+            job_index = self.queue.get()
+            track_info = self.__job_data__.get(job_index)
+            if job_index is None or track_info is None:
+                break
+            # Rip track job_index
+            wavfile = media.track_to_targetpath(self.__basepath__, track_info, 'wav')
+            rv = media.wav_to_mp3(wavfile, self.__basepath__, track_info, self.progress_adaption)
+            os.remove(wavfile)
+            if rv != 0:
+                msg = f"Unable to encode: {wavfile}"
+                logger.exception(msg)
+                self.set_error(msg)
+                break
+            if job_index in self.__job_data__:
+                del self.__job_data__[job_index]
+            self.job_finished.emit(job_index)
+        self.all_jobs_finished.emit()
+
+
+class RipMainWindow(MainWindow):
     def read_it(self):
         self.set_status("Reading data from disc...")
         disc_data = media.get_media_data(media.get_disc_device(), self.cddb_choose_dialog)
@@ -66,59 +165,21 @@ class RipMainWindow(MainWindow):
                 self.set_status(msg)
                 QMessageBox.critical(self, "Error!", msg)
 
-    def single_track_selected(self, list_item):
+    def get_job_data(self, list_index):
-        self.rip_a_track(self.list_widget.row(list_item))
+        list_item = self.list_widget.item(list_index)
-
-    def rip_a_track(self, track_num):
-        list_item = self.list_widget.item(track_num)
         row_widget = self.list_widget.itemWidget(list_item)
         track_info = {}
-        track_info[media.KEY_TRACK] = track_num + 1
+        track_info[media.KEY_TRACK] = list_index + 1
         for key in MEDIA_GUI_DICT:
             track_info[MEDIA_GUI_DICT[key]] = self.get_album_field(key)
         track_info[media.KEY_TITLE] = row_widget.text_edit.toPlainText()
-        wavfile = media.track_to_targetpath(self.__basepath__, track_info, 'wav')
+        return track_info
-        try:
-            fstools.mkdir(os.path.dirname(wavfile))
-        except PermissionError:
-            msg = f"Unable to create ripping target path: {os.path.dirname(wavfile)}"
-            logger.exception(msg)
-            self.set_status(msg)
-            QMessageBox.critical(self, "Error!", msg)
-            self.button_rip.setDisabled(False)
-            self.button_read.setDisabled(False)
-            return
-        self.set_status("Ripping track %02d..." % (track_num + 1))
-        rv = media.disc_track_rip(track_num + 1, wavfile, self.progress_callback_rip)
-        if rv == 0:
-            self.set_status("Encoding track %02d..." % (track_num + 1))
-            rv = media.wav_to_mp3(wavfile, self.__basepath__, track_info, self.progress_callback_enc)
-            os.remove(wavfile)
-        if rv != 0:
-            msg = f"Error while ripping or encoding track {track_num + 1}"
-            logger.error(msg)
-            self.set_status(msg)
-            self.set_title_progress(track_num, 0)
-            QMessageBox.critical(self, "Error!", msg)
-            self.button_rip.setDisabled(False)
-            self.button_read.setDisabled(False)
-        return rv
 
-    def __rip_it__(self):
+    def get_rip_worker(self):
-        self.button_rip.setDisabled(True)
+        return RipQueueWorker(self.__basepath__)
-        self.button_read.setDisabled(True)
-        for i in range(self.list_widget.count()):
-            if self.rip_a_track(i) != 0:
-                return
-        self.set_status("Ripping finished!")
-        self.button_rip.setDisabled(False)
-        self.button_read.setDisabled(False)
 
-    def progress_callback_rip(self, p: float, track_num: int = None):
+    def get_encode_worker(self):
-        self.set_title_progress(track_num - 1, 50 * p)
+        return EncodeQueueWorker(self.__basepath__)
-
-    def progress_callback_enc(self, p: float, track_num: int = None):
-        self.set_title_progress(track_num - 1, 50 + 50 * p)
 
 
 if __name__ == "__main__":
@@ -147,9 +208,4 @@ if __name__ == "__main__":
     # Start GUI
     #
     window.show()
-    try:
-        window.join()
-    finally:
-        window.stop()
-
     sys.exit(app.exec())