PyQt5におけるscheduleモジュール制御方法

Python
2022.06.05

pythonのscheduleモジュールは、曜日単位で起動時間を簡単に設定できる便利なモジュールであり、番組録音などのプログラムには重宝します。

ところがscheduleモジュールは常に起動時間を監視するループ関数ですから、不意にループのなかで制御するとイベントループとなってエラー終了してしまうという、やっかいな制約があります。

さらにQt(PyQt5)やtkinterを使ったGUI上のボタンでスケジュールの起動・停止をする場合は、意外と手間がかかりますので動作検証用プログラムを用いて解説します。

動作検証用に作成したプログラムは以下のとおりです。

# schedule_cancel_qt.py
# Copyright (c) 2022 falconblog.org
#
# Released under the MIT license.
# see https://opensource.org/licenses/MIT
#
import sys
from PyQt5.QtWidgets import QApplication, QWidget, QPushButton, QGridLayout, QLabel, QMessageBox
import datetime
import schedule
import threading
import time

#テスト用スケジュール-秒
MONITOR_SECOND = 2

def now():
    return datetime.datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

class qt_schedule(QWidget):

    def __init__(self):
        super().__init__()
        self.initUI()

    def initUI(self):
        self.resize(250, 150)
        self.move(300, 300)
        self.setWindowTitle('sample')

        self.running = True
        self.job_enable = False

        # buttonの設定
        self.start_button = QPushButton('Start')
        self.start_label = QLabel('Stopped')

        self.stop_button = QPushButton('Stop')
        
        self.quit_button = QPushButton('Quit')        
        self.time_label = QLabel(now())

        # buttonのclickでラベルをクリア
        self.start_button.clicked.connect(self.start)
        self.stop_button.clicked.connect(self.stop)
        self.quit_button.clicked.connect(self.quit)
        
        # レイアウト配置
        self.grid = QGridLayout()
        self.grid.addWidget(self.start_button, 0, 0, 0, 1)
        self.grid.addWidget(self.stop_button, 0, 1, 0, 1)
        self.grid.addWidget(self.quit_button, 0, 2, 0, 1)
        self.grid.addWidget(self.start_label, 1, 0, 1, 1)
        self.grid.addWidget(self.time_label, 1, 1, 1, 1)
        self.setLayout(self.grid)

        # 表示
        self.show()
        
        # ジョブのスケジューリング, 8 = ５秒おきに時刻表示
        self.week_select(8,'06:30')

        # スレッドの開始
        self.thread = threading.Thread(target=self.run_monitor)
        self.thread.setDaemon(True)
        self.thread.start()
    '''        
# スケジュール定義
# 1: 月－金　
# 2: 月－水
# 3: 月火
# 4: 木金
# 5: 土日
# 6: 土
# 7: 日
# 8: 秒
    '''
    def week_select(self, WEEK_NO, s_time):
        schedule.clear()
        if WEEK_NO == 0:
            schedule.every().day.at(s_time).do(self.job)
        elif WEEK_NO == 1:
            schedule.every().monday.at(s_time).do(self.job)
            schedule.every().tuesday.at(s_time).do(self.job)
            schedule.every().wednesday.at(s_time).do(self.job)
            schedule.every().thursday.at(s_time).do(self.job)
            schedule.every().friday.at(s_time).do(self.job)
        elif WEEK_NO == 2:
            schedule.every().monday.at(s_time).do(self.job)
            schedule.every().tuesday.at(s_time).do(self.job)
            schedule.every().wednesday.at(s_time).do(self.job)
        elif WEEK_NO == 3:
            schedule.every().monday.at(s_time).do(self.job)
            schedule.every().tuesday.at(s_time).do(self.job)
        elif WEEK_NO == 4:
            schedule.every().thursday.at(s_time).do(self.job)
            schedule.every().friday.at(s_time).do(self.job)
        elif WEEK_NO == 5:
            schedule.every().saturday.at(s_time).do(self.job)
            schedule.every().sunday.at(s_time).do(self.job)
        elif WEEK_NO == 6:
            schedule.every().saturday.at(s_time).do(self.job)
        elif WEEK_NO == 7:
            schedule.every().sunday.at(s_time).do(self.job)
        elif WEEK_NO == 8:
            schedule.every(MONITOR_SECOND).seconds.do(self.job)

#schedule thread contoroll section
    def run_monitor(self):
        while self.running:
            schedule.run_pending()
            time.sleep(0.2)
        self.close()

    def job(self):
        if self.job_enable:
            self.time_label.setText(str(now()))

    def start(self):
        self.job_enable = True
        self.start_label.setText("Started")

    def stop(self):
        self.job_enable = False
        self.time_label.setText("Job is disabled")
        self.start_label.setText("Stopped")

    def quit(self):
        self.running = False

