PySide6でデスクトップアプリを作る #6 スレッドとタイマー — 止まらないUI

読了 7分

これまで作ってきた画面は、すべての処理が一瞬で終わるものでした。ところが実際のアプリには、時間のかかる処理が必ず入ってきます。数百のファイルを変換し、ネットワークからデータを取得し、重い計算を回します。こうした処理をボタンのハンドラーにそのままつなぐと、処理が終わるまでウィンドウ全体が応答を停止します。今回はその理由を構造から理解し、解決策であるQThreadとQTimerを扱います。

全7回で構成します。

  • #1 PySide6とは — QtとPythonでデスクトップアプリ
  • #2 ウィジェットとレイアウト — 画面を組み立てる
  • #3 シグナルとスロット — イベント処理の核心
  • #4 Qt DesignerとUIファイル — 画面を描いて読み込む
  • #5 モデルとビュー — リスト・テーブルにデータをつなぐ
  • #6 スレッドとタイマー — 止まらないUI ← この記事
  • #7 パッケージングと配布 — PyInstallerで実行ファイルを作る

問題の再現 — ウィンドウが応答を停止するコード #

まず問題を自分の手で作ってみます。ボタンを押すと5秒かかる処理が実行されるウィンドウです。

main.py — UIが止まる例
import sys
import time

from PySide6.QtWidgets import (
    QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget, QLabel,
)

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.label = QLabel("待機中")
        self.button = QPushButton("処理を開始")
        self.button.clicked.connect(self.run_task)

        layout = QVBoxLayout()
        layout.addWidget(self.label)
        layout.addWidget(self.button)
        container = QWidget()
        container.setLayout(layout)
        self.setCentralWidget(container)

    def run_task(self):
        self.label.setText("処理中...")
        time.sleep(5)              # 重い処理のシミュレーション
        self.label.setText("完了")

app = QApplication(sys.argv)
window = MainWindow()
window.show()
app.exec()

ボタンを押すと、5秒間ウィンドウを動かすことも閉じることもできません。ラベルも「処理中…」に変わらず、5秒後にいきなり「完了」になります。OSはこのアプリを応答なしの状態として表示します。

理由は#3で扱ったイベントループにあります。app.exec()が回すイベントループは一度に1つのイベントだけを処理します。クリックイベントを処理するrun_taskが5秒間返らない間、画面更新の要求もマウスイベントもすべてキューで待たされます。ラベルのテキスト変更も次の画面更新で反映されるため、「処理中…」が見えなかったのです。

原則 — UIに触れるのはメインスレッドだけ #

解決の方向は、重い処理を別のスレッドに移すことです。その前に、Qtの大前提を刻んでおく必要があります。ウィジェットの生成と変更はメインスレッドだけで行います。ワーカースレッドからself.label.setText(...)を直接呼ぶと、そのときは動いているように見えても、予告なくクラッシュするコードになります。

では、ワーカースレッドは結果をどうやって画面に反映するのでしょうか。答えは#3で学んだシグナルです。シグナルはスレッドの境界を越えるとき自動的にキューに入り、スロットはメインスレッドで実行されます。ワーカースレッドはシグナルを発信するだけにして、ウィジェットの操作はメインスレッドのスロットが担当する構造です。

QThreadの定石パターン — WorkerとmoveToThread #

QThreadを継承してrunをオーバーライドする例も広く出回っていますが、公式に推奨される形は処理をQObject(Worker)に分離してmoveToThreadでスレッドに配置するパターンです。進捗表示まで含めた全体のコードです。

worker.py — 処理を持つWorker
import time

from PySide6.QtCore import QObject, Signal, Slot

class Worker(QObject):
    progress = Signal(int)      # 進捗率 (0~100)
    finished = Signal(str)      # 完了メッセージ
    failed = Signal(str)        # エラーメッセージ

    @Slot()
    def run(self):
        try:
            for i in range(1, 101):
                time.sleep(0.05)         # 実際はファイル処理、ネットワークなど
                self.progress.emit(i)
            self.finished.emit("100件の処理が完了")
        except Exception as e:
            self.failed.emit(str(e))
main.py — スレッドの結線
from PySide6.QtCore import QThread
from PySide6.QtWidgets import QProgressBar

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.progress_bar = QProgressBar()
        self.button = QPushButton("処理を開始")
        self.button.clicked.connect(self.start_task)
        # ... レイアウト構成は前の例と同じ ...

    def start_task(self):
        self.button.setEnabled(False)

        self.thread = QThread()
        self.worker = Worker()
        self.worker.moveToThread(self.thread)

        # スレッドが始まったらWorker.runを実行
        self.thread.started.connect(self.worker.run)

        # 進捗と結果はすべてシグナルで受信
        self.worker.progress.connect(self.progress_bar.setValue)
        self.worker.finished.connect(self.on_finished)
        self.worker.failed.connect(self.on_failed)

        # 後始末 — 処理が終わったらスレッドを終了して解放
        self.worker.finished.connect(self.thread.quit)
        self.worker.failed.connect(self.thread.quit)
        self.thread.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)

        self.thread.start()

    def on_finished(self, message: str):
        self.statusBar().showMessage(message)
        self.button.setEnabled(True)

    def on_failed(self, error: str):
        self.statusBar().showMessage(f"失敗: {error}")
        self.button.setEnabled(True)

ボタンを押してもウィンドウは自由に動き、プログレスバーがリアルタイムに進みます。構造を整理すると3行です。

  • Workerは計算だけを行い、結果をシグナルとして発信します。ウィジェットのことは一切知りません。
  • メインウィンドウはシグナルをスロットにつなぎ、メインスレッドでウィジェットを更新します。
  • エラーも例外を外に投げず、failedシグナルで伝えます。ワーカースレッドで発生した例外は、放置するとそのまま静かに消えてしまうためです。
注記
self.threadself.workerのようにインスタンス属性として保持するのには理由があります。ローカル変数にすると、メソッドが終わった瞬間にPythonのガベージコレクションがオブジェクトを回収し、スレッドが始まった直後に消えてしまう問題が起きます。実行中に参照が保たれるよう、必ず属性として持っておきます。

QTimer — 周期処理と遅延実行 #

スレッドまで持ち出す必要のない軽い繰り返し処理もあります。1秒ごとに時計を更新したり、数秒後にメッセージを消したりする仕事です。このときはQTimerが適しています。QTimerは別のスレッドを作らず、イベントループのスケジュールにスロット呼び出しを登録する方式なので、スロットが一瞬で終わる処理ならUIに影響はありません。

1秒ごとに更新される時計
from PySide6.QtCore import QTimer, QTime

self.clock_label = QLabel()
self.timer = QTimer(self)
self.timer.timeout.connect(self.update_clock)
self.timer.start(1000)          # 1000ms間隔

def update_clock(self):
    now = QTime.currentTime().toString("HH:mm:ss")
    self.clock_label.setText(now)

一度だけ遅らせて実行する用途にはsingleShotが簡単です。

3秒後にステータスメッセージを消す
QTimer.singleShot(3000, self.statusBar().clearMessage)

QTimerのスロットもメインスレッドで実行される点は覚えておく必要があります。スロットの中に重い処理を入れると最初の例のようにUIが止まるため、時間のかかる仕事は引き続きWorkerスレッドの担当です。

さらに先へ — QThreadPoolとasyncio #

複数の処理を並列でこなす必要があるなら、スレッドを自分でいくつも作る代わりにQThreadPoolとQRunnableの組み合わせがあります。スレッドプールが実行数を管理してくれるので、複数のダウンロードや複数のファイル変換のような状況に合います。このシリーズでは存在と用途の紹介にとどめます。

Pythonのasyncioとの関係も整理しておきます。Qtのイベントループとasyncioのイベントループは互いに別々の世界なので、async defのコードをPySide6アプリでそのまま動かすことはできません。両者をつなぐ専用の統合(PySide6のQtAsyncioモジュールなど)はありますが、別途学習が必要なテーマなので、このシリーズの範囲外に置きます。今回のWorkerパターンだけでも、デスクトップアプリの非同期の要求の大半をまかなえます。

まとめ #

今回の核心は3つです。

  • イベントループは一度に1つのイベントだけを処理するため、ハンドラーが長くかかるとウィンドウ全体が応答を停止します。
  • 重い処理はWorker(QObject)+ moveToThreadパターンで分離し、結果もエラーもすべてシグナルでメインスレッドに伝えます。ウィジェットに触れるのはメインスレッドだけです。
  • 軽い周期処理と遅延実行はQTimerで十分です。ただしQTimerのスロットもメインスレッドで動きます。

次回の「PySide6でデスクトップアプリを作る #7 パッケージングと配布 — PyInstallerで実行ファイルを作る」はシリーズの最終回です。ここまで作ってきたアプリを、Pythonが入っていないコンピュータでも動く配布物にする過程を整理します。

X