Created
January 4, 2017 13:14
-
-
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.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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