/usr/lib/python2.7/dist-packages/livereload/watcher.py is in python-livereload 2.2.2-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | # -*- coding: utf-8 -*-
"""
livereload.watcher
~~~~~~~~~~~~~~~~~~
A file watch management for LiveReload Server.
:copyright: (c) 2013 by Hsiaoming Yang
"""
import os
import glob
import time
class Watcher(object):
"""A file watcher registery."""
def __init__(self):
self._tasks = {}
self._mtimes = {}
# filepath that is changed
self.filepath = None
self._start = time.time()
def ignore(self, filename):
"""Ignore a given filename or not."""
_, ext = os.path.splitext(filename)
return ext in ['.pyc', '.pyo', '.o', '.swp']
def watch(self, path, func=None):
"""Add a task to watcher."""
self._tasks[path] = func
def start(self, callback):
"""Start the watcher running, calling callback when changes are observed. If this returns False,
regular polling will be used."""
return False
def examine(self):
"""Check if there are changes, if true, run the given task."""
# clean filepath
self.filepath = None
for path in self._tasks:
if self.is_changed(path):
func = self._tasks[path]
# run function
func and func()
return self.filepath
def is_changed(self, path):
if os.path.isfile(path):
return self.is_file_changed(path)
elif os.path.isdir(path):
return self.is_folder_changed(path)
return self.is_glob_changed(path)
def is_file_changed(self, path):
if not os.path.isfile(path):
return False
if self.ignore(path):
return False
mtime = os.path.getmtime(path)
if path not in self._mtimes:
self._mtimes[path] = mtime
self.filepath = path
return mtime > self._start
if self._mtimes[path] != mtime:
self._mtimes[path] = mtime
self.filepath = path
return True
self._mtimes[path] = mtime
return False
def is_folder_changed(self, path):
for root, dirs, files in os.walk(path, followlinks=True):
if '.git' in dirs:
dirs.remove('.git')
if '.hg' in dirs:
dirs.remove('.hg')
if '.svn' in dirs:
dirs.remove('.svn')
if '.cvs' in dirs:
dirs.remove('.cvs')
for f in files:
if self.is_file_changed(os.path.join(root, f)):
return True
return False
def is_glob_changed(self, path):
for f in glob.glob(path):
if self.is_file_changed(f):
return True
return False
class INotifyWatcher(Watcher):
def __init__(self):
Watcher.__init__(self)
import pyinotify
self.wm = pyinotify.WatchManager()
self.notifier = None
self.callback = None
def watch(self, path, func=None):
import pyinotify
flag = pyinotify.IN_CREATE | pyinotify.IN_DELETE | pyinotify.IN_MODIFY
self.wm.add_watch(path, flag, rec=True, do_glob=True, auto_add=True)
Watcher.watch(self, path, func)
def inotify_event(self, event):
self.callback()
def start(self, callback):
if not self.notifier:
self.callback = callback
import pyinotify
from tornado import ioloop
self.notifier = pyinotify.TornadoAsyncNotifier(
self.wm, ioloop.IOLoop.instance(),
default_proc_fun=self.inotify_event
)
callback()
return True
|