heat_transfer_workers.py

"""
This module is responsible for defining Workers that enable multihreading.
"""

import traceback
import sys
from PyQt5 import QtCore  # type: ignore


class WorkerSignals(QtCore.QObject):
    '''
    Defines the signals available from a running worker thread.
    Supported signals are:

    finished
        Just a signal, not carrying any data.

    error
        `tuple` (exctype, value, traceback.format_exc() )

    result
        `object` data returned from processing, can be any object

    progress
        `int` - indicating whether the simulation is running (1) or not (0)
    '''

    finished = QtCore.pyqtSignal()
    error = QtCore.pyqtSignal(tuple)
    result = QtCore.pyqtSignal(object)
    progress = QtCore.pyqtSignal(int)


class Worker(QtCore.QRunnable):
    '''
    Worker thread

    Inherits from QRunnable to handler worker thread setup, signals and wrap-up.

    :param callback: The function callback to run on this worker thread.
        Supplied args and kwargs will be passed through to the runner.
    :type callback: function
    :param args: Arguments to pass to the callback function
    :param kwargs: Keywords to pass to the callback function

    '''

    def __init__(self, fn, *args, **kwargs):
        super(Worker, self).__init__()

        # Store constructor arguments (re-used for processing)
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

        self.signals = WorkerSignals()
        # Add the callback to our kwargs, so we are notified of the progress
        self.kwargs['progress_callback'] = self.signals.progress

    @QtCore.pyqtSlot()
    def run(self):
        '''
        Initialise the runner function with passed args, kwargs.
        Running the function.
        '''

        # Retrieve args/kwargs here; and fire processing using them
        try:
            result = self.fn(*self.args, **self.kwargs)
        # In case of any exception transfer them as signals
        except Exception as e:
            print(e)
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit((exctype, value, traceback.format_exc()))
        # If we did not encounter any exception, return the result
        else:
            self.signals.result.emit(result)
        # In any case return the finished signal
        finally:
            self.signals.finished.emit()