Source code for cosedaUI.indexintegrate_sections.helpers

"""Helper classes for index/integrate orchestration."""

import sys
import traceback

from PyQt6.QtCore import QObject, pyqtSignal

from coseda.ici import orchestrator as orch




class GuiStream:
    """Forward writes to a callback for GUI logging."""

    def __init__(self, callback):
        self._callback = callback



    def write(self, text):
        if text:
            self._callback(text)




    def flush(self):
        pass






class OrchestratorWorker(QObject):
    """Run orch.main(argv) in a thread and forward output/progress to the GUI."""

    text_ready = pyqtSignal(str)
    finished = pyqtSignal(int)
    progress_init = pyqtSignal(int)
    progress_step = pyqtSignal(int)
    progress_done = pyqtSignal()

    def __init__(self, argv, should_stop=None, parent=None):
        super().__init__(parent)
        self.argv = argv
        self.should_stop = should_stop



    def run(self):
        orig_out, orig_err = sys.stdout, sys.stderr
        gui_stream = GuiStream(self.text_ready.emit)
        sys.stdout = gui_stream
        sys.stderr = gui_stream
        callbacks = {
            "should_stop": self.should_stop,
            "progress_init": self.progress_init.emit,
            "progress_step": self.progress_step.emit,
            "progress_done": self.progress_done.emit,
        }

        exit_code = 0
        try:
            exit_code = orch.main(self.argv, callbacks=callbacks)
        except SystemExit as e:
            try:
                exit_code = int(e.code)
            except Exception:
                exit_code = 1
        except Exception:
            tb = traceback.format_exc()
            self.text_ready.emit(tb)
            exit_code = 1
        finally:
            sys.stdout = orig_out
            sys.stderr = orig_err

        self.finished.emit(exit_code)