LoginSignup
150
174

More than 5 years have passed since last update.

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

Last updated at Posted at 2015-01-11

暫く前のモノだが、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なツールでも極普通にできる。それらに関してもいずれまとめたいな。

150
174
3

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
150
174