Open1

PySide6で定期処理(Python, Qt)

okometabetaiokometabetai

背景

PySide6アプリケーションで schedule モジュールを使って定期処理を行いたい場合、いくつかの注意点があります。

まず、schedule.run_pending() はジョブを逐次実行するためにループで呼び出す必要がありますが、これをGUIスレッド(メインスレッド)で行うと、GUIの操作や表示がブロックされてしまう可能性があります。特に schedule のジョブの中で QMessageBox などのGUI要素を呼び出すと、処理が詰まってループが止まってしまうことがあります。

また、一般的に schedule.run_pending()while True: のような無限ループで回すことが多いですが、この方法では、アプリケーション終了時にループを止める手段がなく、スレッドが正常に終了できないという問題もあります。

解決策

このような問題に対処するため、以下のような設計が必要です:

  1. schedule.run_pending() は別スレッドで実行する
    → メインスレッドをブロックしないため。

  2. QTimerで定期的にスレッドを動かす
    → 無限ループの代わりに、QTimer を使って1秒ごとに schedule.run_pending() を呼ぶようにする。

  3. QTimerはメインスレッドでしか使えない
    → Qtの公式ドキュメントでは以下のように記載されています:

    You must start and stop the timer in its thread.
    It is not possible to start a timer from another thread.

    つまり、QTimerは生成されたスレッド内でしかstartできないため、サブスレッドで使うことはできません。

  4. GUI要素の操作は必ずメインスレッドで行う
    QMessageBox のようなUIコンポーネントは、メインスレッド以外から呼び出すと不安定な挙動になることがあります。

以上の理由から、次のような構成が安全かつ実用的です:

  • GUIとQTimerはメインスレッドで動かす
  • schedule による定期処理は別スレッドで実行する
  • スレッドからGUIへ通知するには、QThreadSignal または threading.Thread+カスタムイベントを使う

このアーキテクチャにより、GUIの安定性を保ちつつ、schedule によるジョブも正しく定期実行できるようになります。

実装例1:QThread + Signal

Qt流に正統派で書くならこちら。QThreadからSignalを発行し、GUI側で受け取って処理します。

import schedule
from PySide6 import QtCore, QtWidgets
import sys


class AlertRegularly(QtCore.QTimer):
    def __init__(self) -> None:
        super().__init__()
        self.scheduler_thread = SchedulerThread(self)

        # 1秒ごとにスレッドを動かす
        self.setInterval(1000)
        self.timeout.connect(self.scheduler_thread.start)

        # スレッドからのSignalを受けてGUIを更新
        self.scheduler_thread.show_alert_time.connect(self.alert)

        self.start()

    def alert(self, message: str):
        QtWidgets.QMessageBox.information(
            None, "通知", message,
            QtWidgets.QMessageBox.Yes, QtWidgets.QMessageBox.NoButton
        )

    def __del__(self) -> None:
        self.scheduler_thread.stop()


class SchedulerThread(QtCore.QThread):
    show_alert_time = QtCore.Signal(str)

    def __init__(self, parent: QtCore.QObject) -> None:
        super().__init__(parent)
        schedule.every(4).seconds.do(self.alert)

    def alert(self) -> None:
        self.show_alert_time.emit("時間です")

    def run(self) -> None:
        schedule.run_pending()

    def stop(self):
        self.quit()
        self.wait()


class MainWindow(QtWidgets.QWidget):
    def __init__(self) -> None:
        super().__init__()
        self.init_widget()
        self.alert = AlertRegularly()

    def init_widget(self) -> None:
        self.setGeometry(300, 200, 400, 300)
        self.setWindowTitle("Close button")
        button = QtWidgets.QPushButton("close", self)
        button.move(150, 100)
        button.clicked.connect(self.close)


def main():
    app = QtWidgets.QApplication(sys.argv)
    w = MainWindow()
    w.show()
    app.exec()


if __name__ == "__main__":
    main()

実装例2:threading + カスタムイベント

もっとシンプルに書きたいならこちら。
Signalの代わりに「カスタムイベント」をGUIに投げて処理します。
スレッドをデーモン化しているので、アプリ終了時に自動的に終了します。

import schedule
import threading
from PySide6 import QtCore, QtWidgets
import sys
import time


class QEventCustomShowMessageBox(QtCore.QEvent):
    custom_type = QtCore.QEvent.Type(QtCore.QEvent.registerEventType())
    def __init__(self) -> None:
        super().__init__(self.custom_type)


class MainWindow(QtWidgets.QWidget):
    def __init__(self) -> None:
        super().__init__()
        self.init_widget()
        self.scheduler_thread = SchedulerThread(self)

    def init_widget(self) -> None:
        self.setGeometry(500, 300, 400, 270)
        self.setWindowTitle("test")
        button = QtWidgets.QPushButton("close", self)
        button.move(150, 70)
        button.clicked.connect(self.close)

    def event(self, event: QtCore.QEvent) -> bool:
        if isinstance(event, QEventCustomShowMessageBox):
            QtWidgets.QMessageBox.information(
                None, "通知", "インフォメーションです。",
                QtWidgets.QMessageBox.Yes, QtWidgets.QMessageBox.NoButton
            )
            return True
        return super().event(event)


class SchedulerThread(threading.Thread):
    def __init__(self, main_window: MainWindow) -> None:
        super().__init__(daemon=True)
        self.main_window = main_window
        schedule.every(4).seconds.do(self.alert)
        self.start()

    def alert(self) -> None:
        QtCore.QCoreApplication.postEvent(
            self.main_window, QEventCustomShowMessageBox()
        )

    def run(self) -> None:
        while True:
            schedule.run_pending()
            time.sleep(1)


def main():
    app = QtWidgets.QApplication(sys.argv)
    w = MainWindow()
    w.show()
    app.exec()


if __name__ == "__main__":
    main()

まとめ

  • schedule.run_pending()メインループで回すのは危険
  • GUI操作は必ず メインスレッド で行う
  • 定期処理は 別スレッド に切り出す
  • スレッドからGUIへは Signal または カスタムイベント で通知する