Python 周期処理: Timerを使ったスレッド処理とコールバック関数活用

みなさん、プログラムを制作していて、ここの処理を周期的に呼び出したいなぁって思うようなことないですか？

私は普段組み込みシステムの開発を行っているので、WindowsアプリやLinuxアプリ・Pythonアプリの制作を使用していて、あーここ周期処理にできたら簡単なんだけどなーって思ったりよくします。

シリアルやEthernet通信等の通信アプリをプログラミングするときは特に思います。

Timerとは

ですので、今回は周期処理を可能とするTimer機能を紹介します。
今回使用するTimerは周期的にコールバック関数を呼び出ししてくれるものとなり、Threadの機能を使用しますので、メインのプロセスの処理に影響与えず使用できます。

threading ライブラリ を使用しますので、追加でライブラリ・モジュールをインストールする必要がありません。

サンプルコード

はい。ではいつも通りまずはコード全体を記載します。

import time
import threading
from logging import NullHandler

streamstop = False
tm = NullHandler

def tm_callback():
    global tm
    tm.cancel()
    del tm
    tm = NullHandler
    
    print( time.time() )
    if streamstop == False:
        tm = threading.Timer( 1 , tm_callback )
        tm.start()

def main_func( ):
    global tm
    tm = threading.Timer( 1 , tm_callback )
    tm.start()

    while True:
        try:
            pass
        except KeyboardInterrupt:
            print("stop")
            streamstop = True
            tm.cancel()
            tm = NullHandler
            break
    

if __name__ == '__main__':
 result = main_func()

実行結果

ではこのプログラムを実行してみます。

以下の結果から、約1秒周期で呼ばれているのがお分かりいただけると思います。

WindowsでかつPythonでこの処理を行っていることとおそらくタイマの停止や再起動等の処理で、50ms程度ずつ遅延していますが、Windows上で作成するアプリとしてはシビアなことをしない限りは大きな問題にはならないと思います。

続いてはプログラムの解説ですが、解説は次ページにて記載します。

Timer使い方

今回のタイマーは標準ライブラリである、threading ライブラリを使用して行います。
タイマーを起動してコールバックを受け取る流れは以下です。

①タイマのインスタンスを生成する。(この時に時間を指定します。)

②タイマーを起動する。

時間が経過すると、①で指定した関数を呼び出ししてくれますので、タイマを止めて、行いたい処理を実行します。
残念ながら勝手に再起動はしてくれないようなので、再度時間を計測したい場合は①と②を再度実施します。
具体的な処理は以下です。

タイマの起動方法解説

まず、 threading.Timer で、コールバック関数と時間を設定してインスタンスを取得します。
ここでは、第1引数に1秒 第2引数で tm_callback 関数を設定しています。

時間の指定は、秒単位で指定することになります。
ここで以下のような疑問に思うか方もいるかもしれません。
あれ、秒単位でないとタイマー使えないの？？
安心してください。この時間指定は、Float型を指定しますので、1msから時間を設定できます。
1msのタイマを起動したい場合は、1/1000秒ですので、0.001を指定します。

tm = threading.Timer( 1 , tm_callback )   # 周期時間を1秒で指定してコールバックを設定
tm.start()                                # タイマを起動する

そして、以下のようにコールバック関数を用意します。
コールバックの中身は以下のように、 tm.cancel でタイマを一度停止します。
タイマを停止したら、使い終わったtmをdelで一旦解放します。
※このインスタンスを一旦解放処理は、私の環境では終了イベント受けてもタイマが止まらないことがありましたので、タイマが止まらなかったらやってみてください。

そして、最後の3行で再起動をして 、周期的にこの関数がよばれるようにしています。

def tm_callback():
    global tm
    tm.cancel()
    del tm
    tm = NullHandler
    
    print( time.time() )
    if streamstop == False:
        tm = threading.Timer( 1 , tm_callback )
        tm.start()

まとめ

多少の誤差や遅延はあるもののざっくり指定時間で動作していますしかつ最短1msでの処理も行えるので、ちょっと周期で処理したいなぁと思った時には簡単でとても便利だと思います。

ぜひみなさんも周期処理したいぁって思った時は、試してみてください。

Pythonについて勉強したい人は以下がおすすめです。私も持っていてたまに眺めて勉強していますものですのでぜひ購入して学習してみてください。