Source code for temfield_mpylab.EUT

from PySide6.QtCore import QRunnable, Slot, Signal, QObject

import sys
import time
import traceback




def simple_eut_status(progress_callback, dw=1):
    # print("Dwell-Time: ", dw)
    start = now = time.time()
    end = start + dw
    while now < end:
        time.sleep(0.01)
        now = time.time()
        percentage = round((now - start)/dw, 2) * 100
        progress_callback.emit(percentage)
    return "Passed"




class EUT_status_Signal(QObject):
    """Defines the signals available from a running EUT_status thread.

    Supported signals are:
    finished -> No data
    error -> tuple (exctype, value, traceback.format_exc() )
    result -> object data returned from processing, anything
    progress -> int indicating % progress
    """
    finished = Signal()
    error = Signal(tuple)
    result = Signal(object)
    progress = Signal(int)





class EUT_status(QRunnable):
    """EUT_status 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().__init__()

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

        # Add the callback to our kwargs
        self.kwargs['progress_callback'] = self.signals.progress




    @Slot()
    def run(self):
        """
        Initialise the runner function with passed args, kwargs.
        """

        # Retrieve args/kwargs here; and fire processing using them
        try:
            result = self.fn(*self.args, **self.kwargs)
        except:
            traceback.print_exc()
            exctype, value = sys.exc_info()[:2]
            self.signals.error.emit((exctype, value, traceback.format_exc()))
        else:
            self.signals.result.emit(result)  # Return the result of the processing
        finally:
            self.signals.finished.emit()  # Done