はじめに
JavaScriptで作ったタイピングゲームをPythonに移植しようと思いました。JavaScript版では、カウントダウンタイマーをSetIntervalで実装していますが、同等の実装をPythonでどうするか調べてみました。
結論
- PythonはJavaスレッドモデルをベースにしたthreadingパッケージがある
- 一定時間後に、関数を呼び出したい場合は、threading.Timer。
- 定期的に関数を呼び出したい場合は、threading.Threadを継承し、スレッドモデルで実装すればよい。
インターネットを調べていくと、質問が「SetIntervalのような定期実行をしたい」という質問のため、回答が関数を再帰的に呼び出し、スレッドを毎回生成しているコードを見つけました。(Python Equivalent of setInterval()?) これがPythonとして良いのかどうかはわかりませんが、毎回スレッドを生成するのに違和感があったので、もう少し調べたところ、threading.Threadを継承したサンプルを多く見つけました。
SetTimeoutのように、一定時間後に処理を実行する場合は、threading.Timer
threading.Timer(delay, task)で、待機時間を指定します。また、スレッドプログラミングなので、joinも使えます。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
# -*- coding: utf-8 -*- import threading from typing import Callable def task_hello(): print(f"mytask:{threading.current_thread().ident}") print("hello") def set_timeout(task: Callable, delay: int = 0): t = threading.Timer(delay, task) t.start() return t if __name__ == '__main__': print(f"main:{threading.current_thread().ident}") child_id = set_timeout(task_hello, 1) # 待機 child_id.join() # joinしているので、子スレッドわ終わるまで以下は表示されない。 print("mainスレッド終了") |
Pythonで定期的に関数を実行したいのでSetIntervalを調べる
標準ライブラリーでは用意されていないようです。実装するにはthreading.Event(),event.set(),event.wait()の知識が必要なようです。
- https://stackoverflow.com/questions/9812344/cancellable-threading-timer-in-python
- https://stackoverflow.com/questions/2697039/python-equivalent-of-setinterval/48709380#48709380
- おまいらのthreading.Eventの使い方は間違っている
少し調べていると、PythonのスレッドはJavaのスレッドモデルを参考にしているので、JavaScriptのSetIntervalで定期的に実行するのではなく、スレッドモデルの書きかたで実装すればよいだけでした。
もう少し具体的に言うと、threading.Threadクラスを継承したクラスを作成し、runメソッドを好みに書き換えればよいだけです。Pythonには、スレッド間通信するための便利なEventオブジェクトがあります。これでwaitの管理などをします。
ちなみにJavaはThread.sleep,Thread.join,Thread.interruptのようにEventオブジェクトではなく、Threadもしくはそのインスタンスでwaitなどを管理します。
基本的なコード
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
import threading import time class TimerClass(threading.Thread): def __init__(self): threading.Thread.__init__(self) self.event = threading.Event() self.count = 10 def run(self): while self.count > 0 and not self.event.is_set(): print(f"timerスレッド:{threading.current_thread().ident},count={self.count}") self.count -= 1 self.event.wait(1) def stop(self): self.event.set() tmr = TimerClass() tmr.start() print(f"mainスレッド:{threading.current_thread().ident}, timer.sleep(3)呼び出し前") time.sleep(3) print(f"mainスレッド:{threading.current_thread().ident}, timer.sleep(3)呼び出し後") tmr.stop() print(f"mainスレッド:{threading.current_thread().ident}, 終了") |
カウントダウンタイマー
上記ソースコードをベースに、カウントダウンタイマーを作りました。Countdownクラスを初期化するときの引数にcountで残り秒数を設定できるようにしました。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
import threading import time class Countdown(threading.Thread): def __init__(self, count = 0): threading.Thread.__init__(self) self.event = threading.Event() self.count = count def run(self): while self.count > 0 and not self.event.is_set(): #print(f"timerスレッド:{threading.current_thread().ident},count={self.count}") self.count -= 1 self.event.wait(1) def stop(self): self.event.set() print(f"mainスレッド:{threading.current_thread().ident}, 開始") workerThread = Countdown(count = 10) workerThread.start() # メインスレッドが先に終わらないようにする場合は、joinを使えばよい workerThread.join() #workerThread.stop() print(f"mainスレッド:{threading.current_thread().ident}, 終了") |
一定間隔で関数を呼び出す
厳密ではないですが、ひとまず一定間隔で呼び出せる。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 |
import threading import time import datetime class Worker(threading.Thread): def __init__(self, fn, count = 0): threading.Thread.__init__(self) self.event = threading.Event() self.count = count self.fn = fn def run(self): while self.count > 0 and not self.event.is_set(): print(f"timerスレッド:{threading.current_thread().ident},count={self.count}") self.fn() self.count -= 1 self.event.wait(1) def stop(self): self.event.set() def task_hello(): print(f"\t hello task: {datetime.datetime.now()}") print(f"mainスレッド:{threading.current_thread().ident}, 開始") workerThread = Worker(task_hello, count = 10) workerThread.start() # メインスレッドが先に終わらないようにする場合は、joinを使えばよい workerThread.join() #workerThread.stop() print(f"mainスレッド:{threading.current_thread().ident}, 終了") |
厳密な間隔で呼び出したいとき
runメソッドのwhileループのチェックで、time.time()を導入します。以下をまずはコピペしてprintでいろいろ出力しながら試すのが簡単だと思います。
https://stackoverflow.com/questions/2697039/python-equivalent-of-setinterval/48709380#48709380
