Help us understand the problem. What is going on with this article?

ファイルの更新をきっかけにコマンド実行 (python編)

More than 1 year has passed since last update.

暫く前のモノだが、Qiitaにこんな記事があるのを発見した。というか、@wpythonnewsで流れてきた。

ファイルを保存した瞬間ユニットテストを実行

ここで紹介されているやり方は、最終更新のTimestampを取っておいて、それと対象ディレクトリの下にあるファイルの更新時刻を1つずつチェックしていくという方法。

チェック自体は100ms毎に実行し、フルでCPUをぶん回すことにはならないが、それでもそれなりの負荷がかかる。手元のMacで77%くらいのロード。さらに、このコードだと100msの間に複数のファイルが更新された時に動作が読めない気がする(評価される順番によって結果が変わる)。

一方でこの手の「ファイルの更新確認」は何かと必要とされる場面があるのも確か。そのため、最近のOSではカーネルレベルでのサポートがある。

  • inotify (Linux)
  • FSEvents (Mac OS X)
  • ReadDirectoryChangesW (Windows Win32)
  • System.IO.FileSystemWatcher (Windows .NET)
  • epoll (Linux)
  • kqueue (BSD, Mac OS X)

おそらくこれらを使ったPythonのモジュールがあるはず。ということで、調べてみると色々出てくる。

ちなみにepollやkqueueは他のに比べるとより低レベルのI/FでPythonの標準モジュール(select)でサポートされている。

上記のモジュールのなかで、watchdogは使うAPIをプラットフォーム毎に使い分けていて汎用のプログラムを書くのであればこれが一番良さそう。ということで、試しに使ってみた。

導入にはいつものpipを使用。

$ pip install watchdog

そしてサンプルコードはこんな感じ。

#!/usr/bin/env python
from __future__ import print_function

import sys
import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler


class MyHandler(PatternMatchingEventHandler):
    def __init__(self, command, patterns):
        super(MyHandler, self).__init__(patterns=patterns)
        self.command = command

    def _run_command(self):
        subprocess.call([self.command, ])

    def on_moved(self, event):
        self._run_command()

    def on_created(self, event):
        self._run_command()

    def on_deleted(self, event):
        self._run_command()

    def on_modified(self, event):
        self._run_command()


def watch(path, command, extension):
    event_handler = MyHandler(command, ["*"+extension])
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()


if __name__ == "__main__":
    if 4 > len(sys.argv):
        print("Usage:", sys.argv[0], "dir_to_watch command extension")
    else:
        watch(sys.argv[1], sys.argv[2], sys.argv[3])

若干長くなったが、基本は簡単で、用意されているイベントハンドラークラスの一つを継承したクラスを作り、on_moved, on_created, on_deleted, on_modifiedの中身を実装するだけ。それぞれファイルを移動・作成・消去・変更した時に呼ばれるmethodだ。

ちなみに、用意されているイベントハンドラークラスは以下の4つ。

  • FileSystemEventHandler
  • PatternMatchingEventHandler
  • RegexMatchingEventHandler
  • LoggingEventHandler

一つ目がファイル変更のイベント処理を行う基本クラスとなっていて、それにパターンマッチングあるいは正規表現でファイルを絞り込む機能が追加されたのが二つ目と三つ目。4つ目はファイル変更のイベントをログとして書き出すハンドラーが実装されたもの。

ここでは拡張子で絞り込むことになっているので、PatternMatchingEventHandlerを継承し、4つのhandler methodどれが呼ばれてもコマンドが発行されるように実装している。

そしてObserverクラスのインスタンスを作り、イベントハンドラーと監視するディレクトリをschedule()に渡して、start()するだけ。その後に無限ループで1秒毎に起きる処理が入っているが、これはキーイベントをキャプチャし、Ctrl-Cで実行を止められるようにするためのもの。これで少しCPUを消費してしまうが、手元で10%くらい。ま、許容範囲か。間隔を長くすればもっと減らせる。

で、サンプルの実行だがコマンドの引数は「ファイルを保存した瞬間ユニットテストを実行のdirwatchと合わせてある。実行はこんな感じ。

$ python dirwatch2.py <directory_to_watch> <command> <extension>

指定したディレクトリ以下の指定した拡張子を持つファイル変更されると指定したコマンドが発行される。

なお、このwatchdogというPythonモジュールには同じようなことをするためのツールwatchmedoが付属している。それを使うとこのように書ける。

$ watchmedo shell-command \
    --patterns="*"$3 \
    --command $2 \
    --recursive \
    $1

拍子抜けするくらい簡単だった。

なお、同じように「ファイルやディレクトリを監視して変化があったら何かする」というのはGruntやGulpなどNode.jsなツールでも極普通にできる。それらに関してもいずれまとめたいな。

Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
No comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした