🍣

PySide6(Qt for Python) のGUIアプリで、scheduleモジュールで定期実行。(QThread,QTimer)

2023/03/08に公開

Python

注意したこと

scheduleは別スレッドで動かす。そうしないとscheduleが、gui要素関連の処理を呼び出す時、schedule.run_pending()のループが止まってしまう。
通常は、schedule.run_pending()をwhile文で無限ループするが、そうすると、プログラムを終了する時、Scheduler_Threadが終了できない。
そのためにwhile文ではなくQTimerで、1秒ごとにschedule.run_pending()を呼び出す。
しかし、別スレッドでQTimerは利用できない。↓

You must start and stop the timer in its thread.
It is not possible to start a timer from another thread.
（公式ドキュメントより）

また、メインスレッド以外で、QMessageBoxなどGUI要素を動かすことは推奨されない。予期しない動作をする可能性がある。
そのために、QTimerや、GUIはメインスレッドで動かし、scheduleは別スレッドで動かす。

ソースコードの例

以下は、scheduleモジュールで、定期的に QMessageBox を表示するプログラムである。


import schedule
from PySide6 import QtCore, QtWidgets, QtGui
import sys


class alert_regularly(QtCore.QTimer):
    
    def __init__(self) -> None:
        super().__init__()
        self.scheduler_thread = Scheduler_Thread(self)
	
	self.setInterval(1000)
        self.timeout.connect(self.scheduler_thread.start)
	
        self.scheduler_thread.show_alart_time.connect(self.alart)
	
        self.start()
	
    def alart(self,message:str):
        return QtWidgets.QMessageBox.information(None, "通知", message, QtWidgets.QMessageBox.Yes, QtWidgets.QMessageBox.NoButton)

    # 念のため
    def __del__(self) -> None:
        self.scheduler_thread.stop()

class Scheduler_Thread(QtCore.QThread):
    show_alart_time = QtCore.Signal(str)

    def __init__(self, object : QtCore.QObject) -> None:
        super().__init__(object)
        self.object : QtCore.QObject = object
        schedule.every(4).seconds.do(self.alart)

    def alart(self) -> None:
        # signalで送る
        self.show_alart_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 = alert_regularly()

    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()

QThreadではなく、threadingを利用しても良いが、Signalなどを使えないため、カスタムイベントを作る必要がある。
スレッドをデーモンとすることで、プログラム終了時、スレッドも一緒に終了させる事ができるため、QTimerは必要ない。

main.py

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


class QEvent_Custom_ShowMessageBox(QtCore.QEvent):
    # QMessageBox を表示するためのカスタムイベント
    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 = Scheduler_Thread(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,QEvent_Custom_ShowMessageBox):
            QtWidgets.QMessageBox.information(None, "通知", "インフォメーションです。", QtWidgets.QMessageBox.Yes, QtWidgets.QMessageBox.NoButton)
            return True
        return super().event(event)


class Scheduler_Thread(threading.Thread):
    # 定期的に QMessageBox を表示するためのスレッド
    def __init__(self, main_window: MainWindow) -> None:
        super().__init__(daemon=True)
        self.main_window = main_window
        # schedule.every().day.at("23:00").do(self.alart)
        schedule.every(4).seconds.do(self.alart)
        self.start()

    def alart(self) -> None:
        # QMessageBox を表示するカスタムイベントをポストする
        QtCore.QCoreApplication.postEvent(self.main_window, QEvent_Custom_ShowMessageBox())

    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()

参考までに。

注意したこと

ソースコードの例

Discussion