Skip to content

Instantly share code, notes, and snippets.

@ajdavis
Created January 4, 2017 13:14
Show Gist options
  • Save ajdavis/1abab8258f785dddf07c7a44a2ed7163 to your computer and use it in GitHub Desktop.
Save ajdavis/1abab8258f785dddf07c7a44a2ed7163 to your computer and use it in GitHub Desktop.
Script based on the Python "watchdog" module to run tasks when files change.
#!/usr/bin/env python
import os
import re
import threading
import time
import subprocess
from os.path import splitext, expanduser, normpath
import click
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
try:
from Queue import Queue
except ImportError:
from queue import Queue
class Handler(FileSystemEventHandler):
def __init__(self, pattern, exclude, coalesce, command, verbose, notify):
self.pattern = re.compile(pattern or '.*')
if exclude:
self.exclude = normpath(expanduser(exclude))
else:
self.exclude = None
self.coalesce = coalesce
self.command = command
self.verbose = verbose
self.notify = notify
self.thread = None
self.q = Queue()
def start(self):
self.thread = threading.Thread(target=self._process_q)
self.thread.daemon = True
self.thread.start()
def on_any_event(self, event):
global stopped
if not event.is_directory and self.pattern.match(event.src_path):
norm = normpath(expanduser(event.src_path))
if self.exclude and norm.startswith(self.exclude):
return
self.q.put((event, time.time()))
def trigger(self, src_path):
if src_path is not None:
path, ext = splitext(src_path)
else:
path, ext = '', ''
cmd = self.command % {'src_path': src_path, 'path': path}
print(cmd)
try:
subprocess.check_call([cmd], shell=True)
subprocess.check_call([
'/usr/bin/osascript',
'-e',
'display notification "Triggered from %s" with title "Watcher"'
% (src_path or '--trigger')
])
except OSError as exc:
print(exc)
subprocess.check_call([
'/usr/bin/osascript',
'-e',
'display notification "%s from %s" with title "Watcher"'
% (exc, src_path)
])
def _process_q(self):
last_ts = 0
while True:
event, ts = self.q.get()
if self.coalesce and ts < last_ts:
continue
if self.verbose:
print('WATCHER: %s %s' % (event.src_path, event.event_type))
self.trigger(event.src_path)
last_ts = time.time()
@click.command()
@click.option('--verbose', '-v', help='Verbose output.', is_flag=True)
@click.option('--pattern', help='Match filenames.')
@click.option('--exclude', type=click.Path(), help='Exclude directories.')
@click.option('--coalesce', '-c', is_flag=True,
help='Multiple events trigger one command')
@click.option('--trigger/--no-trigger', '-t/-n', default=False,
help='Trigger once at startup.')
@click.option('--notify', '-n', is_flag=True, help="Display notification")
@click.argument('command')
def watcher(pattern, exclude, coalesce, trigger, command, verbose, notify):
observer = Observer()
handler = Handler(pattern, exclude, coalesce, command, verbose, notify)
if trigger:
handler.trigger(None)
handler.start()
observer.schedule(handler, '.', recursive=True)
observer.start()
try:
while True:
time.sleep(0.25)
except KeyboardInterrupt:
observer.stop()
observer.join()
if __name__ == '__main__':
watcher()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment