Skip to content

Instantly share code, notes, and snippets.

@emirotin
Created December 29, 2010 20:08
Show Gist options
  • Save emirotin/759002 to your computer and use it in GitHub Desktop.
Save emirotin/759002 to your computer and use it in GitHub Desktop.
A class to perform polling directory watch and call a specific process, the script to watch directory for HAML files, and the script to support CoffeeScript inside of them
# The script to monitor directory changes and execute specific files on changed files
# Tasks are executed one by one, on keyboard interrupt the last task is waited to finish (not terminated)
# written by Eugene Mirotin, December 2010
# heavily based on http://www.bryceboe.com/2010/08/26/python-multiprocessing-and-keyboardinterrupt/
import sys
import os
import multiprocessing, Queue
from signal import signal, SIGINT, SIG_IGN
os_walk = os.walk
pjoin = os.path.join
realpath = os.path.realpath
os_stat = os.stat
class DirectoryWatcher(object):
def __init__(self, start_dir, mask=None):
self.start_dir = realpath(start_dir)
self.mask = mask
def __enum_files(self):
for (path, dirs, files) in os_walk(self.start_dir):
for file in files:
if self.mask is not None and not self.mask.match(file):
continue
yield pjoin(path, file)
@staticmethod
def __process_file(job_queue, process):
signal(SIGINT, SIG_IGN)
if job_queue.empty():
return
try:
file = job_queue.get(block=False)
process(file)
except Queue.Empty:
pass
def watch(self, process):
print 'Watching {0}'.format(self.start_dir)
filestamps = {}
job_queue = multiprocessing.Queue()
current_worker = None
try:
while True:
workers = []
for file_name in self.__enum_files():
try:
mtime = os_stat(file_name).st_mtime
except OSError:
continue
if file_name not in filestamps or mtime > filestamps[file_name]:
filestamps[file_name] = mtime
job_queue.put(file_name)
worker = multiprocessing.Process(target=DirectoryWatcher.__process_file,
args=(job_queue, process))
workers.append(worker)
for worker in workers:
current_worker = worker
worker.start()
worker.join()
current_worker = None
except KeyboardInterrupt:
if current_worker:
print 'Received ctrl-c, waiting for running task to complete'
current_worker.join()
print ' ...done.'
else:
print 'Received ctrl-c, exiting'
return
require 'rubygems'
require 'active_support/core_ext/object/blank'
require 'haml'
require 'haml/filters/coffee'
template = ARGV.length > 0 ? File.read(ARGV.shift) : STDIN.read
haml_engine = Haml::Engine.new(template)
file = ARGV.length > 0 ? File.open(ARGV.shift, 'w') : STDOUT
file.write(haml_engine.render)
file.close
from directory_watcher import DirectoryWatcher
import os
import time
from subprocess import call, STDOUT
pjoin = os.path.join
splitext = os.path.splitext
relpath = os.path.relpath
realpath = os.path.realpath
class HamlCompiler(object):
def __init__(self, source_dir, dest_dir):
self.source_dir = realpath(source_dir)
self.dest_dir = realpath(dest_dir)
def process(self, source_file):
dest_file = pjoin(self.dest_dir, relpath(source_file, self.source_dir))
dest_file = splitext(dest_file)[0] + '.html'
print 'Recompiling {0} to {1}'.format(source_file, dest_file)
retcode = call(["ruby", "haml.rb", source_file, dest_file], stderr=STDOUT)
print ' ...done. Return code: {0}'.format(retcode)
if __name__ == "__main__":
import re
import sys
import signal
def terminate(*args):
raise KeyboardInterrupt
signal.signal(signal.SIGINT, terminate)
print 'HAML watcher is running'
re_haml = re.compile('.*\.haml$')
processor = HamlCompiler(sys.argv[1], sys.argv[2])
watcher = DirectoryWatcher(sys.argv[1], re_haml)
watcher.watch(processor.process)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment