2022.02.06  

【Python】threading.Threadを使ってファイル監視ツールを作成(並行処理)

Python    

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()
コメント
現在コメントはありません。
コメントする
コメント入力

名前 (※ 必須)

メールアドレス (※ 必須 画面には表示されません)

送信