threading.Threadの動作確認のため、ファイル監視ツールを作成しました。
このツールを実行すると次の処理が同時に実行されます。
・任意のディレクトリにファイルが作成されているか監視
・5秒待機した後に監視対象ディレクトリにファイルを作成
・ファイル監視が終了するまで処理実行時間を表示(2秒間隔)
file_monitor.py実行結果
監視処理を開始します
処理実行時間を計測します
ファイル監視中
処理実行時間:2秒経過
ファイル監視中
処理実行時間:4秒経過
ファイルを作成します
ファイルが配置されました
処理実行時間:6秒経過
監視処理が終了しました
file_monitor.py
import threading
import time
import os
import math
class file_monitor:
def __init__(self):
#ループ制御用のフラグ
self.Loop = True
# 監視対象のファイル名とパスを設定
self.file_path = os.getcwd() + "/monitor_dir/test.txt"
def main(self):
print("監視処理を開始します")
# targetで指定した処理をスレッド化する
file_monitor = threading.Thread(target=self.file_monitor_execute)
file_create = threading.Thread(target=self.file_create, args=(5,))
# スレッドを実行する(ファイル監視処理とファイル作成処理が同時実行される)
file_monitor.start()
file_create.start()
# 並行して別の処理を実行する
print("処理実行時間を計測します")
start_time = time.time()
while self.Loop:
time.sleep(2)
exec_time = time.time()
print(f"処理実行時間:{math.floor(exec_time - start_time)}秒経過")
print("監視処理が終了しました")
# 3秒ごとにmonitor_dirにファイルが配置されているかチェックする
def file_monitor_execute(self):
while self.Loop:
print("ファイル監視中")
time.sleep(3)
# monitor_dirにtest.txtが存在するか監視する
if os.path.isfile(self.file_path):
print('ファイルが配置されました')
self.Loop = False
# 指定した秒数後にファイル作成を行う
def file_create(self, second):
time.sleep(second)
f = open(self.file_path, 'w')
print("ファイルを作成します")
f.write('')
f.close()
if __name__ == '__main__':
file_monitor = file_monitor()
file_monitor.main()