src/CalculationWorker.py - master - Flipper/Marginal

src/CalculationWorker.py
69 строк2 KB
Alex
Marginal v1.0.0 RELEASE
11 янв 2026, 18:02
11 янв 2026, 18:025e1eff8
from PySide6.QtCore import QObject, Signal, Slot, QTimer, QThread
import traceback


class CalculationWorker(QObject):
    """
    Background worker for executing calculations in a separate Qt thread.

    Handles safe execution, logging, error reporting, and lifecycle
    management without blocking the UI thread.
    """

    finished = Signal(bool)
    error = Signal(str)
    log = Signal(str, bool)

    def __init__(self, backbone, args):
        """
        Initialize the calculation worker.

        Args:
            backbone: Application backbone instance.
            args (tuple): Arguments forwarded to Backbone.runCalculation().
        """
        super().__init__()
        self.backbone = backbone
        self.args = args
        self._is_running = False

    @Slot()
    def run(self):
        """
        Entry point for the worker thread.

        Ensures the worker is not already running and schedules execution
        once the Qt event loop is fully active.
        """
        if self._is_running:
            return

        self._is_running = True

        try:
            self.log.emit("Calculations worker connected.", True)

            # Small delay ensures the worker thread event loop is ready
            QTimer.singleShot(10, self._execute)

        except Exception as e:
            self.error.emit(f"Worker setup failed: {e}")
            self.finished.emit(False)

    def _execute(self):
        """
        Perform the calculation task and emit completion signals.

        This method is executed inside the worker thread.
        """
        try:
            result = self.backbone.runCalculation(*self.args)
            self.finished.emit(bool(result))

        except Exception as e:
            errorMsg = f"{str(e)}\n{traceback.format_exc()}"
            self.error.emit(errorMsg)
            self.finished.emit(False)

        finally:
            self._is_running = False