Qt4のQFileSystemWatcherでファイルの更新検知を行う

Python

プログラマの方々なら、ファイルを更新したときに自動的にあるアクションを実行したいと思うことが、一度はあると思います。
そのようなタスクを実行するプログラムを書くには、ファイルシステムを監視して、ファイルの更新を検知する必要があります。


ファイルの更新を検知する方法として、２つの方法が考えられます。
１つ目の方法は、一定時間間隔でファイルの更新時間をポーリングする方法、２つ目の方法は、OSのAPI(Linuxのinotify等)を使う方法です。
１つ目の方法の場合、監視したいファイルが少数の場合は問題ありませんが、監視したいファイルが増えた場合にパフォーマンスが悪化すると思われます。
２つの方法の場合、OSのAPIを使うためにC言語を書く必要があるため、実装が容易ではありません。
また、基本的にファイル監視のAPIはOS固有のAPIであるため、複数OSで動作しないという問題点もあります。


OSのAPIを直接利用すると、特定OSに依存したプログラムになってしまうことが問題だったわけですが、Qt4のQFileSystemWatcherというライブラリがOSの違いを吸収したインターフェースを提供してくれているので、これを利用することによりファイルの更新検知をOS非依存で行うことができます。
また、Qt4自体はC++のライブラリですが、Python等の他の言語からQt4を利用するためのライブラリも存在も存在するため、それらを利用することで、C/C++といった低水準のプログラミング言語を書くことなく、ファイルシステム監視のAPIを利用することが出来ます。


Qt4のPythonバインディングであるPyQt4を使って、特定のファイルを更新したときに、任意のコマンドを実行するプログラムを書いてみたので、ここで紹介したいと思います。

GUIバージョン

スナップショットはこんな感じです。

監視するファイルと実行するコマンドを入力し、監視開始を押すと、ファイルの更新検知を開始します。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import subprocess
from PyQt4.QtCore import *
from PyQt4.QtGui  import *

class FileWatchDialog(QWidget):
  def __init__(self, parent=None):
    super(FileWatchDialog, self).__init__(parent)
    self.watcher = QFileSystemWatcher(self)
    self.filename = None
    self.mtime = None
    self.command = None
    self.file_edit = QLineEdit(self)
    self.command_edit = QLineEdit(self)
    self.start_button = QPushButton(u"開始", self)
    self.start_button.setCheckable(True)
    self.browse_button = QPushButton(u"参照", self)

    top_layout = QHBoxLayout()
    top_layout.addWidget(QLabel(u"監視するファイル:", self))
    top_layout.addWidget(self.file_edit)
    top_layout.addWidget(self.browse_button)
    middle_layout = QHBoxLayout()
    middle_layout.addWidget(QLabel(u"コマンド:", self))
    middle_layout.addWidget(self.command_edit)
    bottom_layout = QHBoxLayout()
    bottom_layout.addStretch()
    bottom_layout.addWidget(self.start_button)
    bottom_layout.addStretch()
    main_layout = QVBoxLayout()
    main_layout.addLayout(top_layout)
    main_layout.addLayout(middle_layout)
    main_layout.addLayout(bottom_layout)
    self.setLayout(main_layout)

    self.start_button.toggled.connect(self._on_start_button_toggled)
    self.browse_button.clicked.connect(self._on_browse_button_clicked)
    self.watcher.fileChanged.connect(self._on_file_changed)

  def _on_start_button_toggled(self, checked):
    if checked == True:
      self._start_watching()
    else:
      self._end_watching()

  def _start_watching(self):
    self.filename = unicode(self.file_edit.text())
    if not os.path.exists(self.filename):
      QMessageBox.information(self,
                              u"指定したファイルは存在しません",
                              u"指定したファイルは存在しません")
      return
    self.mtime = os.path.getmtime(self.filename)
    self.command = unicode(self.command_edit.text())
      
    self.start_button.setText(u"終了")
    self.file_edit.setEnabled(False)
    self.browse_button.setEnabled(False)
    self.command_edit.setEnabled(False)
    self.watcher.addPath(self.filename)

  def _end_watching(self):
    self.start_button.setText(u"開始")
    self.file_edit.setEnabled(True)
    self.browse_button.setEnabled(True)
    self.command_edit.setEnabled(True)
    self.watcher.removePath(self.filename)

  def _on_browse_button_clicked(self):
    filename = QFileDialog.getOpenFileName(self)
    if not filename.isNull():
      self.file_edit.setText(filename)

  def _on_file_changed(self, _):
    mtime = os.path.getmtime(self.filename)
    if self.mtime < mtime:
      self.mtime = mtime
      print self.command
      subprocess.call(self.command, shell=True)

if __name__ == '__main__':
  import sys
  app = QApplication(sys.argv)
  win = FileWatchDialog()
  win.show()
  sys.exit( app.exec_() )

CUIバージョン

第一引数に監視するファイル名を、第二引数にファイル更新時に実行するコマンドを指定します。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import os
import subprocess
from PyQt4.QtCore import *
from PyQt4.QtGui  import *

global filename, command, mtime
filename = None
command  = None
mtime = None

def do_command(_):
  global filename, command, mtime
  new_mtime = os.path.getmtime(filename)
  if mtime < new_mtime:
    mtime = new_mtime
    print command
    subprocess.call(command, shell=True)

def excepthook(type, value, traceback):
  if type is KeyboardInterrupt:
    qApp.quit()
  else:
    sys.__excepthook__(type, value, traceback)
sys.excepthook = excepthook

def usage():
  print "usage: python %s <filename> <command>" % __file__

if __name__ == '__main__':
  if len(sys.argv) < 3:
    usage()
    sys.exit(0)

  filename = sys.argv[1]
  command  = sys.argv[2]
  new_mtime = os.path.getmtime(filename)
  
  app = QApplication(sys.argv)
  watcher = QFileSystemWatcher()
  watcher.addPath(filename)
  watcher.fileChanged.connect(do_command)
  timer = QTimer()
  timer.timeout.connect(lambda : 0)
  timer.setInterval(100)
  timer.setSingleShot(False)
  timer.start()
  sys.exit( app.exec_() )

CUIバージョンでは、Ctrl-Cでプログラムが終了するようにsys.excepthookを書き換えるというハックを行っています。
また、PyQt4はC++のQt4の単なるラッパーであるため、イベントループ中はC++のコードが動いています。
そのためか、Ctrl-Cを押しても、その瞬間にはKeyboardInterruptは発生せず、イベントが発生したときにKeyboardInterruptが発生します。
これは、おそらく、Pythonが非同期シグナルを同期的に処理しているため、Pythonのコードが動いていないとKeyboardInterruptが発生しないためだと考えられます、そのため、タイマーを使って定期的にPythonコードを動かし、Ctrl-Cが押された時にKeyboardInterruptがきちんと発生するようにしています。