if __name__ == '__main__':

    app = QApplication(sys.argv)
    ew = qt_schedule()    
    sys.exit(app.exec_())

このサンプルプログラムを実行すると、次のようなウィンドウが表示されて、ボタンクリックによりスケジュールの起動・停止ができることがわかります。

GUIプログラムは、ソースコードだけではなかなか理解できないものですから、プログラムを動かして挙動を見ながら確認するのが良いでしょう。

プログラムの解説

NHKラジオの語学放送は、平日(weekday)、月～水、木金、土曜日・日曜日のみ、といった変則的な番組放送スケジュールが組まれていますので、事前に全ての放送パターンを調べて、if文で設定を選べるように組むことにします。

week_select関数は、曜日テーブル番号と開始時刻を引数入力することにより、変則的なスケジュールを一度に設定できるようにしています。

GUIからスケジュールを再設定する場合を考慮して、関数の最初にschedule.clear()を入れておき、現在のスケジュールをクリアしてから再設定するようにしています。

def week_select(self, WEEK_NO, s_time):
    schedule.clear()
    if WEEK_NO == 0:
            schedule.every().day.at(s_time).do(self.job)
        elif WEEK_NO == 1: #平日(weekday)のスケジュール
            schedule.every().monday.at(s_time).do(self.job)
            schedule.every().tuesday.at(s_time).do(self.job)
            schedule.every().wednesday.at(s_time).do(self.job)
            schedule.every().thursday.at(s_time).do(self.job)
            schedule.every().friday.at(s_time).do(self.job)
        elif WEEK_NO == 2: #月～水のスケジュール
            schedule.every().monday.at(s_time).do(self.job)
            schedule.every().tuesday.at(s_time).do(self.job)
            schedule.every().wednesday.at(s_time).do(self.job)

scheduleの起動・停止制御

scheduleモジュールは、schedule.run.pending()で設定時刻待ちをすることになりますが、一度起動すると止める手段がないため、個別に停止する手段を構築する必要があります。

対処法としては、schedule.run.pending()が含む関数をマルチスレッドで起動させておき、停止操作が割り込めるように制御します。またマルチスレッドとするため他のタスクも同時動作できるようにデーモン（daemon)化しておきバックグラウンド動作にします。

deamon化しなくてもプログラムは動作しますが、Qtの場合は「✕ボタン」や「Quitボタン」でウィンドウを閉じたときにスレッドが残ったままとなり、以後コンソールが使えなくなりますから、忘れずに行います。（tkinterはマルチスレッド動作に制限があるため、✕ボタンでスレッドを終了してくれる仕様のようです。）

# スレッドの開始
self.thread = threading.Thread(target=self.run_monitor)
self.thread.setDaemon(True)
self.thread.start()

マルチスレッドしたrun_monitor関数では、スケジュールの起動と停止制御を行っています。

スケジュールは、起動時にself.running = Tureとしてpending状態で動作させ、time.sleep(0.2)でプログラムの動作に時間制限をかけておきます。

スケジュールで実行させるプログラムは、self.job_enableというフラグ制御がされたjob関数により動作を制御します。このサンプルでは現在時刻をGUI上にテキスト表示させています。

スケジュール時刻が到来するとjob関数が動作します。job関数は、Start/Stopのボタンに紐付けられたjob_enableフラグで動作が変わるため、Startボタンをクリック = 「時刻をGUIに表示」、Stopボタンをクリック= 「jobはpass（何もしない）」という挙動になるという仕組みです。

schedule thread contoroll section
    def run_monitor(self):
        while self.running:
            schedule.run_pending()
            time.sleep(0.2)
        self.close()

    def job(self):
        if self.job_enable:
            self.time_label.setText(str(now()))

    def start(self):
        self.job_enable = True
        self.start_label.setText("Started")

    def stop(self):
        self.job_enable = False
        self.time_label.setText("Job is disabled")
        self.start_label.setText("Stopped")

    def quit(self):
        self.running = False

Quitボタンをクリックすると、self.running =Falseになり、run_monitorのwhile文から脱出して、self.close()によりプログラム終了となります。

注意点としては、マルチスレッドはデーモン化の制御もきちんと行うことにあります。Qtはデーモン化をしていないと、終了時コマンドラインに戻らず開発効率が激落ちしてしまいます。

以上がPyQt5でのschedule起動制御です。このプログラムを使ってネット語学レコーダを作成しましたので、実アプリへの応用方法を確認したい方はダウンロードページからソースコードを確認してみてください。

無線やネットラジオの受信ではスケジュール制御が必須となりますので、これからいろいろと応用してみようと思います。

スポンサーリンク