PySide6 QRunnable 진행률 업데이트 가이드, QThreadPool과 QueuedConnection으로 안전한 GUI 반응성 확보

from PySide6.QtCore import QObject, Signal, QRunnable, QThreadPool, Slot, Qt
from PySide6.QtWidgets import QApplication, QWidget, QPushButton, QVBoxLayout, QProgressBar, QTextEdit

class WorkerSignals(QObject):
    progress = Signal(int)      # 0~100
    message = Signal(str)       # 상태 메시지
    finished = Signal()         # 작업 완료

class Worker(QRunnable):
    def __init__(self, steps=100):
        super().__init__()
        self.setAutoDelete(True)
        self.steps = steps
        self.signals = WorkerSignals()

    def run(self):
        # 무거운 작업 시뮬레이션
        import time
        for i in range(self.steps + 1):
            time.sleep(0.01)
            self.signals.progress.emit(i)
            if i % 10 == 0:
                self.signals.message.emit(f"processing {i}%")
        self.signals.finished.emit()

class Window(QWidget):
    def __init__(self):
        super().__init__()
        self.btn = QPushButton("Start")
        self.bar = QProgressBar()
        self.log = QTextEdit(); self.log.setReadOnly(True)

        lay = QVBoxLayout(self)
        for w in (self.btn, self.bar, self.log): lay.addWidget(w)

        self.pool = QThreadPool.globalInstance()
        self.btn.clicked.connect(self.start_job)

    def start_job(self):
        self.btn.setEnabled(False)
        worker = Worker(steps=100)

        # QueuedConnection 보장: cross-thread UI 업데이트
        worker.signals.progress.connect(self.on_progress, Qt.ConnectionType.QueuedConnection)
        worker.signals.message.connect(self.on_message,  Qt.ConnectionType.QueuedConnection)
        worker.signals.finished.connect(self.on_finished, Qt.ConnectionType.QueuedConnection)

        self.pool.start(worker)

    @Slot(int)
    def on_progress(self, value: int):
        self.bar.setValue(value)

    @Slot(str)
    def on_message(self, msg: str):
        self.log.append(msg)

    @Slot()
    def on_finished(self):
        self.log.append("done")
        self.btn.setEnabled(True)

if __name__ == "__main__":
    app = QApplication([])
    w = Window(); w.resize(360, 260); w.show()
    app.exec()

class WorkerSignals(QObject):
    progress = Signal(int)
    message = Signal(str)
    finished = Signal()
    error = Signal(str)

def run(self):
    try:
        for i in range(101):
            # CPU-bound 작업 예시
            result = i ** 2  
            if i % 2 == 0:
                self.signals.progress.emit(i)
            if i % 20 == 0:
                self.signals.message.emit(f"{i}% 완료")
        self.signals.finished.emit()
    except Exception as e:
        self.signals.error.emit(str(e))

from PySide6.QtCore import Qt, Slot

# 워커 연결부 (메인 스레드)
worker.signals.progress.connect(self.on_progress, Qt.ConnectionType.QueuedConnection)
worker.signals.message.connect(self.on_message,   Qt.ConnectionType.QueuedConnection)
worker.signals.finished.connect(self.on_finished, Qt.ConnectionType.QueuedConnection)

@Slot(int)
def on_progress(self, v: int):
    # 항상 메인 스레드 문맥에서 실행
    self.bar.setValue(v)

@Slot(str)
def on_message(self, m: str):
    self.log.append(m)

@Slot()
def on_finished(self):
    self.log.append("작업 완료")
    self.btn.setEnabled(True)

from PySide6.QtCore import QObject, Signal, QRunnable, QThreadPool, Qt
from PySide6.QtWidgets import QPushButton

class WorkerSignals(QObject):
    progress = Signal(int)
    message  = Signal(str)
    finished = Signal()
    canceled = Signal()
    error    = Signal(str)

class CancelableWorker(QRunnable):
    def __init__(self, steps=100):
        super().__init__()
        self.steps = steps
        self.signals = WorkerSignals()
        self._stopped = False
        self.setAutoDelete(True)

    def stop(self):
        self._stopped = True

    def run(self):
        try:
            for i in range(self.steps + 1):
                # 협조적 취소 체크
                if self._stopped:
                    self.signals.message.emit("canceled by user")
                    self.signals.canceled.emit()
                    return
                # 계산 또는 I/O
                if i % 5 == 0:
                    self.signals.message.emit(f"step {i}")
                self.signals.progress.emit(i)
            self.signals.finished.emit()
        except Exception as e:
            self.signals.error.emit(str(e))

# 메인 위젯 예시
pool = QThreadPool.globalInstance()
pool.setMaxThreadCount(4)  # 앱 성격에 맞춰 조정

start_btn = QPushButton("Start")
cancel_btn = QPushButton("Cancel")
worker = CancelableWorker(steps=100)

start_btn.clicked.connect(lambda: pool.start(worker))
cancel_btn.clicked.connect(lambda: worker.stop())  # 협조적 취소

from PySide6.QtCore import Qt, QTimer, Slot
from PySide6.QtWidgets import QProgressBar, QLabel, QTextEdit

class UiSyncMixin:
    def __init__(self):
        self._last_progress = -1
        self._log_buffer = []
        self._log_timer = QTimer(self)
        self._log_timer.setInterval(80)  # 80ms마다 배치 플러시
        self._log_timer.timeout.connect(self._flush_logs)

    @Slot(int)
    def on_progress(self, v: int):
        # 중복 호출 억제
        if v == self._last_progress:
            return
        # 부드러운 보간(옵션): 큰 점프일 때만 중간값을 빠르게 채움
        step = 3 if v - self._last_progress > 10 else 0
        if step > 0 and self._last_progress >= 0:
            for x in range(self._last_progress + step, v, step):
                self.bar.setValue(x)
        self.bar.setValue(v)
        self._last_progress = v
        self.status.setText(f"{v}%")

    @Slot(str)
    def on_message(self, msg: str):
        # 배치 버퍼에 저장
        self._log_buffer.append(msg)
        if not self._log_timer.isActive():
            self._log_timer.start()

    def _flush_logs(self):
        if not self._log_buffer:
            self._log_timer.stop()
            return
        # 한 번에 묶어 추가 → append 비용/스크롤 이벤트 최소화
        chunk = "\n".join(self._log_buffer)
        self.log.moveCursor(self.log.textCursor().End)
        self.log.insertPlainText(chunk + "\n")
        self._log_buffer.clear()
        # 메모리 보호: 오래된 로그는 자동 제거
        max_blocks = 2000
        if self.log.document().blockCount() > max_blocks:
            self.log.document().setMaximumBlockCount(max_blocks)